fix(signal_v2-phase4-task2): code review fixes — sell-first ordering + anomaly test + defensive .get

- generate_signals now evaluates sell before buy; buy candidates with a same-cycle
  sell signal are skipped (resolves silent overwrite of state.signals[ticker]).
- Added test_sell_signal_triggers_on_anomaly_path covering _try_anomaly path
  (previously 0% covered).
- Fixed stale test comment referencing deprecated relative spread formula.
- _check_buy_hard_gate uses dict.get(..., 0) for defense against partial upstream state.
- _compute_buy_confidence clamps screener_norm to >= 0 for future Top-N changes.
This commit is contained in:
2026-05-17 13:18:22 +09:00
parent b9def06993
commit e574074ca8
2 changed files with 37 additions and 7 deletions

View File

@@ -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