package com.goldenchart.websocket; import com.goldenchart.dto.CandleBarDto; import com.goldenchart.service.BarCloseStrategyEvaluationService; import com.goldenchart.storage.Ta4jStorage; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import org.ta4j.core.Bar; import org.ta4j.core.BarSeries; import org.ta4j.core.bars.TimeBarBuilderFactory; import org.ta4j.core.num.DoubleNumFactory; import java.time.Duration; import java.time.Instant; import java.time.ZoneId; import java.time.ZonedDateTime; import java.time.ZoneOffset; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; /** * 업비트 WebSocket 틱(Trade) 데이터를 수신하여 분봉 단위 캔들을 조립하는 빌더. * * 명세서 3.2 준수: * - 동일 분(minute) 내 틱 → series.addPrice(price, volume) 로 현재 Bar 갱신 * - 분이 바뀌는 틱 → 이전 Bar 를 확정하고 series.addBar(newBar) 로 새 캔들 열기 * - 1분봉 확정 시 상위 타임프레임(3m, 5m, 10m, 15m, 30m, 1h, 4h, 1w, 1d)으로 집계 전파 * * 지원 캔들 타입: 1m, 3m, 5m, 10m, 15m, 30m, 1h, 4h, 1w, 1d */ @Component @RequiredArgsConstructor @Slf4j public class BarBuilder { private final Ta4jStorage ta4jStorage; private final TradingWebSocketBroker broker; private final BarCloseStrategyEvaluationService barCloseEvaluation; private final com.goldenchart.service.PaperOrderMatcherService paperOrderMatcher; /** * [항목4] Phase 2 비동기 실행용 스레드 풀. * 웹소켓 수신 스레드가 전략 평가 연산에 블로킹되지 않도록 분리한다. */ @org.springframework.beans.factory.annotation.Autowired @Qualifier("barCloseEvalExecutor") private Executor barCloseEvalExecutor; /** 상위 집계 타임프레임 (분 단위 집계 주기: 3, 5, 10, 15, 30, 60, 240, 1440) */ private static final int[] UPPER_MINUTES = {3, 5, 10, 15, 30, 60, 240, 10080, 1440}; private static final String[] UPPER_TYPES = {"3m", "5m", "10m", "15m", "30m", "1h", "4h", "1w", "1d"}; /** 진행 중인 1분봉 상태 (market → PartialBar) */ private final ConcurrentHashMap partialBars = new ConcurrentHashMap<>(); // ── 공개 API ────────────────────────────────────────────────────────────── /** * 업비트 trade 틱을 처리하여 1분봉을 조립한다. * * @param market 마켓 코드 (e.g. "KRW-BTC") * @param tradePrice 체결가 * @param tradeVolume 체결량 * @param tradeTimeMs 체결 Unix 밀리초 */ public void onTick(String market, double tradePrice, double tradeVolume, long tradeTimeMs) { ZonedDateTime tickTime = ZonedDateTime.ofInstant( java.time.Instant.ofEpochMilli(tradeTimeMs), ZoneOffset.UTC); // 해당 분의 시작 시간 (초 이하 제거) ZonedDateTime barStart = tickTime.withSecond(0).withNano(0); PartialBar current = partialBars.get(market); if (current == null || !current.barStart.equals(barStart)) { // ── 분 변경: 이전 Bar 확정 및 신규 Bar 열기 ───────────────────── if (current != null) { commitBar(market, current); // 이전 Bar 확정 } // 신규 Bar 시작 PartialBar newBar = new PartialBar(barStart, tradePrice, tradeVolume); partialBars.put(market, newBar); // 새 분이 시작되는 첫 틱을 Ta4jStorage 에 provisional bar 로 추가. // 이렇게 해야 이후 동일 분 틱에서 updateLastBar 가 직전 확정봉을 오염시키지 않는다. Duration d1m = Duration.ofMinutes(1); ZonedDateTime endTime = barStart.plus(d1m); Bar provisionalBar = buildTa4jBar(newBar, d1m, endTime); ta4jStorage.addBar(market, "1m", provisionalBar); } else { // ── 동일 분: 진행 중인 Bar 갱신 ──────────────────────────────── current.update(tradePrice, tradeVolume); // Ta4jStorage 마지막 봉(현재 진행 중인 provisional bar)을 실시간 갱신 ta4jStorage.updateLastBar(market, "1m", tradePrice, tradeVolume); } // 현재 1분봉 상태를 STOMP 로 즉시 발행 (실시간 틱 렌더링용) publishCurrentBar(market); paperOrderMatcher.onPriceTick(market, tradePrice); } // ── 내부 처리 ────────────────────────────────────────────────────────────── /** * 완성된 1분봉을 Ta4jStorage 에 저장하고 상위 타임프레임에 전파한다. * *

[항목2] 2-Phase 처리 — 멀티 타임프레임 레이스 컨디션 해소: *

* 기존 방식(1m 저장 → 1m 전략 평가 → 상위봉 저장 → 상위봉 전략 평가)에서는 * 1m 전략이 1h CrossSeriesRule 을 참조할 때 1h 봉이 아직 확정 전이어서 * 잘못된 인덱스를 바라보는 레이스 컨디션이 있었다. */ private void commitBar(String market, PartialBar partial) { Duration d1m = Duration.ofMinutes(1); ZonedDateTime endTime = partial.barStart.plus(d1m); Bar bar = buildTa4jBar(partial, d1m, endTime); // ── Phase 1: 모든 타임프레임 스토리지 업데이트 ──────────────────────── ta4jStorage.addBar(market, "1m", bar); log.debug("[BarBuilder] 1m Bar 확정: {} @ {}", market, endTime); // 마감된 상위봉 정보(타입, maturedIndex, bar) — Phase 2 에서 사용 java.util.List closedUppers = new java.util.ArrayList<>(); for (int i = 0; i < UPPER_MINUTES.length; i++) { int minutes = UPPER_MINUTES[i]; String type = UPPER_TYPES[i]; ZonedDateTime upperStart = alignToInterval(partial.barStart, minutes); ZonedDateTime upperEnd = upperStart.plusMinutes(minutes); BarSeries upperSeries = ta4jStorage.getOrCreate(market, type); if (upperSeries.isEmpty()) { Bar upperBar = buildTa4jBar(partial, Duration.ofMinutes(minutes), upperEnd); ta4jStorage.addBar(market, type, upperBar); } else { Instant lastUpperEndInstant = upperSeries.getLastBar().getEndTime(); Instant upperEndInstant = upperEnd.toInstant(); if (lastUpperEndInstant.equals(upperEndInstant)) { ta4jStorage.updateLastBar(market, type, partial.close, partial.volume); } else if (lastUpperEndInstant.isBefore(upperEndInstant)) { // 상위봉 마감 → 새 봉 추가 Bar upperBar = buildTa4jBar(partial, Duration.ofMinutes(minutes), upperEnd); ta4jStorage.addBar(market, type, upperBar); BarSeries upperAfter = ta4jStorage.getOrCreate(market, type); int maturedIdx = Math.max(upperAfter.getBeginIndex(), upperAfter.getEndIndex() - 1); closedUppers.add(new UpperCloseInfo(type, maturedIdx, upperAfter.getBar(maturedIdx))); } else { log.warn("[BarBuilder] 상위봉 endTime 역행 {} {} last={} expected={} — 갱신만 수행", market, type, lastUpperEndInstant, upperEndInstant); ta4jStorage.updateLastBar(market, type, partial.close, partial.volume); } } } // ── Phase 2: 모든 봉이 확정된 후 전략 평가 트리거 (비동기) ────────────── // // [항목4] 웹소켓 수신 스레드(업비트 연결 유지 담당)가 전략 평가 연산 때문에 블로킹되면 // 다음 분봉 시작 틱을 제때 수신하지 못할 수 있다. Phase 1 완료 후 Phase 2 를 // 전용 스레드 풀(barCloseEvalExecutor)에 제출하여 WebSocket 파이프라인을 즉시 반환한다. // // 스레드 안전성: Ta4jStorage·positionOpenCache 는 ConcurrentHashMap, maturedIdx 는 // Phase 1 에서 이미 확정된 값이므로 비동기 실행 중 Phase 1 이 재실행되어 인덱스가 바뀌어도 // 올바른 봉을 참조한다. final Bar confirmedBar = bar; final java.util.List closedUppersSnap = java.util.Collections.unmodifiableList(closedUppers); barCloseEvalExecutor.execute(() -> { try { if (barCloseEvaluation.shouldEvaluateOnClose(market, "1m")) { BarSeries series1m = ta4jStorage.getOrCreate(market, "1m"); barCloseEvaluation.onMaturedBarClose(market, "1m", series1m.getEndIndex(), confirmedBar); } for (UpperCloseInfo info : closedUppersSnap) { if (barCloseEvaluation.shouldEvaluateOnClose(market, info.type())) { barCloseEvaluation.onMaturedBarClose(market, info.type(), info.maturedIdx(), info.bar()); } } } catch (Exception e) { log.error("[BarBuilder] Phase 2 전략 평가 오류 market={}: {}", market, e.getMessage(), e); } }); } /** Phase 1 에서 마감된 상위봉 정보를 Phase 2 로 전달하는 내부 레코드 */ private record UpperCloseInfo(String type, int maturedIdx, Bar bar) {} /** 현재 진행 중인 1분봉을 STOMP 로 발행 (실시간 틱 렌더링용, 시그널 없음). */ private void publishCurrentBar(String market) { PartialBar p = partialBars.get(market); if (p == null) return; CandleBarDto dto = CandleBarDto.builder() .time(p.barStart.toEpochSecond()) .open(p.open) .high(p.high) .low(p.low) .close(p.close) .volume(p.volume) .build(); broker.publish(market, "1m", dto); publishUpperTimeframes(market); } /** 상위 분봉 구독자용 — Ta4jStorage 마지막 봉 스냅샷 (틱마다 1m 과 함께 발행) */ private void publishUpperTimeframes(String market) { for (String type : UPPER_TYPES) { try { BarSeries series = ta4jStorage.getOrCreate(market, type); if (series.isEmpty()) continue; Bar last = series.getLastBar(); long barStart = last.getEndTime().getEpochSecond() - last.getTimePeriod().getSeconds(); broker.publish(market, type, CandleBarDto.builder() .time(barStart) .open(last.getOpenPrice().doubleValue()) .high(last.getHighPrice().doubleValue()) .low(last.getLowPrice().doubleValue()) .close(last.getClosePrice().doubleValue()) .volume(last.getVolume().doubleValue()) .build()); } catch (Exception e) { log.trace("[BarBuilder] upper publish skip {}/{}: {}", market, type, e.getMessage()); } } } /** PartialBar → Ta4j Bar 변환 */ private Bar buildTa4jBar(PartialBar partial, Duration duration, ZonedDateTime endZdt) { BarSeries tmp = new org.ta4j.core.BaseBarSeriesBuilder() .withBarBuilderFactory(new TimeBarBuilderFactory()) .withNumFactory(DoubleNumFactory.getInstance()) // Ta4j 0.22 .build(); // Ta4j 0.22: barBuilder().endTime() 은 Instant 파라미터 Instant endInstant = endZdt.toInstant(); return tmp.barBuilder() .timePeriod(duration) .endTime(endInstant) .openPrice(partial.open) .highPrice(partial.high) .lowPrice(partial.low) .closePrice(partial.close) .volume(partial.volume) .build(); } /** 업비트 일봉 기준 시각: KST (UTC+9) */ private static final ZoneId ZONE_KST = ZoneId.of("Asia/Seoul"); /** * 분 단위 시간을 상위 interval 의 시작으로 정렬. * * 일봉(minutes == 1440)은 업비트와 동일하게 KST 00:00 기준으로 정렬한다. * 그 외 모든 타임프레임은 UTC epoch 기준 정렬(기존 방식 유지). */ private ZonedDateTime alignToInterval(ZonedDateTime time, int minutes) { if (minutes == 1440) { // 일봉: KST 자정(00:00) 기준 — 업비트 일봉 경계와 일치 ZonedDateTime kst = time.withZoneSameInstant(ZONE_KST); return kst.toLocalDate().atStartOfDay(ZONE_KST); } long epochMin = time.toEpochSecond() / 60; long aligned = (epochMin / minutes) * minutes; return Instant.ofEpochSecond(aligned * 60).atZone(ZoneOffset.UTC); } // ── 내부 상태 ────────────────────────────────────────────────────────────── /** 진행 중인 단일 캔들 상태 */ static class PartialBar { final ZonedDateTime barStart; final double open; double high; double low; double close; double volume; PartialBar(ZonedDateTime barStart, double price, double volume) { this.barStart = barStart; this.open = price; this.high = price; this.low = price; this.close = price; this.volume = volume; } void update(double price, double vol) { if (price > this.high) this.high = price; if (price < this.low) this.low = price; this.close = price; this.volume += vol; } } }