643 lines
22 KiB
Markdown
643 lines
22 KiB
Markdown
# ai_trade Hotfix — Code Review F1·F2·F3·F4 Implementation Plan
|
|
|
|
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
|
|
|
**Goal:** ai_trade(V2) 코드 리뷰 7개 finding 중 High 3건(F1·F2·F3) + Medium 1건(F4)을 TDD로 수정. F5/F6은 별도 plan, F7은 pushback.
|
|
|
|
**Architecture:** 모두 ai_trade/ 내부 단일 모듈 수정. (1) config.py default 경로 — legacy/ 경유. (2) kis_client.py — asyncio.Lock으로 `_throttle()` 직렬화. (3) scheduler.py + pull_worker.py — post-close를 시간 윈도우가 아닌 "일 1회 + 16:00 이후" 상태기반으로 변경. (4) chronos_predictor.py — confidence 산식을 absolute spread 기반으로 통일.
|
|
|
|
**Tech Stack:** Python 3.12, asyncio, pytest + pytest-asyncio + respx, httpx.
|
|
|
|
**Test runner:** `.venv` 한글 경로 깨짐 + 리뷰어 Python 312 경로 부재 보고로, 시스템 Python 사용. 정확한 경로는 `where python` 으로 우선 확인. 기본 시도 순서:
|
|
1. `python -m pytest ai_trade/tests -q` (PATH의 Python)
|
|
2. `py -3.12 -m pytest ai_trade/tests -q` (py launcher)
|
|
3. 둘 다 실패 시 환경 셋업이 선행 작업 — plan 진행 중단하고 박재오에게 보고.
|
|
|
|
**Working directory:** `C:\Users\jaeoh\Desktop\workspace\web-ai` (web-ai repo). Commit/push도 이 디렉토리에서만.
|
|
|
|
---
|
|
|
|
## File Map
|
|
|
|
| 파일 | 변경 종류 | 책임 |
|
|
|------|-----------|------|
|
|
| `ai_trade/config.py` | Modify L31-36 | V1_TOKEN_PATH default를 `legacy/signal_v1/data/kis_token.json`로 |
|
|
| `ai_trade/kis_client.py` | Modify L40-62 | `_throttle_lock: asyncio.Lock` 추가, `_throttle()`을 lock 안에서 실행 |
|
|
| `ai_trade/scheduler.py` | Modify L79-84 | `_is_post_close_trigger(now, last_post_close_date)` 시그니처 변경 — 상태기반 |
|
|
| `ai_trade/pull_worker.py` | Modify L1-58 | `poll_loop`에 `last_post_close_date` state 추가, 호출부 갱신 |
|
|
| `ai_trade/chronos_predictor.py` | Modify L106, L127 | spread 계산을 absolute (q90-q10)로, confidence 산식 `max(0, 1 - spread/0.6)` |
|
|
| `ai_trade/tests/test_kis_client.py` | Add 1 test | concurrent gather throttle test |
|
|
| `ai_trade/tests/test_scheduler.py` | Add 3 tests | post-close 상태기반 트리거 |
|
|
| `ai_trade/tests/test_pull_worker.py` | Add 1 test | 첫 호출 안 됐다가 16:00 이후 5분 cycle에서 호출됨 |
|
|
| `ai_trade/tests/test_chronos_predictor.py` | Add 2 tests | median≈0에서도 conf 정상, spread 클수록 conf↓ |
|
|
| `ai_trade/tests/test_main.py` | Modify | v1_token_path default 변경 반영 (있다면) |
|
|
|
|
---
|
|
|
|
## Task 1: F1 — KIS 토큰 경로 default를 legacy/ 경유로
|
|
|
|
**Files:**
|
|
- Modify: `ai_trade/config.py:31-36`
|
|
- Test: `ai_trade/tests/test_config_token_path.py` (Create)
|
|
|
|
- [ ] **Step 1: Write the failing test**
|
|
|
|
```python
|
|
# ai_trade/tests/test_config_token_path.py
|
|
"""F1 — V1_TOKEN_PATH default가 legacy/signal_v1/ 경유인지 검증."""
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from ai_trade.config import Settings
|
|
|
|
|
|
def test_v1_token_default_path_uses_legacy_dir(monkeypatch):
|
|
"""env에 V1_TOKEN_PATH 없으면 legacy/signal_v1/data/kis_token.json"""
|
|
monkeypatch.delenv("V1_TOKEN_PATH", raising=False)
|
|
settings = Settings()
|
|
expected_suffix = Path("legacy") / "signal_v1" / "data" / "kis_token.json"
|
|
assert str(settings.v1_token_path).endswith(str(expected_suffix)), (
|
|
f"expected default to end with {expected_suffix}, got {settings.v1_token_path}"
|
|
)
|
|
|
|
|
|
def test_v1_token_env_override_wins(monkeypatch, tmp_path):
|
|
"""env로 명시한 경로가 default를 덮어씀."""
|
|
custom = tmp_path / "custom_token.json"
|
|
monkeypatch.setenv("V1_TOKEN_PATH", str(custom))
|
|
settings = Settings()
|
|
assert settings.v1_token_path == custom
|
|
```
|
|
|
|
- [ ] **Step 2: Run test to verify it fails**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_config_token_path.py -v
|
|
```
|
|
|
|
Expected: `test_v1_token_default_path_uses_legacy_dir` FAILs (default가 `signal_v1/...` 임). env override는 PASS.
|
|
|
|
- [ ] **Step 3: Fix config.py**
|
|
|
|
`ai_trade/config.py:31-36` 변경:
|
|
|
|
```python
|
|
v1_token_path: Path = field(
|
|
default_factory=lambda: Path(
|
|
os.getenv("V1_TOKEN_PATH",
|
|
str(Path(__file__).parent.parent / "legacy" / "signal_v1" / "data" / "kis_token.json"))
|
|
)
|
|
)
|
|
```
|
|
|
|
- [ ] **Step 4: Run test to verify it passes**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_config_token_path.py -v
|
|
```
|
|
|
|
Expected: 2 passed.
|
|
|
|
- [ ] **Step 5: Verify full test suite still passes**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests -q
|
|
```
|
|
|
|
Expected: 모든 기존 테스트 PASS (token path 기본값 변경이 다른 test에 영향 없는지 확인).
|
|
|
|
- [ ] **Step 6: Commit**
|
|
|
|
```bash
|
|
git add ai_trade/config.py ai_trade/tests/test_config_token_path.py
|
|
git commit -m "fix(ai_trade): V1_TOKEN_PATH default를 legacy/signal_v1/ 경유로 수정 (F1)"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 2: F2 — KIS throttle을 asyncio.Lock으로 직렬화
|
|
|
|
**Files:**
|
|
- Modify: `ai_trade/kis_client.py:40-62`
|
|
- Test: `ai_trade/tests/test_kis_client.py` (Modify — 새 test 추가)
|
|
|
|
- [ ] **Step 1: Write the failing test**
|
|
|
|
`ai_trade/tests/test_kis_client.py` 파일 끝에 추가:
|
|
|
|
```python
|
|
import asyncio
|
|
import time as time_module
|
|
|
|
|
|
@respx.mock
|
|
async def test_throttle_serializes_concurrent_gather(kis_client_factory):
|
|
"""5개 동시 요청이 asyncio.gather로 들어와도 0.5초 간격으로 직렬화되어야 함 (F2).
|
|
|
|
초당 2회 = 0.5초 간격. 5개 요청이면 최소 (5-1)*0.5 = 2.0초 소요.
|
|
Race condition 있으면 5개가 거의 동시에 나가서 2초 훨씬 안쪽에 끝남.
|
|
"""
|
|
sample = {"output2": []}
|
|
respx.get(
|
|
"https://openapivts.koreainvestment.com:29443/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"
|
|
).mock(return_value=httpx.Response(200, json=sample))
|
|
|
|
client = kis_client_factory()
|
|
try:
|
|
start = time_module.monotonic()
|
|
await asyncio.gather(*[client.get_minute_ohlcv(f"00593{i}") for i in range(5)])
|
|
elapsed = time_module.monotonic() - start
|
|
# 5개 throttle = 최소 (5-1)*0.5 = 2.0초. tolerance 0.3초.
|
|
assert elapsed >= 1.7, (
|
|
f"throttle race condition: 5 concurrent calls took only {elapsed:.2f}s, "
|
|
f"expected >=1.7s (0.5s * 4 inter-call gaps)"
|
|
)
|
|
finally:
|
|
await client.close()
|
|
```
|
|
|
|
- [ ] **Step 2: Run test to verify it fails**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_kis_client.py::test_throttle_serializes_concurrent_gather -v
|
|
```
|
|
|
|
Expected: FAIL — elapsed가 0.5초 이하 (race로 동시 깸).
|
|
|
|
- [ ] **Step 3: Add asyncio.Lock to KISClient**
|
|
|
|
`ai_trade/kis_client.py:40` `__init__` 끝에 한 줄 추가:
|
|
|
|
```python
|
|
self._token_cache: tuple[str, float] | None = None # (token, file_mtime)
|
|
self._last_throttle_at = 0.0
|
|
self._throttle_lock = asyncio.Lock()
|
|
```
|
|
|
|
그리고 `_throttle()` (L58-62)을 lock으로 감쌈:
|
|
|
|
```python
|
|
async def _throttle(self) -> None:
|
|
async with self._throttle_lock:
|
|
elapsed = time.monotonic() - self._last_throttle_at
|
|
if elapsed < _THROTTLE_INTERVAL:
|
|
await asyncio.sleep(_THROTTLE_INTERVAL - elapsed)
|
|
self._last_throttle_at = time.monotonic()
|
|
```
|
|
|
|
- [ ] **Step 4: Run test to verify it passes**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_kis_client.py::test_throttle_serializes_concurrent_gather -v
|
|
```
|
|
|
|
Expected: PASS — elapsed >= 1.7s.
|
|
|
|
- [ ] **Step 5: Verify full kis_client suite still passes**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_kis_client.py -v
|
|
```
|
|
|
|
Expected: 모든 test PASS (기존 429 retry 등 영향 없는지 확인).
|
|
|
|
- [ ] **Step 6: Commit**
|
|
|
|
```bash
|
|
git add ai_trade/kis_client.py ai_trade/tests/test_kis_client.py
|
|
git commit -m "fix(ai_trade): KIS throttle을 asyncio.Lock으로 직렬화 (F2)"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 3: F3 — post-close 트리거를 상태기반으로 변경
|
|
|
|
**Files:**
|
|
- Modify: `ai_trade/scheduler.py:79-84`
|
|
- Modify: `ai_trade/pull_worker.py:1-58`
|
|
- Test: `ai_trade/tests/test_scheduler.py` (add 3 tests)
|
|
|
|
**Why state-based:** 16:00:00-16:00:59 윈도우는 5분 sleep + 비결정적 cycle 시작 시각과 충돌. "오늘 아직 post-close 안 돌렸고 현재 시각 ≥ 16:00 이면 trigger 후 today 표시" 로 변경.
|
|
|
|
- [ ] **Step 1: Write the failing tests**
|
|
|
|
`ai_trade/tests/test_scheduler.py` 파일 끝에 추가:
|
|
|
|
```python
|
|
from datetime import date as _date
|
|
from ai_trade.scheduler import _is_post_close_trigger
|
|
|
|
|
|
def test_post_close_trigger_fires_at_1601_if_not_yet_today():
|
|
"""16:01에 깬 cycle도 오늘 아직 안 돌렸으면 trigger (F3)."""
|
|
now = _kst(2026, 5, 18, 16, 1)
|
|
assert _is_post_close_trigger(now, last_post_close_date=None) is True
|
|
|
|
|
|
def test_post_close_trigger_skips_if_already_today():
|
|
"""이미 오늘 돌렸으면 trigger 안 함."""
|
|
now = _kst(2026, 5, 18, 16, 5)
|
|
today = _date(2026, 5, 18)
|
|
assert _is_post_close_trigger(now, last_post_close_date=today) is False
|
|
|
|
|
|
def test_post_close_trigger_skips_before_1600():
|
|
"""16:00 전에는 trigger 안 함."""
|
|
now = _kst(2026, 5, 18, 15, 59)
|
|
assert _is_post_close_trigger(now, last_post_close_date=None) is False
|
|
|
|
|
|
def test_post_close_trigger_fires_next_day_after_reset():
|
|
"""다음 영업일이 되면 last_post_close_date < today.date() 이므로 다시 trigger."""
|
|
now = _kst(2026, 5, 19, 16, 0)
|
|
yesterday = _date(2026, 5, 18)
|
|
assert _is_post_close_trigger(now, last_post_close_date=yesterday) is True
|
|
|
|
|
|
def test_post_close_trigger_skips_on_holiday():
|
|
"""휴장일에는 trigger 안 함 (2026-05-05 어린이날)."""
|
|
now = _kst(2026, 5, 5, 16, 30)
|
|
assert _is_post_close_trigger(now, last_post_close_date=None) is False
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify they fail**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_scheduler.py -v -k post_close
|
|
```
|
|
|
|
Expected: FAIL — `_is_post_close_trigger`가 신규 시그니처(`last_post_close_date` 인자) 미지원.
|
|
|
|
- [ ] **Step 3: Modify scheduler.py:79-84**
|
|
|
|
```python
|
|
def _is_post_close_trigger(now: datetime, last_post_close_date) -> bool:
|
|
"""16:00 KST 이후 오늘 아직 post-close cycle 안 돌렸으면 True (F3 상태기반).
|
|
|
|
Args:
|
|
now: 현재 KST datetime.
|
|
last_post_close_date: 마지막 post-close 실행 영업일 date 객체 (None=미실행).
|
|
"""
|
|
if not _is_market_day(now):
|
|
return False
|
|
if now.time() < time(16, 0):
|
|
return False
|
|
today = now.date()
|
|
return last_post_close_date != today
|
|
```
|
|
|
|
- [ ] **Step 4: Run scheduler tests**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_scheduler.py -v
|
|
```
|
|
|
|
Expected: 신규 5개 PASS. 기존 test도 PASS (다른 함수 영향 없음).
|
|
|
|
- [ ] **Step 5: Update pull_worker.py to track last_post_close_date**
|
|
|
|
`ai_trade/pull_worker.py` 의 `poll_loop` (L18-58)을 다음으로 교체:
|
|
|
|
```python
|
|
async def poll_loop(
|
|
client: StockClient, state: PollState, shutdown: asyncio.Event,
|
|
kis_client: KISClient | None = None,
|
|
chronos=None,
|
|
dedup=None,
|
|
settings=None,
|
|
) -> None:
|
|
"""FastAPI lifespan 에서 asyncio.create_task 로 시작."""
|
|
logger.info("poll_loop started")
|
|
last_post_close_date = None
|
|
while not shutdown.is_set():
|
|
now = datetime.now(KST)
|
|
if _is_market_day(now) and _is_polling_window(now):
|
|
try:
|
|
await _run_polling_cycle(client, state, kis_client=kis_client)
|
|
except Exception:
|
|
logger.exception("poll cycle failed")
|
|
# Minute momentum 갱신 (매 cycle)
|
|
try:
|
|
update_minute_momentum_for_all(state)
|
|
except Exception:
|
|
logger.exception("minute momentum update failed")
|
|
# Post-close trigger (상태기반: 16:00 이후 + 오늘 미실행)
|
|
if (
|
|
_is_post_close_trigger(now, last_post_close_date)
|
|
and chronos is not None and kis_client is not None
|
|
):
|
|
try:
|
|
await _run_post_close_cycle(kis_client, chronos, state)
|
|
last_post_close_date = now.date()
|
|
except Exception:
|
|
logger.exception("post-close cycle failed")
|
|
# Phase 4: generate signals
|
|
if dedup is not None and settings is not None:
|
|
try:
|
|
from ai_trade.signal_generator import generate_signals
|
|
generate_signals(state, dedup, settings)
|
|
except Exception:
|
|
logger.exception("generate_signals failed")
|
|
interval = _next_interval(now)
|
|
try:
|
|
await asyncio.wait_for(shutdown.wait(), timeout=interval)
|
|
break
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
logger.info("poll_loop ended")
|
|
```
|
|
|
|
- [ ] **Step 6: Add pull_worker test**
|
|
|
|
`ai_trade/tests/test_pull_worker.py` 파일 끝에 추가:
|
|
|
|
```python
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from datetime import datetime as _dt
|
|
from zoneinfo import ZoneInfo as _ZI
|
|
import asyncio as _asyncio
|
|
|
|
|
|
async def test_post_close_fires_at_1601_when_not_yet_today(monkeypatch):
|
|
"""16:01에 깬 cycle도 post_close 안 돌렸으면 호출됨 (F3 회귀)."""
|
|
from ai_trade import pull_worker
|
|
|
|
_kst = _ZI("Asia/Seoul")
|
|
now_at_1601 = _dt(2026, 5, 18, 16, 1, tzinfo=_kst)
|
|
|
|
real_dt = _dt
|
|
|
|
class FrozenDateTime:
|
|
@staticmethod
|
|
def now(tz=None):
|
|
return now_at_1601
|
|
|
|
monkeypatch.setattr(pull_worker, "datetime", FrozenDateTime)
|
|
monkeypatch.setattr(
|
|
pull_worker, "_is_market_day", lambda n: True,
|
|
)
|
|
monkeypatch.setattr(
|
|
pull_worker, "_is_polling_window", lambda n: True,
|
|
)
|
|
monkeypatch.setattr(
|
|
pull_worker, "_next_interval", lambda n: 0.01,
|
|
)
|
|
monkeypatch.setattr(
|
|
pull_worker, "_run_polling_cycle", AsyncMock(),
|
|
)
|
|
monkeypatch.setattr(
|
|
pull_worker, "update_minute_momentum_for_all", lambda s: None,
|
|
)
|
|
post_close = AsyncMock()
|
|
monkeypatch.setattr(pull_worker, "_run_post_close_cycle", post_close)
|
|
|
|
state = MagicMock()
|
|
chronos = MagicMock()
|
|
kis = MagicMock()
|
|
shutdown = _asyncio.Event()
|
|
|
|
async def _stop_soon():
|
|
await _asyncio.sleep(0.05)
|
|
shutdown.set()
|
|
|
|
_asyncio.create_task(_stop_soon())
|
|
await pull_worker.poll_loop(
|
|
client=MagicMock(),
|
|
state=state,
|
|
shutdown=shutdown,
|
|
kis_client=kis,
|
|
chronos=chronos,
|
|
dedup=None,
|
|
settings=None,
|
|
)
|
|
|
|
assert post_close.await_count >= 1, "post-close가 16:01에 호출되지 않음 (F3 회귀)"
|
|
```
|
|
|
|
- [ ] **Step 7: Run pull_worker test**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_pull_worker.py::test_post_close_fires_at_1601_when_not_yet_today -v
|
|
```
|
|
|
|
Expected: PASS.
|
|
|
|
- [ ] **Step 8: Run full ai_trade suite**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests -q
|
|
```
|
|
|
|
Expected: 모두 PASS.
|
|
|
|
- [ ] **Step 9: Commit**
|
|
|
|
```bash
|
|
git add ai_trade/scheduler.py ai_trade/pull_worker.py ai_trade/tests/test_scheduler.py ai_trade/tests/test_pull_worker.py
|
|
git commit -m "fix(ai_trade): post-close trigger를 상태기반으로 변경 (F3)"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 4: F4 — Chronos confidence를 absolute spread 기반으로 통일
|
|
|
|
**Files:**
|
|
- Modify: `ai_trade/chronos_predictor.py:106, 127`
|
|
- Test: `ai_trade/tests/test_chronos_predictor.py` (add 2 tests)
|
|
|
|
**Why absolute:** Phase 4 spec amendment (web-ui commit 534ded5)가 absolute spread로 hard gate를 결정. confidence도 같은 철학으로. 새 산식: `conf = max(0, min(1, 1 - spread / SPREAD_THRESHOLD))` — spread가 0.6에 도달하면 conf=0, 0이면 conf=1.
|
|
|
|
- [ ] **Step 1: Write the failing tests**
|
|
|
|
기존 `ai_trade/tests/test_chronos_predictor.py` 끝에 추가 (파일이 없거나 비어있으면 신규):
|
|
|
|
```python
|
|
import numpy as np
|
|
import pytest
|
|
import torch
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_pipeline():
|
|
"""predict_quantiles만 stub하는 가짜 pipeline."""
|
|
class FakePipeline:
|
|
def __init__(self, q10_price, q50_price, q90_price):
|
|
self._q10, self._q50, self._q90 = q10_price, q50_price, q90_price
|
|
def predict_quantiles(self, contexts, prediction_length, quantile_levels):
|
|
n = len(contexts)
|
|
tensor = torch.tensor(
|
|
[[[self._q10, self._q50, self._q90]]] * n,
|
|
dtype=torch.float32,
|
|
)
|
|
return tensor, None
|
|
return FakePipeline
|
|
|
|
|
|
def _make_predictor_with(pipeline_obj):
|
|
"""ChronosPredictor 인스턴스 (실제 모델 안 부르고 pipeline만 주입)."""
|
|
from ai_trade.chronos_predictor import ChronosPredictor
|
|
p = ChronosPredictor.__new__(ChronosPredictor)
|
|
p._pipeline = pipeline_obj
|
|
p._device = "cpu"
|
|
return p
|
|
|
|
|
|
def test_confidence_high_when_spread_near_zero(fake_pipeline):
|
|
"""median≈0, spread≈0 (q10=q90=last_close)일 때 conf≈1 (F4)."""
|
|
last_close = 100000.0
|
|
p = _make_predictor_with(fake_pipeline(last_close, last_close, last_close))
|
|
ohlcv = {"A": [{"close": last_close}] * 30}
|
|
out = p.predict_batch(ohlcv)
|
|
assert out["A"].conf > 0.95, (
|
|
f"median≈0 + spread≈0인데 conf={out['A'].conf} (F4 회귀: relative spread로 폭증)"
|
|
)
|
|
|
|
|
|
def test_confidence_drops_with_spread(fake_pipeline):
|
|
"""spread 0.3일 때 conf≈0.5 (1 - 0.3/0.6 = 0.5)."""
|
|
last_close = 100000.0
|
|
# q10=85000 → -0.15, q90=115000 → 0.15, spread=0.30
|
|
p = _make_predictor_with(fake_pipeline(85000.0, 100000.0, 115000.0))
|
|
ohlcv = {"A": [{"close": last_close}] * 30}
|
|
out = p.predict_batch(ohlcv)
|
|
# 1 - 0.30/0.60 = 0.50
|
|
assert 0.45 < out["A"].conf < 0.55, (
|
|
f"absolute spread 0.30에서 conf={out['A'].conf} (expected ≈0.5)"
|
|
)
|
|
|
|
|
|
def test_confidence_zero_at_threshold_spread(fake_pipeline):
|
|
"""spread가 threshold(0.6) 이상이면 conf=0."""
|
|
last_close = 100000.0
|
|
# q10=70000 → -0.30, q90=130000 → 0.30, spread=0.60
|
|
p = _make_predictor_with(fake_pipeline(70000.0, 100000.0, 130000.0))
|
|
ohlcv = {"A": [{"close": last_close}] * 30}
|
|
out = p.predict_batch(ohlcv)
|
|
assert out["A"].conf < 0.05, (
|
|
f"spread=threshold에서 conf={out['A'].conf} (expected ≈0)"
|
|
)
|
|
```
|
|
|
|
- [ ] **Step 2: Run tests to verify they fail**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_chronos_predictor.py -v -k confidence
|
|
```
|
|
|
|
Expected: `test_confidence_high_when_spread_near_zero` 가 FAIL — 현행 relative spread 산식 때문에 median≈0에서 conf가 0으로 폭락.
|
|
|
|
- [ ] **Step 3: Fix chronos_predictor.py**
|
|
|
|
`ai_trade/chronos_predictor.py` 상단에 상수 추가 (L13 근처):
|
|
|
|
```python
|
|
_SPREAD_THRESHOLD = 0.6 # F4: signal_generator hard gate와 동일 (absolute return spread)
|
|
```
|
|
|
|
L106 (modern API 경로) 변경:
|
|
|
|
```python
|
|
# shape: [num_series, prediction_length, 3]
|
|
for i, ticker in enumerate(tickers):
|
|
q10_price, q50_price, q90_price = quantiles_np[i, 0, :]
|
|
last_close = daily_ohlcv_dict[ticker][-1]["close"]
|
|
median = float((q50_price - last_close) / last_close)
|
|
q10 = float((q10_price - last_close) / last_close)
|
|
q90 = float((q90_price - last_close) / last_close)
|
|
# F4: absolute spread (q90-q10) 기반 — signal_generator hard gate와 통일.
|
|
# median≈0 zero-shot 케이스에서 conf가 0으로 폭락하던 relative 산식 제거.
|
|
spread = q90 - q10
|
|
conf = float(max(0.0, min(1.0, 1.0 - spread / _SPREAD_THRESHOLD)))
|
|
results[ticker] = ChronosPrediction(
|
|
median=median, q10=q10, q90=q90, conf=conf, as_of=now_iso,
|
|
)
|
|
return results
|
|
```
|
|
|
|
L127 (legacy API 경로) 동일하게 변경:
|
|
|
|
```python
|
|
spread = q90 - q10
|
|
conf = float(max(0.0, min(1.0, 1.0 - spread / _SPREAD_THRESHOLD)))
|
|
```
|
|
|
|
- [ ] **Step 4: Run tests to verify they pass**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests/test_chronos_predictor.py -v
|
|
```
|
|
|
|
Expected: 신규 3개 모두 PASS. 기존 test도 PASS.
|
|
|
|
- [ ] **Step 5: Run full ai_trade suite**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests -q
|
|
```
|
|
|
|
Expected: 모두 PASS. signal_generator 테스트(`_compute_buy_confidence` 가 `pred["conf"]` 사용) 도 영향 받을 수 있으니 주시.
|
|
|
|
- [ ] **Step 6: Commit**
|
|
|
|
```bash
|
|
git add ai_trade/chronos_predictor.py ai_trade/tests/test_chronos_predictor.py
|
|
git commit -m "fix(ai_trade): Chronos confidence를 absolute spread 기반으로 통일 (F4)"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 5: 전체 회귀 확인 + push
|
|
|
|
- [ ] **Step 1: Run full ai_trade suite + count**
|
|
|
|
```
|
|
python -m pytest ai_trade/tests -v
|
|
```
|
|
|
|
Expected:
|
|
- 기존 56 tests + 신규 (config 2 + kis_client 1 + scheduler 5 + pull_worker 1 + chronos_predictor 3) = **68 tests** 정도 PASS.
|
|
|
|
- [ ] **Step 2: Quick sanity — server boot smoke test (시간 허용 시)**
|
|
|
|
```
|
|
cd ai_trade && python -c "from main import app; print('app import OK')"
|
|
```
|
|
|
|
Expected: no import errors.
|
|
|
|
- [ ] **Step 3: Push**
|
|
|
|
```bash
|
|
git push origin main
|
|
```
|
|
|
|
---
|
|
|
|
## Self-Review Checklist
|
|
|
|
이 plan을 다 작성한 뒤 다음을 확인:
|
|
|
|
1. **F1**: config.py default + 2 test (default + env override) ✅
|
|
2. **F2**: `_throttle_lock` 추가 + 1 concurrent test ✅
|
|
3. **F3**: `_is_post_close_trigger(now, last_post_close_date)` 시그니처 변경 + `poll_loop` 상태 추적 + 5 scheduler test + 1 pull_worker test ✅
|
|
4. **F4**: `_SPREAD_THRESHOLD=0.6` 상수 + 두 분기(modern + legacy) 모두 absolute spread 적용 + 3 chronos_predictor test ✅
|
|
|
|
**누락 가능 항목**:
|
|
- `test_main.py` 가 `v1_token_path` default를 직접 검증한다면 Task 1에서 같이 갱신. 위 patch는 Settings 객체 통해서만 다루므로 영향 없음(검증 완료).
|
|
- Task 3 pull_worker test의 `FrozenDateTime.now`는 `datetime.now(KST)` 호출만 stub함. 다른 datetime 사용 부분 영향 없음 (verified L28).
|
|
- Task 4 test는 ChronosPredictor `__new__`로 우회 — 실제 HuggingFace 모델 로딩 안 함.
|
|
|
|
---
|
|
|
|
## Execution Handoff
|
|
|
|
**Plan complete and saved to `docs/superpowers/plans/2026-05-25-ai-trade-hotfix.md`.**
|
|
|
|
두 가지 실행 옵션:
|
|
|
|
**1. Subagent-Driven (recommended)** — task 별 fresh subagent dispatch + two-stage review. F2/F3 같이 미묘한 동시성/상태 변경에 유리.
|
|
|
|
**2. Inline Execution** — 현 세션에서 직접 task별 진행 + checkpoint.
|
|
|
|
박재오 결정 대기.
|