web-page/docs/superpowers/specs/2026-05-16-signal-v2-phase3b-chronos-momentum.md

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_ID FHKST03010100
• ② 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_MODEL env
• ⑨ 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.40
• chronos-forecasting>=1.4
• torch>=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 → ChronosPrediction
• test_conf_high_when_distribution_narrow — narrow → conf ≈ 1
• test_conf_low_when_distribution_wide — wide → conf ≈ 0
• test_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_volume
• test_weak_up_3of5_green_normal_volume
• test_neutral_mixed
• test_weak_down_low_green_low_volume
• test_strong_down_5_consecutive_red_high_volume
• test_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.py get_daily_ohlcv 메서드 추가
• signal_v2/chronos_predictor.py 신규
• signal_v2/momentum_classifier.py 신규
• signal_v2/pull_worker.py post-close cycle + momentum 갱신
• signal_v2/scheduler.py _is_post_close_trigger
• signal_v2/state.py 3 필드 추가
• signal_v2/main.py lifespan ChronosPredictor 로드
• signal_v2/config.py CHRONOS_MODEL env
• requirements.txt 3 의존성 추가
• 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 회피)