Macbook
302 lines
14 KiB
Java
302 lines
14 KiB
Java
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<String, PartialBar> 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 에 저장하고 상위 타임프레임에 전파한다.
|
|
*
|
|
* <p>[항목2] 2-Phase 처리 — 멀티 타임프레임 레이스 컨디션 해소:
|
|
* <ul>
|
|
* <li><b>Phase 1</b>: 1m 봉과 모든 상위봉을 Ta4jStorage 에 먼저 확정한다.</li>
|
|
* <li><b>Phase 2</b>: 모든 저장이 완료된 후에야 전략 평가를 트리거한다.</li>
|
|
* </ul>
|
|
* 기존 방식(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<UpperCloseInfo> 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<UpperCloseInfo> 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;
|
|
}
|
|
}
|
|
}
|