goldenChart/backend/src/main/java/com/goldenchart/trading/pipeline/CandleGapBackfillService.java

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<String, Object> monitorSnapshot() {
        Map<String, Object> 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<String> 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<Map<String, Object>> 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<String, Object> 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());
    }
}