goldenChat base source add

2026-05-23 15:11:48 +09:00
commit a4ea7762b5
2081 changed files with 1155760 additions and 0 deletions
@@ -0,0 +1,261 @@
+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);
+        }
+    }
+
+    /**
+     * 실시간 전략 체크 등 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();
+    }
+}