diff --git a/controlroom/app.py b/controlroom/app.py index ccf3b38..fd11e6f 100644 --- a/controlroom/app.py +++ b/controlroom/app.py @@ -9,14 +9,13 @@ untouched until this reaches parity (their Phase 3-5: manual orders, settings di audit trail). Data flow: read-only projections from the shared Postgres event store + Alpaca SIP -for live prices; actions route through the existing dashboard REST (approve/reject) -and the broker API (panic = cancel all open orders, typed confirmation).""" +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 -import urllib.request from datetime import date, datetime from dotenv import load_dotenv @@ -31,17 +30,19 @@ 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 -DASH_API = os.environ.get("DASHBOARD_API", "http://127.0.0.1:8765") 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: - trig_rows, snap_ts, armed_n = _triggers_snapshot(con) + 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: @@ -77,7 +78,9 @@ def fetch_prices(symbols: list[str]) -> dict[str, float]: def fetch_account() -> dict: try: a = client.get_account() - return {"equity": float(a.get("equity") or 0), "cash": float(a.get("cash") or 0)} + 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 {} @@ -85,7 +88,8 @@ def fetch_account() -> dict: def fetch_positions() -> list[dict]: try: pos = client.get_positions(fresh=True) - naked = {n["symbol"] for n in (_unprotected_positions(con) or [])} + 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), @@ -100,9 +104,11 @@ def fetch_positions() -> list[dict]: def fetch_gates() -> list[dict]: today = date.today().isoformat() hist: dict[str, int] = {} - for (pj,) in con.execute( + with DB_LOCK: + rows = con.execute( "SELECT payload_json FROM events WHERE event_type='risk_rule_triggered' " - "AND timestamp >= ?", [today]).fetchall(): + "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])] @@ -124,19 +130,44 @@ def fetch_attribution() -> list[dict]: return out -def fetch_approvals() -> list[dict]: - return [{"order_id": q.get("order_id"), "symbol": q.get("symbol"), - "mode": q.get("approval_mode"), - "entry": (q.get("execution_request") or {}).get("entry_price"), - "qty": (q.get("execution_request") or {}).get("quantity")} - for q in query_approval_queue(store)] +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 post_dashboard(path: str, payload: dict) -> str: - req = urllib.request.Request(DASH_API + path, data=json.dumps(payload).encode(), - headers={"Content-Type": "application/json"}) - with urllib.request.urlopen(req, timeout=10) as r: - return r.read().decode()[:200] +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: @@ -190,11 +221,40 @@ def main_page(): t_live = ui.tab("Live") t_pos = ui.tab("Positions") t_gates = ui.tab("Gates & Attribution") - t_appr = ui.tab("Approvals") + 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") @@ -302,48 +362,21 @@ def main_page(): att_grid.update() ui.button("Load / refresh attribution", on_click=load_attr).props("outline") - # ---- APPROVALS ---- - with ui.tab_panel(t_appr): - appr_grid = ui.aggrid({"columnDefs": [ - {"headerName": "Order", "field": "order_id", "width": 300, "cellClass": "font-mono"}, - {"headerName": "Sym", "field": "symbol", "width": 90}, - {"headerName": "Qty", "field": "qty", "width": 80}, - {"headerName": "Entry", "field": "entry", "width": 95, "cellClass": "font-mono"}, - {"headerName": "Mode", "field": "mode", "width": 90}], - "rowData": [], "rowSelection": "single", - "domLayout": "autoHeight"}).classes("w-full") + # ---- 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 act(kind: str): - sel = await appr_grid.get_selected_row() - if not sel: - ui.notify("select a row first", type="warning") - return - with ui.dialog() as d, ui.card().classes("p-6"): - ui.label(f"{kind.upper()} {sel['symbol']} ({sel['qty']} sh)?") \ - .classes("text-lg font-bold") - with ui.row().classes("justify-end gap-2 mt-4"): - ui.button(kind, on_click=lambda: d.submit(True)) \ - .props(f"color={'positive' if kind == 'approve' else 'negative'}") - ui.button("Back", on_click=lambda: d.submit(False)) - if await d: - try: - res = await run.io_bound( - post_dashboard, f"/api/trading/approvals/{kind}", - {"order_id": sel["order_id"]}) - ui.notify(f"{kind}: {res}", type="positive") - except Exception as exc: # noqa: BLE001 - ui.notify(f"{kind} failed: {exc}", type="negative") - - with ui.row().classes("gap-2 mt-2"): - ui.button("Approve selected", on_click=lambda: act("approve")) \ - .props("color=positive") - ui.button("Reject selected", on_click=lambda: act("reject")) \ - .props("color=negative") - - async def refresh_appr(): - appr_grid.options["rowData"] = await run.io_bound(fetch_approvals) - appr_grid.update() - ui.timer(2.0, refresh_appr) + 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():