""" fast_entry_gate.py [FEATURE 2026-09-11] Entry validation + short live-confirmation layer for the "fast_prediction" decision engine -- see fast_prediction_engine.py's module docstring for the full design history this replaces (structure_ engine.py/prediction_engine.py/entry_score.py's floor-gated architecture). fast_prediction_engine.py's Direction/Confidence is descriptive only and never gates entry by itself, per explicit instruction ("Confidence describes the prediction. It does not control the trade by itself."). This module is the ONLY thing that decides BUY / WAIT / REJECT. STATE MACHINE (caller-held persistence_state dict per symbol, same contract as entry_score.py's confirmation_state -- monitor.py owns one per symbol via Monitor._fast_engine_state): - A HARD disqualifier (price loses VWAP, VWAP slope turns negative, price acceleration turns sharply negative, resistance is TOO_CLOSE with no breakout in progress, spread unacceptable, or severely extended while momentum deteriorates) -> REJECT, persistence_state fully cleared. Per explicit instruction this does NOT ban the symbol -- the very next poll evaluates it completely fresh ("reset and rewatch in place"). No separate cooldown/benching invented here. - No hard disqualifier AND direction == UP -> the setup is "alive": the persistence timer starts (or keeps running) and only flips to BUY once it has held continuously for persistence_seconds (15-20s range, per explicit instruction -- NOT the 60-90s originally floated, and NOT the old architecture's multi-minute structure-score ramp-up). A SOFT condition being momentarily off (e.g. volume_ acceleration cooling, EMA9/EMA20 context misaligned) does not reset the timer -- only a HARD disqualifier does, matching the explicit "one minor condition disappears -> WAIT, don't kill the whole attempt" instruction. - direction != UP with no hard disqualifier -> WAIT, timer cleared (nothing bullish is actually being timed yet). """ from dataclasses import dataclass, field from datetime import datetime, timezone import market_time from config_loader import get_config from fast_prediction_engine import DIRECTION_UP, RESISTANCE_TOO_CLOSE, EXTENSION_SEVERE STATE_BUY = "BUY" STATE_WAIT = "WAIT" STATE_REJECT = "REJECT" DEFAULT_CONFIG = { # [persistence] "aim for seconds, not minutes" / "signal persistence # should be about 15-20 seconds" -- explicit instruction, replacing # the originally-floated 60-90s window. "persistence_seconds": 18, "max_spread_pct": 1.0, # price acceleration (stream_features.py's acceleration. # momentum_acceleration, %/min-basis cross-horizon waterfall) below # this is "sharply negative" -- a hard disqualifier, not just a # confidence penalty. Tune live against real sessions. "hard_price_acceleration_negative": -0.5, # [FEATURE 2026-09-14] relative_volume (stream_features.py) is # cumulative volume since this symbol started streaming, divided by # the PRIOR full day's total -- naturally low for everyone early in # the session regardless of how liquid a stock actually is, so a flat # floor on it would be confounded by time-of-day. Normalizing by how # much of TODAY's session has elapsed (market_time.minutes_since_open() # / market_time.session_length_minutes()) gives a "pace vs. this # symbol's own normal day" ratio that's comparable at any time of day. # Added after TNON (2026-09-14): entered at 19.9 min into the session # with relative_volume=0.013 -- a pace ratio of ~0.25x, the lowest of # every entry that day by a wide margin (next-lowest was ~0.70x; most # were 0.9x-2.6x, one breakout at 19.7x). The subsequent exit, a plain # market order, slipped 5.6% (vs. a few cents on every other exit that # day) because the book was too thin to absorb it -- this stock should # never have been entered. 0.4 is a first-cut floor sized to clear # every other 2026-09-14 entry with margin while catching TNON; # unvalidated against any other day, tune once there's more data. "min_volume_pace_ratio": 0.4, # [BUGFIX 2026-09-14] persistence_state's "since" is just a saved # timestamp -- evaluate_fast_entry() only ever computes `now - since`, # with no check that the symbol was actually evaluated on every poll # in between. Confirmed live: ALLT held a UP-since-09:49:25 timestamp # through a 180s bench, then bought instantly on the first re-check # with zero continuous re-confirmation; JBS did the same across a # ~3min health-STALE stretch; FUBO's worst case carried a since # timestamp across a 2h26m (8804s) health-STALE gap and bought the # instant health recovered. max_evaluation_gap_seconds bounds how # long a since timestamp survives NOT being actively re-evaluated # (benched, health-ineligible, or symbol had an open/closing position) # before it's discarded as stale and persistence restarts from zero. # Sized to tolerate one or two missed ~5s poll ticks, not a bench # (180s) or a health-state gap (minutes to hours). "max_evaluation_gap_seconds": 15, } @dataclass class FastGateDecision: symbol: str state: str = STATE_WAIT should_enter: bool = False persistence_seconds_elapsed: float = 0.0 reasons_for: list = field(default_factory=list) reasons_against: list = field(default_factory=list) persistence: dict = field(default_factory=dict) def _hard_disqualifiers(prediction, features, cfg): """Every item here is one of the explicit REJECT examples: 'price falls below VWAP', 'price acceleration turns strongly negative', 'VWAP slope turns negative', 'resistance rejects price with heavy selling' (modeled here as TOO_CLOSE-with-no-breakout-in-progress -- the prediction engine's own resistance classification already accounts for the breakout exception), 'spread becomes unacceptable', 'extreme overextension occurs while momentum deteriorates'.""" reasons = [] price = features.price.get("last") vwap_value = features.vwap.get("value") if price is not None and vwap_value is not None and price < vwap_value: reasons.append(f"price ${price:.2f} fell below VWAP ${vwap_value:.2f}") vwap_slope = features.vwap.get("slope") if vwap_slope is not None and vwap_slope < 0: reasons.append("VWAP slope turned negative") accel = features.acceleration.get("momentum_acceleration") if accel is not None and accel < cfg["hard_price_acceleration_negative"]: reasons.append(f"price acceleration sharply negative ({accel})") spread_pct = features.quote.get("spread_pct") if spread_pct is not None and spread_pct > cfg["max_spread_pct"]: reasons.append(f"spread {spread_pct:.2f}% exceeds max {cfg['max_spread_pct']}%") if prediction.resistance == RESISTANCE_TOO_CLOSE: dist = prediction.resistance_distance_pct reasons.append(f"resistance too close ({dist:.2f}% away, no breakout in progress)" if dist is not None else "resistance too close") if prediction.extension == EXTENSION_SEVERE: reasons.append("severely extended while momentum deteriorates -- chase protection") relative_volume = features.volume.get("relative_volume") if relative_volume is not None: elapsed_fraction = market_time.minutes_since_open() / market_time.session_length_minutes() if elapsed_fraction > 0: pace_ratio = relative_volume / elapsed_fraction if pace_ratio < cfg["min_volume_pace_ratio"]: reasons.append( f"volume pace {pace_ratio:.2f}x normal (rvol={relative_volume}, " f"{elapsed_fraction * 100:.1f}% of session elapsed) below floor " f"{cfg['min_volume_pace_ratio']}x -- too illiquid to safely exit") return reasons def evaluate_fast_entry(symbol: str, prediction, features, persistence_state: dict, as_of: datetime = None, cfg: dict = None) -> FastGateDecision: """ persistence_state: caller-held {"since": datetime_or_None, "last_seen": datetime_or_None}, fed back in on every call for the same symbol (fresh {} = cold start). "last_seen" is the as_of time of the last call that reached the UP-tracking branch below -- used only to detect a stale "since" (see max_evaluation_gap_seconds above), not exposed outside this module. as_of: real wall-clock time by default; a replay/backtest caller must pass the bar's own timestamp, same contract as every other as_of-taking engine in this project (entry_score.compute_entry_score, stream_features.compute_features). """ cfg = {**DEFAULT_CONFIG, **(cfg or get_config().get("fast_prediction_gate", {}))} now = as_of or datetime.now(timezone.utc) persistence_state = persistence_state or {} if getattr(prediction, "insufficient_data", False): return FastGateDecision(symbol=symbol, state=STATE_WAIT, reasons_against=prediction.reasons, persistence={"since": None}) hard_reasons = _hard_disqualifiers(prediction, features, cfg) if hard_reasons: # [BUGFIX 2026-09-11] Previously dropped prediction.reasons here -- # exactly the REJECT case where the momentum/volume/vwap/ema9/ # resistance/extension classification snapshot is most useful to # have on record, since it's what the rest of the pool didn't get # to see either. Every other branch below already includes # prediction.reasons; this one silently didn't. return FastGateDecision(symbol=symbol, state=STATE_REJECT, reasons_against=hard_reasons + prediction.reasons, persistence={"since": None}) if prediction.direction != DIRECTION_UP: return FastGateDecision( symbol=symbol, state=STATE_WAIT, reasons_against=[f"direction={prediction.direction}, not UP"] + prediction.reasons, persistence={"since": None}) since = persistence_state.get("since") last_seen = persistence_state.get("last_seen") if since is not None and last_seen is not None: gap = (now - last_seen).total_seconds() if gap > cfg["max_evaluation_gap_seconds"]: # Stale: this symbol wasn't actually re-checked continuously # since "since" was set (benched, health-ineligible, or had # an open/closing position in between) -- the elapsed time # since then isn't real continuous confirmation. Restart. since = None since = since or now elapsed = (now - since).total_seconds() if elapsed >= cfg["persistence_seconds"]: return FastGateDecision( symbol=symbol, state=STATE_BUY, should_enter=True, persistence_seconds_elapsed=elapsed, reasons_for=[f"held {elapsed:.1f}s (>= {cfg['persistence_seconds']}s)"] + prediction.reasons, persistence={"since": since, "last_seen": now}) return FastGateDecision( symbol=symbol, state=STATE_WAIT, persistence_seconds_elapsed=elapsed, reasons_for=[f"holding {elapsed:.1f}s / {cfg['persistence_seconds']}s"] + prediction.reasons, persistence={"since": since, "last_seen": now})