""" exit_cents_grace_experimental.py [2026-09-20] NOT wired into monitor.py's live loop -- backtest-only. Brings back a mechanism that actually ran live in the sibling "premarket" bot (see premarket_backup_2026-09-08_2010/config.json's `stop` section and position_manager.py's `_check_fade_confirmation`, added there 2026-08-18): a flat $0.10 trailing stop off the position's own peak, with ONE $0.05 "grace" extension if the stock's own health signals don't confirm real deterioration at the moment the stop is touched. That project's own note on raising its ATR multiplier (2026-09-02) records 14/14 trades touching the stop and needing the grace extension just to survive the first touch on one real day -- i.e. even there this was a live, load-bearing mechanism, not a rarely used edge case. Ported here rather than copied verbatim: the original's "health" gate was a separate intraday_health.py module (price/VWAP slope + volume participation + structure classification) that doesn't exist in this project and pulls in unrelated dependencies. This version reuses THIS bot's own already-existing "healthy" definition instead -- exit.py's own Layer 3 vote (price vs VWAP, VWAP slope, price trend, lower-highs/lower-lows) literally logs "healthy (N/4 deterioration signals)" when fewer than deterioration_min_signals agree. Using that same vote here means "still healthy" means the same thing everywhere in this codebase, not a second, differently-tuned health concept. Genuinely different SHAPE of rule than everything else tried this session (PRF, giveback-vote, resistance-timer, velocity): a flat DOLLAR distance (not a percentage, not ATR-scaled, not a fixed time window), with a bounded one-time forgiveness gated on the bot's own existing structural vote rather than order-flow/imbalance (which the 2026-09-18 cross-symbol analysis found to be non-predictive) or decline velocity (which underperformed baseline in every configuration that actually fired). """ from dataclasses import dataclass, field from indicators import vwap, vwap_slope, atr as calc_atr, normalized_slope_pct, classify_slope, is_lower_highs_lower_lows STATE_HOLD = "HOLD" STATE_EXIT = "EXIT" DEFAULT_CONFIG = { "base_giveback_cents": 0.10, # the original project's flat trailing distance "grace_cents": 0.05, # the original project's grace_distance "max_grace_extensions": 1, # the original project's max_grace_extensions "deterioration_min_signals": 2, # matches exit.py's own DEFAULT_CONFIG -- same "healthy" bar bot-wide "structure_lookback_bars": 4, "min_flat_threshold_pct": 0.05, "flat_threshold_atr_fraction": 0.25, "atr_period": 14, } @dataclass class CentsGraceDecision: symbol: str state: str = STATE_HOLD should_exit: bool = False reason: str = "" metrics: dict = field(default_factory=dict) state_out: dict = field(default_factory=dict) def _merge_cfg(cfg): return {**DEFAULT_CONFIG, **(cfg or {})} def evaluate(symbol: str, bars: list, entry_price: float, state_in: dict, cfg: dict = None) -> CentsGraceDecision: """ bars: same short rolling 1-min-bar window exit.py takes. state_in: caller-held state fed back every poll -- {"peak_price", "fade_grace_active", "grace_extensions_used"}. Fresh {} at entry. """ cfg = _merge_cfg(cfg) state_in = dict(state_in or {}) if len(bars) < 3: return CentsGraceDecision(symbol=symbol, reason="insufficient bar data", state_out=state_in) price = bars[-1]["c"] # Anchored at entry_price, not the first bar's close -- matches the # original project's initial_distance/trailing_distance both being # driven by the same flat-cents value from the moment of entry. prior_peak = state_in.get("peak_price", entry_price) peak_price = max(prior_peak, price) fade_grace_active = state_in.get("fade_grace_active", 0.0) grace_extensions_used = state_in.get("grace_extensions_used", 0) real_stop = peak_price - cfg["base_giveback_cents"] effective_floor = real_stop - fade_grace_active metrics = {"price": price, "peak_price": round(peak_price, 4), "real_stop": round(real_stop, 4), "effective_floor": round(effective_floor, 4), "fade_grace_active": round(fade_grace_active, 4), "grace_extensions_used": grace_extensions_used} if price > real_stop: # Recovered above the real (un-widened) trailing stop -- grace # resets outright, same as the original project's contract, so a # later, unrelated dip always gets a fresh grace budget. return CentsGraceDecision( symbol=symbol, state=STATE_HOLD, reason=f"above trailing stop ${real_stop:.4f}" + (" -- grace reset" if fade_grace_active > 0 else ""), metrics=metrics, state_out={"peak_price": peak_price, "fade_grace_active": 0.0, "grace_extensions_used": 0}) if price > effective_floor: # Below the real stop but still inside previously-granted grace # room -- hold without re-checking health every poll. return CentsGraceDecision( symbol=symbol, state=STATE_HOLD, reason=f"below stop ${real_stop:.4f} but within granted grace floor ${effective_floor:.4f}", metrics=metrics, state_out={"peak_price": peak_price, "fade_grace_active": fade_grace_active, "grace_extensions_used": grace_extensions_used}) if grace_extensions_used >= cfg["max_grace_extensions"]: return CentsGraceDecision( symbol=symbol, state=STATE_EXIT, should_exit=True, reason=(f"floor ${effective_floor:.4f} touched, grace extensions exhausted " f"({grace_extensions_used}/{cfg['max_grace_extensions']})"), metrics=metrics, state_out={"peak_price": peak_price, "fade_grace_active": 0.0, "grace_extensions_used": 0}) # Floor touched, grace budget remains -- check whether this bot's own # deterioration vote (exit.py's Layer 3 signals) actually confirms # real fading, or whether this looks like ordinary noise. vwap_value = vwap(bars) vslope = vwap_slope(bars) a = calc_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["min_flat_threshold_pct"], atr_pct * cfg["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))) signals = [price < vwap_value, vslope < 0, trend == "negative", lhl] agreeing = sum(signals) metrics.update({"vwap": round(vwap_value, 4), "vwap_slope": round(vslope, 4), "trend": trend, "lower_highs_lower_lows": lhl, "agreeing": agreeing}) if agreeing >= cfg["deterioration_min_signals"]: return CentsGraceDecision( symbol=symbol, state=STATE_EXIT, should_exit=True, reason=(f"floor ${effective_floor:.4f} touched, health confirms fading " f"({agreeing}/4 deterioration signals)"), metrics=metrics, state_out={"peak_price": peak_price, "fade_grace_active": 0.0, "grace_extensions_used": 0}) new_grace = fade_grace_active + cfg["grace_cents"] new_used = grace_extensions_used + 1 return CentsGraceDecision( symbol=symbol, state=STATE_HOLD, reason=(f"floor ${effective_floor:.4f} touched but still healthy ({agreeing}/4 signals) -- " f"granting ${cfg['grace_cents']:.2f} grace ({new_used}/{cfg['max_grace_extensions']}), " f"new floor ${real_stop - new_grace:.4f}"), metrics=metrics, state_out={"peak_price": peak_price, "fade_grace_active": new_grace, "grace_extensions_used": new_used})