diff --git a/signal_v2/signal_generator.py b/signal_v2/signal_generator.py index df1c59d..fe24070 100644 --- a/signal_v2/signal_generator.py +++ b/signal_v2/signal_generator.py @@ -20,9 +20,9 @@ MOMENTUM_SCORES = { def generate_signals(state, dedup, settings) -> None: - """Phase 4 entry — state mutating. 매수/매도 룰 적용.""" - _evaluate_buy_signals(state, dedup, settings) + """Phase 4 entry — state-mutating. Evaluation order: sell first (priority), then buy. A ticker receiving a sell signal in this cycle is excluded from buy evaluation to avoid silent overwrite.""" _evaluate_sell_signals(state, dedup, settings) + _evaluate_buy_signals(state, dedup, settings) # ----- 매수 ----- @@ -30,6 +30,9 @@ def generate_signals(state, dedup, settings) -> None: def _evaluate_buy_signals(state, dedup, settings) -> None: candidates = _buy_candidates(state) for ticker, name, rank in candidates: + existing = state.signals.get(ticker) + if existing is not None and existing.get("action") == "sell": + continue if not _check_buy_hard_gate(state, ticker, settings): continue confidence = _compute_buy_confidence(state, ticker, rank) @@ -65,16 +68,16 @@ def _buy_candidates(state) -> list[tuple[str, str, int | None]]: def _check_buy_hard_gate(state, ticker: str, settings) -> bool: pred = state.chronos_predictions.get(ticker) - if pred is None or pred["median"] <= 0: + if pred is None or pred.get("median", 0) <= 0: return False - spread = pred["q90"] - pred["q10"] + spread = pred.get("q90", 0) - pred.get("q10", 0) if spread >= settings.chronos_spread_threshold: return False momentum = state.minute_momentum.get(ticker) if momentum != settings.min_momentum_for_buy: return False ap = state.asking_price.get(ticker) - if ap is None or ap["bid_ratio"] < settings.asking_bid_ratio_threshold: + if ap is None or ap.get("bid_ratio", 0) < settings.asking_bid_ratio_threshold: return False return True @@ -83,7 +86,7 @@ def _compute_buy_confidence(state, ticker: str, rank: int | None) -> float: pred = state.chronos_predictions[ticker] chronos_conf = pred["conf"] minute_score = MOMENTUM_SCORES.get(state.minute_momentum.get(ticker, "neutral"), 0.5) - screener_norm = 1 - (rank - 1) / 20 if rank is not None else 0.0 + screener_norm = max(0.0, 1 - (rank - 1) / 20) if rank is not None else 0.0 return chronos_conf * 0.5 + minute_score * 0.3 + screener_norm * 0.2 diff --git a/signal_v2/tests/test_signal_generator.py b/signal_v2/tests/test_signal_generator.py index 7b70d03..00e9ad2 100644 --- a/signal_v2/tests/test_signal_generator.py +++ b/signal_v2/tests/test_signal_generator.py @@ -85,7 +85,7 @@ def test_silent_when_chronos_median_negative(dedup_mock): def test_silent_when_distribution_spread_too_wide(dedup_mock): - # spread = (0.5 - (-0.5)) / max(0.001, 0.001) = 1000 → > 0.6 + # 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, ) @@ -143,3 +143,30 @@ def test_silent_when_dedup_recently_sent(dedup_mock): 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