""" smart_engine.py The live entry decision engine for this project -- answers one question per symbol, per call: BUY / WAIT / REJECT. Built from the 3-stage sequence worked out in the 2026-09-16 design conversation: Stage 1 -- "is this stock worth trading right now?" Continuously re-checked, every call. A HARD disqualifier (lost VWAP support, VWAP sloping down, spread too wide, a confirmed negative trend, lower-highs/lower-lows structure, or volume pace too thin) fails Stage 1 immediately and clears Stage 2's persistence timer -- same "reset and rewatch fresh next call" contract as sip_bot's fast_entry_gate.py, not a ban or cooldown. Passing Stage 1 also requires a genuinely constructive read, not just "nothing bad yet": either a real positive trend, or a flat- but-constructive one (EMA9 above EMA20, holding above a rising- or-flat VWAP) -- the second branch exists specifically so a naturally quiet, low-ATR grinder isn't judged by the same bar as an actively reversing stock (see trend classification below). Stage 2 -- "is momentum actually developing?" Direction has to hold continuously (Stage 1 passing on every call) for persistence_seconds before Stage 3 even gets checked. Same caller-held persistence_state contract as fast_entry_gate.py ({"since", "last_seen"}, fed back in on every call). Stage 3 -- "is there room to run?" Checked fresh the instant Stage 2's hold completes (not before -- resistance/reward:risk can shift during the hold, so checking it early would be stale). If Stage 3 fails, the decision is WAIT (not REJECT) and the Stage 2 timestamp is preserved -- momentum genuinely confirmed, just not enough room yet -- so it gets rechecked next call without losing the hold, unless Stage 1 breaks in the meantime. [2026-09-17] Two interchangeable implementations, chosen by cfg["use_reward_risk_gate"]: - reward:risk (the original): finds the nearest known resistance level still above price, computes an ATR-based stop, and requires the resulting reward:risk ratio to clear a minimum. - trade-flow imbalance (now the default -- reward:risk is DISABLED pending more days of live data): uses trade_imbalance, the (buy_vol-sell_vol)/(buy_vol+sell_vol) read from stream.py over the last imbalance_window_seconds, as the "is this a real breakout" confirmation instead of a resistance distance calc. Two tiers: imbalance >= imbalance_buy_threshold passes outright (order flow alone is confirmation enough, don't wait for a level to cross); imbalance >= imbalance_cross_threshold requires price to have ALSO crossed the nearest REAL resistance level first (the momentary "session_high" tracked live is deliberately excluded from this check -- see _evaluate_stage3_imbalance's docstring for why). Traced from real 2026-09-17 trade data: at the instant of each symbol's actual breakout, ABSI/SECZ/CRML (genuine moves) read +30% to +71% imbalance; USDE (a breakout that immediately faded, never entered on until an hour later at a much worse price) read -71% at the same instant. Thresholds are a first cut off that one day -- expect to retune both as more days of live data come in. Deliberately takes already-computed bars/quote/resistance-levels as input and performs NO I/O of its own (no Alpaca calls, no file writes, no knowledge of where its inputs came from). That's what makes this safe to drop into whatever the live-monitoring loop ends up looking like later -- a real stream, a replay, or a manual test all look identical to this module. """ from dataclasses import dataclass, field from datetime import datetime, timezone from config_loader import get_config import setup_analyzer import volatility from indicators import ( vwap, vwap_slope, ema, atr, relative_volume, volume_acceleration, is_lower_highs_lower_lows, pullback_then_continuation, distance_from_high_pct, spread_pct as _spread_pct, normalized_slope_pct, classify_slope, ) STATE_BUY = "BUY" STATE_WAIT = "WAIT" STATE_REJECT = "REJECT" DEFAULT_CONFIG = { "atr_period": 14, "trend": {"min_flat_threshold_pct": 0.05, "flat_threshold_atr_fraction": 0.25}, "structure_lookback_bars": 4, "pullback_max_retrace_pct": 50.0, "max_spread_pct": 1.0, "min_volume_pace_ratio": 0.4, "persistence_seconds": 18, "max_evaluation_gap_seconds": 15, "atr_multiplier_stop": 1.2, "min_reward_risk_ratio": 1.5, "clear_air_requires_target": False, "clear_air_atr_target_multiplier": 3.0, "use_reward_risk_gate": False, "imbalance_window_seconds": 20, "imbalance_buy_threshold": 0.50, "imbalance_cross_threshold": 0.30, # [2026-09-22] Live entry-quality score, checked the instant Stage 3 # passes -- see _compute_entry_score()'s docstring for what it measures # and why these particular weights/bands. min_entry_score=0.0 is a # deliberate no-op default: entry_score is always logged in metrics # either way, but nothing is actually gated on it until backtesting # across more days picks a real floor -- see simulate.py's # --min-entry-score sweep. "min_entry_score": 0.0, # [2026-09-23] Recent-slope rule (user request after the SLS re-entry # attempt 35s after its exit): Stage 1's own trend reads the regression # over EVERY bar since the open, and its flat branch let SLS through at # -0.078%/bar. When enabled, the slope of the last lookback_bars 1-min # closes (forming bar included) must be sloping up on every read, or # Stage 1 fails and Stage 2's hold resets -- so the 18s hold means ~5 # consecutive sloping-up reads, for every entry and re-entry alike. # mode "positive": above the same ATR-scaled flat band Stage 1 uses # (i.e. up, not flat); mode "above_zero": any slope > 0. "entry_slope_rule": {"enabled": False, "lookback_bars": 5, "mode": "positive"}, # [2026-09-23] Per-tick trade plan (setup_analyzer.py) -- user chose to # REPLACE Stage 3 (imbalance) + entry_score with it, require R:R >= 1.5, # and use the plan's stop instead of 3x 1-min ATR. Stage 1 (incl. the # 5-bar slope rule) and Stage 2's hold still apply first. "use_setup_plan": False, # [2026-09-25] User rule after APPS (bought +5.6% above the open, at the # high of day, 4 ATR over VWAP): price may not be more than base_pct # above today's open; +1% allowed if volume pace >= strong_pace_ratio, # +1% more if the 20s trade imbalance >= strong_imbalance (so 3% / 4% / # 5% by default). Checked in Stage 1, every read, every setup. # [2026-09-28] anchor "vwap": measure the stretch from VWAP instead of the # open (vwap_base_pct, same +step for strong volume / buying) -- the open # anchor blocks trend days that moved early and then based above VWAP # (TGB, AVPT, IMMX 9/28). "max_extension_from_open": {"enabled": False, "base_pct": 3.0, "step_pct": 1.0, "strong_pace_ratio": 3.0, "strong_imbalance": 0.5, "anchor": "open", "vwap_base_pct": 2.0}, # [2026-09-23] Volatility-scaled minimum stop distance (volatility.py); # widens whichever stop the active Stage 3 / plan produced. "min_stop": {}, "entry_score": { "weight_imbalance": 0.30, "weight_volume_pace": 0.25, "weight_volatility": 0.20, "weight_slope": 0.15, "weight_volume_accel": 0.10, "pace_full_credit_ratio": 3.0, "pace_zero_credit_ratio": 15.0, "atr_full_credit_pct": 0.3, "atr_zero_credit_pct": 2.0, "slope_full_credit_pct": 0.2, "slope_zero_credit_pct": 1.2, "vol_accel_full_credit": 1.0, "vol_accel_zero_credit": 3.0, }, } @dataclass class SmartEngineDecision: symbol: str state: str = STATE_WAIT should_enter: bool = False stage1_passed: bool = False stage2_elapsed: float = 0.0 stage3_passed: bool = None reasons_for: list = field(default_factory=list) reasons_against: list = field(default_factory=list) metrics: dict = field(default_factory=dict) persistence: dict = field(default_factory=dict) def _merge_cfg(cfg): base = {**DEFAULT_CONFIG, **(cfg or get_config().get("smart_engine", {}))} base["trend"] = {**DEFAULT_CONFIG["trend"], **base.get("trend", {})} base["entry_score"] = {**DEFAULT_CONFIG["entry_score"], **base.get("entry_score", {})} base["entry_slope_rule"] = {**DEFAULT_CONFIG["entry_slope_rule"], **base.get("entry_slope_rule", {})} base["min_stop"] = volatility.merge_cfg(base.get("min_stop")) base["max_extension_from_open"] = {**DEFAULT_CONFIG["max_extension_from_open"], **base.get("max_extension_from_open", {})} return base def _clamp01(x): return max(0.0, min(1.0, x)) def _band_credit(value: float, full_credit_at: float, zero_credit_at: float) -> float: """1.0 at/below full_credit_at, 0.0 at/beyond zero_credit_at, linear between -- used for every 'lower is better, but don't cliff-edge it' sub-score below (see _compute_entry_score).""" if zero_credit_at == full_credit_at: return 1.0 if value <= full_credit_at else 0.0 return _clamp01((zero_credit_at - value) / (zero_credit_at - full_credit_at)) def _compute_entry_score(stage1_metrics: dict, stage3: dict, cfg: dict) -> dict: """ [2026-09-22] A continuous 0-100 read of "how good is THIS entry," checked the instant Stage 3 passes -- separate from and in addition to Stage 1-3's existing hard pass/fail gates. Motivated by the 2026-09-21 giveback_room backtest finding: when a slot frees up mid-day, monitor.py currently enters whichever next candidate happens to clear Stage 1-3 first, with no notion of "clears the bar, but is this one actually a GOOD setup" -- unlike scanner.py's premarket candidate_score, which only ever runs once, before the open, and (per the user, 2026-09-22) is understood to only really apply to the day's first entries, not entries taken hours in. Built from the 5 live Stage1/Stage3 metrics that showed the (weak -- see caveat below) most consistent SIGN of correlation with real P/L across the 39 real trades from 2026-09-18 + 2026-09-21: imbalance strength (higher better), volume_pace_ratio (lower better -- chasing an already-way-overtraded stock underperformed), atr_pct (lower better -- tighter/calmer setups outperformed choppier ones), slope_pct (lower/steadier better -- a very steep spike right at entry underperformed a modest climb), volume_acceleration (lower better -- a sudden last-second volume spike looked more like a climax than continuation in this sample). CAVEAT, load-bearing: every one of those correlations is weak (|r| < 0.3, n=39, NOT statistically significant on its own) -- this score is a hypothesis built from the SIGN of thin data, not a validated model. That's exactly why min_entry_score defaults to 0.0 (a no-op floor -- entry_score is computed and logged on every entry either way, but nothing is rejected on it) until backtesting across more real days either confirms a real floor is worth setting or shows this doesn't hold up. """ esc = cfg["entry_score"] imbalance = stage3.get("imbalance") imbalance_floor = cfg["imbalance_cross_threshold"] imbalance_sub = _clamp01((imbalance - imbalance_floor) / (1.0 - imbalance_floor)) \ if imbalance is not None else 0.5 # reward:risk Stage 3 has no imbalance reading -- neutral credit pace = stage1_metrics.get("volume_pace_ratio") pace_sub = _band_credit(pace, esc["pace_full_credit_ratio"], esc["pace_zero_credit_ratio"]) \ if pace is not None else 0.5 atr_sub = _band_credit(stage1_metrics["atr_pct"], esc["atr_full_credit_pct"], esc["atr_zero_credit_pct"]) slope_sub = _band_credit(stage1_metrics["slope_pct"], esc["slope_full_credit_pct"], esc["slope_zero_credit_pct"]) vola_sub = _band_credit(stage1_metrics["volume_acceleration"], esc["vol_accel_full_credit"], esc["vol_accel_zero_credit"]) score = 100.0 * ( esc["weight_imbalance"] * imbalance_sub + esc["weight_volume_pace"] * pace_sub + esc["weight_volatility"] * atr_sub + esc["weight_slope"] * slope_sub + esc["weight_volume_accel"] * vola_sub ) return { "entry_score": round(score, 2), "breakdown": { "imbalance_sub": round(imbalance_sub, 3), "volume_pace_sub": round(pace_sub, 3), "volatility_sub": round(atr_sub, 3), "slope_sub": round(slope_sub, 3), "volume_accel_sub": round(vola_sub, 3), }, } def _vol_note(vol: dict, price: float, stop) -> str: dist = (price - stop) * 100.0 if stop is not None else None pct = f"{vol['daily_atr_pct']:.1f}% daily ATR" if vol["daily_atr_pct"] is not None else "daily ATR unknown" return (f"volatility {vol['class']} ({pct}), stop distance " + (f"{dist:.1f}c ({dist / price:.2f}%)" if dist is not None else "n/a")) def _trend_class(bars: list, atr_pct: float, cfg: dict): """Classifies price trend as positive/flat/negative -- but unlike a single fixed threshold for every symbol, the flat/trend boundary scales with THIS symbol's own ATR%. See config.json's smart_engine.trend note for why.""" closes = [b["c"] for b in bars] slope_pct = normalized_slope_pct(closes) flat_thresh = max(cfg["trend"]["min_flat_threshold_pct"], atr_pct * cfg["trend"]["flat_threshold_atr_fraction"]) return classify_slope(slope_pct, flat_thresh), slope_pct, flat_thresh def _nearest_resistance_above(price: float, resistance_levels: dict): """resistance_levels: {label: level_or_None}, e.g. {"session_high": .., "premarket_high": .., "prev_day_high": .., "range_20d_high": ..} -- any subset, any labels. Returns (level, label) for whichever supplied level is nearest but still above price, or (None, None) if every supplied level is at or below price ("clear air").""" candidates = [(level, label) for label, level in (resistance_levels or {}).items() if volatility.is_level_key(label) and level and level > price] if not candidates: return None, None return min(candidates, key=lambda pair: pair[0]) def _session_elapsed_fraction(now: datetime) -> float: """How far into today's regular session `now` is (0-1), computed from `now` itself -- NOT market_time.minutes_since_open(), which always reads the real wall clock and would silently break replay/ backtest calls that pass a historical `as_of`. This is what makes the module's own as_of contract (see evaluate()'s docstring) actually true instead of aspirational.""" from zoneinfo import ZoneInfo sched = get_config()["schedule"] tz = ZoneInfo(sched["timezone"]) now_et = now.astimezone(tz) oh, om, os_ = (int(x) for x in sched["market_open_time"].split(":")) ch, cm, cs = (int(x) for x in sched["market_close_time"].split(":")) open_dt = now_et.replace(hour=oh, minute=om, second=os_, microsecond=0) close_dt = now_et.replace(hour=ch, minute=cm, second=cs, microsecond=0) session_len = (close_dt - open_dt).total_seconds() / 60.0 if session_len <= 0: return 0.0 return (now_et - open_dt).total_seconds() / 60.0 / session_len def _evaluate_stage1(symbol: str, bars: list, cumulative_volume_today: float, avg_vol_baseline: float, quote: dict, cfg: dict, now: datetime, trade_imbalance: float = None) -> dict: """Pure per-call read -- no state. Returns a dict of metrics plus hard_reasons (any -> Stage 1 fails and Stage 2's timer resets) and `passed` (hard_reasons empty AND a genuinely constructive read).""" price = bars[-1]["c"] vwap_value = vwap(bars) vslope = vwap_slope(bars) a = atr(bars, period=min(cfg["atr_period"], max(2, len(bars) - 1))) atr_pct = (a / price * 100.0) if price else 0.0 trend, slope_pct, flat_thresh = _trend_class(bars, atr_pct, cfg) closes = [b["c"] for b in bars] ema9 = ema(closes, 9) ema20 = ema(closes, 20) lhl = is_lower_highs_lower_lows(bars, lookback=min(cfg["structure_lookback_bars"], len(bars))) pullback_ok = pullback_then_continuation(bars, cfg["pullback_max_retrace_pct"]) vol_accel = volume_acceleration([b["v"] for b in bars]) dist_from_high = distance_from_high_pct(price, max(b["h"] for b in bars)) spread = _spread_pct(quote["bid"], quote["ask"]) if quote else None rvol = relative_volume(cumulative_volume_today, avg_vol_baseline) if avg_vol_baseline else 0.0 elapsed_fraction = _session_elapsed_fraction(now) pace = (rvol / elapsed_fraction) if elapsed_fraction > 0 else None hard_reasons = [] if price < vwap_value: hard_reasons.append(f"price ${price:.2f} below VWAP ${vwap_value:.2f}") if vslope < 0: hard_reasons.append("VWAP sloping down") if spread is not None and spread > cfg["max_spread_pct"]: hard_reasons.append(f"spread {spread:.2f}% exceeds max {cfg['max_spread_pct']}%") if trend == "negative": hard_reasons.append(f"trend negative (slope {slope_pct:.3f}%, flat band {flat_thresh:.3f}%)") if lhl: hard_reasons.append("lower-highs/lower-lows structure") if pace is not None and pace < cfg["min_volume_pace_ratio"]: hard_reasons.append(f"volume pace {pace:.2f}x normal below floor {cfg['min_volume_pace_ratio']}x") ext = cfg["max_extension_from_open"] open_px = bars[0]["o"] if bars else None ext_pct = (price / open_px - 1.0) * 100.0 if open_px else None vwap_ext_pct = (price / vwap_value - 1.0) * 100.0 if vwap_value else None allowed = None if ext["enabled"]: by_vwap = ext.get("anchor", "open") == "vwap" val = vwap_ext_pct if by_vwap else ext_pct if val is not None: strong_vol = pace is not None and pace >= ext["strong_pace_ratio"] strong_buy = trade_imbalance is not None and trade_imbalance >= ext["strong_imbalance"] base = ext.get("vwap_base_pct", 2.0) if by_vwap else ext["base_pct"] allowed = base + ext["step_pct"] * (int(strong_vol) + int(strong_buy)) if val > allowed: ref = f"VWAP ${vwap_value:.2f}" if by_vwap else f"open ${open_px:.2f}" hard_reasons.append( f"extended {val:+.1f}% above {ref} (max {allowed:.0f}%: " f"volume pace {'strong' if strong_vol else 'normal'}, buying {'strong' if strong_buy else 'normal'})") rule = cfg["entry_slope_rule"] recent_slope = normalized_slope_pct(closes[-rule["lookback_bars"]:]) if rule["enabled"]: bar = flat_thresh if rule["mode"] == "positive" else 0.0 if recent_slope <= bar: hard_reasons.append(f"last {rule['lookback_bars']} bars not sloping up " f"(slope {recent_slope:.3f}%, needs > {bar:.3f}%)") constructive = (trend == "positive") or ( trend == "flat" and ema9 > ema20 and price >= vwap_value and vslope >= 0) metrics = { "price": price, "vwap": round(vwap_value, 4), "vwap_slope": round(vslope, 4), "atr": round(a, 4), "atr_pct": round(atr_pct, 3), "trend": trend, "slope_pct": round(slope_pct, 4), "flat_threshold_pct": round(flat_thresh, 4), "recent_slope_pct": round(recent_slope, 4), "ext_from_open_pct": round(ext_pct, 2) if ext_pct is not None else None, "max_ext_allowed_pct": allowed, "ext_from_vwap_pct": round(vwap_ext_pct, 2) if vwap_ext_pct is not None else None, "ema9": round(ema9, 4), "ema20": round(ema20, 4), "lower_highs_lower_lows": lhl, "pullback_then_continuation": pullback_ok, "volume_acceleration": round(vol_accel, 3), "distance_from_high_pct": round(dist_from_high, 3), "spread_pct": round(spread, 3) if spread is not None else None, "rvol": round(rvol, 3), "volume_pace_ratio": round(pace, 3) if pace is not None else None, } return { "passed": not hard_reasons and constructive, "constructive": constructive, "hard_reasons": hard_reasons, "metrics": metrics, } def _evaluate_stage3(price: float, atr_val: float, resistance_levels: dict, cfg: dict) -> dict: """Checked once, fresh, the instant Stage 2's hold completes.""" target, target_label = _nearest_resistance_above(price, resistance_levels) stop = price - atr_val * cfg["atr_multiplier_stop"] risk = max(price - stop, 0.01) if target is None: if cfg["clear_air_requires_target"]: target = price + atr_val * cfg["clear_air_atr_target_multiplier"] target_label = "clear_air_atr_projection" else: return { "passed": True, "target": None, "target_label": None, "stop": round(stop, 4), "risk": round(risk, 4), "reward": None, "reward_risk_ratio": None, "reason": "clear air -- no known resistance above price, reward:risk gate skipped", } reward = target - price ratio = reward / risk passed = ratio >= cfg["min_reward_risk_ratio"] reason = (f"target {target:.2f} ({target_label}), stop {stop:.2f}, " f"reward:risk {ratio:.2f} (min {cfg['min_reward_risk_ratio']})") return { "passed": passed, "target": round(target, 4), "target_label": target_label, "stop": round(stop, 4), "risk": round(risk, 4), "reward": round(reward, 4), "reward_risk_ratio": round(ratio, 3), "reason": reason, } def _evaluate_stage3_imbalance(price: float, atr_val: float, resistance_levels: dict, trade_imbalance, cfg: dict) -> dict: """[2026-09-17] Trade-flow-imbalance Stage 3 -- see evaluate()'s docstring for the two-tier logic. Still computes stop/risk off the same ATR formula reward:risk used, since position_manager.py sizes every position off this stop regardless of which Stage 3 is active. The "has price crossed resistance" check deliberately excludes "session_high" from resistance_levels: session_high is max(bar.high for bar in bars) recomputed live every poll, so during an active rally it's never more than a few cents above price by construction -- it isn't a real level, it's just "wherever price was a moment ago," and treating it as resistance would make the 30%-tier's "cross resistance first" condition nearly impossible to satisfy during a genuine breakout (this is Bug #4 from the 2026-09-17 review). The static, pre-scan levels (premarket_high, prev_day_high, range_20d_high) are what this checks against. """ stop = price - atr_val * cfg["atr_multiplier_stop"] risk = max(price - stop, 0.01) real_levels = {label: level for label, level in (resistance_levels or {}).items() if label != "session_high" and volatility.is_level_key(label)} target, target_label = _nearest_resistance_above(price, real_levels) crossed = target is None # no real level left above price -- already through everything known base = {"stop": round(stop, 4), "risk": round(risk, 4), "target": round(target, 4) if target is not None else None, "target_label": target_label, "crossed_resistance": crossed, "imbalance": round(trade_imbalance, 4) if trade_imbalance is not None else None} if trade_imbalance is None: return {**base, "passed": False, "reason": "no trade-flow imbalance data yet (cold stream buffer)"} if trade_imbalance < 0: # [2026-09-18] Explicit veto, not just "below the cross threshold" -- # net selling into the move means this isn't confirmed as a real # breakout at all (USDE's actual breakout instant on 2026-09-17 # read -71%, a fakeout that never moved further in its favor). # Kept as its own branch (rather than relying on it falling through # imbalance_cross_threshold, which happens to be positive today) # so this stays true even if that threshold is ever tuned to <= 0. return {**base, "passed": False, "reason": f"imbalance {trade_imbalance:+.0%} is negative (net selling) -- refusing entry"} if trade_imbalance >= cfg["imbalance_buy_threshold"]: return {**base, "passed": True, "reason": (f"imbalance {trade_imbalance:+.0%} >= " f"{cfg['imbalance_buy_threshold']:.0%} buy threshold, stop {stop:.2f}")} if trade_imbalance >= cfg["imbalance_cross_threshold"]: if crossed: return {**base, "passed": True, "reason": (f"imbalance {trade_imbalance:+.0%} >= " f"{cfg['imbalance_cross_threshold']:.0%} and resistance cleared, stop {stop:.2f}")} return {**base, "passed": False, "reason": (f"imbalance {trade_imbalance:+.0%} confirms momentum but hasn't cleared " f"{target_label} ${target:.2f} yet" if target is not None else f"imbalance {trade_imbalance:+.0%} confirms momentum, awaiting resistance data")} return {**base, "passed": False, "reason": (f"imbalance {trade_imbalance:+.0%} below " f"{cfg['imbalance_cross_threshold']:.0%} confirmation floor")} def evaluate(symbol: str, bars: list, cumulative_volume_today: float, avg_vol_baseline: float, resistance_levels: dict, persistence_state: dict, quote: dict = None, trade_imbalance: float = None, as_of: datetime = None, cfg: dict = None) -> SmartEngineDecision: """ bars: recent 1-min bars for this symbol, oldest first, each {"t","o","h","l","c","v"} -- enough history for the configured atr_period/structure_lookback_bars (a short rolling window, not the whole day). cumulative_volume_today: total volume for this symbol since the session opened (kept by the caller/stream, NOT derived from `bars` -- `bars` is a short window, this is a running total). avg_vol_baseline: real prior-day volume for this symbol. resistance_levels: {label: level_or_None}, e.g. session_high, premarket_high, prev_day_high, range_20d_high -- any subset. persistence_state: caller-held {"since": datetime_or_None, "last_seen": datetime_or_None} for THIS symbol, fed back in on every call (fresh {} = cold start). Same contract as fast_entry_gate.evaluate_fast_entry(). quote: {"bid": float, "ask": float}, or None to skip the spread check (fails open, same as fast_entry_gate's features.quote contract when a quote isn't available). trade_imbalance: (buy_vol-sell_vol)/(buy_vol+sell_vol) over the last cfg["imbalance_window_seconds"] of trades (stream.py's get_trade_imbalance()), or None if there's no trade data yet. Only used when cfg["use_reward_risk_gate"] is False (the default) -- ignored entirely by the reward:risk Stage 3. as_of: real wall-clock time by default; pass the bar's own timestamp for a replay/backtest, same contract as every other as_of-taking function in this codebase. """ cfg = _merge_cfg(cfg) now = as_of or datetime.now(timezone.utc) persistence_state = persistence_state or {} if len(bars) < 3: return SmartEngineDecision(symbol=symbol, state=STATE_WAIT, reasons_against=["insufficient bar data"], persistence={"since": None}) stage1 = _evaluate_stage1(symbol, bars, cumulative_volume_today, avg_vol_baseline, quote, cfg, now, trade_imbalance=trade_imbalance) price_now = stage1["metrics"]["price"] spread_abs = (quote["ask"] - quote["bid"]) if quote and quote.get("ask") and quote.get("bid") else 0.0 vol = volatility.stop_floor(price_now, (resistance_levels or {}).get("daily_atr"), spread_abs, cfg["min_stop"]) stage1["metrics"]["volatility"] = {"class": vol["class"], "daily_atr_pct": vol["daily_atr_pct"], "min_stop": round(vol["floor"], 4), "min_stop_enabled": cfg["min_stop"]["enabled"]} plan = None if cfg["use_setup_plan"]: # Recomputed every call so the plan (levels, stop, target, triggers) # tracks the chart tick by tick, whatever Stage 1/2 say this poll. plan = setup_analyzer.analyze(symbol, bars, resistance_levels, quote, now, stop_floor=vol["floor"] if cfg["min_stop"]["enabled"] else None, trade_imbalance=trade_imbalance) stage1["metrics"]["plan"] = plan.to_dict() # [2026-09-23] Context rules: an orderly pullback IS a short run of # lower highs/lows -- don't let Stage 1's 4-bar structure check # reject a stock the plan is actively watching as a pullback. ctx = setup_analyzer._merge_cfg(None)["context_rules"] if (ctx["enabled"] and ctx["pullback_exempt_lower_highs"] and plan.metrics.get("pullback_candidate")): stage1["hard_reasons"] = [r for r in stage1["hard_reasons"] if r != "lower-highs/lower-lows structure"] if stage1["hard_reasons"]: return SmartEngineDecision( symbol=symbol, state=STATE_REJECT, stage1_passed=False, reasons_against=stage1["hard_reasons"], metrics=stage1["metrics"], persistence={"since": None}) if not stage1["constructive"]: return SmartEngineDecision( symbol=symbol, state=STATE_WAIT, stage1_passed=False, reasons_against=["not yet constructive (trend/EMA/VWAP alignment not bullish)"], metrics=stage1["metrics"], persistence={"since": None}) # Stage 1 passed -- Stage 2 persistence timer (mirrors # fast_entry_gate.py's since/last_seen/stale-gap logic exactly). 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"]: since = None since = since or now elapsed = (now - since).total_seconds() if elapsed < cfg["persistence_seconds"]: return SmartEngineDecision( symbol=symbol, state=STATE_WAIT, stage1_passed=True, stage2_elapsed=elapsed, reasons_for=[f"holding {elapsed:.1f}s / {cfg['persistence_seconds']}s"], metrics=stage1["metrics"], persistence={"since": since, "last_seen": now}) # Stage 2 hold complete -- Stage 3 checked fresh, right now. price = stage1["metrics"]["price"] atr_val = stage1["metrics"]["atr"] if plan is not None: stage3 = {"passed": plan.state == setup_analyzer.BUY, "stop": plan.stop, "target": plan.target, "reward_risk": plan.reward_risk, "setup": plan.setup, "reason": plan.summary()} metrics = {**stage1["metrics"], "stage3": stage3} if plan.state == setup_analyzer.BUY: return SmartEngineDecision( symbol=symbol, state=STATE_BUY, should_enter=True, stage1_passed=True, stage2_elapsed=elapsed, stage3_passed=True, reasons_for=[f"held {elapsed:.1f}s", plan.summary(), _vol_note(vol, price, plan.stop)], metrics=metrics, persistence={"since": since, "last_seen": now}) return SmartEngineDecision( symbol=symbol, state=STATE_WAIT, stage1_passed=True, stage2_elapsed=elapsed, stage3_passed=False, reasons_against=[f"held {elapsed:.1f}s, plan says {plan.summary()}"], metrics=metrics, persistence={"since": since, "last_seen": now}) if cfg["use_reward_risk_gate"]: stage3 = _evaluate_stage3(price, atr_val, resistance_levels, cfg) else: stage3 = _evaluate_stage3_imbalance(price, atr_val, resistance_levels, trade_imbalance, cfg) if cfg["min_stop"]["enabled"] and stage3.get("stop") is not None: floored = price - vol["floor"] if floored < stage3["stop"]: stage3["stop"] = round(floored, 4) stage3["risk"] = round(price - floored, 4) stage3["reason"] += f" (stop widened to {vol['class']}-volatility floor {floored:.2f})" metrics = {**stage1["metrics"], "stage3": stage3} if stage3["passed"]: entry_score = _compute_entry_score(stage1["metrics"], stage3, cfg) metrics["entry_score"] = entry_score["entry_score"] metrics["entry_score_breakdown"] = entry_score["breakdown"] if entry_score["entry_score"] < cfg["min_entry_score"]: # Same "confirmed but not yet good enough, keep the hold alive" # shape as the room-to-run WAIT below -- entry_score can change # poll to poll (imbalance/pace/slope all do), so don't discard # Stage 2's hold over a single low read. return SmartEngineDecision( symbol=symbol, state=STATE_WAIT, stage1_passed=True, stage2_elapsed=elapsed, stage3_passed=True, reasons_against=[f"held {elapsed:.1f}s, room-to-run passed, but entry_score " f"{entry_score['entry_score']:.1f} below floor {cfg['min_entry_score']:.1f}"], metrics=metrics, persistence={"since": since, "last_seen": now}) return SmartEngineDecision( symbol=symbol, state=STATE_BUY, should_enter=True, stage1_passed=True, stage2_elapsed=elapsed, stage3_passed=True, reasons_for=[f"held {elapsed:.1f}s", stage3["reason"], _vol_note(vol, price, stage3["stop"]), f"entry_score {entry_score['entry_score']:.1f}"], metrics=metrics, persistence={"since": since, "last_seen": now}) # Momentum genuinely confirmed, just not enough room yet -- WAIT, # keep the hold alive (don't reset "since") so this rechecks Stage 3 # next call instead of re-timing Stage 2 from scratch. return SmartEngineDecision( symbol=symbol, state=STATE_WAIT, stage1_passed=True, stage2_elapsed=elapsed, stage3_passed=False, reasons_against=[f"momentum confirmed ({elapsed:.1f}s) but room-to-run failed: {stage3['reason']}"], metrics=metrics, persistence={"since": since, "last_seen": now})