Review sweep: money-path fixes + security + delete 805MB dead weight

Consolidated 4-tool review (2 correctness agents + Codex + over-eng audit).

Money-path (regression-tested):
- _flatten_all: moved the fresh-broker close from close_session INTO the
  shared function + added a broker-held guard on entry cancellation, so
  the daily-loss CIRCUIT BREAKER (which called _flatten_all raw) no
  longer strips a just-filled bracket's legs or leaves a sync-gap fill
  naked. One guard, all callers.
- restart state rebuild: _fresh_entries cap, _requested_symbols dedup,
  and losing benches now rebuild from today's events on boot — a
  mid-session restart no longer resets the cap to 0 or re-fires a
  resting order into a duplicate (the 2026-07-06 incident).
- POST /orders no longer retries a lost-response URLError (was
  re-sending duplicate live orders); GET/DELETE still retry.
- expire_stale_entries now @_locked (raced the trigger thread).
- _session_open: ET conversion instead of a hardcoded 13:30-13:33Z
  window that only matched EDT — the fast path armed nothing all winter.

Security:
- admin panel fail-CLOSED on empty ADMIN_TOKEN (was fail-open on a panel
  fronting the docker socket); header-only token, constant-time compare.
- admin API responses redact secret override VALUES (live keys would
  have echoed after a paper->live switch).
- dashboard: removed wildcard CORS (cross-origin drive-by could drive
  trade actions); optional DASHBOARD_TOKEN gate on POST.

Deletions (~805MB + dead code shipped in the prod image): momentum-
rebuilt.zip, archive venv/htmlcov, and schwab/ core/ telemetry/ models/
runtime/providers/ storage/replay.py controlroom/ + three dead
evaluation stubs + package.json — all verified zero-importer.

+4 regression tests. 99 money-path/trigger/exit tests green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

admin_api.py

@@ -16,6 +16,7 @@ Run: python admin_api.py   (ADMIN_HOST/ADMIN_PORT, default 0.0.0.0:8767)
 
 from __future__ import annotations
 
+import hmac
 import http.client
 import json
 import os
@@ -70,6 +71,14 @@ SEED_PROFILES = {
 }
 
 
+def _safe_overrides() -> dict:
+    """Overrides for API responses with secret VALUES redacted — a /api/mode live
+    switch writes ALPACA_LIVE_* into overrides.env; without this they'd echo back
+    in every /api/status and /api/config response."""
+    return {k: ("***" if any(m in k for m in SECRET_MARKERS) else v)
+            for k, v in env_overrides.read_overrides().items()}
+
+
 def _editable(key: str) -> bool:
     if key in EDITABLE_EXACT:
         return True
@@ -194,10 +203,12 @@ class Handler(BaseHTTPRequestHandler):
         self.wfile.write(raw)
 
     def _authed(self) -> bool:
+        # FAIL CLOSED: no token configured => deny (this panel fronts the docker
+        # socket + the paper/live switch; an unset env var must not open it to the
+        # LAN). Header only — a ?token= query would leak into proxy/access logs.
         if not ADMIN_TOKEN:
-            return True  # explicit opt-out: no token configured
+            return False
-        return (self.headers.get("X-Admin-Token") == ADMIN_TOKEN
+        return hmac.compare_digest(self.headers.get("X-Admin-Token", ""), ADMIN_TOKEN)
-                or f"token={ADMIN_TOKEN}" in (self.path.split("?", 1) + [""])[1])
 
     def do_GET(self):  # noqa: N802
         path = self.path.split("?")[0]
@@ -208,7 +219,7 @@ class Handler(BaseHTTPRequestHandler):
             return self._send(401, {"error": "bad or missing X-Admin-Token"})
         if path == "/api/status":
             return self._send(200, {"containers": list_containers(), "checks": checks(),
-                                    "overrides": env_overrides.read_overrides()})
+                                    "overrides": _safe_overrides()})
         if path == "/api/config":
             eff = {k: v for k, v in sorted(os.environ.items()) if _editable(k)}
             secrets = {k: bool(os.environ.get(k)) for k in
@@ -216,7 +227,7 @@ class Handler(BaseHTTPRequestHandler):
                         "ALPACA_LIVE_SECRET_KEY", "TELEGRAM_BOT_TOKEN", "YOUTUBE_API_KEY",
                         "FMP_API_KEY")}
             return self._send(200, {"effective": eff, "secrets_present": secrets,
-                                    "overrides": env_overrides.read_overrides()})
+                                    "overrides": _safe_overrides()})
         if path == "/api/profiles":
             return self._send(200, profiles())
         return self._send(404, {"error": "unknown route"})

alpaca_paper/client.py

@@ -137,8 +137,14 @@ class AlpacaPaperClient:
                     continue
                 raise AlpacaApiError(exc.code, exc.read().decode(errors="replace")) from exc
             except urllib.error.URLError:
-                # transient network/DNS (e.g. gaierror) — retry a couple times
+                # transient network/DNS (e.g. gaierror) — retry a couple times.
-                if attempt >= self.max_retries:
+                # NOT for POST: urllib wraps a lost RESPONSE (read timeout/reset)
+                # in URLError too, so attempt 1 may have SUCCEEDED at the broker
+                # with the reply lost; re-sending POST /orders then rests a
+                # duplicate live order (or 422s and looks like a failure while a
+                # real bracket rests untracked). Idempotent GET/DELETE retry; POST
+                # raises so the caller reconciles against broker positions.
+                if method == "POST" or attempt >= self.max_retries:
                     raise
                 _time.sleep(0.4 * (attempt + 1))
         raise RuntimeError("request retries exhausted")  # unreachable

api/main.py

@@ -297,7 +297,6 @@ class DashboardHandler(BaseHTTPRequestHandler):
             self.send_response(status)
             self.send_header("Content-Type", "application/json")
             self.send_header("Content-Length", str(len(body)))
-            self.send_header("Access-Control-Allow-Origin", "*")
             self.end_headers()
             self.wfile.write(body)
         except (BrokenPipeError, ConnectionResetError):
@@ -346,7 +345,6 @@ class DashboardHandler(BaseHTTPRequestHandler):
         self.send_header("Content-Type", "text/event-stream")
         self.send_header("Cache-Control", "no-cache")
         self.send_header("Connection", "keep-alive")
-        self.send_header("Access-Control-Allow-Origin", "*")
         self.end_headers()
 
         shutdown = getattr(type(self), "shutdown_event", None)
@@ -410,6 +408,13 @@ class DashboardHandler(BaseHTTPRequestHandler):
             self._send_json({"error": str(exc)}, 500)
 
     def do_POST(self):  # noqa: N802
+        # optional shared-secret gate on trade actions (approve/reject/exit/watch).
+        # Off by default (autonomous stack auto-approves); set DASHBOARD_TOKEN to
+        # require X-Dashboard-Token on every POST. CORS wildcard was removed so a
+        # cross-origin LAN page can't drive these either way.
+        _tok = os.environ.get("DASHBOARD_TOKEN")
+        if _tok and not hmac.compare_digest(self.headers.get("X-Dashboard-Token", ""), _tok):
+            return self._send_json({"error": "bad or missing X-Dashboard-Token"}, 401)
         try:
             route = urlparse(self.path).path
             body = self._read_body()
@@ -443,7 +448,6 @@ class DashboardHandler(BaseHTTPRequestHandler):
 
     def do_OPTIONS(self):  # noqa: N802
         self.send_response(204)
-        self.send_header("Access-Control-Allow-Origin", "*")
         self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
         self.send_header("Access-Control-Allow-Headers", "Content-Type")
         self.end_headers()

controlroom/app.py

@@ -1,399 +0,0 @@
-"""Momentum Trading Control Room — NiceGUI live-analysis dashboard (Milestone 5, core).
-
-Phases 1+2 of the staged control-room plan, adapted to THIS backend: a real-time
-operator view with LIVE TICKERS (SIP last-trade prices at 1s), per-symbol analysis
-(state, gap, RVOL, trigger distance, quality, population-gate check), live positions
-P&L, today's gate activity + attribution, and the approval queue with confirm-flow
-actions. Runs as its OWN service on :8766 — the existing dashboard on :8765 stays
-untouched until this reaches parity (their Phase 3-5: manual orders, settings dials,
-audit trail).
-
-Data flow: read-only projections from the shared Postgres event store + Alpaca SIP
-for live prices; fully AUTONOMOUS ops (owner directive: no approvals/interaction needed) — the only
-action is the emergency panic brake (cancel all open orders, typed confirmation)."""
-
-from __future__ import annotations
-
-import json
-import os
-from datetime import date, datetime
-
-from dotenv import load_dotenv
-load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))
-
-from nicegui import run, ui  # noqa: E402
-
-from alpaca_paper.client import AlpacaPaperClient  # noqa: E402
-from alpaca_paper.settings import AlpacaPaperSettings  # noqa: E402
-from momentum_cli import _market_clock, _triggers_snapshot, _unprotected_positions  # noqa: E402
-from storage.db_pg import get_connection  # noqa: E402
-from storage.event_store import EventStore  # noqa: E402
-from storage.projections import query_approval_queue, query_ready_signals_snapshot  # noqa: E402
-
-POP_GAP, POP_VOL = 0.10, 300_000        # display mirror of the live population gate
-
-import threading
-con = get_connection("momentum")
-store = EventStore("momentum")
-client = AlpacaPaperClient(AlpacaPaperSettings.from_env())
-DB_LOCK = threading.Lock()   # fetchers run in io_bound THREADS; one psycopg2 conn is not thread-safe
-
-
-# ---------------- data fetchers (sync; called via run.io_bound) ----------------
-def fetch_board() -> dict:
-    with DB_LOCK:
-        trig_rows, snap_ts, armed_n = _triggers_snapshot(con)
-    signals = {s["symbol"]: s for s in query_ready_signals_snapshot(store)}
-    rows = []
-    for r in trig_rows:
-        sym = r.get("symbol")
-        sig = signals.get(sym, {})
-        gap = r.get("gap")                      # percent units from the book
-        gap_frac = (gap / 100.0) if isinstance(gap, (int, float)) else sig.get("gap_pct")
-        vol = sig.get("cum_volume")
-        pop_ok = None
-        if gap_frac is not None:
-            pop_ok = bool(gap_frac >= POP_GAP and (vol is None or vol >= POP_VOL))
-        rows.append({
-            "symbol": sym, "state": r.get("state"), "price": r.get("price"),
-            "gap": gap, "rvol": r.get("rvol"), "trigger": r.get("trigger"),
-            "to_trig": r.get("→trigger") or r.get("to_trigger"), "stop": r.get("stop"),
-            "quality": sig.get("quality_score"), "vwap": sig.get("vwap"),
-            "pop": ("✓" if pop_ok else "✗" if pop_ok is not None else "—"),
-            "catalyst": (r.get("catalyst") or "")[:40],
-        })
-    return {"rows": rows, "ts": str(snap_ts)[11:19] if snap_ts else "—", "armed": armed_n}
-
-
-def fetch_prices(symbols: list[str]) -> dict[str, float]:
-    if not symbols:
-        return {}
-    try:
-        trades = client.get_latest_trades(symbols)
-        return {s: float(t.get("p")) for s, t in (trades or {}).items() if t and t.get("p")}
-    except Exception:  # noqa: BLE001
-        return {}
-
-
-def fetch_account() -> dict:
-    try:
-        a = client.get_account()
-        eq = float(a.get("equity") or 0)
-        last = float(a.get("last_equity") or 0)
-        return {"equity": eq, "day_pnl": (eq - last) if last else 0.0}
-    except Exception:  # noqa: BLE001
-        return {}
-
-
-def fetch_positions() -> list[dict]:
-    try:
-        pos = client.get_positions(fresh=True)
-        with DB_LOCK:
-            naked = {n["symbol"] for n in (_unprotected_positions(con) or [])}
-        return [{
-            "symbol": p.get("symbol"), "qty": p.get("qty"),
-            "avg": float(p.get("avg_entry_price") or 0),
-            "price": float(p.get("current_price") or 0),
-            "pnl": float(p.get("unrealized_pl") or 0),
-            "protected": "⚠ NAKED" if p.get("symbol") in naked else "✓",
-        } for p in pos]
-    except Exception:  # noqa: BLE001
-        return []
-
-
-def fetch_gates() -> list[dict]:
-    today = date.today().isoformat()
-    hist: dict[str, int] = {}
-    with DB_LOCK:
-        rows = con.execute(
-            "SELECT payload_json FROM events WHERE event_type='risk_rule_triggered' "
-            "AND timestamp >= ?", [today]).fetchall()
-    for (pj,) in rows:
-        rt = (json.loads(pj or "{}").get("rule_type")) or "?"
-        hist[rt] = hist.get(rt, 0) + 1
-    return [{"gate": k, "cuts": v} for k, v in sorted(hist.items(), key=lambda x: -x[1])]
-
-
-def fetch_attribution() -> list[dict]:
-    from research.attribution import compute_attribution
-    att = compute_attribution(days=7)
-    out = []
-    for d in sorted(att["days"], reverse=True):
-        st = att["days"][d]
-        c, r = st.get("counts", {}), st.get("rating", {})
-        out.append({"day": d, "disc": c.get("symbol_discovered", 0),
-                    "ready": c.get("signal_ready", 0),
-                    "early": f"{r.get('signals_early_pct', '—')}%",
-                    "trades": st.get("outcomes", {}).get("trades", 0),
-                    "win": f"{r.get('win_rate_pct', '—')}%",
-                    "pnl": r.get("pnl")})
-    return out
-
-
-def fetch_activity(limit: int = 40) -> list[dict]:
-    """Merged live event feed for autonomous ops: entries, exits, fills, gate cuts."""
-    kinds = ("position_closed", "order_submitted", "order_cancelled", "signal_ready",
-             "risk_rule_triggered", "order_filled")
-    ph = ",".join(["?"] * len(kinds))
-    with DB_LOCK:
-        rows = con.execute(
-            f"SELECT timestamp, event_type, message FROM events "
-            f"WHERE event_type IN ({ph}) ORDER BY timestamp DESC LIMIT ?",
-            [*kinds, limit]).fetchall()
-    return [{"time": str(ts)[11:19], "type": et, "message": (m or "")[:140]}
-            for ts, et, m in rows]
-
-
-def fetch_summary() -> dict:
-    today = date.today().isoformat()
-    with DB_LOCK:
-        closes = con.execute(
-            "SELECT payload_json FROM events WHERE event_type='position_closed' "
-            "AND timestamp >= ?", [today]).fetchall()
-        subs = con.execute(
-            "SELECT count(*) FROM events WHERE event_type='order_submitted' "
-            "AND timestamp >= ?", [today]).fetchone()[0]
-        cuts = con.execute(
-            "SELECT count(*) FROM events WHERE event_type='risk_rule_triggered' "
-            "AND timestamp >= ?", [today]).fetchone()[0]
-        hb = con.execute(
-            "SELECT max(timestamp) FROM events WHERE event_type='module_tick'").fetchone()[0]
-    pnls = []
-    for (pj,) in closes:
-        try:
-            pnls.append(float(json.loads(pj or "{}").get("realized_pnl") or 0))
-        except Exception:  # noqa: BLE001
-            pass
-    age = (datetime.now() - hb).total_seconds() if hb else None
-    return {"trades": len(pnls), "wins": sum(1 for x in pnls if x > 0),
-            "realized": round(sum(pnls), 2), "submissions": subs, "cuts": cuts,
-            "loop_age": age}
-
-
-def cancel_all_orders() -> str:
-    return str(client._trading("DELETE", "/orders"))[:200]
-
-
-# ---------------- UI ----------------
-@ui.page("/")
-def main_page():
-    ui.add_head_html("""
-    <style>
-      body { background: #0d1520; }
-      .font-mono { font-family: ui-monospace, Menlo, monospace; }
-      .text-green { color: #22C55E !important; font-weight: bold; }
-      .text-red { color: #EF4444 !important; font-weight: bold; }
-      @keyframes flash-green { 0% { background:#052e16; } 100% { background:transparent; } }
-      @keyframes flash-red { 0% { background:#450a0a; } 100% { background:transparent; } }
-      .flash-green { animation: flash-green 1s ease-out; }
-      .flash-red { animation: flash-red 1s ease-out; }
-    </style>""")
-    prices: dict[str, float] = {}
-    board_syms: list[str] = []
-
-    with ui.header().classes("items-center justify-between px-4 bg-blue-900 text-white"):
-        with ui.row().classes("items-center gap-3"):
-            ui.label("Momentum Control Room").classes("text-lg font-bold")
-            ui.badge("🔵 PAPER — Alpaca (Simulated)").classes("text-sm px-2")
-        with ui.row().classes("items-center gap-4"):
-            phase_lbl = ui.label("—").classes("text-sm")
-            eq_lbl = ui.label("equity —").classes("font-mono text-sm")
-            hb_lbl = ui.label("♥ —").classes("font-mono text-xs")
-
-            async def panic():
-                with ui.dialog() as d, ui.card().classes("p-6"):
-                    ui.label("⛔ Cancel ALL open orders?").classes("text-lg font-bold")
-                    ui.label('Type "STOP" to confirm:').classes("mt-2")
-                    typed = ui.input("")
-                    with ui.row().classes("justify-end gap-2 mt-4"):
-                        btn = ui.button("CANCEL ALL", on_click=lambda: d.submit(True)) \
-                            .props("color=negative")
-                        btn.bind_enabled_from(typed, "value", backward=lambda v: v == "STOP")
-                        ui.button("Back", on_click=lambda: d.submit(False))
-                if await d:
-                    res = await run.io_bound(cancel_all_orders)
-                    ui.notify(f"cancel-all sent: {res}", type="warning")
-            ui.button("⛔ STOP ALL", on_click=panic).props("color=negative outline") \
-                .classes("font-bold")
-            clock_lbl = ui.label("--:--:--").classes("font-mono")
-
-    with ui.tabs().classes("w-full") as tabs:
-        t_live = ui.tab("Live")
-        t_pos = ui.tab("Positions")
-        t_gates = ui.tab("Gates & Attribution")
-        t_act = ui.tab("Activity")
-
-    with ui.tab_panels(tabs, value=t_live).classes("w-full"):
-        # ---- LIVE: real-time tickers + analysis ----
-        with ui.tab_panel(t_live):
-            with ui.row().classes("gap-4 mb-2"):
-                def _card(title):
-                    with ui.card().classes("p-3 min-w-32"):
-                        ui.label(title).classes("text-xs text-grey-6")
-                        return ui.label("—").classes("text-lg font-bold font-mono")
-                c_pnl = _card("Day P&L")
-                c_trades = _card("Trades today")
-                c_win = _card("Win rate")
-                c_subs = _card("Submissions")
-                c_cuts = _card("Gate cuts")
-                c_loop = _card("Loop ♥")
-
-            async def refresh_summary():
-                sm = await run.io_bound(fetch_summary)
-                acct = await run.io_bound(fetch_account)
-                pnl = acct.get("day_pnl", 0.0)
-                c_pnl.text = f"${pnl:+,.0f}"
-                c_pnl.classes(replace="text-lg font-bold font-mono " +
-                              ("text-green" if pnl >= 0 else "text-red"))
-                c_trades.text = str(sm["trades"])
-                c_win.text = (f"{100*sm['wins']/sm['trades']:.0f}%" if sm["trades"] else "—")
-                c_subs.text = str(sm["submissions"])
-                c_cuts.text = str(sm["cuts"])
-                age = sm.get("loop_age")
-                c_loop.text = f"{age:.0f}s" if age is not None else "—"
-                c_loop.classes(replace="text-lg font-bold font-mono " +
-                               ("text-green" if age is not None and age < 180 else "text-red"))
-            ui.timer(5.0, refresh_summary)
-
-            with ui.row().classes("items-center gap-4"):
-                ui.label("Live tickers").classes("text-lg font-bold")
-                snap_lbl = ui.label("snapshot —").classes("text-xs font-mono text-grey")
-                armed_lbl = ui.label("").classes("text-xs font-mono")
-            live_grid = ui.aggrid({
-                "columnDefs": [
-                    {"headerName": "Sym", "field": "symbol", "pinned": "left", "width": 90},
-                    {"headerName": "State", "field": "state", "width": 95, "cellClassRules": {
-                        "text-green": 'value === "armed" || value === "fired"',
-                    }},
-                    {"headerName": "Price", "field": "price", "width": 95,
-                     "cellClass": "font-mono", "cellClassRules": {
-                         "flash-green": "data._d > 0", "flash-red": "data._d < 0"}},
-                    {"headerName": "Gap%", "field": "gap", "width": 80, "cellClass": "font-mono"},
-                    {"headerName": "RVOL", "field": "rvol", "width": 80, "cellClass": "font-mono"},
-                    {"headerName": "Trigger", "field": "trigger", "width": 95, "cellClass": "font-mono"},
-                    {"headerName": "→Trig", "field": "to_trig", "width": 85, "cellClass": "font-mono"},
-                    {"headerName": "Stop", "field": "stop", "width": 90, "cellClass": "font-mono"},
-                    {"headerName": "Qual", "field": "quality", "width": 75, "cellClassRules": {
-                        "text-green": "value >= 0.5", "text-red": "value < 0.5"}},
-                    {"headerName": "Pop", "field": "pop", "width": 65, "cellClassRules": {
-                        "text-green": 'value === "✓"', "text-red": 'value === "✗"'}},
-                    {"headerName": "VWAP", "field": "vwap", "width": 90, "cellClass": "font-mono"},
-                    {"headerName": "Catalyst", "field": "catalyst", "flex": 1},
-                ],
-                "rowData": [], "domLayout": "autoHeight", "animateRows": True,
-            }).classes("w-full")
-
-            async def refresh_board():
-                data = await run.io_bound(fetch_board)
-                board_syms[:] = [r["symbol"] for r in data["rows"] if r.get("symbol")]
-                for r in data["rows"]:
-                    lp = prices.get(r["symbol"])
-                    if lp:
-                        old = r.get("price")
-                        r["_d"] = (lp - old) if isinstance(old, (int, float)) else 0
-                        r["price"] = lp
-                live_grid.options["rowData"] = data["rows"]
-                live_grid.update()
-                snap_lbl.text = f"loop snapshot {data['ts']}"
-                armed_lbl.text = f"{data['armed']} armed · {len(data['rows'])} tracked"
-
-            async def refresh_prices():
-                if board_syms:
-                    fresh = await run.io_bound(fetch_prices, list(board_syms))
-                    if fresh:
-                        prices.update(fresh)
-                        rows = live_grid.options["rowData"]
-                        for r in rows:
-                            lp = prices.get(r["symbol"])
-                            if lp and isinstance(r.get("price"), (int, float)):
-                                r["_d"] = round(lp - r["price"], 4)
-                                r["price"] = lp
-                        live_grid.update()
-
-            ui.timer(3.0, refresh_board)
-            ui.timer(1.0, refresh_prices)
-
-        # ---- POSITIONS ----
-        with ui.tab_panel(t_pos):
-            pos_grid = ui.aggrid({
-                "columnDefs": [
-                    {"headerName": "Sym", "field": "symbol", "width": 90},
-                    {"headerName": "Qty", "field": "qty", "width": 80, "cellClass": "font-mono"},
-                    {"headerName": "Avg", "field": "avg", "width": 95, "cellClass": "font-mono"},
-                    {"headerName": "Price", "field": "price", "width": 95, "cellClass": "font-mono"},
-                    {"headerName": "uP&L $", "field": "pnl", "width": 110, "cellClass": "font-mono",
-                     "cellClassRules": {"text-green": "value >= 0", "text-red": "value < 0"}},
-                    {"headerName": "Stop", "field": "protected", "width": 110, "cellClassRules": {
-                        "text-red": 'value !== "✓"'}},
-                ], "rowData": [], "domLayout": "autoHeight",
-            }).classes("w-full")
-
-            async def refresh_pos():
-                pos_grid.options["rowData"] = await run.io_bound(fetch_positions)
-                pos_grid.update()
-            ui.timer(2.0, refresh_pos)
-
-        # ---- GATES + ATTRIBUTION ----
-        with ui.tab_panel(t_gates):
-            ui.label("Today's gate activity").classes("text-lg font-bold")
-            gates_grid = ui.aggrid({"columnDefs": [
-                {"headerName": "Gate", "field": "gate", "flex": 1},
-                {"headerName": "Cuts today", "field": "cuts", "width": 120}],
-                "rowData": [], "domLayout": "autoHeight"}).classes("w-1/2")
-            ui.label("Component attribution (7 days)").classes("text-lg font-bold mt-6")
-            att_grid = ui.aggrid({"columnDefs": [
-                {"headerName": "Day", "field": "day", "width": 110},
-                {"headerName": "Disc", "field": "disc", "width": 80},
-                {"headerName": "Ready", "field": "ready", "width": 80},
-                {"headerName": "Early%", "field": "early", "width": 90},
-                {"headerName": "Trades", "field": "trades", "width": 80},
-                {"headerName": "Win%", "field": "win", "width": 80},
-                {"headerName": "P&L", "field": "pnl", "width": 100, "cellClass": "font-mono",
-                 "cellClassRules": {"text-green": "value >= 0", "text-red": "value < 0"}}],
-                "rowData": [], "domLayout": "autoHeight"}).classes("w-full")
-
-            async def refresh_gates():
-                gates_grid.options["rowData"] = await run.io_bound(fetch_gates)
-                gates_grid.update()
-            ui.timer(5.0, refresh_gates)
-
-            async def load_attr():
-                att_grid.options["rowData"] = await run.io_bound(fetch_attribution)
-                att_grid.update()
-            ui.button("Load / refresh attribution", on_click=load_attr).props("outline")
-
-        # ---- ACTIVITY: autonomous-ops live feed ----
-        with ui.tab_panel(t_act):
-            ui.label("Live activity (autonomous — no input needed)").classes("text-lg font-bold")
-            act_grid = ui.aggrid({"columnDefs": [
-                {"headerName": "Time", "field": "time", "width": 100, "cellClass": "font-mono"},
-                {"headerName": "Event", "field": "type", "width": 170, "cellClassRules": {
-                    "text-green": 'value === "position_closed" || value === "order_filled"',
-                    "text-red": 'value === "risk_rule_triggered"'}},
-                {"headerName": "Detail", "field": "message", "flex": 1}],
-                "rowData": [], "domLayout": "autoHeight"}).classes("w-full")
-
-            async def refresh_act():
-                act_grid.options["rowData"] = await run.io_bound(fetch_activity)
-                act_grid.update()
-            ui.timer(4.0, refresh_act)
-
-    # ---- global header timers ----
-    async def refresh_header():
-        acct = await run.io_bound(fetch_account)
-        if acct:
-            eq_lbl.text = f"equity ${acct['equity']:,.0f}"
-        phase, detail, _ = _market_clock()
-        phase_lbl.text = f"{phase} {detail}"
-        hb_lbl.text = f"♥ {datetime.now().strftime('%H:%M:%S')}"
-    ui.timer(2.0, refresh_header)
-
-    def tick_clock():
-        clock_lbl.text = datetime.now().strftime("%H:%M:%S")
-    ui.timer(1.0, tick_clock)
-
-
-if __name__ in {"__main__", "__mp_main__"}:
-    ui.run(host="0.0.0.0", port=int(os.environ.get("CONTROLROOM_PORT", "8766")),
-           title="Momentum Control Room", dark=True, show=False, reload=False,
-           favicon="🎯")

core/contracts.py

@@ -1,259 +0,0 @@
-"""Module contracts and interfaces for the momentum trading system.
-
-Every module must implement the ModuleContract protocol, defining:
-- input_schema: What the module expects
-- output_schema: What the module produces
-- run(): Execute the module
-- validate_input(): Ensure data meets contracts
-"""
-
-from __future__ import annotations
-
-from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-from datetime import datetime, date
-from typing import Any, Protocol, TypeVar
-
-T_Input = TypeVar("T_Input")
-T_Output = TypeVar("T_Output")
-
-
-@dataclass
-class ModuleMetrics:
-    """Standard metrics every module should track."""
-
-    module_name: str
-    timestamp: datetime
-    duration_ms: float
-    input_count: int = 0
-    output_count: int = 0
-    error_count: int = 0
-    warnings: list[str] = field(default_factory=list)
-    custom_metrics: dict[str, Any] = field(default_factory=dict)
-
-    def to_dict(self) -> dict:
-        return {
-            "module_name": self.module_name,
-            "timestamp": self.timestamp.isoformat(),
-            "duration_ms": self.duration_ms,
-            "input_count": self.input_count,
-            "output_count": self.output_count,
-            "error_count": self.error_count,
-            "warnings": self.warnings,
-            "custom_metrics": self.custom_metrics,
-        }
-
-
-@dataclass
-class ValidationFailure:
-    """Single validation failure with context."""
-
-    field: str
-    value: Any
-    reason: str
-    severity: str = "error"  # "error" or "warning"
-
-
-@dataclass
-class ModuleResult:
-    """Base class for all module results."""
-
-    success: bool
-    timestamp: datetime
-    metrics: ModuleMetrics
-    validation_failures: list[ValidationFailure] = field(default_factory=list)
-
-    def has_errors(self) -> bool:
-        return bool(self.validation_failures or not self.success)
-
-    def error_summary(self) -> str:
-        """Human-readable error summary."""
-        if not self.validation_failures:
-            return "No errors"
-        errors = [f for f in self.validation_failures if f.severity == "error"]
-        if not errors:
-            return f"{len(self.validation_failures)} warnings only"
-        return f"{len(errors)} error(s), {len([f for f in self.validation_failures if f.severity == 'warning'])} warning(s)"
-
-
-class ModuleContract(Protocol):
-    """Contract every module must fulfill."""
-
-    def run(self, input: Any) -> ModuleResult:
-        """Execute the module.
-
-        Args:
-            input: Module-specific input conforming to input_schema
-
-        Returns:
-            Module-specific result inheriting from ModuleResult
-        """
-        ...
-
-    def validate_input(self, input: Any) -> tuple[bool, list[ValidationFailure]]:
-        """Validate input before processing.
-
-        Args:
-            input: Data to validate
-
-        Returns:
-            (is_valid, failures)
-        """
-        ...
-
-    def get_metrics(self) -> ModuleMetrics:
-        """Get last execution metrics."""
-        ...
-
-
-# ============================================================================
-# INGESTION MODULE CONTRACTS
-# ============================================================================
-
-
-@dataclass
-class BarIngestRequest:
-    """Input contract for ingestion module."""
-
-    symbols: list[str]
-    session_date: date
-    lookback_minutes: int = 120  # how many minutes of history to fetch
-    source: str = "alpaca"  # "alpaca", "test", "file"
-
-
-@dataclass
-class BarIngestResult(ModuleResult):
-    """Output contract for ingestion module."""
-
-    minute_rows_inserted: int = 0
-    symbols_updated: list[str] = field(default_factory=list)
-    source_health: dict[str, str] = field(default_factory=dict)  # {source: "healthy"|"error"}
-
-
-# ============================================================================
-# RESEARCH MODULE CONTRACTS
-# ============================================================================
-
-
-@dataclass
-class ResearchRequest:
-    """Input contract for research module."""
-
-    symbols: list[str]
-    session_date: date
-    lookback_days: int = 30  # for daily history
-
-
-@dataclass
-class ResearchResult(ModuleResult):
-    """Output contract for research module."""
-
-    symbols_researched: list[str] = field(default_factory=list)
-    symbols_with_data: list[str] = field(default_factory=list)
-    symbols_missing_data: list[str] = field(default_factory=list)
-    total_bars_available: int = 0
-
-
-# ============================================================================
-# EVALUATION MODULE CONTRACTS
-# ============================================================================
-
-
-@dataclass
-class CriterionDetail:
-    """Result of evaluating one criterion."""
-
-    name: str
-    passed: bool
-    value: float  # actual metric value
-    threshold: float | None = None  # minimum to pass
-    reason: str = ""
-    confidence: float = 1.0  # 0-1, how confident in this result
-
-
-@dataclass
-class EvaluationRequest:
-    """Input contract for evaluation module."""
-
-    symbol: str
-    bars_df: Any  # pd.DataFrame with {timestamp, open, high, low, close, volume}
-    previous_close: float
-    avg_daily_volume: float
-    session_date: date
-
-
-@dataclass
-class EvaluationResult(ModuleResult):
-    """Output contract for evaluation module."""
-
-    symbol: str = ""
-    status: str = "late"  # "ready", "blocked", "late", "error"
-    score: float = 0.0  # 0-100
-    criteria: list[CriterionDetail] = field(default_factory=list)
-    entry_price: float | None = None
-    stop_loss: float | None = None
-    target_price: float | None = None
-    quality_score: float = 0.0  # 0-1
-    blocking_reason: str | None = None
-    metrics_dict: dict[str, float] = field(default_factory=dict)  # gap%, rvol, etc.
-
-
-# ============================================================================
-# EXECUTION MODULE CONTRACTS
-# ============================================================================
-
-
-@dataclass
-class ExecutionRequest:
-    """Input contract for execution module."""
-
-    symbol: str
-    entry_price: float
-    stop_price: float
-    target_price: float
-    score: float
-    criteria_count: int
-    account_equity: float
-    current_positions: int
-    today_realized_pnl: float
-
-
-@dataclass
-class ExecutionResult(ModuleResult):
-    """Output contract for execution module."""
-
-    symbol: str = ""
-    status: str = ""  # "approved", "rejected", "pending_approval", "error"
-    reason: str = ""  # why it was approved/rejected
-    order_id: str | None = None
-    shares: int | None = None
-    risk_dollars: float | None = None
-    risk_checks_applied: list[str] = field(default_factory=list)
-
-
-# ============================================================================
-# BROKER MODULE CONTRACTS
-# ============================================================================
-
-
-@dataclass
-class BrokerRequest:
-    """Input contract for broker module."""
-
-    symbol: str
-    order_id: str
-    entry_price: float
-    stop_price: float
-    target_price: float
-    shares: int
-    order_type: str = "limit"  # "limit" or "market"
-
-
-@dataclass
-class BrokerResult(ModuleResult):
-    """Output contract for broker module."""
-
-    symbol: str = ""
-    status: str = ""  # "submitted", "rejected", "error"
-    broker_order_id: str | None = None
-    submitted_at: datetime | None = None

core/telemetry.py

@@ -1,274 +0,0 @@
-"""Telemetry collection and event emission for module observability."""
-
-from __future__ import annotations
-
-import time
-import logging
-from dataclasses import dataclass, field
-from datetime import datetime
-from typing import TYPE_CHECKING, Any, Callable
-
-if TYPE_CHECKING:
-    from core.contracts import ModuleMetrics
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass
-class TelemetryCollector:
-    """Collect metrics during module execution."""
-
-    module_name: str
-    start_time: float = field(default_factory=time.time)
-    input_count: int = 0
-    output_count: int = 0
-    error_count: int = 0
-    warnings: list[str] = field(default_factory=list)
-    custom_metrics: dict[str, Any] = field(default_factory=dict)
-
-    def record_input(self, count: int = 1) -> None:
-        """Record input items processed."""
-        self.input_count += count
-
-    def record_output(self, count: int = 1) -> None:
-        """Record output items produced."""
-        self.output_count += count
-
-    def record_error(self) -> None:
-        """Record an error occurred."""
-        self.error_count += 1
-
-    def record_warning(self, message: str) -> None:
-        """Record a warning."""
-        self.warnings.append(message)
-
-    def record_metric(self, name: str, value: float | int | str) -> None:
-        """Record a custom metric."""
-        self.custom_metrics[name] = value
-
-    def finalize(self) -> ModuleMetrics:
-        """Finalize and return metrics."""
-        duration_ms = (time.time() - self.start_time) * 1000
-        return ModuleMetrics(
-            module_name=self.module_name,
-            timestamp=datetime.now(),
-            duration_ms=duration_ms,
-            input_count=self.input_count,
-            output_count=self.output_count,
-            error_count=self.error_count,
-            warnings=self.warnings,
-            custom_metrics=self.custom_metrics,
-        )
-
-
-class TelemetryEmitter:
-    """Emit telemetry events to event store and observability systems."""
-
-    def __init__(self, event_store=None):
-        """Initialize emitter.
-
-        Args:
-            event_store: Optional EventStore for persisting telemetry events
-        """
-        self.event_store = event_store
-        self.callbacks: list[Callable[[dict], None]] = []
-
-    def add_callback(self, callback: Callable[[dict], None]) -> None:
-        """Add a callback to receive telemetry events."""
-        self.callbacks.append(callback)
-
-    def emit_module_tick(
-        self,
-        module_name: str,
-        stage: str,
-        metrics: ModuleMetrics | None = None,
-        errors: list[dict] | None = None,
-        correlation_id: str | None = None,
-    ) -> None:
-        """Emit a module tick event.
-
-        Args:
-            module_name: Name of module (e.g., "ingestion", "evaluation")
-            stage: Stage name (e.g., "started", "processing", "completed")
-            metrics: Optional ModuleMetrics object
-            errors: Optional list of error details
-            correlation_id: Correlation ID linking to signal_ready event
-        """
-        event = {
-            "type": "module_tick",
-            "timestamp": datetime.now().isoformat(),
-            "module": module_name,
-            "stage": stage,
-            "metrics": metrics.to_dict() if metrics else None,
-            "errors": errors or [],
-            "correlation_id": correlation_id,
-        }
-        self._emit(event)
-
-    def emit_data_validation(
-        self,
-        module_name: str,
-        validation_type: str,
-        valid: bool,
-        details: dict,
-        correlation_id: str | None = None,
-    ) -> None:
-        """Emit a data validation event.
-
-        Args:
-            module_name: Name of module performing validation
-            validation_type: Type of validation ("schema", "checksum", "business_rules")
-            valid: Whether validation passed
-            details: Validation details (errors, warnings, statistics)
-            correlation_id: Optional correlation ID
-        """
-        event = {
-            "type": "data_validation",
-            "timestamp": datetime.now().isoformat(),
-            "module": module_name,
-            "validation_type": validation_type,
-            "valid": valid,
-            "details": details,
-            "correlation_id": correlation_id,
-        }
-        self._emit(event)
-
-    def emit_symbol_evaluation_detail(
-        self,
-        symbol: str,
-        stage: str,
-        metrics: dict[str, Any],
-        passed: bool,
-        blocking_reason: str | None = None,
-        correlation_id: str | None = None,
-    ) -> None:
-        """Emit detailed evaluation event for a symbol.
-
-        Args:
-            symbol: Symbol being evaluated
-            stage: Evaluation stage (e.g., "bars_loaded", "gap_checked", "rvol_calculated")
-            metrics: Metrics at this stage
-            passed: Whether this stage passed
-            blocking_reason: If not passed, why
-            correlation_id: Optional correlation ID
-        """
-        event = {
-            "type": "symbol_evaluation_detail",
-            "timestamp": datetime.now().isoformat(),
-            "symbol": symbol,
-            "stage": stage,
-            "metrics": metrics,
-            "passed": passed,
-            "blocking_reason": blocking_reason,
-            "correlation_id": correlation_id,
-        }
-        self._emit(event)
-
-    def _emit(self, event: dict) -> None:
-        """Emit event to all registered callbacks and event store."""
-        logger.debug(f"Telemetry: {event['type']} - {event.get('module', event.get('symbol', 'n/a'))}")
-
-        # Call all registered callbacks
-        for callback in self.callbacks:
-            try:
-                callback(event)
-            except Exception as e:
-                logger.error(f"Telemetry callback failed: {e}")
-
-        # Emit to event store if available
-        if self.event_store:
-            try:
-                # Import here to avoid circular dependency
-                from storage.event_schema import ModuleTickEvent, DataValidationEvent
-
-                if event["type"] == "module_tick":
-                    self.event_store.emit(
-                        ModuleTickEvent(
-                            timestamp=datetime.now(),
-                            module=event["module"],
-                            stage=event["stage"],
-                            duration_ms=event["metrics"]["duration_ms"]
-                            if event["metrics"]
-                            else 0,
-                            input_count=event["metrics"]["input_count"]
-                            if event["metrics"]
-                            else 0,
-                            output_count=event["metrics"]["output_count"]
-                            if event["metrics"]
-                            else 0,
-                            metrics=event["metrics"]["custom_metrics"]
-                            if event["metrics"]
-                            else {},
-                            errors=event["errors"],
-                            correlation_id=event.get("correlation_id"),
-                        )
-                    )
-                elif event["type"] == "data_validation":
-                    self.event_store.emit(
-                        DataValidationEvent(
-                            timestamp=datetime.now(),
-                            module=event["module"],
-                            validation_type=event["validation_type"],
-                            valid=event["valid"],
-                            details=event["details"],
-                            correlation_id=event.get("correlation_id"),
-                        )
-                    )
-            except Exception as e:
-                logger.error(f"Failed to emit telemetry event to store: {e}")
-
-
-# Global instance (can be overridden in tests)
-_telemetry_emitter: TelemetryEmitter | None = None
-
-
-def set_telemetry_emitter(emitter: TelemetryEmitter) -> None:
-    """Set the global telemetry emitter."""
-    global _telemetry_emitter
-    _telemetry_emitter = emitter
-
-
-def get_telemetry_emitter() -> TelemetryEmitter:
-    """Get the global telemetry emitter."""
-    global _telemetry_emitter
-    if _telemetry_emitter is None:
-        _telemetry_emitter = TelemetryEmitter()
-    return _telemetry_emitter
-
-
-def emit_module_tick(
-    module_name: str,
-    stage: str,
-    metrics: ModuleMetrics | None = None,
-    errors: list[dict] | None = None,
-    correlation_id: str | None = None,
-) -> None:
-    """Convenience function to emit module tick."""
-    get_telemetry_emitter().emit_module_tick(module_name, stage, metrics, errors, correlation_id)
-
-
-def emit_data_validation(
-    module_name: str,
-    validation_type: str,
-    valid: bool,
-    details: dict,
-    correlation_id: str | None = None,
-) -> None:
-    """Convenience function to emit data validation."""
-    get_telemetry_emitter().emit_data_validation(
-        module_name, validation_type, valid, details, correlation_id
-    )
-
-
-def emit_symbol_evaluation_detail(
-    symbol: str,
-    stage: str,
-    metrics: dict[str, Any],
-    passed: bool,
-    blocking_reason: str | None = None,
-    correlation_id: str | None = None,
-) -> None:
-    """Convenience function to emit symbol evaluation detail."""
-    get_telemetry_emitter().emit_symbol_evaluation_detail(
-        symbol, stage, metrics, passed, blocking_reason, correlation_id
-    )

core/validation.py

@@ -1,284 +0,0 @@
-"""Data validation utilities for module boundaries."""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Callable
-
-if TYPE_CHECKING:
-    from core.contracts import ValidationFailure
-
-
-@dataclass
-class SchemaValidator:
-    """Validate data against a schema."""
-
-    required_fields: dict[str, type]  # {field_name: expected_type}
-    optional_fields: dict[str, type] | None = None
-
-    def __post_init__(self):
-        if self.optional_fields is None:
-            self.optional_fields = {}
-
-    def validate(self, data: dict) -> tuple[bool, list[dict]]:
-        """Validate data against schema.
-
-        Args:
-            data: Dictionary to validate
-
-        Returns:
-            (is_valid, list of failure dicts)
-        """
-        failures = []
-
-        # Check required fields
-        for field_name, expected_type in self.required_fields.items():
-            if field_name not in data:
-                failures.append(
-                    {
-                        "field": field_name,
-                        "value": None,
-                        "reason": "Required field missing",
-                        "severity": "error",
-                    }
-                )
-            elif not isinstance(data[field_name], expected_type):
-                failures.append(
-                    {
-                        "field": field_name,
-                        "value": data[field_name],
-                        "reason": f"Expected {expected_type.__name__}, got {type(data[field_name]).__name__}",
-                        "severity": "error",
-                    }
-                )
-
-        # Check optional fields if present
-        if self.optional_fields:
-            for field_name, expected_type in self.optional_fields.items():
-                if field_name in data and not isinstance(data[field_name], expected_type):
-                    failures.append(
-                        {
-                            "field": field_name,
-                            "value": data[field_name],
-                            "reason": f"Expected {expected_type.__name__}, got {type(data[field_name]).__name__}",
-                            "severity": "warning",
-                        }
-                    )
-
-        return len([f for f in failures if f.severity == "error"]) == 0, failures
-
-
-class RangeValidator:
-    """Validate numeric values are within expected ranges."""
-
-    def __init__(self, min_val: float | None = None, max_val: float | None = None):
-        self.min_val = min_val
-        self.max_val = max_val
-
-    def validate(self, value: float, field_name: str = "value") -> tuple[bool, dict | None]:
-        """Validate value is in range.
-
-        Args:
-            value: Numeric value to validate
-            field_name: Name of field for error message
-
-        Returns:
-            (is_valid, failure dict or None)
-        """
-        if self.min_val is not None and value < self.min_val:
-            return False, {
-                "field": field_name,
-                "value": value,
-                "reason": f"Value {value} below minimum {self.min_val}",
-                "severity": "error",
-            }
-
-        if self.max_val is not None and value > self.max_val:
-            return False, {
-                "field": field_name,
-                "value": value,
-                "reason": f"Value {value} above maximum {self.max_val}",
-                "severity": "error",
-            }
-
-        return True, None
-
-
-class ChecksumValidator:
-    """Validate data integrity using checksums."""
-
-    @staticmethod
-    def compute_checksum(data: dict | list) -> str:
-        """Compute simple checksum for data.
-
-        Args:
-            data: Dictionary or list to checksum
-
-        Returns:
-            Hex checksum string
-        """
-        import hashlib
-        import json
-
-        data_str = json.dumps(data, sort_keys=True, default=str)
-        return hashlib.md5(data_str.encode()).hexdigest()
-
-    @staticmethod
-    def validate_checksum(
-        data: dict | list, expected_checksum: str
-    ) -> tuple[bool, dict | None]:
-        """Validate data matches checksum.
-
-        Args:
-            data: Data to validate
-            expected_checksum: Expected checksum value
-
-        Returns:
-            (is_valid, failure dict or None)
-        """
-        actual_checksum = ChecksumValidator.compute_checksum(data)
-        if actual_checksum != expected_checksum:
-            return False, {
-                "field": "checksum",
-                "value": actual_checksum,
-                "reason": f"Checksum mismatch: expected {expected_checksum}, got {actual_checksum}",
-                "severity": "error",
-            }
-        return True, None
-
-
-class BusinessRuleValidator:
-    """Validate data against business logic rules."""
-
-    def __init__(self):
-        self.rules: dict[str, Callable[[Any], tuple[bool, str]]] = {}
-
-    def add_rule(self, rule_name: str, rule_fn: Callable[[Any], tuple[bool, str]]) -> None:
-        """Add a business rule.
-
-        Args:
-            rule_name: Name of the rule
-            rule_fn: Function that takes data and returns (is_valid, reason)
-        """
-        self.rules[rule_name] = rule_fn
-
-    def validate(self, data: Any) -> tuple[bool, list[dict]]:
-        """Apply all business rules.
-
-        Args:
-            data: Data to validate
-
-        Returns:
-            (is_valid, list of failure dicts)
-        """
-        failures = []
-        for rule_name, rule_fn in self.rules.items():
-            try:
-                is_valid, reason = rule_fn(data)
-                if not is_valid:
-                    failures.append(
-                        {
-                            "field": "business_rule",
-                            "value": None,
-                            "reason": f"{rule_name}: {reason}",
-                            "severity": "error",
-                        }
-                    )
-            except Exception as e:
-                failures.append(
-                    {
-                        "field": "business_rule",
-                        "value": None,
-                        "reason": f"{rule_name} raised exception: {e}",
-                        "severity": "error",
-                    }
-                )
-
-        return len(failures) == 0, failures
-
-
-# Concrete validators for common cases
-
-class BarDataValidator:
-    """Validate OHLCV bar data."""
-
-    @staticmethod
-    def validate_bar(bar: dict) -> tuple[bool, list[dict]]:
-        """Validate a single bar.
-
-        Args:
-            bar: Bar dictionary with {timestamp, open, high, low, close, volume}
-
-        Returns:
-            (is_valid, list of failure dicts)
-        """
-        failures = []
-
-        # Check required fields
-        required = {"timestamp", "open", "high", "low", "close", "volume"}
-        for field in required:
-            if field not in bar:
-                failures.append(
-                    {
-                        "field": field,
-                        "value": None,
-                        "reason": "Missing required field",
-                        "severity": "error",
-                    }
-                )
-
-        if failures:
-            return False, failures
-
-        # Check value consistency
-        open_val = bar["open"]
-        high_val = bar["high"]
-        low_val = bar["low"]
-        close_val = bar["close"]
-        volume_val = bar["volume"]
-
-        # High must be >= all others
-        if high_val < open_val or high_val < close_val or high_val < low_val:
-            failures.append(
-                {
-                    "field": "high",
-                    "value": high_val,
-                    "reason": f"High ({high_val}) must be >= open ({open_val}), close ({close_val}), low ({low_val})",
-                    "severity": "error",
-                }
-            )
-
-        # Low must be <= all others
-        if low_val > open_val or low_val > close_val or low_val > high_val:
-            failures.append(
-                {
-                    "field": "low",
-                    "value": low_val,
-                    "reason": f"Low ({low_val}) must be <= open ({open_val}), close ({close_val}), high ({high_val})",
-                    "severity": "error",
-                }
-            )
-
-        # Volume should be non-negative
-        if volume_val < 0:
-            failures.append(
-                {
-                    "field": "volume",
-                    "value": volume_val,
-                    "reason": "Volume cannot be negative",
-                    "severity": "error",
-                }
-            )
-
-        # Warn on zero volume
-        if volume_val == 0:
-            failures.append(
-                {
-                    "field": "volume",
-                    "value": volume_val,
-                    "reason": "Zero volume bar",
-                    "severity": "warning",
-                }
-            )
-
-        return len([f for f in failures if f.severity == "error"]) == 0, failures

deploy/systemd/momentum-controlroom.service

@@ -1,17 +0,0 @@
-[Unit]
-Description=Momentum Trading Control Room (NiceGUI live-analysis dashboard, :8766)
-StartLimitIntervalSec=0
-
-[Service]
-Type=simple
-WorkingDirectory=/mnt/c/code/momentum
-Environment=PYTHONPATH=/mnt/c/code/momentum
-ExecStart=/home/philip/.venvs/momentum/bin/python -u controlroom/app.py
-Restart=always
-RestartSec=10
-StandardOutput=append:/mnt/c/code/momentum/data/controlroom.log
-StandardError=append:/mnt/c/code/momentum/data/controlroom.log
-SyslogIdentifier=momentum-controlroom
-
-[Install]
-WantedBy=default.target

models/base.py

@@ -1,26 +0,0 @@
-"""Models for Milestone 2.
-
-Event and data models for the event-first architecture.
-"""
-
-from pydantic import BaseModel, Field
-
-
-class EventMode(str):
-    """Runtime modes for events."""
-
-    REPLAY = "replay"
-    PAPER = "paper"
-    LIVE = "live"
-
-
-class BaseEvent(BaseModel):
-    """Base event with common fields."""
-
-    timestamp: str = Field(description="Event timestamp (ISO format)")
-    mode: EventMode = Field(description="Runtime mode (replay/paper/live)")
-    event_type: str = Field(description="Event type identifier")
-    correlation_id: str | None = Field(
-        default=None, description="Correlation ID for causal chain"
-    )
-    message: str = Field(description="Human-readable message")

package-lock.json

@@ -1,18 +0,0 @@
-{
-  "name": "momentum",
-  "lockfileVersion": 3,
-  "requires": true,
-  "packages": {
-    "": {
-      "dependencies": {
-        "claude": "^0.1.1"
-      }
-    },
-    "node_modules/claude": {
-      "version": "0.1.1",
-      "resolved": "https://registry.npmjs.org/claude/-/claude-0.1.1.tgz",
-      "integrity": "sha512-j7oSibqQdIODNhkI1sEJzHMiPsF43L/GqNbcA+eDDyGM10+x2sH9NW/PK6vM3z0J2tLDKMBcc5ZjVaoRinhuCA==",
-      "license": "ISC"
-    }
-  }
-}

package.json

@@ -1,5 +0,0 @@
-{
-  "dependencies": {
-    "claude": "^0.1.1"
-  }
-}

run_live_paper.py

@@ -411,13 +411,21 @@ def _session_open(con, symbol: str, session_date) -> float | None:
             "AND is_regular_hours ORDER BY timestamp LIMIT 1", [symbol, session_date]
         ).fetchall()
         if r:
-            # trust it ONLY if it is really the opening bar (<= 09:33 ET = 13:33Z):
+            # trust it ONLY if it is really the 09:30 ET opening bar: mid-day-
-            # mid-day-discovered names briefly have bars starting mid-session, and
+            # discovered names briefly have bars starting mid-session, and treating
-            # treating that first bar as "the open" is how TVRD/FRTT armed as fake
+            # that first bar as "the open" is how TVRD/FRTT armed as fake 20%
-            # 20% gappers on 2026-07-08.
+            # gappers on 2026-07-08. minute_bars.timestamp is UTC-naive, so the
-            ts = str(r[0][1])
+            # open is 13:30Z in summer (EDT) but 14:30Z in winter (EST) — convert
-            hhmm = ts[11:16]
+            # to ET instead of a fixed UTC window, else the whole fast path arms
-            if "13:30" <= hhmm <= "13:33":
+            # nothing all winter.
+            from datetime import datetime, timezone
+            from zoneinfo import ZoneInfo
+            raw = r[0][1]
+            ts = raw if isinstance(raw, datetime) else datetime.fromisoformat(str(raw))
+            if ts.tzinfo is None:
+                ts = ts.replace(tzinfo=timezone.utc)
+            et_hhmm = ts.astimezone(ZoneInfo("America/New_York")).strftime("%H:%M")
+            if "09:30" <= et_hhmm <= "09:33":
                 _SESSION_OPEN_CACHE[key] = float(r[0][0])
                 return _SESSION_OPEN_CACHE[key]
     except Exception:  # noqa: BLE001 — fail-open; gap falls back to window/day_open

runtime/providers/__init__.py

@@ -1,10 +0,0 @@
-"""Market-data provider registry."""
-
-from runtime.providers.registry import (
-    ProviderKind,
-    ProviderConfig,
-    ProviderRegistry,
-    default_registry,
-)
-
-__all__ = ["ProviderKind", "ProviderConfig", "ProviderRegistry", "default_registry"]

runtime/providers/registry.py

@@ -1,112 +0,0 @@
-"""Provider registry for market-data sources.
-
-(Repaired: ProviderConfig is a proper top-level dataclass and the registry
-initializes its mapping correctly — the original nested the dataclass inside
-the enum and used an invalid `self.providers: {}` annotation-as-assignment.)
-"""
-
-from __future__ import annotations
-
-import os
-from dataclasses import dataclass, field
-from enum import Enum
-
-
-class ProviderKind(str, Enum):
-    """Supported market-data providers."""
-
-    ALPACA = "alpaca"
-    SCHWAB = "schwab"
-    POLYGON = "polygon"
-    FINNHUB = "finnhub"
-    SYNTHETIC = "synthetic"
-
-
-@dataclass
-class ProviderConfig:
-    """Configuration for one provider."""
-
-    kind: ProviderKind
-    enabled: bool = True
-    priority: int = 100  # lower = preferred
-    rate_limit_per_minute: int = 200
-    supports_minute_bars: bool = True
-    supports_quotes: bool = True
-    env_keys: tuple[str, ...] = field(default_factory=tuple)
-
-    def is_configured(self, env: dict[str, str] | None = None) -> bool:
-        """A provider is configured when all of its env keys are present."""
-        values = dict(os.environ)
-        if env is not None:
-            values.update(env)
-        return all(values.get(key) for key in self.env_keys)
-
-
-class ProviderRegistry:
-    """Registry of provider configs with priority-based selection."""
-
-    def __init__(self) -> None:
-        self.providers: dict[ProviderKind, ProviderConfig] = {}
-
-    def register(self, config: ProviderConfig) -> None:
-        self.providers[config.kind] = config
-
-    def get(self, kind: ProviderKind | str) -> ProviderConfig | None:
-        if isinstance(kind, str):
-            try:
-                kind = ProviderKind(kind)
-            except ValueError:
-                return None
-        return self.providers.get(kind)
-
-    def enabled_providers(self) -> list[ProviderConfig]:
-        return sorted(
-            (p for p in self.providers.values() if p.enabled),
-            key=lambda p: p.priority,
-        )
-
-    def best_configured(
-        self, env: dict[str, str] | None = None
-    ) -> ProviderConfig | None:
-        for provider in self.enabled_providers():
-            if provider.is_configured(env):
-                return provider
-        return None
-
-
-def default_registry() -> ProviderRegistry:
-    """Registry with the standard provider stack."""
-    registry = ProviderRegistry()
-    registry.register(
-        ProviderConfig(
-            kind=ProviderKind.ALPACA,
-            priority=10,
-            env_keys=("ALPACA_API_KEY", "ALPACA_SECRET_KEY"),
-        )
-    )
-    registry.register(
-        ProviderConfig(
-            kind=ProviderKind.SCHWAB,
-            priority=20,
-            env_keys=("SCHWAB_MARKET_DATA_APP_KEY", "SCHWAB_MARKET_DATA_APP_SECRET"),
-        )
-    )
-    registry.register(
-        ProviderConfig(
-            kind=ProviderKind.POLYGON, priority=30, env_keys=("POLYGON_API_KEY",)
-        )
-    )
-    registry.register(
-        ProviderConfig(
-            kind=ProviderKind.FINNHUB, priority=40, env_keys=("FINNHUB_API_KEY",)
-        )
-    )
-    registry.register(
-        ProviderConfig(
-            kind=ProviderKind.SYNTHETIC,
-            priority=1000,
-            env_keys=(),
-            supports_quotes=False,
-        )
-    )
-    return registry

storage/replay.py

@@ -1,148 +0,0 @@
-import json
-
-"""Replay/rebuild tooling for Milestone 2.
-
-Reconstruct state from event store for analysis and testing.
-"""
-
-import logging
-from typing import TYPE_CHECKING
-
-from .event_store import EventStore
-
-if TYPE_CHECKING:
-    pass
-
-logger = logging.getLogger(__name__)
-
-
-class StateRebuilder:
-    """Rebuild trading state from event store."""
-
-    def __init__(self, store: EventStore):
-        self.store = store
-
-    def rebuild_session(self, session_id: str) -> dict:
-        """Rebuild complete state for a session from events."""
-        logger.info(f"Rebuilding session {session_id}")
-
-        events = self.store.query_events(session_id=session_id)
-
-        state = {
-            "session_id": session_id,
-            "symbols": {},
-            "orders": {},
-            "positions": {},
-            "events": [],
-        }
-
-        for event in events:
-            event_type = event["event_type"]
-            payload = json.loads(event.get("payload_json", "{}"))
-            symbol = payload.get("symbol")
-
-            self._process_event(state, event_type, symbol, payload, event)
-
-        logger.info(f"Rebuilt state for {len(events)} events")
-        return state
-
-    def _process_event(
-        self,
-        state: dict,
-        event_type: str,
-        symbol: str | None,
-        payload: dict,
-        event: dict,
-    ):
-        """Process individual event and update state."""
-        if event_type == "symbol_discovered":
-            state["symbols"][symbol] = {
-                "state": "discovered",
-                "discovered_at": event["timestamp"],
-            }
-
-        elif event_type == "symbol_state_changed":
-            if symbol not in state["symbols"]:
-                state["symbols"][symbol] = {
-                    "state": payload.get("previous_state") or "discovered",
-                    "discovered_at": event["timestamp"],
-                }
-            state["symbols"][symbol]["state"] = payload["new_state"]
-            state["symbols"][symbol]["state_history"] = state["symbols"][symbol].get(
-                "state_history", []
-            )
-            state["symbols"][symbol]["state_history"].append(
-                {
-                    "state": payload["new_state"],
-                    "timestamp": event["timestamp"],
-                    "reason": payload.get("state_reason"),
-                }
-            )
-
-        elif event_type == "signal_ready":
-            if symbol in state["symbols"]:
-                state["symbols"][symbol]["signals"] = state["symbols"][symbol].get(
-                    "signals", []
-                )
-                state["symbols"][symbol]["signals"].append(
-                    {
-                        "signal_type": payload["signal_type"],
-                        "confidence": payload["confidence"],
-                        "timestamp": event["timestamp"],
-                    }
-                )
-
-        elif event_type == "position_opened":
-            position_id = payload["position_id"]
-            state["positions"][position_id] = {
-                "symbol": symbol,
-                "status": "open",
-                "entry_price": payload["entry_price"],
-                "quantity": payload["quantity"],
-                "stop_loss_price": payload["stop_loss_price"],
-                "opened_at": event["timestamp"],
-            }
-
-        elif event_type == "position_closed":
-            position_id = payload["position_id"]
-            if position_id in state["positions"]:
-                state["positions"][position_id].update(
-                    {
-                        "status": "closed",
-                        "exit_price": payload["exit_price"],
-                        "exit_reason": payload["exit_reason"],
-                        "realized_pnl": payload["realized_pnl"],
-                        "closed_at": event["timestamp"],
-                    }
-                )
-
-        elif event_type == "order_filled":
-            state["events"].append(
-                {
-                    "type": event_type,
-                    "order_id": payload["order_id"],
-                    "symbol": symbol,
-                    "timestamp": event["timestamp"],
-                }
-            )
-
-        elif event_type == "order_submitted":
-            order_id = payload["order_id"]
-            state["orders"][order_id] = {
-                "symbol": symbol,
-                "status": "submitted",
-                "side": payload.get("side"),
-                "quantity": payload.get("quantity"),
-                "price": payload.get("price"),
-                "updated_at": event["timestamp"],
-            }
-
-        elif event_type == "order_cancelled":
-            order_id = payload["order_id"]
-            state["orders"][order_id] = {
-                **state["orders"].get(order_id, {}),
-                "symbol": symbol,
-                "status": "cancelled",
-                "cancel_reason": payload.get("cancel_reason"),
-                "updated_at": event["timestamp"],
-            }

strategy/evaluation/llm_integration.py

@@ -1,36 +0,0 @@
-"""Optional LLM (Ollama) enrichment hooks. Disabled unless configured."""
-
-from __future__ import annotations
-
-import json
-import logging
-import urllib.request
-
-logger = logging.getLogger(__name__)
-
-
-def summarize_setup_with_llm(
-    setup: dict,
-    host: str = "http://localhost:11434",
-    model: str = "mistral",
-    timeout: int = 15,
-) -> str | None:
-    """Ask a local Ollama model to summarize a setup. Returns None on failure."""
-    prompt = (
-        "Summarize this momentum trading setup in one sentence:\n"
-        + json.dumps(setup, default=str)
-    )
-    try:
-        req = urllib.request.Request(
-            f"{host}/api/generate",
-            data=json.dumps(
-                {"model": model, "prompt": prompt, "stream": False}
-            ).encode(),
-            headers={"Content-Type": "application/json"},
-        )
-        with urllib.request.urlopen(req, timeout=timeout) as resp:
-            body = json.loads(resp.read().decode())
-        return body.get("response")
-    except Exception:
-        logger.debug("LLM enrichment unavailable", exc_info=True)
-        return None

strategy/evaluation/storage.py

@@ -1,33 +0,0 @@
-"""Persist setup evaluations into the research DB (setup_events table)."""
-
-from __future__ import annotations
-
-import uuid
-from datetime import datetime
-
-
-def store_setup_event(con, symbol: str, session_date, result: dict) -> str:
-    """Insert one evaluation result into setup_events. Returns setup_id."""
-    setup_id = str(uuid.uuid4())
-    setups = result.get("setups") or []
-    best = setups[0] if setups else {}
-    con.execute(
-        """
-        INSERT INTO setup_events
-            (setup_id, symbol, setup_time, session_date, setup_name,
-             entry_reference_price, invalidation_price, gap_pct, relative_volume)
-        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
-        """,
-        [
-            setup_id,
-            symbol,
-            datetime.now(),
-            session_date,
-            best.get("setup_type") or "first_pullback",
-            best.get("entry_price"),
-            best.get("stop_loss_price"),
-            result.get("gap_pct"),
-            result.get("relative_volume"),
-        ],
-    )
-    return setup_id

strategy/evaluation/tracking.py

@@ -1,32 +0,0 @@
-"""Lightweight in-memory tracking of evaluation history per symbol."""
-
-from __future__ import annotations
-
-from collections import defaultdict
-
-
-class EvaluationTracker:
-    """Tracks recent evaluation results per symbol for debounce/analysis."""
-
-    def __init__(self, max_history: int = 50):
-        self.max_history = max_history
-        self._history: dict[str, list[dict]] = defaultdict(list)
-
-    def record(self, symbol: str, result: dict) -> None:
-        history = self._history[symbol]
-        history.append(result)
-        if len(history) > self.max_history:
-            del history[: len(history) - self.max_history]
-
-    def last(self, symbol: str) -> dict | None:
-        history = self._history.get(symbol)
-        return history[-1] if history else None
-
-    def consecutive_status(self, symbol: str, status: str) -> int:
-        count = 0
-        for result in reversed(self._history.get(symbol, [])):
-            if result.get("status") == status:
-                count += 1
-            else:
-                break
-        return count

telemetry/events.py

@@ -1,102 +0,0 @@
-"""Telemetry event emission and querying for runtime observability."""
-
-from __future__ import annotations
-
-import json
-import logging
-import uuid
-from enum import Enum
-
-logger = logging.getLogger(__name__)
-
-
-class TelemetryEventType(str, Enum):
-    SOURCE_DEGRADED = "SOURCE_DEGRADED"
-    SOURCE_RECOVERED = "SOURCE_RECOVERED"
-    SOURCE_COOLDOWN = "SOURCE_COOLDOWN"
-    WATCHLIST_ADD = "WATCHLIST_ADD"
-    WATCHLIST_DROP = "WATCHLIST_DROP"
-    CANDIDATE_STATE_CHANGE = "CANDIDATE_STATE_CHANGE"
-    FETCH_ATTEMPT = "FETCH_ATTEMPT"
-    ENRICHMENT_QUEUED = "ENRICHMENT_QUEUED"
-    ENRICHMENT_DONE = "ENRICHMENT_DONE"
-    ENRICHMENT_FAILED = "ENRICHMENT_FAILED"
-
-
-def emit_event(
-    con,
-    event_type: TelemetryEventType | str,
-    session_id: str | None = None,
-    symbol: str | None = None,
-    source: str | None = None,
-    from_state: str | None = None,
-    to_state: str | None = None,
-    reason: str | None = None,
-    metadata: dict | None = None,
-) -> None:
-    event_id = str(uuid.uuid4())
-    event_type_val = (
-        event_type.value
-        if isinstance(event_type, TelemetryEventType)
-        else str(event_type)
-    )
-    metadata_json = json.dumps(metadata) if metadata is not None else None
-    try:
-        con.execute(
-            """
-            INSERT INTO telemetry_events
-                (id, event_type, session_id, symbol, source, from_state, to_state, reason, metadata_json)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
-            """,
-            [
-                event_id,
-                event_type_val,
-                session_id,
-                symbol,
-                source,
-                from_state,
-                to_state,
-                reason,
-                metadata_json,
-            ],
-        )
-    except Exception:
-        logger.exception(
-            "telemetry emit_event failed — event_type=%s symbol=%s",
-            event_type_val,
-            symbol,
-        )
-
-
-def query_events(
-    con,
-    session_id: str | None = None,
-    symbol: str | None = None,
-    source: str | None = None,
-    limit: int = 100,
-) -> list[dict]:
-    clauses: list[str] = []
-    params: list = []
-
-    if session_id is not None:
-        clauses.append("session_id = ?")
-        params.append(session_id)
-    if symbol is not None:
-        clauses.append("symbol = ?")
-        params.append(symbol)
-    if source is not None:
-        clauses.append("source = ?")
-        params.append(source)
-
-    where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
-    sql = f"SELECT * FROM telemetry_events {where} ORDER BY created_at DESC LIMIT ?"
-    params.append(limit)
-
-    rows = con.execute(sql, params).fetchall()
-    cols = [
-        desc[0]
-        for desc in con.execute(
-            f"SELECT * FROM telemetry_events {where} LIMIT 0", params[:-1]
-        ).description
-    ]
-    return [dict(zip(cols, row)) for row in rows]

tests/unit/alpaca_paper/test_execution_and_sync.py

@@ -231,3 +231,31 @@ def test_reconcile_seeds_from_persisted_snapshot_after_restart(store):
     rows = _closed(store)
     assert len(rows) == 1 and rows[0]["symbol"] == "REST"
     assert rows[0]["realized_pnl"] == 5.0       # (2.5 - 2.0) * 10
+
+
+def test_post_not_retried_on_urlerror(monkeypatch):
+    """Lost-response URLError on POST /orders must NOT re-send (duplicate live
+    order, 2026-07-06). GET/DELETE still retry."""
+    import urllib.error
+    import urllib.request
+    from alpaca_paper.client import AlpacaPaperClient
+    from alpaca_paper.settings import AlpacaPaperSettings
+    cli = AlpacaPaperClient.__new__(AlpacaPaperClient)
+    cli.timeout = 1
+    cli.max_retries = 2
+    cli._headers = lambda: {}
+    calls = {"n": 0}
+
+    def boom(req, timeout=None):
+        calls["n"] += 1
+        raise urllib.error.URLError("connection reset")
+
+    monkeypatch.setattr(urllib.request, "urlopen", boom)
+    import pytest
+    with pytest.raises(urllib.error.URLError):
+        cli._request("POST", "https://x/v2/orders", body={"symbol": "AAA"})
+    assert calls["n"] == 1, "POST must not retry a lost-response URLError"
+    calls["n"] = 0
+    with pytest.raises(urllib.error.URLError):
+        cli._request("GET", "https://x/v2/account")
+    assert calls["n"] > 1, "idempotent GET should still retry"

tests/unit/test_trading_execution.py

@@ -875,3 +875,59 @@ def test_session_gate_bars_after_hours_entries():
     assert svc._outside_session(datetime(2026, 7, 9, 9, 30, tzinfo=ET)) is None       # the open
     svc.settings = ExecutionSettings(session_only=False)
     assert svc._outside_session(datetime(2026, 7, 9, 20, 9, tzinfo=ET)) is None       # opt-out works
+
+
+def test_flatten_all_closes_broker_position_missed_by_snapshot(store):
+    """Naked-through-breaker fix: a position that filled in the sync gap isn't in
+    the event-store held set, so _flatten_all's step-2 skips it. Step 3 (moved out
+    of close_session so the daily-loss breaker gets it too) closes it from fresh
+    broker positions."""
+    from alpaca_paper.execution import AlpacaPaperExecutor
+
+    class _Cli:
+        def __init__(self):
+            self.closed = []
+            self.pos = [{"symbol": "GAP", "qty": "100"}]  # filled, snapshot missed it
+
+        def get_positions(self, fresh=False):
+            return list(self.pos)
+
+        def close_position(self, symbol, qty=None, percentage=None):
+            self.closed.append(symbol)
+            self.pos = [p for p in self.pos if p["symbol"] != symbol]
+            return {"id": "c"}
+
+        def get_orders(self, **kw):
+            return []
+
+        def cancel_order(self, oid):
+            pass
+
+    cli = _Cli()
+    svc = TradingExecutionService(
+        store, executor=AlpacaPaperExecutor(store, client=cli),
+        settings=ExecutionSettings(auto_approve=True), session_id="t")
+    result = svc._flatten_all("daily_loss_circuit_breaker")
+    assert "GAP" in cli.closed, "breaker flatten must close broker positions the snapshot missed"
+    assert "GAP" in result["closed_positions"]
+
+
+def test_daily_state_rebuilds_from_events_on_restart(store):
+    """Mid-session restart must not reset the entry cap to 0 or forget resting
+    orders (re-fire => duplicate order, 2026-07-06). A new service instance
+    rebuilds fresh-fill count, requested symbols, and losing benches from today."""
+    from storage.event_schema import OrderSubmittedEvent, OrderFilledEvent, PositionClosedEvent
+    now = datetime.now()
+    store.emit(OrderSubmittedEvent(timestamp=now, mode=EventMode.PAPER, correlation_id="t",
+        message="sub", order_id="o1", symbol="AAA", side="buy", quantity=10, price=5.0,
+        broker_order_id="b1"))
+    store.emit(OrderFilledEvent(timestamp=now, mode=EventMode.PAPER, correlation_id="t",
+        message="fill", order_id="o1", symbol="AAA", fill_price=5.0, fill_quantity=10,
+        broker_order_id="b1"))
+    store.emit(PositionClosedEvent(timestamp=now, mode=EventMode.PAPER, correlation_id="t",
+        message="closed", position_id="p1", symbol="AAA", side="buy", quantity=10,
+        entry_price=5.0, exit_price=4.5, realized_pnl=-5.0, exit_reason="stop"))
+    svc = make_service(store)  # boot AFTER the events land
+    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"

trading_execution.py

@@ -403,6 +403,59 @@ class TradingExecutionService:
         )
         # serializes state mutations across the main loop and the trigger thread
         self._lock = threading.RLock()
+        # rebuild the memory-only daily state from today's events so a mid-session
+        # restart doesn't reset the entry cap to 0, un-bench stopped-out names, or
+        # 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()
+
+    def _rebuild_daily_state(self) -> None:
+        """Restore per-day counters/benches from today's event stream on boot.
+
+        ponytail: rebuilds the SLOW-path dedup + cap + bench only. The fast
+        trigger book (run_live_paper) is separately memory-only; the standing
+        broker-open-orders diff after a restart is still the backstop for a
+        resting fast-path order (see momentum-live-loop-architecture memory).
+        """
+        try:
+            import json as _json
+            today = self._now().date()
+            since = today.isoformat()
+
+            def _payload(e):
+                raw = e.get("payload") or e.get("payload_json")
+                if isinstance(raw, str):
+                    try:
+                        return _json.loads(raw)
+                    except ValueError:
+                        return {}
+                return raw or {}
+
+            # order_filled carries no side; map fills back to their BUY submit so
+            # sell/exit fills don't inflate the fresh-entry cap
+            buy_order_ids = set()
+            for e in self.store.query_events(event_type="order_submitted", since=since, limit=500):
+                p = _payload(e)
+                sym, oid = p.get("symbol"), p.get("order_id")
+                if sym:
+                    self._requested_symbols.add(str(sym))
+                if oid and str(p.get("side", "buy")).lower() == "buy":
+                    buy_order_ids.add(str(oid))
+            n_fresh = sum(
+                1 for e in self.store.query_events(event_type="order_filled", since=since, limit=500)
+                if str(_payload(e).get("order_id")) in buy_order_ids)
+            self._fresh_entries = {"date": today, "n": n_fresh}
+            for e in self.store.query_events(event_type="position_closed", since=since, limit=500):
+                p = _payload(e)
+                sym = p.get("symbol")
+                # bench only names that closed at a LOSS (mirror the live bench rule)
+                if sym and float(p.get("realized_pnl") or 0) < 0:
+                    self._exited_today.add(str(sym))
+            if n_fresh or self._requested_symbols or self._exited_today:
+                logger.info("restored daily state: %d fresh fills, %d requested, %d benched",
+                            n_fresh, len(self._requested_symbols), len(self._exited_today))
+        except Exception as exc:  # noqa: BLE001 — a rebuild failure must not block boot
+            logger.warning("daily-state rebuild skipped: %s", exc)
 
     # -- pipeline -----------------------------------------------------------
 
@@ -499,10 +552,18 @@ class TradingExecutionService:
         block the rest. Resting bracket legs are cancelled before each close so
         the held quantity is freed (else close_position 403s)."""
         result: dict = {"cancelled_entries": 0, "closed_positions": [], "errors": []}
+        client = getattr(self.executor, "client", None)
+        # broker truth up front: an entry that FILLED in the sync gap is still in
+        # _armed; cancelling its bracket parent would strip the stop/TP legs and
+        # leave the position naked. Never cancel a name the broker holds — flatten
+        # it in step 2 instead. None => positions unreadable, stay conservative.
+        broker_held = self._broker_held_symbols() or set()
         # 1) cancel unfilled (armed) entries we are tracking
         for order_id in list(self._armed):
             armed = self._armed.pop(order_id)
             symbol = armed.get("symbol", "")
+            if symbol in broker_held:
+                continue  # filled — leave the bracket intact; step 2/3 closes it
             try:
                 self.executor.cancel_entry(
                     order_id, armed.get("broker_order_id"), symbol, reason
@@ -511,8 +572,7 @@ class TradingExecutionService:
             except Exception as exc:  # noqa: BLE001
                 result["errors"].append(f"cancel {symbol}: {exc}")
             self._requested_symbols.discard(symbol)
-        # 2) market-close open positions
+        # 2) market-close open positions from the store snapshot
-        client = getattr(self.executor, "client", None)
         for symbol in sorted(self._held_symbols()):
             if client is None or not hasattr(client, "close_position"):
                 result["errors"].append(f"close {symbol}: no broker client to flatten")
@@ -522,6 +582,22 @@ class TradingExecutionService:
                 result["closed_positions"].append(symbol)  # only on a confirmed close
             except Exception as exc:  # noqa: BLE001
                 result["errors"].append(f"close {symbol}: {exc}")
+        # 3) belt-and-suspenders: close any BROKER position the store snapshot
+        # missed (filled within the ~60s sync lag). Was in close_session only, so
+        # the daily-loss breaker flatten skipped it — moved here so EVERY caller
+        # (breaker + EOD) gets it. This is the naked-through-breaker fix.
+        if client is not None and hasattr(client, "get_positions"):
+            try:
+                for p in client.get_positions(fresh=True) or []:
+                    sym = p.get("symbol")
+                    if sym and sym not in result["closed_positions"]:
+                        try:
+                            self._release_and_close(client, sym)
+                            result["closed_positions"].append(sym)
+                        except Exception as exc:  # noqa: BLE001
+                            result["errors"].append(f"close {sym}: {exc}")
+            except Exception as exc:  # noqa: BLE001
+                result["errors"].append(f"positions: {exc}")
         return result
 
     @_locked
@@ -1396,21 +1472,7 @@ class TradingExecutionService:
         halt new entries for the rest of the session.
         """
         self._session_closed = True
-        result = self._flatten_all(reason)
+        result = self._flatten_all(reason)  # fresh-broker belt now lives inside
-        # belt-and-suspenders: close any broker position the store snapshot missed
-        client = getattr(self.executor, "client", None)
-        if client is not None and hasattr(client, "get_positions"):
-            try:
-                for p in client.get_positions():
-                    sym = p.get("symbol")
-                    if sym and sym not in result["closed_positions"]:
-                        try:
-                            self._release_and_close(client, sym)
-                            result["closed_positions"].append(sym)
-                        except Exception as exc:  # noqa: BLE001
-                            result["errors"].append(f"close {sym}: {exc}")
-            except Exception as exc:  # noqa: BLE001
-                result["errors"].append(f"positions: {exc}")
         self.store.emit(
             RiskRuleTriggeredEvent(
                 timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id,
@@ -1576,6 +1638,7 @@ class TradingExecutionService:
                 and self._fresh_entries["n"] > 0):
             self._fresh_entries["n"] -= 1
 
+    @_locked
     def expire_stale_entries(self) -> list[dict]:
         """Back out of unfilled entries that timed out or broke their trigger.