전략평가 시 부하문제 개선

2026-06-05 23:53:00 +09:00
parent 032ec69290
commit 1d4e606f77
7 changed files with 120 additions and 26 deletions
@@ -19,6 +19,7 @@ import org.ta4j.core.num.DoubleNumFactory;

 import java.time.Duration;
 import java.time.Instant;
+import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.util.Comparator;
@@ -51,6 +52,15 @@ public class CandleGapBackfillService {
    private long requestDelayMs;

    private static final String[] BACKFILL_TYPES = {"1m", "3m", "5m", "10m", "15m", "30m", "1h", "4h", "1w", "1d"};
+    /** 업비트 일봉 기준 시각: KST (UTC+9) — BarBuilder.alignToInterval(1440) 과 동일 */
+    private static final ZoneId ZONE_KST = ZoneId.of("Asia/Seoul");
+
+    /** Bar 생성용 공유 팩토리 — 매번 BarSeries를 new 하는 낭비 제거 */
+    private static final BarSeries BAR_FACTORY_SERIES =
+            new BaseBarSeriesBuilder()
+                    .withBarBuilderFactory(new TimeBarBuilderFactory())
+                    .withNumFactory(DoubleNumFactory.getInstance())
+                    .build();

    private volatile long lastRunAtMs;
    private final AtomicInteger lastRunMarkets = new AtomicInteger();
@@ -120,18 +130,26 @@ public class CandleGapBackfillService {
        int merged = 0;
        for (Map<String, Object> r : page) {
            long epochSec = parseEpoch((String) r.get("candle_date_time_utc"));
-            Instant endInstant = Instant.ofEpochSecond(epochSec).plus(duration);
+            // 1d 봉: Upbit REST candle_date_time_utc 는 UTC 자정 기준이지만
+            // BarBuilder.alignToInterval(1440) 은 KST 자정 기준으로 endTime 을 계산한다.
+            // UTC open 시각이 속하는 KST 날짜의 종료 시각(KST 다음날 00:00)으로 변환하여
+            // BarBuilder 와 동일한 경계를 사용한다.
+            Instant endInstant;
+            if ("1d".equals(candleType)) {
+                ZonedDateTime kstDay = Instant.ofEpochSecond(epochSec)
+                        .atZone(ZoneOffset.UTC)
+                        .withZoneSameInstant(ZONE_KST);
+                endInstant = kstDay.toLocalDate().atStartOfDay(ZONE_KST).plusDays(1).toInstant();
+            } else {
+                endInstant = Instant.ofEpochSecond(epochSec).plus(duration);
+            }
            double open = toDouble(r.get("opening_price"));
            double high = toDouble(r.get("high_price"));
            double low = toDouble(r.get("low_price"));
            double close = toDouble(r.get("trade_price"));
            double vol = toDouble(r.get("candle_acc_trade_volume"));

-            Bar bar = new BaseBarSeriesBuilder()
-                    .withBarBuilderFactory(new TimeBarBuilderFactory())
-                    .withNumFactory(DoubleNumFactory.getInstance())
-                    .build()
-                    .barBuilder()
+            Bar bar = BAR_FACTORY_SERIES.barBuilder()
                    .timePeriod(duration)
                    .endTime(endInstant)
                    .openPrice(open).highPrice(high).lowPrice(low)
@@ -21,6 +21,14 @@ public class TickActivityTracker {
        return lastTickMs.getOrDefault(market, 0L);
    }

+    /**
+     * WebSocket 재연결 시 모든 타임스탬프 초기화.
+     * 이전 연결의 오래된 타임스탬프가 남아 재연결 직후 stale 판단 → 재연결 루프를 유발하는 것을 방지한다.
+     */
+    public void resetAll() {
+        lastTickMs.clear();
+    }
+
    public Map<String, Long> snapshot() {
        return Map.copyOf(lastTickMs);
    }
@@ -29,8 +29,8 @@ public class TradingPipelineProperties {
    /** BarSeries 메모리 상한 (OOM 방지) */
    private int maxBarsPerSeries = 500;

-    /** 종목별 틱 미수신 시 재연결 요청(초) */
-    private int staleTickSeconds = 30;
+    /** 종목별 틱 미수신 시 재연결 요청(초) — 저유동성 종목은 수 분간 체결 없는 게 정상이므로 넉넉히 설정 */
+    private int staleTickSeconds = 120;

    /** 갭 백필 스케줄 (Spring cron) */
    private String gapBackfillCron = "0 */5 * * * *";
@@ -59,18 +59,62 @@ public class WebSocketStalenessMonitor {
        );
    }

+    /**
+     * 재연결 후 틱 수신까지 여유를 주기 위한 연결 직후 유예 기간 (ms).
+     * - 재연결 직후 이전 타임스탬프가 초기화되지만, 업비트가 새 구독에 응답하는데 수 초 소요
+     * - 유예 기간 동안은 stale 체크를 건너뜀으로써 재연결 루프를 차단
+     */
+    private static final long CONNECT_GRACE_MS = 90_000; // 90초
+
+    /**
+     * WS가 살아있어도(메시지 수신 중) 일부 저유동성 종목만 stale 일 때는 재연결하지 않는다.
+     * 전체 메시지가 없어야 WS 자체가 죽었다고 판단한다.
+     */
+    private static final long GLOBAL_DEAD_MS = 60_000; // 60초 동안 WS 메시지 없으면 연결 불량
+
    private void check() {
        lastCheckAtMs = System.currentTimeMillis();
        long now = System.currentTimeMillis();
-        long thresholdMs = props.getStaleTickSeconds() * 1000L;
+
+        // ① 재연결 직후 유예 기간: 연결 후 CONNECT_GRACE_MS 이내라면 체크 건너뜀
+        long connectedAt = upbitWebSocketClient.getConnectedAtMs();
+        if (connectedAt > 0 && now - connectedAt < CONNECT_GRACE_MS) {
+            log.debug("[WebSocketStalenessMonitor] 연결 직후 유예 기간 — 체크 건너뜀 ({}ms 경과)",
+                    now - connectedAt);
+            return;
+        }
+
+        // ② WS 전체 헬스 체크: 최근에 메시지가 수신되고 있으면 연결 자체는 살아있음
+        //    이 경우 저유동성 종목의 틱 부재는 정상으로 간주하여 재연결 불필요
+        long lastMsg = upbitWebSocketClient.getLastMessageAtMs();
+        if (lastMsg > 0 && now - lastMsg < GLOBAL_DEAD_MS) {
+            // WS는 살아있음 — stale 정보는 로그만 남기고 재연결하지 않음
+            var subscribed = upbitWebSocketClient.getSubscribedMarkets();
+            long thresholdMs = props.getStaleTickSeconds() * 1000L;
+            List<String> stale = new ArrayList<>();
+            for (String market : subscribed) {
+                long last = activityTracker.lastTickMs(market);
+                if (last == 0) continue;
+                if (now - last > thresholdMs) stale.add(market);
+            }
+            lastStaleCount = stale.size();
+            if (!stale.isEmpty()) {
+                log.debug("[WebSocketStalenessMonitor] 저유동성 종목 틱 부재 {}개 (WS 정상) — 재연결 없음: {}",
+                        stale.size(), stale);
+            }
+            return;
+        }
+
+        // ③ WS 메시지 자체가 끊긴 경우 — 연결 불량으로 판단하여 재연결
        var subscribed = upbitWebSocketClient.getSubscribedMarkets();
        if (subscribed.isEmpty()) return;

+        long thresholdMs = props.getStaleTickSeconds() * 1000L;
        List<String> stale = new ArrayList<>();
        for (String market : subscribed) {
            long last = activityTracker.lastTickMs(market);
            if (last == 0) {
-                // 아직 한 번도 틱 없음 — 구독 직후 30초 유예는 threshold 로 커버
+                // 아직 한 번도 틱 없음 — 구독 직후 유예는 위에서 처리
                continue;
            }
            if (now - last > thresholdMs) {
@@ -79,9 +123,13 @@ public class WebSocketStalenessMonitor {
        }
        lastStaleCount = stale.size();
        if (!stale.isEmpty()) {
-            log.warn("[WebSocketStalenessMonitor] 틱 정체 {}종목 → WS 재연결 요청: {}",
+            log.warn("[WebSocketStalenessMonitor] WS 응답 없음 + 틱 정체 {}종목 → 재연결 요청: {}",
                    stale.size(), stale);
            upbitWebSocketClient.requestReconnect("stale_ticks:" + stale);
+        } else if (lastMsg > 0 && now - lastMsg > GLOBAL_DEAD_MS) {
+            log.warn("[WebSocketStalenessMonitor] WS 메시지 없음 ({}초 이상) → 재연결 요청",
+                    (now - lastMsg) / 1000);
+            upbitWebSocketClient.requestReconnect("no_messages");
        }
    }
 }