feat: live-money prep — equity-floor kill switch + boot order reconciliation

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 <noreply@anthropic.com>

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 == []

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)