""" winner_score.py [Phase 4 -- new] Composite 0-100 ranking score, ADDITIVE ONLY -- see the 2026-09-08 design discussion: entry_score.py's five separate scores (trend/structure/prediction/entry/risk) remain the ONLY thing that gates should_enter, unchanged. WINNER_SCORE never overrides that gate; it exists purely for cross-symbol ranking, the market-state output, and score-velocity tracking (RISING_FAST etc.) -- exactly the two independent fields the project's own requested output schema already has (`winner_score` and `entry_status` side by side, not one derived from the other). Composition follows the project's own suggested starting weights (all configurable, none hard-coded): trend_structure 20, relative_strength 15, volume_expansion 15, vwap_structure 15, premarket_setup 10, breakout_quality 10, volatility_expansion 5, sector_strength 5, market_regime 5. SCORE MOMENTUM (velocity/acceleration/trend) needs a caller-held history -- same ownership pattern as every other stateful piece in this pipeline (entry_score.py's confirmation_state, stream_features.py's `prior`): the caller holds a bounded list of (timestamp, score, velocity) tuples, oldest first, passes it in as `score_history`, and is responsible for appending this call's own (now, score, velocity) to that list afterward -- this function never mutates or stores it. """ from dataclasses import dataclass, field from config_loader import get_config DEFAULT_CONFIG = { "weights": { "trend_structure": 20, "relative_strength": 15, "volume_expansion": 15, "vwap_structure": 15, "premarket_setup": 10, "breakout_quality": 10, "volatility_expansion": 5, "sector_strength": 5, "market_regime": 5, }, "trend_structure_blend": {"trend": 0.6, "structure": 0.4}, "volume_expansion_blend": {"acceleration": 0.6, "relative_volume": 0.4}, "relative_volume_saturation": 2.0, "volatility_expansion_saturation_pct": 50.0, "no_sector_default_score": 50.0, "vwap_classification_scores": { "STRONG_VWAP_SUPPORT": 100, "VWAP_RECLAIM": 75, "VWAP_HOLD": 60, "VWAP_EXTENSION": 50, "VWAP_FLAT": 40, "VWAP_DECLINING": 25, "VWAP_FAILURE": 0, }, "market_regime_credit": { "STRONG_RISK_ON": 1.0, "RISK_ON": 0.85, "NEUTRAL": 0.65, "RISK_OFF": 0.5, "STRONG_RISK_OFF": 0.4, }, "no_market_regime_default_credit": 0.65, "breakout_setup_names": ["BREAKOUT_RETEST", "COMPRESSION_EXPANSION"], "breakout_quality_no_setup_credit": 0.5, # partial credit from breakout_pressure_score alone, pre-breakout "velocity_lookback_readings": 5, "score_trend_threshold_per_min": 1.0, "score_trend_fast_threshold_per_min": 4.0, } @dataclass class WinnerScoreReading: symbol: str winner_score: float = 0.0 velocity_per_min: float = 0.0 acceleration_per_min: float = 0.0 score_trend: str = "STABLE" breakdown: dict = field(default_factory=dict) insufficient_data: bool = False missing: list = field(default_factory=list) def _clamp01(x: float) -> float: return max(0.0, min(1.0, x)) def _time_weighted_slope(points: list) -> float: """Least-squares slope of `points` ([(elapsed_seconds, value), ...]) against real elapsed time -- the general form of indicators. linreg_slope() (which assumes evenly-spaced integer x=0..n-1), needed here because winner_score readings aren't guaranteed evenly spaced in wall-clock time the way 1-min bars are.""" n = len(points) if n < 2: return 0.0 mean_x = sum(p[0] for p in points) / n mean_y = sum(p[1] for p in points) / n num = sum((x - mean_x) * (y - mean_y) for x, y in points) den = sum((x - mean_x) ** 2 for x, _ in points) return num / den if den else 0.0 def _score_trend(velocity_per_min: float, cfg: dict) -> str: fast, normal = cfg["score_trend_fast_threshold_per_min"], cfg["score_trend_threshold_per_min"] if velocity_per_min >= fast: return "RISING_FAST" if velocity_per_min >= normal: return "RISING" if velocity_per_min <= -fast: return "DECLINING_FAST" if velocity_per_min <= -normal: return "DECLINING" return "STABLE" def _breakout_quality(setups: dict, compression, cfg: dict) -> float: best = None if setups: valid = [r for r in setups.values() if r.valid] best = max(valid, key=lambda r: r.confidence, default=None) pressure = compression.breakout_pressure_score if ( compression is not None and not getattr(compression, "insufficient_data", False)) else 0.0 if best is not None and best.setup in cfg["breakout_setup_names"]: return _clamp01((best.confidence * 0.7 + pressure * 0.3) / 100.0) * 100.0 return _clamp01(pressure * cfg["breakout_quality_no_setup_credit"] / 100.0) * 100.0 def compute_winner_score(symbol: str, features, trend, structure, prediction, relative_strength=None, compression=None, setups=None, market_regime=None, premarket_result: dict = None, now=None, score_history: list = None, cfg: dict = None) -> WinnerScoreReading: """ features/trend/structure/prediction: required, same readings the rest of this pipeline already computes. relative_strength: optional relative_strength_engine.RelativeStrengthReading. Omit to fall back to a neutral component (never a hard block, per the project's "sector confirmation is additional, not required" principle extended to RS in general when the market data isn't available yet). compression: optional compression_engine.CompressionReading. setups: optional {name: setup_engine.SetupResult} dict. market_regime: optional relative_strength_engine.MarketRegimeReading. premarket_result: optional scorer.score_premarket_candidate() dict for THIS symbol -- reuses its total_score directly as the premarket_setup component (the "premarket produces the thesis" fix). Omit if unavailable; falls back to a neutral 50. score_history: optional caller-held list of (timestamp, score, velocity_per_min) tuples, oldest first, NOT including this call's own reading -- see module docstring. Omit (or pass []) for a cold start (velocity/acceleration read as 0.0, trend as STABLE). """ from datetime import datetime, timezone cfg = {**DEFAULT_CONFIG, **(cfg or get_config().get("winner_score", {}))} now = now or datetime.now(timezone.utc) if (getattr(features, "insufficient_data", False) or getattr(trend, "insufficient_data", False) or getattr(structure, "insufficient_data", False) or getattr(prediction, "insufficient_data", False)): return WinnerScoreReading(symbol=symbol, insufficient_data=True, missing=["one or more required upstream readings was insufficient"]) w = cfg["weights"] tb = cfg["trend_structure_blend"] trend_structure_score = trend.score * tb["trend"] + structure.score * tb["structure"] rs_score = (relative_strength.final_score if (relative_strength is not None and not getattr(relative_strength, "insufficient_data", False)) else cfg["no_sector_default_score"]) vb = cfg["volume_expansion_blend"] vol_accel = features.volume.get("volume_acceleration") rvol = features.volume.get("relative_volume") accel_credit = _clamp01(((vol_accel or 1.0) - 0.8) / 1.2) * 100.0 rvol_credit = _clamp01((rvol or 0.0) / cfg["relative_volume_saturation"]) * 100.0 volume_expansion_score = accel_credit * vb["acceleration"] + rvol_credit * vb["relative_volume"] vwap_class = features.vwap.get("classification") vwap_structure_score = cfg["vwap_classification_scores"].get(vwap_class, 50) premarket_setup_score = (premarket_result.get("total_score", 50.0) if premarket_result else cfg["no_sector_default_score"]) breakout_quality_score = _breakout_quality(setups or {}, compression, cfg) range_expansion = features.volatility.get("range_expansion_pct") volatility_expansion_score = (_clamp01((range_expansion or 0.0) / cfg["volatility_expansion_saturation_pct"]) * 100.0 if range_expansion is not None else cfg["no_sector_default_score"]) sector_score = cfg["no_sector_default_score"] if relative_strength is not None and not getattr(relative_strength, "insufficient_data", False): if relative_strength.sector_breadth is not None: momentum_credit = _clamp01(((relative_strength.sector_momentum or 0.0) + 2.0) / 4.0) * 100.0 sector_score = relative_strength.sector_breadth * 100.0 * 0.5 + momentum_credit * 0.5 market_regime_score = (cfg["market_regime_credit"].get(market_regime.regime, cfg["no_market_regime_default_credit"]) if (market_regime is not None and not getattr(market_regime, "insufficient_data", False)) else cfg["no_market_regime_default_credit"]) * 100.0 components = { "trend_structure": trend_structure_score, "relative_strength": rs_score, "volume_expansion": volume_expansion_score, "vwap_structure": vwap_structure_score, "premarket_setup": premarket_setup_score, "breakout_quality": breakout_quality_score, "volatility_expansion": volatility_expansion_score, "sector_strength": sector_score, "market_regime": market_regime_score, } breakdown = {k: round(components[k] * w[k] / 100.0, 2) for k in components} weight_total = sum(w.values()) winner_score = round(_clamp01(sum(breakdown.values()) / weight_total) * 100.0, 2) # ---- score momentum, from caller-held history ---- history = score_history or [] points = [((t - history[0][0]).total_seconds(), s) for t, s, _v in history] if history else [] points.append(((now - history[0][0]).total_seconds() if history else 0.0, winner_score)) slope_per_sec = _time_weighted_slope(points) velocity_per_min = round(slope_per_sec * 60.0, 3) prior_velocity = history[-1][2] if history else 0.0 acceleration_per_min = round(velocity_per_min - prior_velocity, 3) return WinnerScoreReading( symbol=symbol, winner_score=winner_score, velocity_per_min=velocity_per_min, acceleration_per_min=acceleration_per_min, score_trend=_score_trend(velocity_per_min, cfg), breakdown={k: round(v, 2) for k, v in components.items()}, insufficient_data=False, )