From 5287c8ebfed4cd35964fb9a1f24c7945a79e0dc6 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Wed, 22 Jul 2026 16:13:26 -0500 Subject: [PATCH] =?UTF-8?q?feat:=20live-money=20prep=20=E2=80=94=20equity-?= =?UTF-8?q?floor=20kill=20switch=20+=20boot=20order=20reconciliation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TRADING_EQUITY_FLOOR (0=off): equity at/below an absolute dollar floor permanently halts the session and flattens the book — the total-drawdown backstop the per-day breaker can't provide (a slow bleed of -2.9% days never trips it). Same flatten path, loud risk event, bounce does not re-open. TRADING_BOOT_RECONCILE (default on): at boot, diff the broker's open orders against today's submitted events — an open BUY on an unheld symbol this process never submitted is a crash-restart orphan (the 2026-07-06 duplicate-order incident) and is cancelled. Orders on held symbols are never touched (cancelling a bracket parent strips its protective stop). Co-Authored-By: Claude Fable 5 --- tests/unit/test_trading_execution.py | 96 ++++++++++++++++++++++++++++ trading_execution.py | 78 ++++++++++++++++++++++ 2 files changed, 174 insertions(+) diff --git a/tests/unit/test_trading_execution.py b/tests/unit/test_trading_execution.py index e91de83..d5c7d7d 100644 --- a/tests/unit/test_trading_execution.py +++ b/tests/unit/test_trading_execution.py @@ -972,3 +972,99 @@ def test_daily_state_rebuilds_from_events_on_restart(store): assert svc._fresh_entries["n"] == 1, "cap count must survive restart" assert "AAA" in svc._requested_symbols, "resting-order dedup must survive restart" assert "AAA" in svc._exited_today, "losing bench must survive restart" + + +def test_equity_floor_kill_switch_halts_below_floor(store): + """Equity at/below TRADING_EQUITY_FLOOR must permanently halt entries — the + total-drawdown backstop the per-day breaker can't provide (slow bleed).""" + from alpaca_paper.execution import AlpacaPaperExecutor + + class _Cli: + equity = "84000" + def get_positions(self, fresh=False): return [] + def get_account(self): return {"equity": self.equity, "last_equity": "85000"} + def submit_order(self, **kw): return {"id": "o", "status": "new", "filled_qty": "0"} + def cancel_order(self, order_id): pass + def get_orders(self, **kw): return [] + + cli = _Cli() + svc = TradingExecutionService( + store, executor=AlpacaPaperExecutor(store, client=cli), + settings=ExecutionSettings(auto_approve=True, equity_floor=85_000.0, + max_daily_loss_pct=0.5), + session_id="t", + ) + assert svc._daily_loss_breach() is True # 84k <= 85k floor -> halted + assert svc._halted is True # permanent for the session + ev = [e for e in store.query_events(event_type="risk_rule_triggered", limit=None) + if "EQUITY FLOOR" in (e.get("message") or "")] + assert ev, "kill switch must emit a risk event" + cli.equity = "90000" # bounce does NOT re-open + assert svc._daily_loss_breach() is True + + +def test_equity_floor_off_by_default(store): + from alpaca_paper.execution import AlpacaPaperExecutor + + class _Cli: + def get_positions(self, fresh=False): return [] + def get_account(self): return {"equity": "50", "last_equity": "50"} + def get_orders(self, **kw): return [] + + svc = TradingExecutionService( + store, executor=AlpacaPaperExecutor(store, client=_Cli()), + settings=ExecutionSettings(auto_approve=True, max_daily_loss_pct=0.5), + session_id="t", + ) + assert svc._daily_loss_breach() is False # floor 0 = off + + +def test_boot_reconcile_cancels_orphan_buy_leaves_held_alone(store): + """Crash-restart backstop: an open BUY on an unheld symbol this process never + submitted is cancelled; anything on a HELD symbol is untouched (bracket legs).""" + from alpaca_paper.execution import AlpacaPaperExecutor + + class _Cli: + def __init__(self): + self.cancelled = [] + self.orders = [ + {"id": "orph-1", "symbol": "GHOST", "side": "buy"}, # orphan -> cancel + {"id": "held-1", "symbol": "KEEP", "side": "buy"}, # held -> leave + {"id": "sell-1", "symbol": "GHOST", "side": "sell"}, # sell leg -> leave + ] + def get_orders(self, **kw): return list(self.orders) + def get_positions(self, fresh=False): return [{"symbol": "KEEP", "qty": "10"}] + def get_account(self): return {"equity": "100000", "last_equity": "100000"} + def cancel_order(self, oid): self.cancelled.append(oid) + + cli = _Cli() + TradingExecutionService( + store, executor=AlpacaPaperExecutor(store, client=cli), + settings=ExecutionSettings(auto_approve=True, boot_reconcile=True, + max_daily_loss_pct=0.5), + session_id="t", + ) + assert cli.cancelled == ["orph-1"] + ev = [e for e in store.query_events(event_type="risk_rule_triggered", limit=None) + if "Boot reconcile" in (e.get("message") or "")] + assert ev and "GHOST" in ev[0]["message"] + + +def test_boot_reconcile_off_by_flag(store): + from alpaca_paper.execution import AlpacaPaperExecutor + + class _Cli: + def __init__(self): self.cancelled = [] + def get_orders(self, **kw): return [{"id": "x", "symbol": "GHOST", "side": "buy"}] + def get_positions(self, fresh=False): return [] + def get_account(self): return {"equity": "1", "last_equity": "1"} + def cancel_order(self, oid): self.cancelled.append(oid) + + cli = _Cli() + TradingExecutionService( + store, executor=AlpacaPaperExecutor(store, client=cli), + settings=ExecutionSettings(auto_approve=True, boot_reconcile=False, + max_daily_loss_pct=0.5), + session_id="t", + ) + assert cli.cancelled == [] diff --git a/trading_execution.py b/trading_execution.py index f471d91..ee574d9 100644 --- a/trading_execution.py +++ b/trading_execution.py @@ -151,6 +151,16 @@ class ExecutionSettings: # halt-resume gap-through can't be chased. entry_fill_model: str = "resting" max_daily_loss_pct: float = 0.03 + # ACCOUNT KILL SWITCH: absolute equity floor in dollars. Equity at/below it + # -> permanent halt + flatten, same path as the daily breaker but anchored to + # TOTAL drawdown, not one day (a slow bleed of -2.9% days never trips the + # daily breaker). Operator-set intent: e.g. 0.90 x starting equity. 0 = off. + equity_floor: float = 0.0 + # boot reconciliation: cancel orphan resting BUY entries at broker that this + # process doesn't know about (crash-restart duplicated a resting order, + # 2026-07-06). Never touches orders on HELD symbols (bracket legs protect + # the position; cancelling a parent strips its stop — the naked-stop bug). + boot_reconcile: bool = True # on a daily-loss breach, also flatten open positions + cancel unfilled entries flatten_on_breach: bool = True # --- entry mechanism (Ross-Cameron-style; all tunable) ----------------- @@ -285,6 +295,8 @@ class ExecutionSettings: risk_per_trade_pct=float(values.get("TRADING_RISK_PER_TRADE_PCT", "0.01")), default_equity=float(values.get("TRADING_DEFAULT_EQUITY", "100000")), max_daily_loss_pct=float(values.get("TRADING_MAX_DAILY_LOSS_PCT", "0.03")), + equity_floor=float(values.get("TRADING_EQUITY_FLOOR", "0")), + boot_reconcile=flag("TRADING_BOOT_RECONCILE", "1"), flatten_on_breach=flag("TRADING_FLATTEN_ON_BREACH", "1"), reward_multiple=float(values.get("TRADING_REWARD_MULTIPLE", "2.0")), entry_order_type=order_type, @@ -479,6 +491,8 @@ class TradingExecutionService: # forget which symbols already have a resting order (re-fire => duplicate # order, the 2026-07-06 incident). Best-effort; a fresh DB is fine. self._rebuild_daily_state() + if self.settings.boot_reconcile: + self.reconcile_broker_orders() def _rebuild_daily_state(self) -> None: """Restore per-day counters/benches from today's event stream on boot. @@ -764,6 +778,23 @@ class TradingExecutionService: return self._data_halt if baseline <= 0: return False + floor = float(self.settings.equity_floor or 0.0) + if floor > 0 and equity <= floor: + self._halted = True + flat = (self._flatten_all("equity_floor_kill_switch") + if self.settings.flatten_on_breach + else {"closed_positions": [], "cancelled_entries": 0}) + self.store.emit(RiskRuleTriggeredEvent( + timestamp=datetime.now(), mode=self.mode, + correlation_id=self.session_id, + message=(f"EQUITY FLOOR kill switch: equity {equity:,.2f} <= floor " + f"{floor:,.2f} — trading halted for the session, book flattened " + f"({len(flat['closed_positions'])} closed, " + f"{flat['cancelled_entries']} cancelled)"), + rule_type="equity_floor", rule_value=floor, + current_state={"equity": equity, "floor": floor}, + action_taken="halted_and_flattened")) + return True pnl_pct = (equity - baseline) / baseline if pnl_pct <= -max_loss: self._halted = True @@ -1149,6 +1180,53 @@ class TradingExecutionService: symbol=ctx.symbol, source=ctx.source, passed=decision.passed, reject_reason=decision.reason, gates=decision.gates)) + def reconcile_broker_orders(self) -> dict: + """Boot backstop for the 2026-07-06 duplicate-order incident: diff the + broker's OPEN orders against this process's knowledge. An open BUY parent + on a symbol we do NOT hold and did NOT submit today is an orphan from a + dead process — cancel it (nothing held = nothing to protect). Orders on + held symbols are always left alone (bracket stop/TP legs). Best-effort: + a broker error logs and returns; it never blocks boot.""" + client = getattr(self.executor, "client", None) + if client is None or not hasattr(client, "get_orders"): + return {"ok": False, "reason": "no broker client"} + try: + open_orders = client.get_orders(status="open", limit=500, nested=True) or [] + held = {str(p.get("symbol")) for p in (client.get_positions() or [])} + except Exception as exc: # noqa: BLE001 + logger.warning("boot reconcile: broker read failed: %s", exc) + return {"ok": False, "reason": str(exc)} + known = set() + for e in self.store.query_events(event_type="order_submitted", + since=self._now().date().isoformat(), limit=500): + raw = e.get("payload") or e.get("payload_json") + try: + p = json.loads(raw) if isinstance(raw, str) else (raw or {}) + except ValueError: + p = {} + if p.get("broker_order_id"): + known.add(str(p["broker_order_id"])) + cancelled = [] + for o in open_orders: + sym = str(o.get("symbol") or "") + oid = str(o.get("id") or "") + side = str(o.get("side") or "").lower() + if side != "buy" or not sym or sym in held or oid in known: + continue + try: + client.cancel_order(oid) + cancelled.append(f"{sym}:{oid[:8]}") + except Exception as exc: # noqa: BLE001 + logger.warning("boot reconcile: cancel %s %s failed: %s", sym, oid, exc) + if cancelled: + self.store.emit(RiskRuleTriggeredEvent( + timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, + message=(f"Boot reconcile: cancelled {len(cancelled)} orphan resting " + f"BUY order(s) unknown to this process: {', '.join(cancelled)}"), + rule_type="boot_reconcile", rule_value=float(len(cancelled)), + current_state={"cancelled": cancelled}, action_taken="cancelled_orphans")) + return {"ok": True, "cancelled": cancelled, "open_orders": len(open_orders)} + def set_operator_paused(self, paused: bool) -> None: """Operator PAUSE/RESUME (command bus). Blocks new entries on both paths.""" self._operator_paused = bool(paused)