Control room: autonomous-ops redesign (owner directive: zero-interaction)
The owner runs this system fully autonomously — no approvals, no manual actions. The control room is therefore a WATCHING instrument, not a steering one: - Approvals tab REMOVED (vestigial: the loop auto-approves; interaction contradicts the operating model). REST plumbing dropped. - ACTIVITY tab added: merged live event feed (entries, fills, exits w/ realized P&L, gate cuts, ready signals) refreshing every 4s — see the autonomous system act. - TODAY summary strip on the Live tab: Day P&L (broker last_equity based, colored), trades + win rate, submissions, gate cuts, and LOOP HEARTBEAT age (green <3min, red otherwise) — at-a-glance trust that the unattended system is alive. - Thread-safety fix: all fetchers run in io_bound THREADS but shared one psycopg2 connection — now serialized behind DB_LOCK (latent race, caught during redesign). - Panic brake stays (emergency-only, typed confirmation) — the one deliberate exception to zero-interaction. Headless-verified: cards live (Day P&L, Loop 0s green), 10 ticker rows, 40-row activity feed, zero JS errors. Service restarted. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
momentum devandClaude Opus 4.8 committed
1 parent
90b4aae1b8
commit
20695d5e87
1 file changed
+95
-62
+95
-62
@@ -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():
|
||||
|
||||
Reference in new issue
Block a user