fez5587/momentum-system
1
0 Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
cad9051a96c23a277dd885ae595a0b8af404d449
momentum-system/controlroom/app.py
T
momentum devandClaude Opus 4.8 20695d5e87 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>
2026-07-04 10:06:18 -04:00

400 lines
19 KiB
Python
Raw Blame History

"""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="🎯")
Reference in new issue View Git Blame Copy Permalink