KIS daily OHLCV fetch (kis_client.get_daily_ohlcv, FHKST03010100) + ChronosPredictor (HuggingFace amazon/chronos-2 zero-shot, env-configurable model, always-loaded) + minute momentum classifier (5-level rule: strong_up/weak_up/neutral/weak_down/strong_down) + post-close cycle trigger (16:00 KST). 12 new tests (33 → 45 total). brainstorming 7 decisions: daily=B(KIS REST) / freq=A(post-close 1x) / model=A(env CHRONOS_MODEL) / momentum=A(5-level rule) / state=B(median+ q10+q90+conf+as_of) / test=A(mock+pure) / scope=integrated. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
17 KiB
Confidence Signal Pipeline V2 — Phase 3b: Chronos-2 + Minute Momentum Design
작성일: 2026-05-16 작성자: gahusb 상태: Approved for implementation 선행 spec:
- Phase 0 architecture (
2026-05-15-confidence-signal-pipeline-v2-architecture.md) - Phase 1 stock WebAI API (
2026-05-15-signal-v2-phase1-webai-api.md) - Phase 2 web-ai pull worker (
2026-05-16-signal-v2-phase2-web-ai-pull-worker.md) - Phase 3a KIS data collection (
2026-05-16-signal-v2-phase3a-kis-data-collection.md)
브레인스토밍 결정 7개:
- daily data 소스 = B (KIS REST
kis_client.get_daily_ohlcv) - 추론 빈도 = A (종가 후 1회 + 메모리 보관)
- 모델 = A (env
CHRONOS_MODEL외부화, 기본amazon/chronos-2, 항상 로드) - 분봉 모멘텀 = A (5-level 룰 기반)
- State output = B (median + q10 + q90 + conf + as_of)
- 테스트 = A (모델 mock + 순수 함수)
- scope = 통합 9 항목 (Phase 3a 와 같은 1주 단위)
1. 목표
Phase 3a 의 데이터 위에 추론 레이어 추가. Chronos-2 zero-shot 으로 다음날 가격 분포 예측 + 1분봉 → 5분봉 aggregate 후 5-level 모멘텀 분류. Phase 4 (signal generator) 가 두 출력 + Phase 3a 의 호가/분봉 + Phase 2 의 portfolio/news_sentiment 를 종합해 매수/매도 신호 룰 적용.
Why: Phase 0 §3 "web-ai = 시점 분석" 책임의 추론 부분. Chronos-2 의 zero-shot quantile 분포 + 분봉 모멘텀 5-level 이 매수/매도 룰의 핵심 입력.
2. 범위
포함 (9 항목)
- ①
kis_client.get_daily_ohlcv(ticker, days=60)— KIS REST TR_IDFHKST03010100 - ②
chronos_predictor.py신규 —ChronosPredictor(HuggingFace 모델 + batch predict) - ③
momentum_classifier.py신규 —aggregate_1min_to_5min+classify_minute_momentum - ④
pull_worker.py확장 —_run_post_close_cycle+update_minute_momentum_for_all - ⑤
scheduler.py확장 —_is_post_close_trigger(16:00 KST) - ⑥
state.py확장 —daily_ohlcv+chronos_predictions+minute_momentum - ⑦
main.py확장 — lifespan 에 ChronosPredictor 로드 - ⑧
config.py확장 —CHRONOS_MODELenv - ⑨
requirements.txt—transformers>=4.40,chronos-forecasting>=1.4,torch>=2.0
범위 외 (NOT)
- Signal generator 매수/매도 룰 (Phase 4)
- agent-office
/signal호출 (Phase 5) - 모델 재학습/fine-tune — zero-shot only
- 다중 horizon 예측 — 1-day median 만, 다른 horizon Phase 7
- 외부 데이터 (yfinance/FDR) — KIS REST 만
- Chronos lazy load — 항상 로드 (Phase 7 모니터링 후 검토)
- 분봉 모멘텀 ML 모델 — 룰 기반만 (Phase 7 백테스트 후 ML 검토)
- WebSocket 동적 subscribe (Phase 3a backlog 그대로)
3. 파일 구조 + 변경 매트릭스
| 파일 | 작업 | 라인 |
|---|---|---|
signal_v2/kis_client.py |
get_daily_ohlcv 메서드 추가 |
+50 |
signal_v2/chronos_predictor.py |
신규 | ~120 |
signal_v2/momentum_classifier.py |
신규 | ~80 |
signal_v2/pull_worker.py |
post-close cycle + momentum 갱신 | +50 |
signal_v2/scheduler.py |
_is_post_close_trigger 헬퍼 |
+20 |
signal_v2/state.py |
3 필드 추가 | +5 |
signal_v2/main.py |
lifespan ChronosPredictor 로드 | +15 |
signal_v2/config.py |
chronos_model 필드 |
+3 |
signal_v2/requirements.txt |
3 의존성 | +3 |
signal_v2/tests/test_kis_client.py |
daily 1 케이스 | +30 |
signal_v2/tests/test_chronos_predictor.py |
신규 4 케이스 | ~120 |
signal_v2/tests/test_momentum_classifier.py |
신규 6 케이스 | ~150 |
signal_v2/tests/test_pull_worker.py |
post-close 1 케이스 | +50 |
합계: 13 파일 변경 (8 코드 + 4 테스트 + 1 requirements), 12 신규 테스트 (33 → 45 total).
외부 의존성 신규
transformers>=4.40chronos-forecasting>=1.4torch>=2.0(CUDA 12.x 빌드, V1 venv 공유 시 재설치 불필요)
모델 다운로드
amazon/chronos-2 HuggingFace 모델 첫 로드 시 ~1GB 다운로드 (~수십 초). ~/.cache/huggingface/ 캐시 후 무영향. Task 7 manual smoke 에 시간 예상 명시.
4. KIS Daily OHLCV (kis_client.get_daily_ohlcv)
async def get_daily_ohlcv(self, ticker: str, days: int = 60) -> list[dict]:
"""KRX 일봉 OHLCV (TR_ID FHKST03010100).
Args:
ticker: 6자리 종목코드
days: 최근 N영업일 (KIS 한도 100영업일)
Returns:
[{"datetime": "2026-05-15", "open": int, "high": int, "low": int,
"close": int, "volume": int}, ...]
시간 오름차순 (가장 최근이 마지막).
"""
path = "/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice"
today = datetime.now(KST).strftime("%Y%m%d")
start_date = (datetime.now(KST) - timedelta(days=days * 2)).strftime("%Y%m%d")
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": ticker,
"FID_INPUT_DATE_1": start_date,
"FID_INPUT_DATE_2": today,
"FID_PERIOD_DIV_CODE": "D",
"FID_ORG_ADJ_PRC": "1",
}
raw = await self._request_with_retry(
"GET", path, tr_id="FHKST03010100", params=params,
)
output2 = raw.get("output2", [])
bars = []
for row in output2:
try:
date = row["stck_bsop_date"]
bars.append({
"datetime": f"{date[:4]}-{date[4:6]}-{date[6:]}",
"open": int(row["stck_oprc"]),
"high": int(row["stck_hgpr"]),
"low": int(row["stck_lwpr"]),
"close": int(row["stck_clpr"]),
"volume": int(row["acml_vol"]),
})
except (KeyError, ValueError):
continue
bars.reverse() # KIS descending → ascending
return bars[-days:]
핵심:
- TR_ID
FHKST03010100(V1 패턴) - 수정주가 (
FID_ORG_ADJ_PRC=1) - start_date 를
days*2로 → 휴장일 + 주말 고려 →[-days:]트리밍
5. ChronosPredictor
@dataclass
class ChronosPrediction:
median: float
q10: float
q90: float
conf: float
as_of: str
class ChronosPredictor:
"""HuggingFace Chronos-2 zero-shot forecaster."""
def __init__(self, model_name: str = "amazon/chronos-2", device: str | None = None):
from chronos import ChronosPipeline
import torch
self._device = device or ("cuda" if torch.cuda.is_available() else "cpu")
logger.info("Loading Chronos pipeline: %s on %s", model_name, self._device)
self._pipeline = ChronosPipeline.from_pretrained(
model_name,
device_map=self._device,
torch_dtype=torch.float16 if self._device == "cuda" else torch.float32,
)
def predict_batch(
self,
daily_ohlcv_dict: dict[str, list[dict]],
prediction_length: int = 1,
num_samples: int = 100,
) -> dict[str, ChronosPrediction]:
"""종목별 1-day return 분포 예측."""
import torch
import numpy as np
tickers = list(daily_ohlcv_dict.keys())
contexts = [
torch.tensor([bar["close"] for bar in daily_ohlcv_dict[t]], dtype=torch.float32)
for t in tickers
]
forecasts = self._pipeline.predict(
context=contexts, prediction_length=prediction_length, num_samples=num_samples,
)
from datetime import datetime
now_iso = datetime.now(KST).isoformat()
results: dict[str, ChronosPrediction] = {}
for i, ticker in enumerate(tickers):
samples = forecasts[i, :, 0].numpy()
last_close = daily_ohlcv_dict[ticker][-1]["close"]
returns = (samples - last_close) / last_close
median = float(np.quantile(returns, 0.5))
q10 = float(np.quantile(returns, 0.1))
q90 = float(np.quantile(returns, 0.9))
spread = (q90 - q10) / max(abs(median), 0.001)
conf = float(max(0.0, min(1.0, 1.0 - spread / 2.0)))
results[ticker] = ChronosPrediction(median, q10, q90, conf, now_iso)
return results
핵심:
- Lazy import (
chronos-forecasting무거움) - GPU 자동 감지 + FP16 (CUDA) / FP32 (CPU)
- Batch predict — 30+ 종목 동시 ~1-2초
- Price → return 변환
- Confidence — 분포 폭 기반 (좁을수록 1)
6. 분봉 모멘텀 분류기
6.1 1분봉 → 5분봉 aggregate
def aggregate_1min_to_5min(minute_bars: list[dict]) -> list[dict]:
"""1분봉 N개 → 5분봉 floor(N/5) 개. 시간 오름차순."""
bars_5min = []
chunks = len(minute_bars) // 5
for i in range(chunks):
chunk = minute_bars[i * 5 : (i + 1) * 5]
bars_5min.append({
"datetime": chunk[0]["datetime"],
"open": chunk[0]["open"],
"high": max(b["high"] for b in chunk),
"low": min(b["low"] for b in chunk),
"close": chunk[-1]["close"],
"volume": sum(b["volume"] for b in chunk),
})
return bars_5min
6.2 5-level 분류
def classify_minute_momentum(minute_bars: deque) -> str:
"""1분봉 deque → strong_up / weak_up / neutral / weak_down / strong_down."""
minute_list = list(minute_bars)
if len(minute_list) < 5 * 5: # 25 bars minimum
return NEUTRAL
bars_5min = aggregate_1min_to_5min(minute_list)
if len(bars_5min) < 5:
return NEUTRAL
recent = bars_5min[-5:] # 직전 5개 5분봉
up_count = sum(1 for b in recent if b["close"] > b["open"])
# 거래량 multiplier — recent 5 vs 60분 평균
recent_vol_avg = sum(b["volume"] for b in recent) / len(recent)
long_window = bars_5min[-12:] # 60분 = 5분봉 12개
long_vol_avg = sum(b["volume"] for b in long_window) / len(long_window)
vol_mult = recent_vol_avg / long_vol_avg if long_vol_avg > 0 else 1.0
if up_count == 5 and vol_mult >= 1.5:
return STRONG_UP
elif up_count >= 3 and vol_mult >= 1.0:
return WEAK_UP
elif up_count == 0 and vol_mult >= 1.5:
return STRONG_DOWN
elif up_count <= 2 and vol_mult < 1.0:
return WEAK_DOWN
else:
return NEUTRAL
7. PollState 확장 + pull_worker
7.1 PollState 추가 필드
@dataclass
class PollState:
# ... 기존 필드 ...
# Phase 3b additions
daily_ohlcv: dict[str, list[dict]] = field(default_factory=dict)
chronos_predictions: dict[str, dict] = field(default_factory=dict)
minute_momentum: dict[str, str] = field(default_factory=dict)
7.2 pull_worker 확장
async def _run_post_close_cycle(
kis_client: KISClient, chronos: ChronosPredictor, state: PollState,
) -> None:
"""16:00 KST 종가 후 1회: daily fetch + chronos predict."""
tickers = list(set(_portfolio_tickers(state)) | set(_screener_tickers(state)))
daily_results = await asyncio.gather(*[
kis_client.get_daily_ohlcv(t, days=60) for t in tickers
], return_exceptions=True)
daily_dict = {}
for ticker, result in zip(tickers, daily_results):
if isinstance(result, list) and len(result) >= 30:
daily_dict[ticker] = result
state.daily_ohlcv[ticker] = result
if daily_dict:
predictions = chronos.predict_batch(daily_dict)
now_iso = datetime.now(KST).isoformat()
for ticker, pred in predictions.items():
state.chronos_predictions[ticker] = {
"median": pred.median, "q10": pred.q10, "q90": pred.q90,
"conf": pred.conf, "as_of": pred.as_of,
}
state.last_updated[f"chronos/{ticker}"] = pred.as_of
def update_minute_momentum_for_all(state: PollState) -> None:
"""매 분봉 cycle 후 호출 — 모든 종목 모멘텀 갱신."""
from signal_v2.momentum_classifier import classify_minute_momentum
for ticker, bars in state.minute_bars.items():
state.minute_momentum[ticker] = classify_minute_momentum(bars)
7.3 scheduler _is_post_close_trigger
def _is_post_close_trigger(now: datetime) -> bool:
"""16:00 KST ±1분 (post-close cycle 트리거)."""
if not _is_market_day(now):
return False
t = now.time()
return time(16, 0) <= t < time(16, 1)
poll_loop 안에서 매 cycle:
if _is_post_close_trigger(now) and chronos is not None:
await _run_post_close_cycle(kis_client, chronos, state)
8. 테스트 (12 신규)
8.1 test_kis_client.py (1)
test_get_daily_ohlcv_returns_60_bars— respx mock 200 → 60 bars 시간 오름차순
8.2 test_chronos_predictor.py (4, 모델 mock)
test_predict_batch_returns_prediction_dict— mock pipeline → ChronosPredictiontest_conf_high_when_distribution_narrow— narrow → conf ≈ 1test_conf_low_when_distribution_wide— wide → conf ≈ 0test_return_computed_from_price_relative_to_last_close— price → return 변환
8.3 test_momentum_classifier.py (6)
test_strong_up_5_consecutive_green_with_high_volumetest_weak_up_3of5_green_normal_volumetest_neutral_mixedtest_weak_down_low_green_low_volumetest_strong_down_5_consecutive_red_high_volumetest_aggregate_1min_to_5min_correctness
8.4 test_pull_worker.py (1)
test_post_close_cycle_updates_chronos_predictions— mock kis + mock chronos → state 갱신
합계: 1 + 4 + 6 + 1 = 12 신규. 기존 33 + 12 = 45 total.
9. 위험 및 완화
| 위험 | 완화 |
|---|---|
| Chronos-2 첫 로드 ~1GB 다운로드 | startup INFO + Task 7 smoke 시간 예상 명시 |
| GPU OOM (Chronos + V1 Ollama 동거) | FP16 ~400MB + Ollama 4GB = 5GB / 15.5GB 여유. Phase 5 Qwen3 추가 시 13.3GB. Phase 6 V1 deprecation 후 해소 |
chronos-forecasting 호환 (transformers 버전) |
명시 버전. 운영 첫 install 검증 |
| KIS daily fetch + V1 Macro 동시 → rate limit (EGW00201) | post-close 16:00 트리거 vs V1 Trading Bot 의 장 마감 cycle 충돌 위험. 운영 검증 후 16:05 으로 조정 가능 |
| Chronos-2 예측 정확도 불확실 | Phase 7 IC 검증 + 신호 hit-rate 추적. 부족 시 model env 변경 또는 Moirai-2.0 |
| 모멘텀 룰 임계값 (1.5x / 5/5) 보수적 | Phase 7 운영 후 임계값 조정 |
| 1분봉 60개 미만 (장 시작 1시간 내) | NEUTRAL 폴백. 09:00-10:00 신호 발생 안 함 (운영 허용) |
| Chronos 모델 다운로드 네트워크 단절 | startup RuntimeError + 운영자 알림 + 재시작. 캐시 후 무관 |
| daily_ohlcv 메모리 누수 | 종목 ~30 × 60일 ~100B = ~180KB. 무시 |
| Chronos 추론 시 V1 Ollama 와 동시 GPU 사용 | 일 1회 + 짧음 (~2초). V1 Ollama 의 GPU 점유 사이에 끼어들 가능성 → 일시 deferred. Phase 7 모니터링 |
10. 운영 영향
| 항목 | 영향 |
|---|---|
| 다운타임 | signal_v2 재기동 ~30초 (첫 모델 로드) |
| 사용자 영향 | 없음 (Phase 3b 도 silent, 신호 발송은 Phase 5) |
.env 갱신 |
optional 1줄 (CHRONOS_MODEL=amazon/chronos-2 — 기본값과 동일 시 미설정 OK) |
| V1 영향 | 0 (별도 process). GPU 메모리만 공유 |
| KIS API 부하 | post-close cycle 일 1회 30 종목 daily fetch ~60 calls. 평소 분봉/호가 cycle 그대로 |
| 모델 다운로드 | 첫 시작 ~1GB / 캐시 |
11. Phase 3b 완료 조건 (DoD)
signal_v2/kis_client.pyget_daily_ohlcv메서드 추가signal_v2/chronos_predictor.py신규signal_v2/momentum_classifier.py신규signal_v2/pull_worker.pypost-close cycle + momentum 갱신signal_v2/scheduler.py_is_post_close_triggersignal_v2/state.py3 필드 추가signal_v2/main.pylifespan ChronosPredictor 로드signal_v2/config.pyCHRONOS_MODELenvrequirements.txt3 의존성 추가- 12 신규 테스트 PASS (총 45)
- 운영 smoke: signal_v2 시작 → Chronos 모델 로드 성공 → 16:00 post-close cycle 1회 실행 → state.chronos_predictions 갱신 확인
- V1 무영향 (GPU OOM 없음)
- git push
12. Phase 4 와의 관계
본 Phase 3b 완료 후 즉시 Phase 4 (Signal Generator) brainstorming. 의존성:
[Phase 3b spec/plan/실행] → [Phase 4 spec/plan/실행]
1주 1주
Phase 4 의 입력 = 본 spec 의 state.chronos_predictions + state.minute_momentum + Phase 3a 의 state.asking_price + Phase 2 의 state.portfolio + state.news_sentiment. Phase 4 산출 = state.signals[ticker] (buy/sell decision + confidence).
13. Backlog (본 spec NOT)
- Chronos lazy load (Phase 5 Qwen3 동거 시 VRAM 압박 검토)
- 다중 horizon (1-day + 5-day + 20-day)
- ML 기반 분봉 모멘텀 (현재 룰 기반만)
- Chronos model A/B (chronos-bolt-base vs chronos-2 비교 실험)
- KIS daily fetch 의 V1 충돌 회피 — file mutex 또는 V2 별도 app_key
- Chronos quantile 의 임의 quantile 지원 (현재 q10/q50/q90 만)
- daily_ohlcv 영속 저장 (재기동 시 reset 회피)