package com.goldenchart.trading.pipeline; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.goldenchart.storage.Ta4jStorage; import com.goldenchart.websocket.UpbitWebSocketClient; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.springframework.web.reactive.function.client.WebClient; 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.Duration; import java.time.Instant; import java.time.ZoneOffset; import java.time.ZonedDateTime; import java.util.Comparator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; /** * 5분마다 REST 로 최근 캔들을 조회해 WS 갭을 멱등적으로 보정한다. */ @Service @RequiredArgsConstructor @Slf4j public class CandleGapBackfillService { private final Ta4jStorage ta4jStorage; private final UpbitWebSocketClient upbitWebSocketClient; private final TradingPipelineProperties props; private final WebClient.Builder webClientBuilder; private final ObjectMapper objectMapper; @Value("${upbit.api.base-url:https://api.upbit.com}") private String upbitBaseUrl; @Value("${upbit.api.request-delay-ms:110}") private long requestDelayMs; private static final String[] BACKFILL_TYPES = {"1m", "3m", "5m", "10m", "15m", "30m", "1h", "4h", "1d"}; private volatile long lastRunAtMs; private final AtomicInteger lastRunMarkets = new AtomicInteger(); private final AtomicLong lastRunBarsMerged = new AtomicLong(); private volatile String lastRunError = ""; public Map monitorSnapshot() { Map m = new LinkedHashMap<>(); m.put("cron", props.getGapBackfillCron()); m.put("candleCount", props.getGapBackfillCandleCount()); m.put("lastRunAtMs", lastRunAtMs); m.put("lastRunMarkets", lastRunMarkets.get()); m.put("lastRunBarsMerged", lastRunBarsMerged.get()); m.put("lastRunError", lastRunError); return m; } @Scheduled(cron = "${goldenchart.trading.pipeline.gap-backfill-cron:0 */5 * * * *}") public void scheduledBackfill() { if (!props.isEnabled()) return; Set markets = upbitWebSocketClient.getSubscribedMarkets(); if (markets.isEmpty()) return; lastRunAtMs = System.currentTimeMillis(); lastRunMarkets.set(markets.size()); lastRunBarsMerged.set(0); lastRunError = ""; int count = props.getGapBackfillCandleCount(); long mergedTotal = 0; for (String market : markets) { for (String type : BACKFILL_TYPES) { if (!ta4jStorage.exists(market, type)) continue; try { Thread.sleep(requestDelayMs); mergedTotal += backfillMarket(market, type, count); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } catch (Exception e) { lastRunError = e.getMessage(); log.debug("[GapBackfill] {} {}: {}", market, type, e.getMessage()); } } } lastRunBarsMerged.set(mergedTotal); } int backfillMarket(String market, String candleType, int count) throws Exception { String path = Ta4jStorage.CANDLE_TYPE_MAP.getOrDefault(candleType, "minutes/1"); Duration duration = Ta4jStorage.parseDuration(candleType); String url = upbitBaseUrl + "/v1/candles/" + path + "?market=" + market + "&count=" + count; String json = webClientBuilder.build() .get().uri(url) .retrieve().bodyToMono(String.class).block(); if (json == null || json.isBlank()) return 0; List> page = objectMapper.readValue(json, new TypeReference<>() {}); if (page.isEmpty()) return 0; page.sort(Comparator.comparingLong(r -> parseEpoch((String) r.get("candle_date_time_utc")))); int merged = 0; for (Map r : page) { long epochSec = parseEpoch((String) r.get("candle_date_time_utc")); Instant 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() .timePeriod(duration) .endTime(endInstant) .openPrice(open).highPrice(high).lowPrice(low) .closePrice(close).volume(vol) .build(); ta4jStorage.addBar(market, candleType, bar); merged++; } if (merged > 0) { BarSeries series = ta4jStorage.getOrCreate(market, candleType); log.debug("[GapBackfill] {} {} — REST {}봉 병합 (series={})", market, candleType, merged, series.getBarCount()); } return merged; } private static long parseEpoch(String utc) { if (utc == null) return 0; String s = utc.endsWith("Z") ? utc : utc + "Z"; return ZonedDateTime.parse(s).toEpochSecond(); } private static double toDouble(Object o) { if (o == null) return 0; if (o instanceof Number n) return n.doubleValue(); return Double.parseDouble(o.toString()); } }