Dashboard: fix snapshot timeout (O(1) reads) + date picker + .env/host

The /api/snapshots endpoint was timing out (the screen showed all '—'): several
projections did query_events(type, limit=None), loading an event type's ENTIRE
history just to take the latest. Timed: watch_states 112s, account_orders 78s,
risk 80s -> ~280s total.

- Latest-state reads (account orders/positions/summary, session_pnl, risk_state)
  now use a DESC-LIMIT-1 index seek (_latest_event_payload) — O(1). Live snapshot
  4.7s -> 1.5s; the heavy account_orders payload no longer loads all history.
- Date picker: /api/snapshots?date=YYYY-MM-DD reviews a past session (windowed
  queries via _evt_window). Default LIVE; header <select> populated from
  available_dates; live-only panels (triggers/watch/approval/ready) skip for a
  historical view (watch_states was the 112s killer + a live concept). Frontend
  setView()/loadHistorical() switch between live stream and a one-shot date load.
- dashboard_api.py now loads .env (DATABASE_URL resolves to the configured
  Postgres) and binds 0.0.0.0 / port 8765 so it's reachable from outside WSL.

Verified: LIVE 1.5s, HISTORICAL(2026-06-18) 0.75s with full P&L/equity/fills.

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

api/main.py

@@ -57,6 +57,20 @@ def _eastern_session_date():
     return datetime.now(ZoneInfo("America/New_York")).date()
 
 
+def _available_dates(store) -> list[str]:
+    """Distinct session dates with account data — populates the date picker so
+    you can switch from LIVE to review a past day."""
+    try:
+        rows = store.con.execute(
+            "SELECT DISTINCT timestamp::date AS d FROM events "
+            "WHERE event_type = 'account_summary_updated' "
+            "ORDER BY d DESC LIMIT 90"
+        ).fetchall()
+        return [str(r[0]) for r in rows]
+    except Exception:  # noqa: BLE001
+        return []
+
+
 class DashboardState:
     """Shared state: event store path + optional live execution service."""
 
@@ -79,7 +93,10 @@ class DashboardState:
     def open_store(self) -> EventStore:
         return EventStore(self.event_db_path)
 
-    def snapshots(self) -> dict:
+    def snapshots(self, for_date: str | None = None) -> dict:
+        # for_date=None/'live' -> the live view (latest snapshots, fast windowed
+        # query); a 'YYYY-MM-DD' -> that past session's end-of-day review.
+        is_live = not for_date or str(for_date).lower() in ("live", "today")
         with self._lock:
             store = self.open_store()
             try:
@@ -87,18 +104,23 @@ class DashboardState:
                     "generated_at": datetime.now(timezone.utc).isoformat(),
                     "execution_mode": self.execution_mode,
                     "has_execution": self.execution_service is not None,
-                    "pnl": query_session_pnl(store),
-                    "risk": query_risk_state(store),
-                    "triggers": query_armed_triggers(store),
-                    "equity_curve": query_equity_curve(store),
-                    "approval_queue": query_approval_queue(store),
-                    "ready_signals": query_ready_signals_snapshot(store),
-                    "watch_states": query_watch_states_snapshot(store),
-                    "accounts": query_account_summary_snapshot(store),
-                    "positions": query_account_positions_snapshot(store),
-                    "orders": query_account_orders_snapshot(store),
+                    "view_date": None if is_live else for_date,
+                    "available_dates": _available_dates(store),
+                    "pnl": query_session_pnl(store, for_date=for_date),
+                    "risk": query_risk_state(store, for_date=for_date),
+                    "triggers": query_armed_triggers(store) if is_live else {"triggers": [], "armed": 0},
+                    "equity_curve": query_equity_curve(store, for_date=for_date),
+                    "approval_queue": query_approval_queue(store) if is_live else [],
+                    "ready_signals": query_ready_signals_snapshot(store) if is_live else [],
+                    # watch_states is a LIVE concept (and the heaviest projection
+                    # — ~99k criteria rows for a whole past day); skip it for a
+                    # historical review, which is about that day's P&L/trades.
+                    "watch_states": query_watch_states_snapshot(store) if is_live else [],
+                    "accounts": query_account_summary_snapshot(store, for_date=for_date),
+                    "positions": query_account_positions_snapshot(store, for_date=for_date),
+                    "orders": query_account_orders_snapshot(store, for_date=for_date),
                     "order_lifecycle": query_order_lifecycle_snapshot(store),
-                    "fills": query_fills_feed(store, limit=30),
+                    "fills": query_fills_feed(store, limit=30, for_date=for_date),
                     "sessions": query_session_summary(store),
                 }
             finally:
@@ -301,7 +323,8 @@ class DashboardHandler(BaseHTTPRequestHandler):
             elif route == "/api/stream":
                 self._stream()
             elif route == "/api/snapshots":
-                self._send_json(self.state.snapshots())
+                d = (self._query().get("date") or [""])[0]
+                self._send_json(self.state.snapshots(for_date=d or None))
             elif route == "/api/approval-queue":
                 self._send_json({"approval_queue": self.state.approval_queue()})
             elif route == "/api/criteria":

api/static/dashboard.html

@@ -55,6 +55,10 @@
     padding: 5px 9px; cursor: pointer;
   }
   .hdr-btn.on { color: var(--accent); border-color: #2a4a78; }
+  .hdr-sel { margin-left: 10px; background: var(--panel, #0d1117); color: var(--text);
+    border: 1px solid var(--border); border-radius: 5px; padding: 4px 8px;
+    font: 600 11px var(--mono); cursor: pointer; }
+  .hdr-sel.histor { color: #f1c40f; border-color: #6b5a16; }
 
   /* P&L strip */
   .pnl-strip {
@@ -211,6 +215,7 @@
 <header>
   <h1>MOMENTUM <span>/ control room</span></h1>
   <span id="mode-badge" class="badge paper">PAPER</span>
+  <select id="date-select" class="hdr-sel" title="View LIVE or review a past session"><option value="">● LIVE</option></select>
   <div class="spacer"></div>
   <button id="sound-toggle" class="hdr-btn" title="Toggle signal alert sound">SOUND OFF</button>
   <button id="kbd-toggle" class="hdr-btn" title="Keyboard shortcuts (?)">KEYS</button>
@@ -555,6 +560,7 @@ function rBar(entry, cur, stop) {
 
 function render(d) {
   latest = d;
+  populateDates(d.available_dates);
 
   const live = d.execution_mode === "schwab_live";
   const badge = $("mode-badge");
@@ -706,7 +712,34 @@ function render(d) {
   lastSignalKey = key;
 }
 
-let es = null, pollTimer = null;
+let es = null, pollTimer = null, viewDate = null;
+
+// populate the date picker from the snapshot's available_dates (LIVE + past days)
+function populateDates(dates) {
+  const sel = document.getElementById("date-select"); if (!sel) return;
+  if (sel.options.length === ((dates || []).length + 1)) return;  // already built
+  sel.innerHTML = '<option value="">● LIVE</option>' +
+    (dates || []).map(d => '<option value="' + d + '">' + d + '</option>').join('');
+  sel.value = viewDate || '';
+}
+// switch between live and a historical session
+function setView(date) {
+  viewDate = date || null;
+  const sel = document.getElementById("date-select");
+  if (sel) sel.classList.toggle("histor", !!viewDate);
+  if (es) { es.close(); es = null; }
+  if (pollTimer) { clearInterval(pollTimer); pollTimer = null; }
+  if (viewDate) loadHistorical(); else startStream();   // live resumes the stream
+}
+async function loadHistorical() {
+  setConn(true, "loading " + viewDate + " ...");
+  try {
+    const r = await fetch("/api/snapshots?date=" + encodeURIComponent(viewDate));
+    if (!r.ok) throw new Error("HTTP " + r.status);
+    render(await r.json());
+    setConn(true, "VIEWING " + viewDate + " (historical)");
+  } catch (e) { setConn(false, "load failed — " + e.message); }
+}
 function setConn(ok, msg) {
   $("conn-dot").className = "dot " + (ok ? "live" : "err");
   $("stamp").textContent = msg;
@@ -741,6 +774,7 @@ $("sound-toggle").onclick = () => {
   $("sound-toggle").className = "hdr-btn" + (soundOn ? " on" : "");
   if (soundOn) beep();
 };
+$("date-select").onchange = (e) => setView(e.target.value);
 $("kbd-toggle").onclick = () => $("kbd-help").classList.toggle("show");
 $("kbd-help").onclick = (e) => { if (e.target.id === "kbd-help") $("kbd-help").classList.remove("show"); };
 

dashboard_api.py

@@ -13,17 +13,28 @@ from __future__ import annotations
 
 import os
 
-from api.main import DashboardState, create_server
+from dotenv import load_dotenv
+
+# Load .env so DATABASE_URL resolves to the configured Postgres (e.g.
+# 192.168.1.5:5432) — the event store reads it. Without this the standalone
+# dashboard fell back to the legacy local DuckDB path and showed nothing.
+load_dotenv()
+
+from api.main import DashboardState, create_server  # noqa: E402
 
 
 def main() -> None:
-    host = os.environ.get("DASHBOARD_HOST", "127.0.0.1")
-    port = int(os.environ.get("DASHBOARD_PORT", "8010"))
-    db_path = os.environ.get("WATCHER_EVENT_DB_PATH", "./data/momentum.duckdb")
+    host = os.environ.get("DASHBOARD_HOST", "0.0.0.0")
+    port = int(os.environ.get("DASHBOARD_PORT", "8765"))
+    # The datastore is Postgres (DATABASE_URL); the event store reads it directly.
+    # WATCHER_EVENT_DB_PATH is a legacy fallback only.
+    db_path = os.environ.get("DATABASE_URL") or os.environ.get(
+        "WATCHER_EVENT_DB_PATH", "./data/momentum.duckdb")
     mode = os.environ.get("TRADING_EXECUTION_MODE", "alpaca_paper")
     state = DashboardState(db_path, execution_service=None, execution_mode=mode)
     server = create_server(state, host, port)
-    print(f"[dashboard] serving http://{host}:{port} (event db: {db_path}, mode: {mode})")
+    db_disp = db_path.split("@")[-1] if "@" in db_path else db_path
+    print(f"[dashboard] serving http://{host}:{port} (event db: {db_disp}, mode: {mode})")
     print("[dashboard] read-only mode — run run_live_paper.py for live actions")
     try:
         server.serve_forever()

storage/projections.py

@@ -11,6 +11,61 @@ if TYPE_CHECKING:
     pass
 
 
+def _evt_window(for_date: str | None = None) -> tuple[str | None, str | None]:
+    """(since, until) event-time bounds for a dashboard view.
+
+    LIVE (None / 'live' / 'today'): the last 45 min — enough to surface the
+    LATEST snapshot of every account/symbol while turning a full-history scan
+    into a tiny range read (the dashboard projections only need the latest, but
+    used to load the ENTIRE history — 112s on watch_states, 78s on orders).
+
+    A 'YYYY-MM-DD': that whole local calendar day, so the date picker can review
+    a past session's end-of-day P&L / trades / activity.
+    """
+    from datetime import datetime, timedelta
+    if for_date and str(for_date).lower() not in ("live", "today", ""):
+        try:
+            d = datetime.fromisoformat(str(for_date)).date()
+            start = datetime(d.year, d.month, d.day)
+            return (start.isoformat(), (start + timedelta(days=1)).isoformat())
+        except (TypeError, ValueError):
+            pass
+    return ((datetime.now() - timedelta(minutes=45)).isoformat(), None)
+
+
+def _state_until(for_date: str | None = None) -> str | None:
+    """Upper time bound for a 'latest state' read: end-of-day for a historical
+    date, None (= now) for live."""
+    from datetime import datetime, timedelta
+    if for_date and str(for_date).lower() not in ("live", "today", ""):
+        try:
+            d = datetime.fromisoformat(str(for_date)).date()
+            return (datetime(d.year, d.month, d.day) + timedelta(days=1)).isoformat()
+        except (TypeError, ValueError):
+            pass
+    return None
+
+
+def _latest_event_payload(store, event_type: str, until: str | None = None):
+    """The single most-recent event payload of a type (optionally before
+    `until`) via a DESC-LIMIT-1 index seek — O(1) vs loading the whole history.
+    Used by the latest-state snapshots (account orders/positions carry big
+    payloads; loading every one just to take the newest was 78s)."""
+    safe_type = "".join(c for c in str(event_type) if c.isalnum() or c == "_")
+    cond = f"event_type = '{safe_type}'"
+    if until:
+        safe_until = str(until).replace("'", "")
+        cond += f" AND timestamp < '{safe_until}'"
+    try:
+        rows = store.con.execute(
+            f"SELECT payload_json, timestamp FROM events WHERE {cond} "
+            f"ORDER BY timestamp DESC LIMIT 1"
+        ).fetchall()
+        return rows[0] if rows else None
+    except Exception:  # noqa: BLE001
+        return None
+
+
 def query_session_summary(store) -> list[dict]:
     """Query all session summaries from events."""
     events = store.query_events(event_type="session_summary")
@@ -163,15 +218,17 @@ def query_ready_signals_snapshot(store, session_id: str | None = None) -> list[d
     )
 
 
-def query_watch_states_snapshot(store, session_id: str | None = None) -> list[dict]:
+def query_watch_states_snapshot(store, session_id: str | None = None,
+                                for_date: str | None = None) -> list[dict]:
+    since, until = _evt_window(for_date)
     state_events = store.query_events(
         event_type="symbol_state_changed",
-        session_id=session_id,
+        session_id=session_id, since=since, until=until,
         limit=None,
     )
     criteria_events = store.query_events(
         event_type="criteria_evaluated",
-        session_id=session_id,
+        session_id=session_id, since=since, until=until,
         limit=None,
     )
 
@@ -213,88 +270,56 @@ def query_watch_states_snapshot(store, session_id: str | None = None) -> list[di
     return sorted(latest_by_symbol.values(), key=lambda row: row["symbol"])
 
 
-def query_account_summary_snapshot(store, broker_name: str | None = None) -> list[dict]:
-    events = store.query_events(event_type="account_summary_updated", limit=None)
-    latest_by_account: dict[tuple[str, str], dict] = {}
-
-    for event in events:
-        payload = json.loads(event.get("payload_json", "{}"))
+def query_account_summary_snapshot(store, broker_name: str | None = None,
+                                   for_date: str | None = None) -> list[dict]:
+    row = _latest_event_payload(store, "account_summary_updated", _state_until(for_date))
+    if not row:
+        return []
+    payload = json.loads(row[0] or "{}")
     broker = str(payload.get("broker_name") or "")
     account_id = str(payload.get("account_id") or "")
-        if not broker or not account_id:
-            continue
-        if broker_name is not None and broker != broker_name:
-            continue
-
-        latest_by_account[(broker, account_id)] = {
-            "broker_name": broker,
-            "account_id": account_id,
+    if not broker or not account_id or (broker_name is not None and broker != broker_name):
+        return []
+    return [{
+        "broker_name": broker, "account_id": account_id,
         "account_desc": payload.get("account_desc"),
         "total_equity": payload.get("total_equity"),
         "cash_balance": payload.get("cash_balance"),
         "buying_power": payload.get("buying_power"),
         "net_liquidating_value": payload.get("net_liquidating_value"),
-            "timestamp": event.get("timestamp"),
-        }
-
-    return sorted(
-        latest_by_account.values(),
-        key=lambda row: (str(row["broker_name"]), str(row["account_id"])),
-    )
+        "timestamp": row[1],
+    }]
 
 
 def query_account_positions_snapshot(
-    store, broker_name: str | None = None
+    store, broker_name: str | None = None, for_date: str | None = None
 ) -> list[dict]:
-    events = store.query_events(event_type="account_positions_updated", limit=None)
-    latest_by_account: dict[tuple[str, str], dict] = {}
-
-    for event in events:
-        payload = json.loads(event.get("payload_json", "{}"))
+    row = _latest_event_payload(store, "account_positions_updated", _state_until(for_date))
+    if not row:
+        return []
+    payload = json.loads(row[0] or "{}")
     broker = str(payload.get("broker_name") or "")
     account_id = str(payload.get("account_id") or "")
-        if not broker or not account_id:
-            continue
-        if broker_name is not None and broker != broker_name:
-            continue
-
-        latest_by_account[(broker, account_id)] = {
-            "broker_name": broker,
-            "account_id": account_id,
-            "positions": payload.get("positions", []),
-            "timestamp": event.get("timestamp"),
-        }
-
-    return sorted(
-        latest_by_account.values(),
-        key=lambda row: (str(row["broker_name"]), str(row["account_id"])),
-    )
+    if not broker or not account_id or (broker_name is not None and broker != broker_name):
+        return []
+    return [{"broker_name": broker, "account_id": account_id,
+             "positions": payload.get("positions", []), "timestamp": row[1]}]
 
 
-def query_account_orders_snapshot(store, broker_name: str | None = None) -> list[dict]:
-    events = store.query_events(event_type="account_orders_updated", limit=None)
-    latest_by_account: dict[tuple[str, str], dict] = {}
-
-    for event in events:
-        payload = json.loads(event.get("payload_json", "{}"))
+def query_account_orders_snapshot(store, broker_name: str | None = None,
+                                  for_date: str | None = None) -> list[dict]:
+    # O(1) latest event — these payloads are huge (every order incl. legs), so
+    # loading the full history to take the newest was the 78s offender.
+    row = _latest_event_payload(store, "account_orders_updated", _state_until(for_date))
+    if not row:
+        return []
+    payload = json.loads(row[0] or "{}")
     broker = str(payload.get("broker_name") or "")
     account_id = str(payload.get("account_id") or "")
-        if not broker or not account_id:
-            continue
-        if broker_name is not None and broker != broker_name:
-            continue
-
-        latest_by_account[(broker, account_id)] = {
-            "broker_name": broker,
-            "account_id": account_id,
-            "orders": payload.get("orders", []),
-            "timestamp": event.get("timestamp"),
-        }
-
-    return sorted(
-        latest_by_account.values(),
-        key=lambda row: (str(row["broker_name"]), str(row["account_id"])),
-    )
+    if not broker or not account_id or (broker_name is not None and broker != broker_name):
+        return []
+    return [{"broker_name": broker, "account_id": account_id,
+             "orders": payload.get("orders", []), "timestamp": row[1]}]
 
 
 def query_order_lifecycle_snapshot(store) -> list[dict]:
@@ -420,11 +445,13 @@ def query_symbol_criteria(store, symbol: str, session_id: str | None = None) ->
     }
 
 
-def query_fills_feed(store, limit: int = 50) -> list[dict]:
+def query_fills_feed(store, limit: int = 50, for_date: str | None = None) -> list[dict]:
     """Chronological fills + submissions for the activity feed (newest first)."""
+    since, until = _evt_window(for_date)
     rows: list[dict] = []
     for event_type in ("order_filled", "order_submitted"):
-        for event in store.query_events(event_type=event_type, limit=None):
+        for event in store.query_events(event_type=event_type, since=since,
+                                        until=until, limit=None):
             payload = json.loads(event.get("payload_json", "{}"))
             rows.append(
                 {
@@ -456,13 +483,14 @@ def _r_multiple(entry, stop, exit_price, side: str) -> float | None:
     return round((exit_price - entry) * direction / risk, 2)
 
 
-def query_session_pnl(store, session_id: str | None = None) -> dict:
-    """Day P&L + trade stats from position_closed events and live positions.
-
-    Realized stats come from closed positions; unrealized comes from the most
-    recent account positions snapshot so the strip matches the broker.
-    """
-    closed = store.query_events(event_type="position_closed", limit=None)
+def query_session_pnl(store, session_id: str | None = None,
+                      for_date: str | None = None) -> dict:
+    """Day P&L + trade stats. Realized comes from the broker equity delta (the
+    closed-event reconstruction reads $0); unrealized from the latest positions
+    snapshot. ``for_date`` scopes it to a past session for the date picker."""
+    since, until = _evt_window(for_date)
+    closed = store.query_events(event_type="position_closed",
+                                since=since, until=until, limit=None)
 
     realized = 0.0
     wins = 0
@@ -502,7 +530,7 @@ def query_session_pnl(store, session_id: str | None = None) -> dict:
     # unrealized from the latest positions snapshot (any broker)
     unrealized = 0.0
     open_positions = 0
-    snapshots = query_account_positions_snapshot(store)
+    snapshots = query_account_positions_snapshot(store, for_date=for_date)
     if snapshots:
         for position in snapshots[-1].get("positions") or []:
             open_positions += 1
@@ -521,9 +549,9 @@ def query_session_pnl(store, session_id: str | None = None) -> dict:
     # the broker equity delta: day P&L = latest equity - prior-session close
     # (last_equity); matched (realized) = day P&L - current unrealized.
     broker_day = None
-    summ = store.query_events(event_type="account_summary_updated", limit=None)
-    if summ:
-        sp = json.loads(summ[-1].get("payload_json", "{}"))  # ASC order -> latest
+    _ar = _latest_event_payload(store, "account_summary_updated", _state_until(for_date))
+    if _ar:
+        sp = json.loads(_ar[0] or "{}")  # latest equity snapshot (O(1))
         try:
             eq = float(sp.get("total_equity") or 0.0)
             le = float(sp.get("last_equity") or 0.0)
@@ -562,7 +590,7 @@ _PROT_ACTIVE = {"held", "new", "accepted", "pending_new",
                 "accepted_for_bidding", "partially_filled"}
 
 
-def query_risk_state(store) -> dict:
+def query_risk_state(store, for_date: str | None = None) -> dict:
     """Portfolio risk/health for the dashboard gauge — the conditions that
     silently cost money this session: concentration vs the caps, day P&L vs the
     daily-loss limit, data freshness, and any UNPROTECTED (naked) long. Returns a
@@ -572,11 +600,11 @@ def query_risk_state(store) -> dict:
 
     equity = None
     day_pnl = None
-    summ = store.query_events(event_type="account_summary_updated", limit=None)
+    _ar = _latest_event_payload(store, "account_summary_updated", _state_until(for_date))
     ts_latest = None
-    if summ:
-        sp = json.loads(summ[-1].get("payload_json", "{}"))
-        ts_latest = summ[-1].get("timestamp")
+    if _ar:
+        sp = json.loads(_ar[0] or "{}")
+        ts_latest = _ar[1]
         try:
             equity = float(sp.get("total_equity") or 0.0) or None
             le = float(sp.get("last_equity") or 0.0)
@@ -585,9 +613,9 @@ def query_risk_state(store) -> dict:
         except (TypeError, ValueError):
             pass
 
-    psnap = query_account_positions_snapshot(store)
+    psnap = query_account_positions_snapshot(store, for_date=for_date)
     positions = (psnap[-1].get("positions") if psnap else []) or []
-    osnap = query_account_orders_snapshot(store)
+    osnap = query_account_orders_snapshot(store, for_date=for_date)
     orders = (osnap[-1].get("orders") if osnap else []) or []
 
     sell_cover: dict = {}
@@ -682,13 +710,17 @@ def query_armed_triggers(store) -> dict:
     return {"triggers": [], "armed": 0, "timestamp": None}
 
 
-def query_equity_curve(store, points: int = 200) -> dict:
-    """Recent total-equity series for an intraday P&L-shape sparkline, plus the
-    prior-session-close baseline (last_equity). Bounded to the last 10h so it's a
-    cheap range scan and naturally scopes to the trading day."""
+def query_equity_curve(store, points: int = 200, for_date: str | None = None) -> dict:
+    """Total-equity series for an intraday P&L-shape sparkline, plus the prior-
+    session-close baseline (last_equity). Live = last 10h (the trading day); a
+    YYYY-MM-DD = that whole day, for the date picker."""
     from datetime import datetime, timedelta
-    since = (datetime.now() - timedelta(hours=10)).isoformat()
-    events = store.query_events(event_type="account_summary_updated", since=since, limit=None)
+    if for_date and str(for_date).lower() not in ("live", "today", ""):
+        since, until = _evt_window(for_date)
+    else:
+        since, until = (datetime.now() - timedelta(hours=10)).isoformat(), None
+    events = store.query_events(event_type="account_summary_updated",
+                                since=since, until=until, limit=None)
     series: list[float] = []
     baseline = None
     for e in events: