diff --git a/api/main.py b/api/main.py index 540b33c..962da1c 100644 --- a/api/main.py +++ b/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": diff --git a/api/static/dashboard.html b/api/static/dashboard.html index 691f6bc..b311d45 100644 --- a/api/static/dashboard.html +++ b/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 @@

MOMENTUM / control room

PAPER +
@@ -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 = '' + + (dates || []).map(d => '').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"); }; diff --git a/dashboard_api.py b/dashboard_api.py index 323b8ed..580d710 100644 --- a/dashboard_api.py +++ b/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() diff --git a/storage/projections.py b/storage/projections.py index 5bd13dc..8e06b80 100644 --- a/storage/projections.py +++ b/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", "{}")) - 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, - "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"])), - ) +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 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": 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", "{}")) - 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"])), - ) + 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 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", "{}")) - 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"])), - ) +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 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: