feat(signal_v2-phase3b): chronos_predictor + 4 mock tests

ChronosPredictor wraps HuggingFace ChronosPipeline. Batch predict
returns ChronosPrediction(median, q10, q90, conf, as_of) per ticker.
Confidence = 1 - clamp(spread/2, 0, 1) where spread = (q90-q10) / |median|.
Lazy import of chronos lib (heavy). GPU auto-detect with FP16.

44 tests pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-16 18:00:46 +09:00
parent c5a88fab66
commit 28f9c8c3a6
2 changed files with 161 additions and 0 deletions

View File

@@ -0,0 +1,83 @@
"""Tests for ChronosPredictor (model mock)."""
from unittest.mock import MagicMock, patch
import numpy as np
import pytest
@pytest.fixture
def mock_pipeline():
"""Mock ChronosPipeline.from_pretrained returning a mock pipeline object."""
with patch("chronos.ChronosPipeline") as cls:
instance = MagicMock()
cls.from_pretrained.return_value = instance
yield instance
@pytest.fixture
def mock_torch_cpu():
with patch("torch.cuda.is_available", return_value=False):
yield
def _daily_ohlcv(close_seq):
return [{"datetime": f"2026-05-{i+1:02d}", "open": c, "high": c, "low": c,
"close": c, "volume": 1000} for i, c in enumerate(close_seq)]
def test_predict_batch_returns_prediction_dict(mock_pipeline, mock_torch_cpu):
"""mock pipeline → dict[ticker, ChronosPrediction]. last_close=100, samples=102 → ~+2% return."""
import torch
samples = np.full((100,), 102.0)
mock_pipeline.predict.return_value = torch.tensor(samples.reshape(1, 100, 1))
from signal_v2.chronos_predictor import ChronosPredictor, ChronosPrediction
predictor = ChronosPredictor(model_name="mock-model")
daily = {"005930": _daily_ohlcv([100] * 60)}
result = predictor.predict_batch(daily)
assert "005930" in result
pred = result["005930"]
assert isinstance(pred, ChronosPrediction)
assert abs(pred.median - 0.02) < 0.001
def test_conf_high_when_distribution_narrow(mock_pipeline, mock_torch_cpu):
"""좁은 distribution → conf ≈ 1."""
import torch
np.random.seed(42)
samples = np.random.normal(102.0, 0.1, 100)
mock_pipeline.predict.return_value = torch.tensor(samples.reshape(1, 100, 1))
from signal_v2.chronos_predictor import ChronosPredictor
predictor = ChronosPredictor(model_name="mock-model")
daily = {"005930": _daily_ohlcv([100] * 60)}
result = predictor.predict_batch(daily)
assert result["005930"].conf > 0.8
def test_conf_low_when_distribution_wide(mock_pipeline, mock_torch_cpu):
"""넓은 distribution → conf ≈ 0."""
import torch
np.random.seed(42)
samples = np.random.normal(100.0, 30.0, 100)
mock_pipeline.predict.return_value = torch.tensor(samples.reshape(1, 100, 1))
from signal_v2.chronos_predictor import ChronosPredictor
predictor = ChronosPredictor(model_name="mock-model")
daily = {"005930": _daily_ohlcv([100] * 60)}
result = predictor.predict_batch(daily)
assert result["005930"].conf < 0.3
def test_return_computed_from_price_relative_to_last_close(mock_pipeline, mock_torch_cpu):
"""price 예측 → last_close 대비 return 변환. last_close=100, samples=110 → return ~+10%."""
import torch
samples = np.full((100,), 110.0)
mock_pipeline.predict.return_value = torch.tensor(samples.reshape(1, 100, 1))
from signal_v2.chronos_predictor import ChronosPredictor
predictor = ChronosPredictor(model_name="mock-model")
# last close in the seq = 100
daily = {"005930": _daily_ohlcv(list(range(41, 101)))} # last = 100
result = predictor.predict_batch(daily)
assert abs(result["005930"].median - 0.10) < 0.001