""" exit.py Answers ONE question for an open position: should we still be in this? Deliberately independent of smart_engine.py's entry logic -- per the breakout_bot design doc's explicit rule, entry and exit must never share one combined decision rule (smart_engine.py answers "should we enter," this answers "should we remain"). Three layers, checked every poll: 1. HARD stop -- always active, exits immediately, no confirmation needed. Pure capital protection. [BUGFIX 2026-09-22] Uses the position's own stop_price fixed at entry (smart_engine.py's Stage 3 stop). It used to be recomputed every poll as current price minus ATR, which always sat below price and trailed it down -- it could never fire; 32 of 70 live trades (9/17-9/22) exited below their entry stop, $455 worse than exiting at it. 2. [2026-09-17, redesigned 2026-09-18] Trade-flow-imbalance decline -- the first of the two SIGNAL-based exits (checked right after the hard stop, ahead of the deterioration layer below). Originally checked once per completed 1-min bar; rebuilt after real-data testing (ABSI, 2026-09-17) showed the 1-min-bar version completely missed a real reversal -- the bar spanning ABSI's actual low read the MOST bullish imbalance of the whole window -- while it also fired falsely later on, during a recovery, on bar-to-bar noise with no price move behind it at all. A continuous, 5-second-cadence read of the SAME live rolling-window imbalance smart_engine.py's Stage 3 uses (stream.py's get_trade_imbalance()) caught the real reversal within 5 seconds of the actual price low -- roughly as fast as the trend classifier itself, and far better than the 1-min-bar version. Now checks every poll (not gated behind price/trend going flat first -- that precondition would have excluded the one real catch found, since trend didn't reclassify until 5 seconds after the fast imbalance streak already completed). Fires immediately (no separate confirm-read wait -- the streak itself IS the confirmation) once imbalance_exit_confirm_reads consecutive polls have each read lower than the one before. 3. SOFT "deterioration" exit -- requires a COMBINATION of signals (not just one) to agree, sustained for exit_confirm_reads consecutive polls, before it fires. This is the direct answer to the 2026-09-16 IIIV lesson: a single flat/quiet reading forced an exit there and caused a whipsaw (exit-then-rebuy-higher, twice) on a stock that was only pausing, not actually reversing. Requiring BOTH multiple agreeing signals AND sustained confirmation (not "any one blip") is what should tell a real breakdown apart from a stock that's just quiet for a bar or two. Caller-held confirmation state per symbol, fed back in every call (same pattern as smart_engine.py's persistence contract and sip_bot's intraday_health.py confirm-reads hysteresis) -- now {"count": int, "imbalance_decline_count": int, "imbalance_prev": float_or_None}. """ from dataclasses import dataclass, field from config_loader import get_config from indicators import vwap, vwap_slope, atr, normalized_slope_pct, classify_slope, is_lower_highs_lower_lows STATE_HOLD = "HOLD" STATE_EXIT = "EXIT" DEFAULT_CONFIG = { "deterioration_min_signals": 2, "deterioration_signals_total": 4, "exit_confirm_reads": 3, "atr_multiplier_stop": 1.2, "atr_period": 14, "structure_lookback_bars": 4, "trend": {"min_flat_threshold_pct": 0.05, "flat_threshold_atr_fraction": 0.25}, "imbalance_window_seconds": 20, "imbalance_exit_confirm_reads": 5, # [2026-09-25] 0 = trend over every bar since the open (original). N > 0: # the deterioration layer's "trend negative" sign uses only the last N # bars -- EMBC sagged for hours while the whole-day slope stayed flat/up. "deterioration_trend_bars": 0, } @dataclass class ExitDecision: symbol: str state: str = STATE_HOLD should_exit: bool = False reason: str = "" metrics: dict = field(default_factory=dict) confirmation: dict = field(default_factory=dict) def _merge_cfg(cfg): base = {**DEFAULT_CONFIG, **(cfg or get_config().get("exit", {}))} base["trend"] = {**DEFAULT_CONFIG["trend"], **base.get("trend", {})} return base def _evaluate_imbalance_decline(trade_imbalance, confirmation_state: dict, cfg: dict) -> dict: """[2026-09-18] First signal-based exit gate, continuous cadence. trade_imbalance: stream.py's get_trade_imbalance(imbalance_window_seconds) live snapshot for this symbol -- the SAME rolling-window read smart_engine.py's Stage 3 uses, just read again every poll instead of once per bar. Compares this poll's reading directly against last poll's (confirmation_state["imbalance_prev"]), so the streak advances every ~poll_interval_seconds rather than every minute. A None reading (cold buffer / no trades in the window) holds the streak steady rather than resetting or advancing it -- fails open, doesn't punish a momentary data gap.""" prior_count = confirmation_state.get("imbalance_decline_count", 0) prior_imb = confirmation_state.get("imbalance_prev") if trade_imbalance is None: # No data this poll -- hold everything steady, INCLUDING the # comparison baseline, so a momentary gap can't itself count as # a decline, and doesn't silently erase the streak's baseline # once data resumes next poll. return {"triggered": False, "count": prior_count, "cur_imbalance": prior_imb, "prev_imbalance": prior_imb} declining = prior_imb is not None and trade_imbalance < prior_imb count = prior_count + 1 if declining else 0 return {"triggered": count >= cfg["imbalance_exit_confirm_reads"], "count": count, "cur_imbalance": trade_imbalance, "prev_imbalance": prior_imb} def evaluate(symbol: str, bars: list, entry_price: float, confirmation_state: dict, trade_imbalance: float = None, cfg: dict = None, stop_price: float = None) -> ExitDecision: """ bars: recent 1-min bars for this symbol, same short rolling window shape smart_engine.py takes. entry_price: this position's own entry price. stop_price: the position's stop fixed at entry -- Layer 1's hard stop. None disables Layer 1. confirmation_state: caller-held state for this symbol, fed back in every call (fresh {} = cold start, e.g. right after entry) -- {"count": int, "imbalance_decline_count": int, "imbalance_prev": float_or_None}. trade_imbalance: stream.py's get_trade_imbalance(imbalance_window_seconds) live read for this symbol -- feeds the imbalance-decline gate below, checked every poll. None simply means that gate can't advance this poll (fails open). """ cfg = _merge_cfg(cfg) confirmation_state = confirmation_state or {} imb = _evaluate_imbalance_decline(trade_imbalance, confirmation_state, cfg) imbalance_state = {"imbalance_decline_count": imb["count"], "imbalance_prev": imb["cur_imbalance"]} if len(bars) < 3: return ExitDecision(symbol=symbol, state=STATE_HOLD, reason="insufficient bar data", confirmation={"count": 0, **imbalance_state}) 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 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"]) trend = classify_slope(slope_pct, flat_thresh) lhl = is_lower_highs_lower_lows(bars, lookback=min(cfg["structure_lookback_bars"], len(bars))) stop = stop_price metrics = { "price": price, "vwap": round(vwap_value, 4), "vwap_slope": round(vslope, 4), "atr": round(a, 4), "trend": trend, "slope_pct": round(slope_pct, 4), "lower_highs_lower_lows": lhl, "stop": round(stop, 4) if stop is not None else None, "imbalance_decline_count": imb["count"], "imbalance_cur": imb["cur_imbalance"], "imbalance_prev": imb["prev_imbalance"], } # Layer 1 -- hard stop, no confirmation needed. if stop is not None and price <= stop: return ExitDecision(symbol=symbol, state=STATE_EXIT, should_exit=True, reason=f"hard stop: price ${price:.4f} <= entry stop ${stop:.4f}", metrics=metrics, confirmation={"count": 0, **imbalance_state}) # Layer 2 -- trade-flow imbalance declining for N consecutive polls, # AND price itself must also be sloping down (slope_pct < 0) -- # [2026-09-18] added after real-data testing showed the imbalance # streak alone fires on order-flow noise with no price move behind # it (ABSI, 9:56 ET) or on early post-entry noise before a real rally # (SECZ, 10:38 ET). Uses the raw slope_pct, not the discretized # `trend` label -- `trend` has a deliberate ATR-scaled flat deadzone # around zero (see _trend_class's docstring) that would mask a real # but modest decline the same way it's designed to ignore noise, and # testing showed ABSI's own real reversal never actually reclassified # trend to "negative" (positive -> flat only) even while price was # genuinely dropping. The streak itself is still the confirmation -- # no separate confirm-read wait -- price merely has to agree at the # moment the streak completes. if imb["triggered"] and slope_pct < 0: return ExitDecision( symbol=symbol, state=STATE_EXIT, should_exit=True, reason=(f"imbalance declining for {imb['count']} consecutive reads " f"({imb['prev_imbalance']:+.0%} -> {imb['cur_imbalance']:+.0%}), " f"price sloping down ({slope_pct:.3f}%)"), metrics=metrics, confirmation={"count": 0, "imbalance_decline_count": 0, "imbalance_prev": imb["cur_imbalance"]}) if imb["triggered"]: # Streak completed but price isn't confirming it -- don't reset # the streak (it's still real order-flow deterioration, keep # watching), just note the near-miss and fall through. metrics["imbalance_reason"] = ( f"imbalance declined {imb['count']} reads but price not sloping down " f"(slope {slope_pct:.3f}%) -- withholding exit") # Layer 3 -- soft deterioration: count how many of the 4 signals agree. det_trend = trend if cfg.get("deterioration_trend_bars"): recent = closes[-cfg["deterioration_trend_bars"]:] det_trend = classify_slope(normalized_slope_pct(recent), flat_thresh) metrics["recent_trend"] = det_trend signals = [ price < vwap_value, vslope < 0, det_trend == "negative", lhl, ] agreeing = sum(signals) if agreeing < cfg["deterioration_min_signals"]: return ExitDecision(symbol=symbol, state=STATE_HOLD, reason=f"healthy ({agreeing}/{cfg['deterioration_signals_total']} deterioration signals)", metrics=metrics, confirmation={"count": 0, **imbalance_state}) count = confirmation_state.get("count", 0) + 1 if count < cfg["exit_confirm_reads"]: return ExitDecision( symbol=symbol, state=STATE_HOLD, reason=(f"deteriorating ({agreeing}/{cfg['deterioration_signals_total']} signals) but not yet " f"confirmed ({count}/{cfg['exit_confirm_reads']} reads)"), metrics=metrics, confirmation={"count": count, **imbalance_state}) return ExitDecision( symbol=symbol, state=STATE_EXIT, should_exit=True, reason=(f"deterioration confirmed: {agreeing}/{cfg['deterioration_signals_total']} signals agreed " f"for {count} consecutive reads"), metrics=metrics, confirmation={"count": 0, **imbalance_state})