Macbook
289 lines
12 KiB
Java
289 lines
12 KiB
Java
package com.goldenchart.websocket;
|
|
|
|
import com.fasterxml.jackson.core.type.TypeReference;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import com.goldenchart.dto.CandleBarDto;
|
|
import com.goldenchart.storage.Ta4jStorage;
|
|
import lombok.RequiredArgsConstructor;
|
|
import lombok.extern.slf4j.Slf4j;
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
import org.springframework.context.event.EventListener;
|
|
import org.springframework.stereotype.Component;
|
|
import org.springframework.web.reactive.function.client.WebClient;
|
|
import org.springframework.web.socket.messaging.SessionSubscribeEvent;
|
|
import org.springframework.web.socket.messaging.SessionUnsubscribeEvent;
|
|
import org.ta4j.core.Bar;
|
|
import org.ta4j.core.BarSeries;
|
|
import org.ta4j.core.BaseBarSeriesBuilder;
|
|
import org.ta4j.core.bars.TimeBarBuilderFactory;
|
|
import org.ta4j.core.num.DoubleNumFactory;
|
|
|
|
import java.time.Instant;
|
|
|
|
import java.time.Duration;
|
|
import java.time.ZonedDateTime;
|
|
import java.time.ZoneOffset;
|
|
import java.util.*;
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
import java.util.concurrent.ExecutorService;
|
|
import java.util.concurrent.Executors;
|
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
import java.util.regex.Matcher;
|
|
import java.util.regex.Pattern;
|
|
|
|
/**
|
|
* 동적 STOMP 구독 관리자 (명세서 3.3).
|
|
*
|
|
* 프론트엔드가 /sub/charts/{market}/{type} 채널을 구독하면:
|
|
* 1. Ta4jStorage 에 해당 종목 BarSeries 가 없으면 업비트 REST API Warm-up
|
|
* 2. UpbitWebSocketClient 구독 목록에 market 추가
|
|
*
|
|
* 구독자 수가 0이 되면:
|
|
* 1. UpbitWebSocketClient 에서 market 제거
|
|
* 2. Ta4jStorage 에서 BarSeries 삭제 (메모리 방어)
|
|
*/
|
|
@Component
|
|
@RequiredArgsConstructor
|
|
@Slf4j
|
|
public class DynamicSubscriptionManager {
|
|
|
|
// /sub/charts/{market}/{type} 패턴 파서
|
|
private static final Pattern TOPIC_PATTERN =
|
|
Pattern.compile("^/sub/charts/([^/]+)/([^/]+)$");
|
|
|
|
private final Ta4jStorage ta4jStorage;
|
|
private final UpbitWebSocketClient upbitWsClient;
|
|
private final WebClient.Builder webClientBuilder;
|
|
private final ObjectMapper objectMapper;
|
|
|
|
@Value("${upbit.api.base-url:https://api.upbit.com}")
|
|
private String upbitBaseUrl;
|
|
|
|
@Value("${upbit.api.candle-limit:200}")
|
|
private int candlePageSize;
|
|
|
|
/** EMA 수렴을 위한 warm-up 목표 봉 수 (페이지네이션으로 달성) */
|
|
@Value("${upbit.api.warmup-count:600}")
|
|
private int warmupTargetCount;
|
|
|
|
@Value("${upbit.api.request-delay-ms:110}")
|
|
private long requestDelayMs;
|
|
|
|
/** 구독자 수 추적: "KRW-BTC:1m" → count */
|
|
private final ConcurrentHashMap<String, AtomicInteger> subscriberCount =
|
|
new ConcurrentHashMap<>();
|
|
|
|
/** 워밍업 중인 키 추적 (중복 요청 방지) */
|
|
private final ConcurrentHashMap<String, Boolean> warmingUp =
|
|
new ConcurrentHashMap<>();
|
|
|
|
/** 워밍업 전용 단일 스레드 — 업비트 API 레이트리밋 보호용 직렬 실행 */
|
|
private final ExecutorService warmUpExecutor = Executors.newSingleThreadExecutor(r -> {
|
|
Thread t = new Thread(r, "warmup-thread");
|
|
t.setDaemon(true);
|
|
return t;
|
|
});
|
|
|
|
// ── STOMP 구독/구독해제 이벤트 리스너 ────────────────────────────────────
|
|
|
|
@EventListener
|
|
public void handleSubscribe(SessionSubscribeEvent event) {
|
|
String dest = getDestination(event.getMessage().getHeaders());
|
|
if (dest == null) return;
|
|
Matcher m = TOPIC_PATTERN.matcher(dest);
|
|
if (!m.matches()) return;
|
|
|
|
String market = m.group(1);
|
|
String candleType = m.group(2);
|
|
String key = market + ":" + candleType;
|
|
|
|
int count = subscriberCount.computeIfAbsent(key, k -> new AtomicInteger(0))
|
|
.incrementAndGet();
|
|
log.info("[DynamicSubscriptionManager] 구독: {} (총 {}명)", key, count);
|
|
|
|
// Warm-up: 메모리에 없고 현재 워밍업 중이 아니면 비동기 초기화
|
|
if (!ta4jStorage.exists(market, candleType)
|
|
&& warmingUp.putIfAbsent(key, Boolean.TRUE) == null) {
|
|
warmUpExecutor.submit(() -> {
|
|
try {
|
|
warmUp(market, candleType);
|
|
} finally {
|
|
warmingUp.remove(key);
|
|
}
|
|
});
|
|
}
|
|
// UpbitWebSocket 구독 추가
|
|
upbitWsClient.addMarket(market);
|
|
if ("1m".equals(candleType)) {
|
|
upbitWsClient.resubscribeAll();
|
|
}
|
|
}
|
|
|
|
@EventListener
|
|
public void handleUnsubscribe(SessionUnsubscribeEvent event) {
|
|
// unsubscribe 이벤트는 구독 ID 기반이므로 별도 매핑 필요 (간소화: 0 도달 시 GC)
|
|
// 실제 구현에서는 sessionId + subscriptionId → topic 역매핑 테이블 필요
|
|
// 여기서는 구독 수 감소만 처리
|
|
String dest = getDestination(event.getMessage().getHeaders());
|
|
if (dest == null) return;
|
|
Matcher m = TOPIC_PATTERN.matcher(dest);
|
|
if (!m.matches()) return;
|
|
|
|
String market = m.group(1);
|
|
String candleType = m.group(2);
|
|
String key = market + ":" + candleType;
|
|
|
|
AtomicInteger counter = subscriberCount.get(key);
|
|
if (counter == null) return;
|
|
|
|
int remaining = counter.decrementAndGet();
|
|
log.info("[DynamicSubscriptionManager] 구독 해제: {} (남은 {}명)", key, remaining);
|
|
|
|
if (remaining <= 0) {
|
|
subscriberCount.remove(key);
|
|
upbitWsClient.removeMarket(market);
|
|
ta4jStorage.remove(market, candleType);
|
|
log.info("[DynamicSubscriptionManager] GC: {} 메모리 해제", key);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* live-conditions API 등 — 최소 봉 수 확보까지 동기 warm-up (종목별 독립).
|
|
*/
|
|
public void ensureBarsReadySync(String market, String candleType, int minBars) {
|
|
upbitWsClient.addMarket(market);
|
|
if (ta4jStorage.getBarCount(market, candleType) >= minBars) {
|
|
return;
|
|
}
|
|
String key = market + ":" + candleType;
|
|
if (warmingUp.putIfAbsent(key, Boolean.TRUE) == null) {
|
|
try {
|
|
warmUp(market, candleType);
|
|
} finally {
|
|
warmingUp.remove(key);
|
|
}
|
|
} else {
|
|
// 다른 스레드 warm-up 중 — 짧게 대기 후 재확인
|
|
for (int i = 0; i < 30 && ta4jStorage.getBarCount(market, candleType) < minBars; i++) {
|
|
try {
|
|
Thread.sleep(100);
|
|
} catch (InterruptedException e) {
|
|
Thread.currentThread().interrupt();
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 실시간 전략 체크 등 STOMP 클라이언트 없이도 백엔드가 틱·캔들을 수집하도록 마켓을 고정 구독한다.
|
|
*/
|
|
public void ensureMarketPinned(String market, String candleType) {
|
|
String key = market + ":" + candleType;
|
|
if (!ta4jStorage.exists(market, candleType)
|
|
&& warmingUp.putIfAbsent(key, Boolean.TRUE) == null) {
|
|
warmUpExecutor.submit(() -> {
|
|
try {
|
|
warmUp(market, candleType);
|
|
} finally {
|
|
warmingUp.remove(key);
|
|
}
|
|
});
|
|
}
|
|
upbitWsClient.addMarket(market);
|
|
log.info("[DynamicSubscriptionManager] 실시간 전략용 마켓 고정: {}:{}", market, candleType);
|
|
}
|
|
|
|
// ── Warm-up ────────────────────────────────────────────────────────────────
|
|
|
|
/**
|
|
* 업비트 REST API 로 최근 N개 캔들을 페이지네이션으로 가져와 Ta4jStorage 를 초기화한다.
|
|
* EMA(200) 수렴에는 약 1000봉이 필요하므로 warmupTargetCount(기본 600)봉을 목표로 한다.
|
|
*/
|
|
private void warmUp(String market, String candleType) {
|
|
String path = Ta4jStorage.CANDLE_TYPE_MAP.getOrDefault(candleType, "minutes/1");
|
|
Duration duration = Ta4jStorage.parseDuration(candleType);
|
|
List<Map<String, Object>> allRaw = new ArrayList<>();
|
|
String toParam = null; // null = 최신부터
|
|
int fetched = 0;
|
|
|
|
try {
|
|
while (fetched < warmupTargetCount) {
|
|
int pageSize = Math.min(candlePageSize, warmupTargetCount - fetched);
|
|
String url = upbitBaseUrl + "/v1/candles/" + path
|
|
+ "?market=" + market + "&count=" + pageSize
|
|
+ (toParam != null ? "&to=" + toParam : "");
|
|
|
|
Thread.sleep(requestDelayMs);
|
|
String json = webClientBuilder.build()
|
|
.get().uri(url)
|
|
.retrieve().bodyToMono(String.class).block();
|
|
if (json == null || json.isBlank()) break;
|
|
|
|
List<Map<String, Object>> page =
|
|
objectMapper.readValue(json, new TypeReference<>() {});
|
|
if (page.isEmpty()) break;
|
|
|
|
allRaw.addAll(page);
|
|
fetched += page.size();
|
|
|
|
// 다음 페이지: 가장 오래된 봉 시각보다 1초 이전을 to 로 설정.
|
|
// 업비트 to 파라미터는 exclusive 이지만, 동일 시각 중복 방지를 위해 1초 뺀다.
|
|
String oldest = (String) page.get(page.size() - 1).get("candle_date_time_utc");
|
|
if (oldest == null) break;
|
|
long oldestEpoch = parseEpoch(oldest);
|
|
toParam = Instant.ofEpochSecond(oldestEpoch - 1)
|
|
.atZone(ZoneOffset.UTC)
|
|
.format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"))
|
|
+ "Z";
|
|
|
|
if (page.size() < pageSize) break; // 더 이상 데이터 없음
|
|
}
|
|
|
|
// 업비트 응답은 최신→과거 순 → 오름차순 정렬 후 Ta4jStorage 에 적재
|
|
allRaw.sort(Comparator.comparingLong(r -> parseEpoch((String) r.get("candle_date_time_utc"))));
|
|
|
|
for (Map<String, Object> r : allRaw) {
|
|
long epochSec = parseEpoch((String) r.get("candle_date_time_utc"));
|
|
Instant endInstant = Instant.ofEpochSecond(epochSec).plus(duration);
|
|
try {
|
|
BarSeries tmp = new BaseBarSeriesBuilder()
|
|
.withBarBuilderFactory(new TimeBarBuilderFactory())
|
|
.withNumFactory(DoubleNumFactory.getInstance()).build();
|
|
Bar bar = tmp.barBuilder()
|
|
.timePeriod(duration).endTime(endInstant)
|
|
.openPrice(toDouble(r.get("opening_price")))
|
|
.highPrice(toDouble(r.get("high_price")))
|
|
.lowPrice(toDouble(r.get("low_price")))
|
|
.closePrice(toDouble(r.get("trade_price")))
|
|
.volume(toDouble(r.get("candle_acc_trade_volume")))
|
|
.build();
|
|
ta4jStorage.addBar(market, candleType, bar);
|
|
} catch (Exception e) {
|
|
log.trace("[DynamicSubscriptionManager] warmUp bar 스킵: {}", e.getMessage());
|
|
}
|
|
}
|
|
log.info("[DynamicSubscriptionManager] Warm-up 완료: {} / {} ({} bars)", market, candleType, allRaw.size());
|
|
} catch (Exception e) {
|
|
log.error("[DynamicSubscriptionManager] Warm-up 실패 {}/{}: {}", market, candleType, e.getMessage());
|
|
}
|
|
}
|
|
|
|
// ── 유틸 ──────────────────────────────────────────────────────────────────
|
|
|
|
private String getDestination(org.springframework.messaging.MessageHeaders headers) {
|
|
Object dest = headers.get("simpDestination");
|
|
return dest instanceof String s ? s : null;
|
|
}
|
|
|
|
private static long parseEpoch(String utcStr) {
|
|
if (utcStr == null) return 0L;
|
|
return ZonedDateTime.parse(utcStr + "Z",
|
|
java.time.format.DateTimeFormatter.ISO_DATE_TIME).toEpochSecond();
|
|
}
|
|
|
|
private static double toDouble(Object v) {
|
|
return v == null ? 0.0 : ((Number) v).doubleValue();
|
|
}
|
|
}
|