""" fast_pipeline.py [FEATURE 2026-09-11] Live wiring for the "fast_prediction" decision engine: stream_features -> fast_prediction_engine -> fast_entry_gate. Same role as prediction_pipeline.py plays for the "prediction_pipeline" decision engine -- a thin orchestrator, all real logic lives in the engines themselves. Selected via config.json's entry.decision_engine == "fast_prediction". See fast_prediction_engine.py's module docstring for why this path exists (2026-09-10's zero-trade session, the structure_score cold-start + extension-cliff interaction it was built to route around) and for confirmation that stream_features.py itself is completely unchanged -- this only changes how its output is consumed downstream. CALLER-HELD STATE: monitor.py keeps one persistence dict per symbol (Monitor._fast_engine_state) across poll cycles and passes it in as `persistence_state`; this function returns the updated value for the caller to store back. A fresh {} for a symbol is a cold start. """ from dataclasses import dataclass, field, asdict from datetime import datetime, timezone from stream_features import compute_features from fast_prediction_engine import compute_fast_prediction from fast_entry_gate import evaluate_fast_entry @dataclass class FastPipelineDecision: symbol: str should_enter: bool confirmation_score: float reasons_for: list = field(default_factory=list) reasons_against: list = field(default_factory=list) # additive, for logging/review -- never gates should_enter direction: str = None confidence: float = None momentum: str = None volume_state: str = None vwap_state: str = None ema9_state: str = None resistance: str = None extension: str = None state: str = None persistence_seconds_elapsed: float = 0.0 def compute_fast_ranking(symbol: str, bars: list, bars_sub: list, quote, avg_vol_baseline: float, state: dict, sub_bucket_seconds: int = 30, as_of=None, premarket_result: dict = None): """ [FEATURE 2026-09-11 v2] Features + fast_prediction only -- NOT the entry gate. Split out of evaluate_fast_pipeline() so a caller can rank a wider pool of candidates by fast_prediction_engine's own Direction/Confidence reading BEFORE deciding which ones are worth spending a full evaluate_fast_entry() persistence-gate call on. Why this exists: monitor.py._scan_for_entries() used to pick its top-5 entry shortlist by intraday_health.health_score, computed before any fast_prediction reading existed for that cycle -- a chicken-and-egg problem for ranking by confidence instead. Per explicit instruction ("the confidence value should be the value used for selecting the top_5"), monitor.py now calls this for EVERY health-eligible candidate first, ranks by prediction.confidence, and only calls evaluate_fast_entry_only() below for the top 5. Returns (prediction, features) -- both fully computed, so a subsequent evaluate_fast_entry_only() call for a chosen candidate never recomputes either. bars / bars_sub / quote / avg_vol_baseline: same shapes monitor.py already hands prediction_pipeline.evaluate_entry_pipeline(). state: caller-held per-symbol dict, see module docstring. Holds {"prior_features": ..., "persistence_state": ...}. Updates state["prior_features"] as a side effect (unconditionally, even for a candidate that doesn't end up in the top 5 -- its features genuinely evolved this cycle regardless of ranking, and the next cycle's acceleration math needs that real prior reading). """ as_of = as_of or datetime.now(timezone.utc) state.setdefault("prior_features", None) state.setdefault("persistence_state", {}) features = compute_features(symbol, bars, bars_sub, quote, avg_vol_baseline=avg_vol_baseline, prior=state["prior_features"], sub_bucket_seconds=sub_bucket_seconds, as_of=as_of) state["prior_features"] = features prediction = compute_fast_prediction(symbol, features, bars, premarket_result=premarket_result) return prediction, features def evaluate_fast_entry_only(symbol: str, prediction, features, state: dict, as_of=None) -> FastPipelineDecision: """ Finishes evaluate_fast_pipeline()'s work given an already-computed (prediction, features) pair from compute_fast_ranking() above -- avoids a redundant compute_features() call for a candidate already scored this cycle. Same return shape/contract as evaluate_fast_pipeline() below. """ as_of = as_of or datetime.now(timezone.utc) decision = evaluate_fast_entry(symbol, prediction, features, state["persistence_state"], as_of=as_of) state["persistence_state"] = decision.persistence return FastPipelineDecision( symbol=symbol, should_enter=decision.should_enter, confirmation_score=prediction.confidence, reasons_for=decision.reasons_for, reasons_against=decision.reasons_against, direction=prediction.direction, confidence=prediction.confidence, momentum=prediction.momentum, volume_state=prediction.volume_state, vwap_state=prediction.vwap_state, ema9_state=prediction.ema9_state, resistance=prediction.resistance, extension=prediction.extension, state=decision.state, persistence_seconds_elapsed=decision.persistence_seconds_elapsed, ) def snapshot_dict(features, prediction) -> dict: """ [FEATURE 2026-09-11 v3] Full raw-indicator snapshot for logging -- not just the classified labels (Strong/Expanding/Bullish/...) that the [FAST] log line and FastPipelineDecision carry, but every underlying numeric/raw value stream_features.py and fast_prediction_engine.py computed this cycle (slope_5m, momentum_acceleration, relative_volume, volume_acceleration, vwap value/slope/classification, price_vs_ema9_pct/ema9_slope, atr_pct, vwap_distance_atr, resistance_level/distance_pct, spread_pct, rsi, etc.). Per explicit instruction ("make sure ... we have all the indicator values logged during the life cycle of the symbol") -- the classified labels alone aren't enough to reconstruct exactly what the engine saw at a given moment; this is the reproducible version. Dumps `features` (a stream_features.StreamFeatures) and `prediction` (a fast_prediction_engine.FastPredictionReading) wholesale via dataclasses.asdict() rather than hand-picking fields, so a new indicator added to either dataclass is captured automatically without this function needing to be updated too. """ f = asdict(features) f["computed_at"] = features.computed_at.isoformat() if features.computed_at else None return {"features": f, "prediction": asdict(prediction)} def evaluate_fast_pipeline(symbol: str, bars: list, bars_sub: list, quote, avg_vol_baseline: float, state: dict, sub_bucket_seconds: int = 30, as_of=None, premarket_result: dict = None) -> FastPipelineDecision: """ Convenience wrapper combining compute_fast_ranking() + evaluate_fast_entry_only() for a caller that doesn't need to rank against other candidates first (e.g. a one-off/test evaluation). monitor.py._scan_for_entries() calls the two split functions directly instead -- see compute_fast_ranking()'s docstring. """ as_of = as_of or datetime.now(timezone.utc) prediction, features = compute_fast_ranking( symbol, bars, bars_sub, quote, avg_vol_baseline, state, sub_bucket_seconds=sub_bucket_seconds, as_of=as_of, premarket_result=premarket_result) return evaluate_fast_entry_only(symbol, prediction, features, state, as_of=as_of)