fix(ai_trade): Chronos confidence를 absolute spread 기반으로 통일 (F4)

코드 리뷰 F4: signal_generator의 hard gate(L79)는 absolute spread(0.6 threshold)를
쓰지만 chronos_predictor:106의 confidence는 relative spread (q90-q10)/max(|median|, 0.001).
zero-shot median≈0 케이스에서 spread가 폭증하여 conf=0으로 눌리고 결국 모든
매수 신호가 confidence_threshold(0.7)를 못 넘김.

산식 통일: conf = max(0, min(1, 1 - spread/_SPREAD_THRESHOLD)). _SPREAD_THRESHOLD=0.6
은 signal_generator hard gate와 동일.

- spread≈0 → conf≈1 (확신)
- spread=0.3 → conf=0.5 (중간)
- spread≥0.6 → conf=0 (거부)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-25 19:39:15 +09:00
parent bea27a75cf
commit c2e77a7310
2 changed files with 59 additions and 4 deletions

View File

@@ -10,6 +10,10 @@ import numpy as np
logger = logging.getLogger(__name__)
KST = ZoneInfo("Asia/Seoul")
# F4: signal_generator hard gate와 동일한 absolute spread threshold.
# zero-shot median≈0에서 conf가 0으로 폭락하던 relative 산식 (spread/abs(median)) 대체.
_SPREAD_THRESHOLD = 0.6
@dataclass
class ChronosPrediction:
@@ -103,8 +107,8 @@ class ChronosPredictor:
median = float((q50_price - last_close) / last_close)
q10 = float((q10_price - last_close) / last_close)
q90 = float((q90_price - last_close) / last_close)
spread = (q90 - q10) / max(abs(median), 0.001)
conf = float(max(0.0, min(1.0, 1.0 - spread / 2.0)))
spread = q90 - q10 # F4: absolute spread
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,
)
@@ -124,8 +128,8 @@ class ChronosPredictor:
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)))
spread = q90 - q10 # F4: absolute spread
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,
)

View File

@@ -90,3 +90,54 @@ def test_return_computed_from_price_relative_to_last_close(mock_pipeline, mock_t
daily = {"005930": _daily_ohlcv(list(range(41, 101)))} # last = 100
result = predictor.predict_batch(daily)
assert abs(result["005930"].median - 0.10) < 0.001
# ----- F4: absolute spread 기반 confidence -----
def test_confidence_high_when_spread_near_zero(mock_pipeline, mock_torch_cpu):
"""F4 — median≈0 + spread≈0 일 때 conf≈1 (현 relative 산식의 회귀 케이스).
한국 주가 100000원, q10=q50=q90=100000 → median=0, spread=0.
Relative 산식 (spread/abs(median))은 0/0.001 보호선이라 spread=0이면 conf=1로
동작하지만, median≈0 + 미세 spread(예 1원) 케이스에서 폭증 → conf=0.
Absolute 산식은 그런 폭증 없음.
"""
quantiles = _mk_quantiles_tensor(100000.0, 100000.0, 100000.0)
mock_pipeline.predict_quantiles.return_value = (quantiles, None)
from ai_trade.chronos_predictor import ChronosPredictor
predictor = ChronosPredictor(model_name="mock-model")
daily = {"005930": _daily_ohlcv([100000] * 60)}
result = predictor.predict_batch(daily)
assert result["005930"].conf > 0.95, (
f"median≈0 + spread≈0인데 conf={result['005930'].conf} (F4 회귀)"
)
def test_confidence_half_at_spread_03(mock_pipeline, mock_torch_cpu):
"""F4 — spread 0.30일 때 conf ≈ 0.5 (1 - 0.3/0.6)."""
# q10=85000 → -0.15, q90=115000 → 0.15, q50=100000 → 0.0
# spread = 0.30, conf = 1 - 0.30/0.60 = 0.50
quantiles = _mk_quantiles_tensor(85000.0, 100000.0, 115000.0)
mock_pipeline.predict_quantiles.return_value = (quantiles, None)
from ai_trade.chronos_predictor import ChronosPredictor
predictor = ChronosPredictor(model_name="mock-model")
daily = {"005930": _daily_ohlcv([100000] * 60)}
result = predictor.predict_batch(daily)
conf = result["005930"].conf
assert 0.45 < conf < 0.55, f"spread=0.30에서 conf={conf} (expected ≈0.5)"
def test_confidence_zero_at_threshold_spread(mock_pipeline, mock_torch_cpu):
"""F4 — spread가 _SPREAD_THRESHOLD(0.6)이면 conf=0."""
quantiles = _mk_quantiles_tensor(70000.0, 100000.0, 130000.0)
mock_pipeline.predict_quantiles.return_value = (quantiles, None)
from ai_trade.chronos_predictor import ChronosPredictor
predictor = ChronosPredictor(model_name="mock-model")
daily = {"005930": _daily_ohlcv([100000] * 60)}
result = predictor.predict_batch(daily)
assert result["005930"].conf < 0.05, (
f"spread=threshold에서 conf={result['005930'].conf} (expected ≈0)"
)