Add dynamic agent registry and options mandate
Introduces a dashboard-managed agent registry (`state/agents.json`) used by nightly research and market-open execution, including support for custom/disabled agents and per-agent time budgets. Expands personas to four agents (adds Maverick), shifts pick targets to include mandatory options, and updates prompts/finalization nudges to enforce option research without hard-failing runs. Strengthens execution safety by adding option-specific guardrails (conviction floor, premium cap, reserved option slot through trade/day/buying-power caps), validates OCC option symbols in schemas, and normalizes invalid option direction cases. The dashboard was refactored into modular views, adds an Agents admin tab, and improves live portfolio visibility with Alpaca-first data and snapshot fallback.
This commit is contained in:
Aryan Singh committed
1 parent
b3821abd69
commit
e1ac514f30
36 files changed
+1476
-294
No files matched your search
+68
-146
@@ -1,183 +1,105 @@
|
||||
"""ENTRYPOINT #4 — local Streamlit dashboard.
|
||||
|
||||
Read-only over the state DB (positions, snapshots, orders, theses), the nightly output JSON,
|
||||
and the log files. Everything the system did, in one place.
|
||||
Read-only over the trading state (positions, snapshots, orders, theses, reports, logs) plus
|
||||
one read-WRITE surface: the Agents tab, which manages the agent roster in state/agents.json
|
||||
(add/edit/disable agents, per-agent research budgets) — the nightly run reads that registry.
|
||||
|
||||
uv run streamlit run web/dashboard.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import pandas as pd # noqa: E402
|
||||
import streamlit as st # noqa: E402
|
||||
|
||||
from agents.schemas import AgentOutput # noqa: E402
|
||||
from config.settings import get_settings # noqa: E402
|
||||
|
||||
S = get_settings()
|
||||
st.set_page_config(page_title="Overnight Trading Agent", layout="wide")
|
||||
|
||||
from web.views import agents_admin, logs, orders, portfolio, reports, theses # noqa: E402
|
||||
from web.views.common import live_overview, query_df # noqa: E402
|
||||
|
||||
def _con() -> sqlite3.Connection | None:
|
||||
if not S.db_path.exists():
|
||||
return None
|
||||
con = sqlite3.connect(str(S.db_path))
|
||||
con.row_factory = sqlite3.Row
|
||||
return con
|
||||
|
||||
|
||||
def _query_df(sql: str, params: tuple = ()) -> pd.DataFrame:
|
||||
con = _con()
|
||||
if con is None:
|
||||
return pd.DataFrame()
|
||||
try:
|
||||
return pd.read_sql_query(sql, con, params=params)
|
||||
except Exception:
|
||||
return pd.DataFrame()
|
||||
finally:
|
||||
con.close()
|
||||
|
||||
|
||||
def _trade_dates() -> list[str]:
|
||||
if not S.output_dir.exists():
|
||||
return []
|
||||
return sorted((p.name for p in S.output_dir.iterdir() if p.is_dir()), reverse=True)
|
||||
|
||||
# ---------------- sidebar ----------------
|
||||
_REFRESH = {"Off": 0, "30s": 30, "60s": 60, "5 min": 300}
|
||||
refresh_choice = st.sidebar.selectbox("Auto-refresh", list(_REFRESH), index=0)
|
||||
st.sidebar.caption("Alpaca is the source of truth for live positions.")
|
||||
|
||||
# ---------------- header ----------------
|
||||
st.title("🌙 Overnight Trading Agent")
|
||||
|
||||
# latest snapshot per account (one book per persona)
|
||||
# latest snapshot per account (one book per persona), plus the prior day's close for deltas
|
||||
_LATEST = """
|
||||
SELECT s.account, s.equity, s.cash, s.buying_power FROM snapshots s
|
||||
JOIN (SELECT account, MAX(ts) mt FROM snapshots GROUP BY account) m
|
||||
ON s.account = m.account AND s.ts = m.mt
|
||||
"""
|
||||
latest = _query_df(_LATEST)
|
||||
c1, c2, c3, c4 = st.columns(4)
|
||||
if not latest.empty:
|
||||
c1.metric("Total Equity", f"${latest['equity'].sum():,.0f}")
|
||||
c2.metric("Total Cash", f"${latest['cash'].sum():,.0f}")
|
||||
c3.metric("Accounts", str(len(latest)))
|
||||
pos = _query_df("SELECT COALESCE(SUM(unrealized_pl),0) AS pl FROM positions")
|
||||
c4.metric("Unrealized PnL", f"${pos.iloc[0]['pl']:,.0f}" if not pos.empty else "$0")
|
||||
_PREV = """
|
||||
SELECT s.account, s.equity FROM snapshots s
|
||||
JOIN (
|
||||
SELECT account, MAX(ts) mt FROM snapshots
|
||||
WHERE DATE(ts) < (SELECT DATE(MAX(ts)) FROM snapshots)
|
||||
GROUP BY account
|
||||
) m ON s.account = m.account AND s.ts = m.mt
|
||||
"""
|
||||
# LIVE-first: pull accounts + positions straight from Alpaca (60s cache); the DB snapshot
|
||||
# (written at the last morning reconcile) is only the offline fallback. Deltas always
|
||||
# compare against the previous day's DB snapshot.
|
||||
live_acc, live_pos = live_overview()
|
||||
if not live_acc.empty:
|
||||
latest, source = live_acc, "live"
|
||||
else:
|
||||
c1.info("No account snapshot yet — run execution once Alpaca keys are set.")
|
||||
latest, source = query_df(_LATEST), "snapshot"
|
||||
prev = query_df(_PREV)
|
||||
prev_eq = dict(zip(prev["account"], prev["equity"])) if not prev.empty else {}
|
||||
|
||||
tab_port, tab_theses, tab_reports, tab_orders, tab_logs = st.tabs(
|
||||
["📊 Portfolio", "🧠 Tonight's Theses", "📝 Reports", "🧾 Orders", "📜 Logs"]
|
||||
if latest.empty:
|
||||
st.info("No account data yet — set Alpaca keys in .env (live) or run execution once (snapshot).")
|
||||
else:
|
||||
st.caption("🟢 live from Alpaca (refreshes ≤60s)" if source == "live"
|
||||
else "🟡 offline — showing last reconciled snapshot")
|
||||
total_eq = latest["equity"].sum()
|
||||
total_prev = sum(prev_eq.get(a, e) for a, e in zip(latest["account"], latest["equity"]))
|
||||
if source == "live":
|
||||
upl = live_pos["unrealized_pl"].sum() if not live_pos.empty else 0.0
|
||||
else:
|
||||
pos = query_df("SELECT COALESCE(SUM(unrealized_pl),0) AS pl FROM positions")
|
||||
upl = pos.iloc[0]["pl"] if not pos.empty else 0.0
|
||||
|
||||
cols = st.columns(2 + len(latest))
|
||||
cols[0].metric(
|
||||
"Total Equity", f"${total_eq:,.0f}",
|
||||
delta=f"${total_eq - total_prev:+,.0f}" if prev_eq else None,
|
||||
)
|
||||
cols[1].metric("Unrealized PnL", f"${upl:,.0f}")
|
||||
for col, (_, acc) in zip(cols[2:], latest.iterrows()):
|
||||
p = prev_eq.get(acc["account"])
|
||||
col.metric(
|
||||
acc["account"], f"${acc['equity']:,.0f}",
|
||||
delta=f"${acc['equity'] - p:+,.0f}" if p is not None else None,
|
||||
)
|
||||
|
||||
tab_port, tab_theses, tab_reports, tab_orders, tab_agents, tab_logs = st.tabs(
|
||||
["📊 Portfolio", "🧠 Tonight's Theses", "📝 Reports", "🧾 Orders", "🤖 Agents", "📜 Logs"]
|
||||
)
|
||||
|
||||
# ---------------- portfolio ----------------
|
||||
with tab_port:
|
||||
if latest.empty:
|
||||
st.info("No positions recorded yet — run execution once Alpaca keys are set.")
|
||||
else:
|
||||
for _, acc in latest.iterrows():
|
||||
st.subheader(f"{acc['account']} · ${acc['equity']:,.0f} equity · ${acc['cash']:,.0f} cash")
|
||||
p = _query_df(
|
||||
"SELECT symbol, qty, avg_entry, market_value, unrealized_pl FROM positions "
|
||||
"WHERE account = ? ORDER BY market_value DESC",
|
||||
(acc["account"],),
|
||||
)
|
||||
if p.empty:
|
||||
st.caption("all cash — no open positions")
|
||||
else:
|
||||
st.dataframe(p, use_container_width=True, hide_index=True)
|
||||
equity = _query_df("SELECT ts, account, equity FROM snapshots ORDER BY ts")
|
||||
if len(equity) > 1:
|
||||
st.subheader("Equity curves")
|
||||
equity["ts"] = pd.to_datetime(equity["ts"])
|
||||
st.line_chart(equity.pivot_table(index="ts", columns="account", values="equity"))
|
||||
|
||||
# ---------------- theses ----------------
|
||||
portfolio.render(latest, live_pos if source == "live" else None)
|
||||
with tab_theses:
|
||||
dates = _trade_dates()
|
||||
if not dates:
|
||||
st.info("No nightly output yet. Run: `uv run python -m orchestration.run_nightly --compressed`")
|
||||
else:
|
||||
date = st.selectbox("Trade date", dates)
|
||||
day_dir = S.output_dir / date
|
||||
files = sorted(f for f in day_dir.glob("*.json") if not f.name.startswith("_"))
|
||||
cols = st.columns(max(len(files), 1))
|
||||
for col, f in zip(cols, files):
|
||||
try:
|
||||
out = AgentOutput.model_validate_json(f.read_text())
|
||||
except Exception as e:
|
||||
col.error(f"{f.name}: {e}")
|
||||
continue
|
||||
with col:
|
||||
st.markdown(f"### {out.agent_name}")
|
||||
st.caption(f"status **{out.status.value}** · {len(out.picks)} picks · "
|
||||
f"{out.steps_used} steps · {out.wall_seconds:.0f}s")
|
||||
for p in out.picks:
|
||||
with st.expander(f"{p.ticker} · conv {p.conviction:.2f} · {p.time_horizon.value}"):
|
||||
st.write(p.thesis)
|
||||
st.caption(
|
||||
f"weight {p.requested_weight_pct:.0%} · stop {p.stop_loss_pct} · "
|
||||
f"take {p.take_profit_pct} · {p.direction.value}"
|
||||
)
|
||||
if p.catalysts:
|
||||
st.caption("catalysts: " + ", ".join(p.catalysts))
|
||||
if out.reasoning_trace:
|
||||
with st.expander("🔍 reasoning trace"):
|
||||
st.code("\n".join(out.reasoning_trace), language="text")
|
||||
if out.errors:
|
||||
st.warning("errors: " + "; ".join(out.errors))
|
||||
|
||||
# ---------------- reports ----------------
|
||||
theses.render()
|
||||
with tab_reports:
|
||||
rdir = S.reports_dir
|
||||
rdates = (
|
||||
sorted((p.name for p in rdir.iterdir() if p.is_dir()), reverse=True)
|
||||
if rdir.exists() else []
|
||||
)
|
||||
if not rdates:
|
||||
st.info("No reports yet — generated during the nightly run (state/reports/).")
|
||||
else:
|
||||
rdate = st.selectbox("Report date", rdates, key="rdate")
|
||||
ddir = rdir / rdate
|
||||
# daily.md first, then each agent
|
||||
mds = sorted(ddir.glob("*.md"), key=lambda p: (p.name != "daily.md", p.name))
|
||||
for md in mds:
|
||||
st.markdown(md.read_text())
|
||||
pdf = md.with_suffix(".pdf")
|
||||
if pdf.exists():
|
||||
st.download_button(
|
||||
f"⬇ Download {pdf.name}", pdf.read_bytes(), file_name=pdf.name,
|
||||
mime="application/pdf", key=f"dl_{md.stem}",
|
||||
)
|
||||
for png in sorted(ddir.glob(f"{md.stem}_*.png")):
|
||||
st.image(str(png))
|
||||
st.divider()
|
||||
|
||||
# ---------------- orders ----------------
|
||||
reports.render()
|
||||
with tab_orders:
|
||||
orders = _query_df(
|
||||
"SELECT trade_date, agent, ticker, side, qty, est_price, stop_price, take_price, "
|
||||
"status, alpaca_order_id, submitted_at FROM orders ORDER BY submitted_at DESC LIMIT 200"
|
||||
)
|
||||
if orders.empty:
|
||||
st.info("No orders placed yet.")
|
||||
else:
|
||||
st.dataframe(orders, use_container_width=True, hide_index=True)
|
||||
|
||||
# ---------------- logs ----------------
|
||||
orders.render()
|
||||
with tab_agents:
|
||||
agents_admin.render()
|
||||
with tab_logs:
|
||||
log_files = sorted(S.log_dir.glob("*.log"), reverse=True) if S.log_dir.exists() else []
|
||||
if not log_files:
|
||||
st.info("No logs yet.")
|
||||
else:
|
||||
choice = st.selectbox("Log file", [f.name for f in log_files])
|
||||
n = st.slider("Lines", 20, 1000, 200, 20)
|
||||
text = (S.log_dir / choice).read_text().splitlines()[-n:]
|
||||
st.code("\n".join(text), language="text")
|
||||
logs.render()
|
||||
|
||||
st.caption("Read-only. Refresh the page to update. Alpaca is the source of truth for live positions.")
|
||||
# ---------------- auto-refresh ----------------
|
||||
_interval = _REFRESH[refresh_choice]
|
||||
if _interval:
|
||||
time.sleep(_interval)
|
||||
st.rerun()
|
||||
Reference in new issue
Block a user