""" entry_score.py [SIMULATION-ONLY 2026-09-05] Combines trend_engine/structure_engine/ prediction_engine/setup_engine/regime_engine into FIVE SEPARATE scores (trend, structure, prediction, entry, risk) and one action decision (WAIT / CONFIRMING / READY) -- deliberately NOT one blended "health" number, per the project's original design brief's explicit section 24 instruction and its own worked example: a symbol can be Trend=94, Prediction=88, Entry=51, Risk=76 and still resolve to WAIT, because Entry (setup/trigger readiness) is what's actually low, and no amount of trend or prediction strength should paper over that. CONFIRMATION WINDOW (section 25): "never enter from one tick." Needs a caller-held confirmation_state dict, same pattern as every other caller-held-history piece built this session (stream_features.py's `prior`, prediction_engine.py's `prior`) -- this module stays stateless, the orchestrator holds {symbol: confirmation_state} across calls. Any single reading that fails the thresholds immediately resets the window (an ABORT, per the design brief's own worked example of a declining reading sequence) rather than averaging through it. Not consumed anywhere in the live bot yet -- see config.json's entry_score._note. opportunity_score is deliberately NOT computed here -- section 26 makes that explicitly opportunity_ranker.py's job (it ranks ACROSS symbols; this module only ever looks at one). """ from dataclasses import dataclass, field from datetime import datetime, timezone from config_loader import get_config ACTION_WAIT = "WAIT" ACTION_CONFIRMING = "CONFIRMING" ACTION_READY = "READY" DEFAULT_CONFIG = { "min_trend_score": 60, "min_structure_score": 50, "min_prediction_score": 60, "min_entry_score": 65, "min_risk_score": 60, "max_spread_pct": 1.0, "confirmation_seconds": 8, "confirmation_min_readings": 2, "min_reward_risk_ratio": 1.5, "entry_weights": {"setup_confidence": 0.5, "regime_allows": 0.3, "prediction_momentum": 0.2}, "risk_weights": {"stop_quality": 0.4, "liquidity": 0.3, "spread": 0.3}, "min_relative_volume": 0.5, } @dataclass class EntryScoreReading: symbol: str computed_at: datetime trend_score: float = 0.0 structure_score: float = 0.0 prediction_score: float = 0.0 entry_score: float = 0.0 risk_score: float = 0.0 action: str = ACTION_WAIT reasons: list = field(default_factory=list) confirmation: dict = field(default_factory=dict) thresholds_cleared: bool = False insufficient_data: bool = False missing: list = field(default_factory=list) def _clamp01(x: float) -> float: return max(0.0, min(1.0, x)) def _compute_entry_component(structure, regime, setups, prediction, cfg: dict) -> tuple: reasons = [] best = None if setups: valid = [r for r in setups.values() if r.valid] best = max(valid, key=lambda r: r.confidence, default=None) setup_confidence = (best.confidence / 100.0) if best is not None else 0.0 if best is None: reasons.append("no setup currently validates") allowed = getattr(regime, "allowed_setups", None) if regime is not None else None if best is not None and allowed is not None: regime_allows = 1.0 if best.setup in allowed else 0.0 if regime_allows == 0.0: reasons.append(f"regime={getattr(regime, 'regime', '?')} does not permit " f"{best.setup} (allowed: {allowed})") elif allowed is not None and not allowed: regime_allows = 0.0 reasons.append(f"regime={getattr(regime, 'regime', '?')} permits no setups right now") else: regime_allows = 0.5 # unknown regime context -- neutral, not a hard block pred_slope = getattr(prediction, "slope", 0.0) if prediction is not None else 0.0 if pred_slope > 0: prediction_momentum = 1.0 elif pred_slope < 0: prediction_momentum = 0.0 reasons.append(f"prediction score deteriorating (slope={pred_slope})") else: prediction_momentum = 0.5 w = cfg["entry_weights"] entry_score = (setup_confidence * w["setup_confidence"] + regime_allows * w["regime_allows"] + prediction_momentum * w["prediction_momentum"]) * 100.0 if best is not None and best.valid and (allowed is None or best.setup in allowed): reasons.append(f"setup {best.setup} valid at {best.confidence}% confidence") return round(entry_score, 2), reasons, best def _compute_risk_component(features, structure, prediction, cfg: dict) -> tuple: """ [FEATURE 2026-09-05] Structure-aware risk quality -- per section 27's explicit "risk_per_share = entry_price - structure_stop" concept. Uses structure.last_swing_low as the natural stop reference (the same level structure_engine already identifies as the last confirmed support) rather than an arbitrary distance. This module only SCORES that quality -- it does not compute actual position size or replace risk_manager.py, per the project's explicit "do not replace the existing risk engine" instruction; risk_manager.py's real compute_position_size()/compute_initial_stop() remain the only functions that ever size a real trade. """ reasons = [] price = features.price.get("last") atr = features.volatility.get("atr") structural_stop = structure.last_swing_low if structural_stop is not None and price is not None and price > structural_stop: risk_per_share_pct = (price - structural_stop) / price * 100.0 elif atr and price: risk_per_share_pct = atr / price * 100.0 # fallback: ATR-based estimate, no swing low available reasons.append("no structural swing-low stop available -- using ATR-based estimate instead") else: risk_per_share_pct = None expected_reward_pct = None if prediction is not None and prediction.expected_move_pct: expected_reward_pct = abs(prediction.expected_move_pct.get("5m", 0.0)) if risk_per_share_pct and risk_per_share_pct > 0 and expected_reward_pct is not None: reward_risk_ratio = expected_reward_pct / risk_per_share_pct stop_quality = _clamp01(reward_risk_ratio / cfg["min_reward_risk_ratio"]) if reward_risk_ratio < cfg["min_reward_risk_ratio"]: reasons.append(f"reward/risk ratio {reward_risk_ratio:.2f} below minimum " f"{cfg['min_reward_risk_ratio']} (stop would be too wide for the expected move)") else: stop_quality = 0.5 # unknown -- neutral, not a hard fail reasons.append("could not compute a reward/risk ratio (missing stop reference or expected move)") rvol = features.volume.get("relative_volume") liquidity = _clamp01(rvol / cfg["min_relative_volume"]) if rvol is not None else 0.5 spread_pct = features.quote.get("spread_pct") if spread_pct is not None: spread_quality = _clamp01(1 - spread_pct / cfg["max_spread_pct"]) if spread_pct > cfg["max_spread_pct"]: reasons.append(f"spread {spread_pct:.2f}% exceeds max {cfg['max_spread_pct']}%") else: spread_quality = 0.5 reasons.append("no quote available -- spread quality unknown") w = cfg["risk_weights"] risk_score = (stop_quality * w["stop_quality"] + liquidity * w["liquidity"] + spread_quality * w["spread"]) * 100.0 return round(risk_score, 2), reasons def update_confirmation_state(confirmation_state: dict, thresholds_cleared: bool, now: datetime) -> dict: """ Caller-held confirmation tracking. Pass the PREVIOUS call's returned dict back in as `confirmation_state` (an empty dict on the first call for a symbol). ANY reading with thresholds_cleared=False resets the window immediately -- no partial credit, matching the design brief's own "81, 78, 69, 57 -> ABORT" example (a declining sequence aborts, it doesn't average out). """ if not thresholds_cleared: return {"since": None, "readings": 0} since = confirmation_state.get("since") if since is None: since = now return {"since": since, "readings": confirmation_state.get("readings", 0) + 1} def compute_entry_score(symbol: str, features, trend, structure, prediction, regime=None, setups: dict = None, confirmation_state: dict = None, as_of: datetime = None, cfg: dict = None) -> EntryScoreReading: """ as_of: [FEATURE 2026-09-05] the moment this reading is "as of" -- drives the confirmation window's elapsed-time tracking. Defaults to real wall-clock time (correct for live use). A REPLAY/BACKTEST caller MUST pass the bar's own timestamp here -- otherwise the confirmation window would measure the REAL time the backtest script happens to take to execute, not simulated market time, making CONFIRMING -> READY transitions meaningless (either impossible if the script runs faster than confirmation_seconds, or requiring an actual multi-second sleep per symbol per bar if it doesn't). """ cfg = {**DEFAULT_CONFIG, **(cfg or get_config().get("entry_score", {}))} confirmation_state = confirmation_state or {} now = as_of or datetime.now(timezone.utc) missing = [] for name, obj in (("features", features), ("trend", trend), ("structure", structure), ("prediction", prediction)): if getattr(obj, "insufficient_data", False): missing.append(name) if missing: new_conf = update_confirmation_state(confirmation_state, False, now) return EntryScoreReading(symbol=symbol, computed_at=now, insufficient_data=True, missing=missing, confirmation=new_conf, reasons=[f"insufficient upstream data: {missing}"]) trend_score = trend.score structure_score = structure.score prediction_score = prediction.score entry_score, entry_reasons, best_setup_result = _compute_entry_component( structure, regime, setups, prediction, cfg) risk_score, risk_reasons, = _compute_risk_component(features, structure, prediction, cfg) reasons = list(entry_reasons) + list(risk_reasons) spread_pct = features.quote.get("spread_pct") spread_ok = spread_pct is None or spread_pct <= cfg["max_spread_pct"] thresholds_cleared = ( trend_score >= cfg["min_trend_score"] and structure_score >= cfg["min_structure_score"] and prediction_score >= cfg["min_prediction_score"] and entry_score >= cfg["min_entry_score"] and risk_score >= cfg["min_risk_score"] and spread_ok ) if not thresholds_cleared: if trend_score < cfg["min_trend_score"]: reasons.append(f"trend_score {trend_score} below minimum {cfg['min_trend_score']}") if structure_score < cfg["min_structure_score"]: reasons.append(f"structure_score {structure_score} below minimum {cfg['min_structure_score']}") if prediction_score < cfg["min_prediction_score"]: reasons.append(f"prediction_score {prediction_score} below minimum {cfg['min_prediction_score']}") if entry_score < cfg["min_entry_score"]: reasons.append(f"entry_score {entry_score} below minimum {cfg['min_entry_score']}") if risk_score < cfg["min_risk_score"]: reasons.append(f"risk_score {risk_score} below minimum {cfg['min_risk_score']}") if not spread_ok: reasons.append(f"spread {spread_pct}% exceeds max {cfg['max_spread_pct']}%") new_conf = update_confirmation_state(confirmation_state, thresholds_cleared, now) if not thresholds_cleared: action = ACTION_WAIT else: elapsed = (now - new_conf["since"]).total_seconds() if new_conf["since"] else 0.0 if elapsed >= cfg["confirmation_seconds"] and new_conf["readings"] >= cfg["confirmation_min_readings"]: action = ACTION_READY reasons.append(f"all thresholds held for {elapsed:.1f}s across {new_conf['readings']} " f"readings -- confirmed") else: action = ACTION_CONFIRMING reasons.append(f"thresholds cleared, confirming ({elapsed:.1f}s / {cfg['confirmation_seconds']}s, " f"{new_conf['readings']} reading(s))") return EntryScoreReading( symbol=symbol, computed_at=now, trend_score=trend_score, structure_score=structure_score, prediction_score=prediction_score, entry_score=entry_score, risk_score=risk_score, action=action, reasons=reasons, confirmation=new_conf, thresholds_cleared=thresholds_cleared, )