diff --git a/runtime/exit_manager.py b/runtime/exit_manager.py index 74b6398..bcd2a99 100644 --- a/runtime/exit_manager.py +++ b/runtime/exit_manager.py @@ -20,7 +20,7 @@ import pandas as pd from runtime.flatten import cancel_protective_and_close from storage.event_schema import EventMode, RiskRuleTriggeredEvent -from strategy.exits import ExitConfig, TRAIL_NONE, manage_live +from strategy.exits import ExitConfig, TRAIL_NONE, catastrophe_triggered, manage_live logger = logging.getLogger(__name__) @@ -38,7 +38,7 @@ def _is_active(cfg: ExitConfig) -> bool: return bool(cfg and (cfg.breakeven_at_r or cfg.breakeven_at_pct or cfg.trail_mode != TRAIL_NONE or cfg.scale_out_r or cfg.first_red_exit - or cfg.profit_lock_tiers)) + or cfg.profit_lock_tiers or cfg.catastrophe_pct)) class LiveExitManager: @@ -146,13 +146,36 @@ class LiveExitManager: actions: list[str] = [] for sym, p in held.items(): + entry = float(p.get("avg_entry_price") or 0.0) + cur = float(p.get("current_price") or 0.0) + # CATASTROPHE STOP — the hard backstop, FIRST and unconditional. Runs on + # EVERY held position, before the bars fetch and before the no-stop skip + # below that otherwise leaves a NAKED position (no working bracket) with + # zero exit management — the exact NIVF case that ran to -23% and was + # ~the entire net loss. Doesn't depend on bars or a live stop. + if (self.cfg.catastrophe_pct and entry > 0 and cur > 0 + and str(p.get("side") or "long").lower() != "short"): + _leg0, cata_stop = self._stop_leg(sym, orders) + if catastrophe_triggered(entry, cur, cata_stop, + self.cfg.catastrophe_pct, + self.cfg.catastrophe_risk_mult): + loss_pct = (entry - cur) / entry * 100 + try: + self._flatten(sym, orders) + self._tracked.pop(sym, None) + self._emit(sym, f"CATASTROPHE exit {sym} @~{cur:.4f} " + f"({loss_pct:.1f}% below entry)", "exit_catastrophe", + {"price": cur, "entry": entry, "stop": cata_stop}) + actions.append(f"{sym} CATASTROPHE-exit @{cur:.2f} (-{loss_pct:.0f}%)") + except Exception as exc: # noqa: BLE001 + logger.warning("catastrophe exit failed %s: %s", sym, exc) + continue try: bars = self.bars_fn(sym) except Exception: # noqa: BLE001 continue if bars is None or len(bars) == 0: continue - entry = float(p.get("avg_entry_price") or 0.0) if entry <= 0: continue if sym not in self._tracked: diff --git a/strategy/exits.py b/strategy/exits.py index 8e6809d..4b767fa 100644 --- a/strategy/exits.py +++ b/strategy/exits.py @@ -70,6 +70,13 @@ class ExitConfig: # trade has been up gain%, the stop never sits below entry*(1+lock%) — so a # winner can only give back down to a locked-in minimum gain. profit_lock_tiers: list = field(default_factory=list) + # CATASTROPHE STOP — a hard market-exit backstop that does NOT depend on the + # broker bracket stop being live. Fires when price is >= catastrophe_pct below + # entry, OR (when a stop is known) >= catastrophe_risk_mult x the intended 1R + # underwater. Catches a missing/failed protective stop (NIVF ran to -23% naked, + # = the entire net loss). 0 = off. + catastrophe_pct: float = 0.10 + catastrophe_risk_mult: float = 1.5 @classmethod def from_env(cls, env: dict | None = None) -> "ExitConfig": @@ -97,6 +104,8 @@ class ExitConfig: scale_out_pct=f("TRADING_EXIT_SCALE_OUT_PCT", "0.5"), first_red_exit=b("TRADING_EXIT_FIRST_RED", "0"), profit_lock_tiers=parse_profit_tiers(v.get("TRADING_EXIT_PROFIT_TIERS", "")), + catastrophe_pct=f("TRADING_CATASTROPHE_STOP_PCT", "0.10"), + catastrophe_risk_mult=f("TRADING_CATASTROPHE_RISK_MULT", "1.5"), ) def describe(self) -> str: @@ -115,9 +124,30 @@ class ExitConfig: if self.profit_lock_tiers: parts.append("lock " + "/".join( f"+{g * 100:g}%->{lk * 100:g}%" for g, lk in self.profit_lock_tiers)) + if self.catastrophe_pct: + parts.append(f"catastrophe@-{self.catastrophe_pct:.0%}/{self.catastrophe_risk_mult:g}xR") return ", ".join(parts) +def catastrophe_triggered(entry: float, current: float, stop: float | None, + pct: float, risk_mult: float) -> bool: + """True if a LONG position is catastrophically underwater and must be + market-exited NOW regardless of the broker bracket — the hard backstop for a + missing or non-firing protective stop. Fires when price is >= ``pct`` below + entry, OR (when a real stop is known) >= ``risk_mult`` x the intended 1R + risk underwater. ``pct`` <= 0 disables the percentage arm.""" + if not entry or not current or current <= 0 or entry <= 0: + return False + loss_frac = (entry - current) / entry # >0 = underwater (long) + if pct and pct > 0 and loss_frac >= pct: + return True + if stop is not None and stop < entry and risk_mult and risk_mult > 0: + risk = entry - stop + if risk > 0 and current <= entry - risk_mult * risk: + return True + return False + + @dataclass class ExitFill: frac: float # fraction of the original position closed by this fill diff --git a/tests/unit/test_exits.py b/tests/unit/test_exits.py index 038c466..b6b4576 100644 --- a/tests/unit/test_exits.py +++ b/tests/unit/test_exits.py @@ -10,6 +10,7 @@ from strategy.exits import ( ExitConfig, TRAIL_PCT, TRAIL_PRIOR_LOW, + catastrophe_triggered, manage_live, parse_profit_tiers, simulate_exit, @@ -21,6 +22,28 @@ def test_parse_profit_tiers(): assert parse_profit_tiers("") == [] +def test_catastrophe_pct_arm(): + # 10% pct floor, no known stop -> the only thing protecting a naked position + assert catastrophe_triggered(1.00, 0.89, None, 0.10, 1.5) is True # -11%, fires + assert catastrophe_triggered(1.00, 0.92, None, 0.10, 1.5) is False # -8%, safe + # NIVF: entry 0.9936 ran to 0.76 (-23.4%) with NO stop -> would have fired + assert catastrophe_triggered(0.9936, 0.7608, None, 0.10, 1.5) is True + + +def test_catastrophe_risk_arm_fires_before_pct(): + # tight 5% stop: -1.5R = -7.5% < the 10% pct floor, so the risk arm catches it sooner + assert catastrophe_triggered(1.00, 0.92, 0.95, 0.10, 1.5) is True # past -1.5R (-1.6R) + assert catastrophe_triggered(1.00, 0.93, 0.95, 0.10, 1.5) is False # -1.4R, not yet + # a stop at/above entry is not a usable risk reference -> only the pct arm applies + assert catastrophe_triggered(1.00, 0.94, 1.00, 0.10, 1.5) is False + + +def test_catastrophe_off_and_in_profit(): + assert catastrophe_triggered(1.00, 0.50, None, 0.0, 1.5) is False # pct arm disabled, no stop + assert catastrophe_triggered(1.00, 1.20, 0.95, 0.10, 1.5) is False # in profit + assert catastrophe_triggered(1.00, 0.0, None, 0.10, 1.5) is False # bad price -> no-op + + def test_profit_lock_keeps_minimum_gain(): # entry 10, stop 9. price runs to +12% (11.2) then fades; +8%->lock+3% tier # pins the stop to 10.30, so the fade exits in profit instead of at the stop.