"""Tests for signal_generator.""" from unittest.mock import MagicMock import pytest from ai_trade.signal_generator import generate_signals from ai_trade.state import PollState def _settings(**overrides): """Build a Settings-like object for tests (avoid env).""" defaults = dict( stop_loss_pct=-0.07, take_profit_pct=0.15, chronos_spread_threshold=0.6, asking_bid_ratio_threshold=0.6, confidence_threshold=0.7, min_momentum_for_buy="strong_up", signal_ttl_seconds=300, ) defaults.update(overrides) m = MagicMock() for k, v in defaults.items(): setattr(m, k, v) return m def _make_state_with_buy_candidate( ticker="005930", name="삼성전자", chronos_median=0.02, chronos_q10=-0.01, chronos_q90=0.04, chronos_conf=0.85, momentum="strong_up", bid_ratio=0.7, current_price=78500, ): state = PollState() state.screener_preview = {"items": [{"ticker": ticker, "name": name}]} state.chronos_predictions[ticker] = { "median": chronos_median, "q10": chronos_q10, "q90": chronos_q90, "conf": chronos_conf, "as_of": "2026-05-17T16:00:00+09:00", } state.minute_momentum[ticker] = momentum state.asking_price[ticker] = { "bid_total": int(bid_ratio * 1000), "ask_total": int((1 - bid_ratio) * 1000), "bid_ratio": bid_ratio, "current_price": current_price, "as_of": "2026-05-17T16:00:01+09:00", } return state def _make_state_with_holding( ticker="005930", name="삼성전자", pnl_pct=0.0, avg_price=75000, current_price=75000, ): state = PollState() state.portfolio = {"holdings": [{ "ticker": ticker, "name": name, "avg_price": avg_price, "current_price": current_price, "pnl_pct": pnl_pct, "profit_rate": pnl_pct * 100, "quantity": 100, "broker": "키움", }]} state.screener_preview = {"items": []} return state @pytest.fixture def dedup_mock(): d = MagicMock() d.is_recent.return_value = False return d def test_buy_signal_when_all_conditions_pass_and_confidence_high(dedup_mock): state = _make_state_with_buy_candidate() generate_signals(state, dedup_mock, _settings()) assert "005930" in state.signals sig = state.signals["005930"] assert sig["action"] == "buy" assert sig["confidence_webai"] > 0.7 dedup_mock.record.assert_called() def test_silent_when_chronos_median_negative(dedup_mock): state = _make_state_with_buy_candidate(chronos_median=-0.01) generate_signals(state, dedup_mock, _settings()) assert "005930" not in state.signals def test_silent_when_distribution_spread_too_wide(dedup_mock): # spread = q90 - q10 = 0.5 - (-0.5) = 1.0 > 0.6 → hard gate fails state = _make_state_with_buy_candidate( chronos_median=0.001, chronos_q10=-0.5, chronos_q90=0.5, ) generate_signals(state, dedup_mock, _settings()) assert "005930" not in state.signals def test_silent_when_momentum_not_strong_up(dedup_mock): state = _make_state_with_buy_candidate(momentum="weak_up") generate_signals(state, dedup_mock, _settings()) assert "005930" not in state.signals def test_silent_when_bid_ratio_below_threshold(dedup_mock): state = _make_state_with_buy_candidate(bid_ratio=0.5) generate_signals(state, dedup_mock, _settings()) assert "005930" not in state.signals def test_silent_when_confidence_below_threshold(dedup_mock): # chronos_conf low + rank=20 → confidence < 0.7 state = _make_state_with_buy_candidate(chronos_conf=0.3) # add 19 fake items to push 005930 rank to 20 state.screener_preview["items"] = ( [{"ticker": f"FAKE{i:03d}"} for i in range(19)] + [{"ticker": "005930", "name": "삼성전자"}] ) generate_signals(state, dedup_mock, _settings()) # confidence_webai = 0.3*0.5 + 1.0*0.3 + 0.05*0.2 = 0.46 < 0.7 assert "005930" not in state.signals def test_sell_signal_when_stop_loss_triggered(dedup_mock): state = _make_state_with_holding(pnl_pct=-0.08, current_price=69000, avg_price=75000) generate_signals(state, dedup_mock, _settings()) assert "005930" in state.signals sig = state.signals["005930"] assert sig["action"] == "sell" assert sig["confidence_webai"] == 1.0 assert sig["pnl_pct"] == -0.08 def test_sell_signal_when_take_profit_triggered(dedup_mock): state = _make_state_with_holding(pnl_pct=0.16, current_price=87000, avg_price=75000) generate_signals(state, dedup_mock, _settings()) assert "005930" in state.signals sig = state.signals["005930"] assert sig["action"] == "sell" assert sig["confidence_webai"] == 0.6 def test_silent_when_dedup_recently_sent(dedup_mock): state = _make_state_with_buy_candidate() dedup_mock.is_recent.return_value = True generate_signals(state, dedup_mock, _settings()) assert "005930" not in state.signals dedup_mock.record.assert_not_called() def test_sell_signal_triggers_on_anomaly_path(dedup_mock): """Anomaly sell: median < -1%, momentum strong_down, low bid_ratio, confidence > threshold.""" state = PollState() state.portfolio = {"holdings": [{ "ticker": "005930", "name": "삼성전자", "avg_price": 75000, "current_price": 70000, "pnl_pct": -0.067, # within stop_loss tolerance (default -0.07): NOT triggering stop_loss "quantity": 100, "broker": "키움", }]} state.screener_preview = {"items": []} state.chronos_predictions["005930"] = { "median": -0.025, "q10": -0.05, "q90": 0.005, "conf": 0.85, } state.minute_momentum["005930"] = "strong_down" state.asking_price["005930"] = {"current_price": 70000, "bid_ratio": 0.30} # bid_ratio 0.30 < (1 - 0.6) = 0.4 → anomaly bid_ratio gate passes # confidence = 0.85*0.5 + 1.0*0.3 + 1.0*0.2 = 0.425 + 0.3 + 0.2 = 0.925 > 0.7 generate_signals(state, dedup_mock, _settings()) assert "005930" in state.signals sig = state.signals["005930"] assert sig["action"] == "sell" assert sig["context"]["sell_reason"] == "anomaly" assert sig["confidence_webai"] > 0.7 # ----- F5: cycle_id + expires_at 부착 ----- def test_emit_attaches_cycle_id_and_expires_at(dedup_mock): """F5 — emit signal에 cycle_id (state.signal_cycle_id) + expires_at 부착.""" from datetime import datetime, timedelta from zoneinfo import ZoneInfo _kst = ZoneInfo("Asia/Seoul") state = _make_state_with_buy_candidate() before = datetime.now(_kst) generate_signals(state, dedup_mock, _settings(signal_ttl_seconds=300)) after = datetime.now(_kst) sig = state.signals["005930"] assert sig["cycle_id"] == 1 assert "expires_at" in sig exp_dt = datetime.fromisoformat(sig["expires_at"]) assert before + timedelta(seconds=295) < exp_dt < after + timedelta(seconds=305) def test_cycle_id_increments_each_call(dedup_mock): """F5 — generate_signals 호출마다 cycle_id += 1 (emit 여부 무관).""" state = _make_state_with_buy_candidate() generate_signals(state, dedup_mock, _settings()) assert state.signal_cycle_id == 1 # 2번째 호출 — dedup이 막아도 cycle_id는 증가 dedup_mock.is_recent.return_value = True generate_signals(state, dedup_mock, _settings()) assert state.signal_cycle_id == 2 def test_sell_signal_also_carries_cycle_id_and_expires_at(dedup_mock): """F5 — sell signal도 동일하게 부착.""" from datetime import datetime state = _make_state_with_holding(pnl_pct=-0.08, current_price=68000) generate_signals(state, dedup_mock, _settings(signal_ttl_seconds=120)) assert "005930" in state.signals sig = state.signals["005930"] assert sig["action"] == "sell" assert sig["cycle_id"] == 1 # parse expires_at as ISO — must succeed datetime.fromisoformat(sig["expires_at"])