momentum devandClaude Fable 5
2e39b756ba
The 07-21 morning outage (loop booted post-EOD on the 20th, survived
midnight, silently blocked every entry while signals went READY) had zero
UI surface. New query_loop_session projection parses the live session id's
embedded date; a red banner fires when it isn't today's — the exact day-2
poison signature — telling the operator to restart the loop. Armed
Triggers rows now also show rvol_tod (the arming signal since 5597250)
next to legacy rvol.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1012 lines
42 KiB
Python
1012 lines
42 KiB
Python
import json
|
|
|
|
"""Projections for Milestone 2 read models.
|
|
|
|
Aggregated views for querying sessions, symbols, approvals, and positions.
|
|
"""
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
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 _hist_window(for_date: str | None = None) -> tuple[str | None, str | None]:
|
|
"""(since, until) for DAY-SCOPED reads (fills, closed trades). Live = UNBOUNDED
|
|
(these event types are tiny — fills/closed are rare), so an idle loop or a
|
|
test's fixed-date fixtures still show. A 'YYYY-MM-DD' = that whole day."""
|
|
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 (None, 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} "
|
|
# created_at DESC tie-breaks events sharing a timestamp (e.g. two
|
|
# account snapshots emitted in the same second) so we return the
|
|
# truly newest by insertion order, not an arbitrary one.
|
|
f"ORDER BY timestamp DESC, created_at 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")
|
|
|
|
summaries = []
|
|
for event in events:
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
if "session_id" in payload:
|
|
summaries.append(
|
|
{
|
|
"session_id": payload["session_id"],
|
|
"event_id": event["id"],
|
|
"timestamp": event["timestamp"],
|
|
}
|
|
)
|
|
|
|
return summaries
|
|
|
|
|
|
def query_approval_queue(store) -> list[dict]:
|
|
requested = store.query_events(event_type="order_approval_requested", limit=None)
|
|
approved = store.query_events(event_type="order_approved", limit=None)
|
|
rejected = store.query_events(event_type="order_rejected", limit=None)
|
|
submitted = store.query_events(event_type="order_submitted", limit=None)
|
|
cancelled = store.query_events(event_type="order_cancelled", limit=None)
|
|
|
|
resolved_ids = {
|
|
str(json.loads(event.get("payload_json", "{}")).get("order_id"))
|
|
for event in approved + rejected + submitted + cancelled
|
|
if json.loads(event.get("payload_json", "{}")).get("order_id") is not None
|
|
}
|
|
|
|
queue = []
|
|
for event in requested:
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
order_id = payload.get("order_id")
|
|
if order_id is None or str(order_id) in resolved_ids:
|
|
continue
|
|
queue.append(
|
|
{
|
|
"event_id": event["id"],
|
|
"order_id": str(order_id),
|
|
"symbol": payload.get("symbol"),
|
|
"requested_by": payload.get("requested_by"),
|
|
"approval_mode": payload.get("approval_mode"),
|
|
"execution_mode": payload.get("execution_mode"),
|
|
"execution_request": payload.get("execution_request", {}),
|
|
"timestamp": event["timestamp"],
|
|
}
|
|
)
|
|
return queue
|
|
|
|
|
|
def query_ready_signals_snapshot(store, session_id: str | None = None) -> list[dict]:
|
|
"""Latest signal_ready per symbol, newest first. Feeds the dashboard."""
|
|
events = store.query_events(
|
|
event_type="signal_ready", session_id=session_id, limit=None
|
|
)
|
|
latest_by_symbol: dict[str, dict] = {}
|
|
for event in events:
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
symbol = payload.get("symbol")
|
|
if not symbol:
|
|
continue
|
|
signal_data = payload.get("signal_data") or {}
|
|
latest_by_symbol[str(symbol)] = {
|
|
"symbol": str(symbol),
|
|
"signal_type": payload.get("signal_type"),
|
|
"confidence": payload.get("confidence"),
|
|
"entry_price": signal_data.get("entry_price"),
|
|
"stop_loss_price": signal_data.get("stop_loss_price"),
|
|
"quality_score": signal_data.get("quality_score"),
|
|
"vwap": signal_data.get("vwap"),
|
|
"above_vwap": signal_data.get("above_vwap"),
|
|
"day_open": signal_data.get("day_open"),
|
|
"cum_volume": signal_data.get("cum_volume"),
|
|
"gap_pct": signal_data.get("gap_pct"),
|
|
"relative_volume": signal_data.get("relative_volume"),
|
|
"timestamp": event.get("timestamp"),
|
|
}
|
|
return sorted(
|
|
latest_by_symbol.values(),
|
|
key=lambda row: str(row.get("timestamp") or ""),
|
|
reverse=True,
|
|
)
|
|
|
|
|
|
def query_watch_states_snapshot(store, session_id: str | None = None,
|
|
for_date: str | None = None) -> list[dict]:
|
|
since, until = _evt_window(for_date)
|
|
# symbol_state_changed is tiny (~2k all-time) and we want the current state
|
|
# of every watched symbol, so load it unbounded; only criteria_evaluated
|
|
# (the ~120k-row monster that made this projection 112s) needs the window.
|
|
state_events = store.query_events(
|
|
event_type="symbol_state_changed",
|
|
session_id=session_id, limit=None,
|
|
)
|
|
criteria_events = store.query_events(
|
|
event_type="criteria_evaluated",
|
|
session_id=session_id, since=since, until=until,
|
|
limit=None,
|
|
)
|
|
|
|
latest_by_symbol: dict[str, dict] = {}
|
|
|
|
def _blank(symbol_key: str) -> dict:
|
|
return {
|
|
"symbol": symbol_key,
|
|
"state": None,
|
|
"previous_state": None,
|
|
"state_reason": None,
|
|
"last_transition_at": None,
|
|
"last_score": None,
|
|
"last_criteria_at": None,
|
|
}
|
|
|
|
for event in criteria_events:
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
symbol = payload.get("symbol")
|
|
if not symbol:
|
|
continue
|
|
symbol_key = str(symbol)
|
|
entry = latest_by_symbol.setdefault(symbol_key, _blank(symbol_key))
|
|
entry["last_score"] = payload.get("success_score_pct")
|
|
entry["last_criteria_at"] = event.get("timestamp")
|
|
|
|
for event in state_events:
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
symbol = payload.get("symbol")
|
|
if not symbol:
|
|
continue
|
|
symbol_key = str(symbol)
|
|
entry = latest_by_symbol.setdefault(symbol_key, _blank(symbol_key))
|
|
entry["state"] = payload.get("new_state")
|
|
entry["previous_state"] = payload.get("previous_state")
|
|
entry["state_reason"] = payload.get("state_reason")
|
|
entry["last_transition_at"] = event.get("timestamp")
|
|
|
|
return sorted(latest_by_symbol.values(), key=lambda row: row["symbol"])
|
|
|
|
|
|
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, for_date: str | None = None
|
|
) -> list[dict]:
|
|
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,
|
|
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]:
|
|
event_types = [
|
|
"order_submitted",
|
|
"order_approval_requested",
|
|
"order_approved",
|
|
"order_rejected",
|
|
"order_filled",
|
|
"order_cancelled",
|
|
]
|
|
latest_by_order: dict[str, dict] = {}
|
|
|
|
for event_type in event_types:
|
|
events = store.query_events(event_type=event_type, limit=None)
|
|
for event in events:
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
order_id = payload.get("order_id")
|
|
if not order_id:
|
|
continue
|
|
entry = latest_by_order.setdefault(
|
|
str(order_id),
|
|
{
|
|
"order_id": str(order_id),
|
|
"symbol": payload.get("symbol"),
|
|
"status": None,
|
|
"submitted_at": None,
|
|
"updated_at": None,
|
|
"side": None,
|
|
"quantity": None,
|
|
"price": None,
|
|
"fill_price": None,
|
|
"fill_quantity": None,
|
|
"approval_mode": None,
|
|
"rejection_reason": None,
|
|
},
|
|
)
|
|
|
|
if event_type == "order_submitted":
|
|
entry["status"] = "submitted"
|
|
entry["submitted_at"] = event.get("timestamp")
|
|
entry["side"] = payload.get("side")
|
|
entry["quantity"] = payload.get("quantity")
|
|
entry["price"] = payload.get("price")
|
|
elif event_type == "order_approval_requested":
|
|
entry["status"] = "approval_requested"
|
|
entry["approval_mode"] = payload.get("approval_mode")
|
|
elif event_type == "order_approved":
|
|
entry["status"] = "approved"
|
|
elif event_type == "order_rejected":
|
|
entry["status"] = "rejected"
|
|
entry["rejection_reason"] = payload.get("rejection_reason")
|
|
elif event_type == "order_filled":
|
|
entry["status"] = "filled"
|
|
entry["fill_price"] = payload.get("fill_price")
|
|
entry["fill_quantity"] = payload.get("fill_quantity")
|
|
elif event_type == "order_cancelled":
|
|
entry["status"] = "cancelled"
|
|
entry["rejection_reason"] = payload.get("cancel_reason")
|
|
|
|
entry["updated_at"] = event.get("timestamp")
|
|
|
|
return sorted(latest_by_order.values(), key=lambda row: row["order_id"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Interactive dashboard projections (criteria detail, session P&L, fills feed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Human-readable labels for the nine setup criteria, in evaluation order.
|
|
CRITERIA_LABELS = {
|
|
"sufficient_data": "Sufficient data",
|
|
"gap": "Gap %",
|
|
"relative_volume": "Relative volume",
|
|
"impulse": "Impulse leg",
|
|
"pullback": "Pullback depth",
|
|
"pullback_volume": "Pullback volume",
|
|
"vwap": "Holding VWAP",
|
|
"candle_quality": "Candle quality",
|
|
"breakout": "Breakout trigger",
|
|
}
|
|
|
|
|
|
def query_symbol_criteria(store, symbol: str, session_id: str | None = None) -> dict:
|
|
"""Latest per-criterion pass/fail breakdown for one symbol.
|
|
|
|
Powers the click-to-expand criteria panel: returns every criterion with a
|
|
passed flag and label so a 'blocked' symbol becomes legible at a glance.
|
|
"""
|
|
events = store.query_events(
|
|
event_type="criteria_evaluated",
|
|
symbol=symbol,
|
|
session_id=session_id,
|
|
limit=None,
|
|
)
|
|
if not events:
|
|
return {"symbol": symbol, "criteria": [], "score": None, "evaluated_at": None}
|
|
|
|
latest = events[-1]
|
|
payload = json.loads(latest.get("payload_json", "{}"))
|
|
results = payload.get("criteria_results") or {}
|
|
passed = set(results.get("passed") or [])
|
|
failed = set(results.get("failed") or [])
|
|
|
|
criteria = []
|
|
for key, label in CRITERIA_LABELS.items():
|
|
if key in passed:
|
|
state = True
|
|
elif key in failed:
|
|
state = False
|
|
else:
|
|
state = None # not evaluated this pass
|
|
criteria.append({"key": key, "label": label, "passed": state})
|
|
|
|
return {
|
|
"symbol": symbol,
|
|
"criteria": criteria,
|
|
"passed_count": payload.get("passed_criteria"),
|
|
"total_count": payload.get("total_criteria"),
|
|
"score": payload.get("success_score_pct"),
|
|
"status": payload.get("status"),
|
|
"evaluated_at": latest.get("timestamp"),
|
|
}
|
|
|
|
|
|
def query_fills_feed(store, limit: int = 50, for_date: str | None = None,
|
|
symbol: str | None = None) -> list[dict]:
|
|
"""Chronological fills + submissions for the activity feed (newest first).
|
|
|
|
``symbol`` optionally scopes to one ticker (used by the /api/chart marker
|
|
fetch, plan 013) — default behavior (None) is unchanged."""
|
|
since, until = _hist_window(for_date)
|
|
rows: list[dict] = []
|
|
for event_type in ("order_filled", "order_submitted"):
|
|
for event in store.query_events(event_type=event_type, since=since,
|
|
until=until, limit=None):
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
if symbol is not None and payload.get("symbol") != symbol:
|
|
continue
|
|
rows.append(
|
|
{
|
|
"kind": event_type,
|
|
"order_id": payload.get("order_id"),
|
|
"symbol": payload.get("symbol"),
|
|
"side": payload.get("side"),
|
|
"quantity": payload.get("fill_quantity") or payload.get("quantity"),
|
|
"price": payload.get("fill_price") or payload.get("price"),
|
|
"timestamp": event.get("timestamp"),
|
|
}
|
|
)
|
|
rows.sort(key=lambda r: str(r.get("timestamp") or ""), reverse=True)
|
|
return rows[:limit]
|
|
|
|
|
|
def query_gate_receipts(store, symbol: str | None = None, limit: int = 100,
|
|
for_date: str | None = None) -> list[dict]:
|
|
"""Latest ``entry_gate_evaluated`` decision receipts (plan 014), newest
|
|
first — the per-gate pass/reject breakdown behind every entry evaluation.
|
|
|
|
LIVE LENS, not an archive: the nightly housekeeping in
|
|
research/premarket_eval.py purges these events after 14 days.
|
|
``symbol`` matches case-insensitively when given (mirrors query_fills_feed's
|
|
in-Python filter idiom rather than store.query_events' exact-match one)."""
|
|
since, until = _hist_window(for_date)
|
|
sym_upper = symbol.upper() if symbol else None
|
|
rows: list[dict] = []
|
|
for event in store.query_events(event_type="entry_gate_evaluated", since=since,
|
|
until=until, limit=None):
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
if sym_upper is not None and str(payload.get("symbol") or "").upper() != sym_upper:
|
|
continue
|
|
rows.append({
|
|
"timestamp": event.get("timestamp"),
|
|
"symbol": payload.get("symbol"),
|
|
"source": payload.get("source"),
|
|
"passed": payload.get("passed"),
|
|
"reject_reason": payload.get("reject_reason"),
|
|
"gates": payload.get("gates") or {},
|
|
})
|
|
rows.sort(key=lambda r: str(r.get("timestamp") or ""), reverse=True)
|
|
return rows[:limit]
|
|
|
|
|
|
def _r_multiple(entry, stop, exit_price, side: str) -> float | None:
|
|
"""Realized R-multiple given entry/stop/exit. None if stop distance is 0."""
|
|
try:
|
|
entry = float(entry)
|
|
stop = float(stop)
|
|
exit_price = float(exit_price)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
risk = abs(entry - stop)
|
|
if risk <= 0:
|
|
return None
|
|
direction = 1.0 if side != "sell" else -1.0
|
|
return round((exit_price - entry) * direction / risk, 2)
|
|
|
|
|
|
def _sum_position_unrealized(positions) -> tuple[float, int]:
|
|
"""(sum of unrealized P&L, position count) for a positions-snapshot list.
|
|
Live sync stores the key as `unrealized_pnl`; accept the alpaca-native
|
|
`unrealized_pl` alias, else derive from qty*(price-cost)."""
|
|
total = 0.0
|
|
count = 0
|
|
for position in positions or []:
|
|
count += 1
|
|
try:
|
|
upl = position.get("unrealized_pnl")
|
|
if upl is None:
|
|
upl = position.get("unrealized_pl")
|
|
if upl is None:
|
|
qty = float(position.get("quantity") or 0.0)
|
|
cur = float(position.get("current_price") or 0.0)
|
|
avg = float(position.get("avg_entry_price") or 0.0)
|
|
upl = qty * (cur - avg)
|
|
total += float(upl or 0.0)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
return total, count
|
|
|
|
|
|
def query_alltime_score(store) -> dict:
|
|
"""Cumulative scorecard across EVERY closed round-trip (all market days).
|
|
Feeds the dashboard's all-time panel: total trades, win rate, total realized,
|
|
avg R / expectancy, the distinct trading days covered, and the best/worst
|
|
single day by realized P&L. Reads position_closed events (live reconcile +
|
|
backfill); {zeros} until any exist."""
|
|
closed = store.query_events(event_type="position_closed", since=None,
|
|
until=None, limit=None)
|
|
wins = losses = 0
|
|
total_realized = 0.0
|
|
r_multiples: list[float] = []
|
|
by_day: dict[str, float] = {}
|
|
for event in closed:
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
pnl = payload.get("realized_pnl")
|
|
if pnl is None:
|
|
continue
|
|
pnl = float(pnl)
|
|
total_realized += pnl
|
|
if pnl > 0:
|
|
wins += 1
|
|
elif pnl < 0:
|
|
losses += 1
|
|
r = _r_multiple(payload.get("entry_price"), payload.get("stop_loss_price"),
|
|
payload.get("exit_price"), payload.get("side") or "buy")
|
|
if r is not None:
|
|
r_multiples.append(r)
|
|
day = str(event.get("timestamp") or "")[:10]
|
|
if day:
|
|
by_day[day] = by_day.get(day, 0.0) + pnl
|
|
|
|
trades = wins + losses
|
|
best_day = worst_day = None
|
|
if by_day:
|
|
bd = max(by_day.items(), key=lambda kv: kv[1])
|
|
wd = min(by_day.items(), key=lambda kv: kv[1])
|
|
best_day = {"date": bd[0], "pnl": round(bd[1], 2)}
|
|
worst_day = {"date": wd[0], "pnl": round(wd[1], 2)}
|
|
return {
|
|
"trades": trades,
|
|
"wins": wins,
|
|
"losses": losses,
|
|
"win_rate": round(wins / trades, 3) if trades else None,
|
|
"total_realized": round(total_realized, 2),
|
|
"avg_r_multiple": round(sum(r_multiples) / len(r_multiples), 2) if r_multiples else None,
|
|
"trading_days": len(by_day),
|
|
"best_day": best_day,
|
|
"worst_day": worst_day,
|
|
}
|
|
|
|
|
|
def query_daily_performance(store, max_days: int = 30) -> list[dict]:
|
|
"""One row per market day for the day-over-day comparison curve:
|
|
{date, day_pnl (broker equity delta), cum_pnl (running), realized (closed
|
|
round-trips that day), trades, win_rate, equity (EOD)}. Built from the
|
|
account summaries (EOD equity per day) joined to the closed-trade journal.
|
|
Flat non-trading days (|day_pnl|<$1 and no trades) are dropped so the curve
|
|
is trading days only. Most-recent ``max_days``."""
|
|
# EOD equity + baseline per day (events are timestamp-ASC, so last wins)
|
|
by_day: dict[str, dict] = {}
|
|
for e in store.query_events(event_type="account_summary_updated", limit=None):
|
|
day = str(e.get("timestamp") or "")[:10]
|
|
if not day:
|
|
continue
|
|
p = json.loads(e.get("payload_json", "{}"))
|
|
try:
|
|
eq = float(p.get("total_equity") or 0.0)
|
|
le = float(p.get("last_equity") or 0.0)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if eq:
|
|
d = by_day.setdefault(day, {"eq": None, "le": None})
|
|
d["eq"] = eq
|
|
if le:
|
|
d["le"] = le
|
|
|
|
# closed-trade stats per day
|
|
trades_by_day: dict[str, dict] = {}
|
|
for e in store.query_events(event_type="position_closed", limit=None):
|
|
day = str(e.get("timestamp") or "")[:10]
|
|
p = json.loads(e.get("payload_json", "{}"))
|
|
pnl = p.get("realized_pnl")
|
|
if not day or pnl is None:
|
|
continue
|
|
pnl = float(pnl)
|
|
t = trades_by_day.setdefault(day, {"n": 0, "w": 0, "realized": 0.0})
|
|
t["n"] += 1
|
|
t["w"] += 1 if pnl > 0 else 0
|
|
t["realized"] += pnl
|
|
|
|
# restrict to NYSE trading days so weekend/holiday mark-drift on held
|
|
# positions doesn't masquerade as a performance day (degrade gracefully)
|
|
from datetime import date as _date
|
|
try:
|
|
from runtime.market_calendar import is_trading_day
|
|
except Exception: # noqa: BLE001
|
|
is_trading_day = None
|
|
|
|
out: list[dict] = []
|
|
cum = 0.0
|
|
for day in sorted(set(by_day) | set(trades_by_day)):
|
|
bd = by_day.get(day)
|
|
t = trades_by_day.get(day, {"n": 0, "w": 0, "realized": 0.0})
|
|
day_pnl = (bd["eq"] - bd["le"]) if (bd and bd["eq"] and bd["le"]) else None
|
|
if day_pnl is None:
|
|
day_pnl = t["realized"] if t["n"] else 0.0
|
|
if t["n"] == 0 and is_trading_day is not None:
|
|
try:
|
|
if not is_trading_day(_date.fromisoformat(day)):
|
|
continue # non-trading day, no trades -> drop
|
|
except (TypeError, ValueError):
|
|
pass
|
|
if abs(day_pnl) < 1.0 and t["n"] == 0: # flat non-trading day -> skip
|
|
continue
|
|
cum += day_pnl
|
|
out.append({
|
|
"date": day,
|
|
"day_pnl": round(day_pnl, 2),
|
|
"cum_pnl": round(cum, 2),
|
|
"realized": round(t["realized"], 2),
|
|
"trades": t["n"],
|
|
"win_rate": round(t["w"] / t["n"], 3) if t["n"] else None,
|
|
"equity": round(bd["eq"], 2) if (bd and bd["eq"]) else None,
|
|
})
|
|
return out[-max_days:]
|
|
|
|
|
|
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 = _hist_window(for_date)
|
|
if since is None and until is None:
|
|
# LIVE: scope the journal to TODAY's session. Before any closes existed
|
|
# this was unbounded (harmless); now that the journal is populated,
|
|
# unbounded would make "live" show the all-time trade list — which is the
|
|
# separate All-Time panel. Today only, so live = this market day.
|
|
from datetime import datetime
|
|
_now = datetime.now()
|
|
since = datetime(_now.year, _now.month, _now.day).isoformat()
|
|
closed = store.query_events(event_type="position_closed",
|
|
since=since, until=until, limit=None)
|
|
|
|
realized = 0.0
|
|
wins = 0
|
|
losses = 0
|
|
r_multiples: list[float] = []
|
|
trades: list[dict] = []
|
|
|
|
for event in closed:
|
|
payload = json.loads(event.get("payload_json", "{}"))
|
|
pnl = payload.get("realized_pnl")
|
|
if pnl is None:
|
|
continue
|
|
pnl = float(pnl)
|
|
realized += pnl
|
|
if pnl > 0:
|
|
wins += 1
|
|
elif pnl < 0:
|
|
losses += 1
|
|
r = _r_multiple(
|
|
payload.get("entry_price"),
|
|
payload.get("stop_loss_price"),
|
|
payload.get("exit_price"),
|
|
payload.get("side") or "buy",
|
|
)
|
|
if r is not None:
|
|
r_multiples.append(r)
|
|
trades.append(
|
|
{
|
|
"symbol": payload.get("symbol"),
|
|
"entry_price": payload.get("entry_price"),
|
|
"exit_price": payload.get("exit_price"),
|
|
"realized_pnl": round(pnl, 2),
|
|
"r_multiple": r,
|
|
"exit_reason": payload.get("exit_reason"),
|
|
"timestamp": event.get("timestamp"),
|
|
}
|
|
)
|
|
|
|
# End-of-window open-position unrealized (the latest snapshot in scope).
|
|
snapshots = query_account_positions_snapshot(store, for_date=for_date)
|
|
unrealized, open_positions = _sum_position_unrealized(
|
|
snapshots[-1].get("positions") if snapshots else [])
|
|
|
|
# START-OF-DAY baseline: the unrealized carried INTO this day on positions
|
|
# held overnight. Without it, realized = broker_day - unrealized fabricated
|
|
# P&L on past/closed days — a position opened days ago still showed its full
|
|
# mark-vs-entry, so a $0 holiday read as "+$330 realized / -$330 unrealized"
|
|
# every day. The day's P&L decomposes as broker_day = realized + (U_eod -
|
|
# U_sod), so we subtract the carry. Live (for_date None) baselines at today's
|
|
# midnight; after the EOD flatten there are no overnight positions so U_sod=0
|
|
# and this is a no-op — it only bites on genuine overnight holds / past days.
|
|
from datetime import datetime
|
|
since, _until = _hist_window(for_date)
|
|
if since is None:
|
|
now = datetime.now()
|
|
day_start = datetime(now.year, now.month, now.day).isoformat()
|
|
else:
|
|
day_start = since
|
|
sod_row = _latest_event_payload(store, "account_positions_updated", day_start)
|
|
unrealized_sod = 0.0
|
|
if sod_row:
|
|
unrealized_sod, _ = _sum_position_unrealized(
|
|
json.loads(sod_row[0] or "{}").get("positions") or [])
|
|
unrealized_day = unrealized - unrealized_sod # this day's open-position move
|
|
|
|
closed_count = wins + losses
|
|
win_rate = round(wins / closed_count, 3) if closed_count else None
|
|
avg_r = round(sum(r_multiples) / len(r_multiples), 2) if r_multiples else None
|
|
|
|
# BROKER-AUTHORITATIVE day P&L. The closed-event `realized` above is $0
|
|
# whenever position_closed events aren't emitted (they currently aren't), so
|
|
# the strip read $0 on a real -$2,322 day — the operator flew blind. Prefer
|
|
# the broker equity delta: day P&L = latest equity - prior-session close
|
|
# (last_equity); matched (realized) = day P&L - this day's unrealized move.
|
|
broker_day = None
|
|
_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)
|
|
if eq and le:
|
|
broker_day = eq - le
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
if broker_day is not None:
|
|
# total = realized + unrealized holds by construction (broker day P&L
|
|
# split into booked vs the day's open-position move).
|
|
realized_out = round(broker_day - unrealized_day, 2)
|
|
unrealized_out = round(unrealized_day, 2)
|
|
total_out = round(broker_day, 2)
|
|
pnl_source = "broker"
|
|
else:
|
|
realized_out = round(realized, 2)
|
|
unrealized_out = round(unrealized, 2)
|
|
total_out = round(realized + unrealized, 2)
|
|
pnl_source = "closed_events"
|
|
|
|
return {
|
|
"realized_pnl": realized_out,
|
|
"unrealized_pnl": unrealized_out,
|
|
"total_pnl": total_out,
|
|
"broker_day_pnl": round(broker_day, 2) if broker_day is not None else None,
|
|
"open_unrealized_pnl": round(unrealized, 2), # full open mark (vs entry)
|
|
"pnl_source": pnl_source,
|
|
"wins": wins,
|
|
"losses": losses,
|
|
"closed_trades": closed_count,
|
|
"open_positions": open_positions,
|
|
"win_rate": win_rate,
|
|
"avg_r_multiple": avg_r,
|
|
"expectancy_r": avg_r,
|
|
"trades": list(reversed(trades))[:20],
|
|
}
|
|
|
|
|
|
_PROT_ACTIVE = {"held", "new", "accepted", "pending_new",
|
|
"accepted_for_bidding", "partially_filled"}
|
|
|
|
|
|
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
|
|
status (ok / warn / critical) plus the underlying numbers."""
|
|
import os
|
|
from datetime import datetime, timezone
|
|
|
|
equity = None
|
|
day_pnl = None
|
|
_ar = _latest_event_payload(store, "account_summary_updated", _state_until(for_date))
|
|
ts_latest = None
|
|
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)
|
|
if equity and le:
|
|
day_pnl = equity - le
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
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, for_date=for_date)
|
|
orders = (osnap[-1].get("orders") if osnap else []) or []
|
|
|
|
sell_cover: dict = {}
|
|
for o in orders:
|
|
if (o.get("side") == "sell" and o.get("status") in _PROT_ACTIVE
|
|
and o.get("type") in ("stop", "stop_limit", "limit", "trailing_stop")):
|
|
q = float(o.get("quantity") or 0) - float(o.get("filled_quantity") or 0)
|
|
sell_cover[o.get("symbol")] = sell_cover.get(o.get("symbol"), 0.0) + max(q, 0.0)
|
|
|
|
open_count = 0
|
|
gross = 0.0
|
|
unprotected: list[str] = []
|
|
for p in positions:
|
|
try:
|
|
qty = float(p.get("quantity") if p.get("quantity") is not None
|
|
else p.get("qty") or 0)
|
|
except (TypeError, ValueError):
|
|
qty = 0.0
|
|
if qty <= 0:
|
|
continue
|
|
open_count += 1
|
|
try:
|
|
gross += abs(float(p.get("market_value") or 0.0))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
if sell_cover.get(p.get("symbol"), 0.0) < qty - 1e-6:
|
|
unprotected.append(p.get("symbol"))
|
|
|
|
gross_pct = (gross / equity) if equity else None
|
|
day_pnl_pct = (day_pnl / equity) if (equity and day_pnl is not None) else None
|
|
|
|
age = None # data freshness from the latest broker-sync timestamp
|
|
if ts_latest:
|
|
try:
|
|
t = (datetime.fromisoformat(ts_latest.replace("Z", "+00:00"))
|
|
if isinstance(ts_latest, str) else ts_latest)
|
|
if t.tzinfo is None:
|
|
# event timestamps are stored NAIVE LOCAL (datetime.now()), so
|
|
# compare against naive-local now — comparing to UTC would be off
|
|
# by the whole timezone offset (the chip read ~4h on fresh data).
|
|
age = (datetime.now() - t).total_seconds()
|
|
else:
|
|
age = (datetime.now(timezone.utc) - t).total_seconds()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
max_conc = int(os.environ.get("TRADING_MAX_CONCURRENT_POSITIONS", "6"))
|
|
max_gross = float(os.environ.get("TRADING_MAX_GROSS_NOTIONAL_PCT", "0.60"))
|
|
daily_loss = float(os.environ.get("TRADING_MAX_DAILY_LOSS_PCT", "0.03"))
|
|
|
|
status = "ok"
|
|
if unprotected:
|
|
status = "critical"
|
|
elif day_pnl_pct is not None and day_pnl_pct <= -daily_loss * 0.8:
|
|
status = "critical"
|
|
elif (open_count >= max_conc
|
|
or (gross_pct is not None and gross_pct >= max_gross * 0.9)
|
|
or (day_pnl_pct is not None and day_pnl_pct <= -daily_loss * 0.5)
|
|
or (age is not None and age > 90)):
|
|
status = "warn"
|
|
|
|
return {
|
|
"status": status,
|
|
"open_positions": open_count,
|
|
"max_concurrent": max_conc,
|
|
"gross_pct": round(gross_pct, 3) if gross_pct is not None else None,
|
|
"max_gross_pct": max_gross,
|
|
"day_pnl": round(day_pnl, 2) if day_pnl is not None else None,
|
|
"day_pnl_pct": round(day_pnl_pct, 4) if day_pnl_pct is not None else None,
|
|
"daily_loss_limit_pct": daily_loss,
|
|
"data_age_s": round(age, 1) if age is not None else None,
|
|
"unprotected": unprotected,
|
|
}
|
|
|
|
|
|
def query_armed_triggers(store) -> dict:
|
|
"""The live armed-trigger book — the heart of the ORB strategy: which gappers
|
|
are loaded and how close each is to firing its opening-range breakout. The
|
|
fast trigger thread emits this every few seconds as a module_tick(triggers);
|
|
the dashboard renders it as a proximity radar. Bounded to the last 15 min so
|
|
it's a cheap range scan, not a full module_tick load."""
|
|
from datetime import datetime, timedelta
|
|
since = (datetime.now() - timedelta(minutes=15)).isoformat()
|
|
events = store.query_events(event_type="module_tick", since=since, limit=None)
|
|
for event in reversed(events): # ASC order -> newest first
|
|
p = json.loads(event.get("payload_json", "{}"))
|
|
if p.get("module") == "triggers":
|
|
m = p.get("metrics") or {}
|
|
return {"triggers": m.get("triggers") or [],
|
|
"armed": m.get("armed", 0),
|
|
"timestamp": event.get("timestamp")}
|
|
return {"triggers": [], "armed": 0, "timestamp": None}
|
|
|
|
|
|
def query_loop_session(store) -> dict:
|
|
"""The live loop's CURRENT session identity — and whether it is STALE (booted a
|
|
prior day and never restarted). The loop's daily flags (_session_closed, entry
|
|
caps, benches) are memory-only: a session whose embedded date != today means a
|
|
post-EOD boot survived midnight and entries are silently blocked (hit live
|
|
2026-07-21: signals READY + book armed, zero orders all morning). Session ids
|
|
look like paper-YYYY-MM-DD-hash; the date is parsed from the id, so this is one
|
|
bounded range scan with no per-session lookup."""
|
|
from datetime import datetime, timedelta
|
|
since = (datetime.now() - timedelta(minutes=15)).isoformat()
|
|
events = store.query_events(since=since, limit=None)
|
|
sid = None
|
|
for e in reversed(events):
|
|
cid = str(e.get("correlation_id") or "")
|
|
if cid.startswith("paper-"):
|
|
sid = cid
|
|
break
|
|
if not sid:
|
|
return {"session_id": None, "session_date": None, "stale": False}
|
|
parts = sid.split("-") # paper-YYYY-MM-DD-hash
|
|
sess_date = "-".join(parts[1:4]) if len(parts) >= 5 else None
|
|
today = datetime.now().strftime("%Y-%m-%d")
|
|
return {"session_id": sid, "session_date": sess_date,
|
|
"stale": bool(sess_date and sess_date != today)}
|
|
|
|
|
|
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
|
|
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:
|
|
p = json.loads(e.get("payload_json", "{}"))
|
|
try:
|
|
eq = float(p.get("total_equity") or 0.0)
|
|
if eq:
|
|
series.append(round(eq, 2))
|
|
if baseline is None:
|
|
le = float(p.get("last_equity") or 0.0)
|
|
if le:
|
|
baseline = round(le, 2)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
if len(series) > points: # keep the most recent N
|
|
series = series[-points:]
|
|
return {"equity": series, "baseline": baseline if baseline else (series[0] if series else None)}
|
|
|
|
|
|
def query_catalyst_advisory(store, lookback_hours: int = 8) -> dict:
|
|
"""Local-LLM news/catalyst read for the dashboard advisory column.
|
|
|
|
{ticker: {catalyst_type, sentiment, conviction, is_dilutive, rationale,
|
|
headline}} for tickers enriched within the window (latest per symbol). Reads
|
|
the news_catalyst_cache table directly (same Postgres datastore) — never
|
|
raises, returns {} on any fault so the dashboard degrades cleanly."""
|
|
from datetime import datetime, timedelta
|
|
cutoff = (datetime.utcnow() - timedelta(hours=lookback_hours))
|
|
try:
|
|
rows = store.con.execute(
|
|
"SELECT symbol, catalyst_type, sentiment, conviction, is_dilutive, "
|
|
"rationale, headline FROM news_catalyst_cache "
|
|
"WHERE symbol <> '' AND enriched_at >= ? ORDER BY enriched_at DESC",
|
|
[cutoff],
|
|
).fetchall()
|
|
except Exception: # noqa: BLE001 — table may not exist / Ollama never ran
|
|
return {}
|
|
out: dict[str, dict] = {}
|
|
for sym, ctype, sentiment, conviction, is_dilutive, rationale, headline in rows:
|
|
key = (sym or "").upper()
|
|
if key and key not in out: # newest first -> first seen is latest
|
|
out[key] = {
|
|
"catalyst_type": ctype,
|
|
"sentiment": round(float(sentiment), 3) if sentiment is not None else None,
|
|
"conviction": round(float(conviction), 3) if conviction is not None else None,
|
|
"is_dilutive": bool(is_dilutive),
|
|
"rationale": rationale,
|
|
"headline": headline,
|
|
}
|
|
return out
|
|
|
|
|
|
def query_catalyst_feed(store, limit: int = 40, lookback_hours: int = 24) -> list[dict]:
|
|
"""The raw stream of what the local LLM actually read and decided — newest
|
|
first — so the operator can see *what* headlines drove each call and *why*
|
|
(the model's rationale), not just the per-symbol summary the advisory gives.
|
|
|
|
Each row: {symbol, catalyst_type, sentiment, conviction, is_dilutive,
|
|
vetoed, rationale, headline, source, model, enriched_at}. ``vetoed`` marks
|
|
the reads the dilution veto would actually block (is_dilutive AND conviction
|
|
at/above the live floor). Never raises — returns [] on any fault so the
|
|
dashboard degrades cleanly when Ollama has never run."""
|
|
from datetime import datetime, timedelta
|
|
cutoff = (datetime.utcnow() - timedelta(hours=lookback_hours))
|
|
try:
|
|
rows = store.con.execute(
|
|
"SELECT symbol, catalyst_type, sentiment, conviction, is_dilutive, "
|
|
"rationale, headline, source, model, enriched_at "
|
|
"FROM news_catalyst_cache WHERE enriched_at >= ? "
|
|
"ORDER BY enriched_at DESC LIMIT ?",
|
|
[cutoff, int(limit)],
|
|
).fetchall()
|
|
except Exception: # noqa: BLE001 — table may not exist / Ollama never ran
|
|
return []
|
|
out: list[dict] = []
|
|
for sym, ctype, sentiment, conv, is_dil, rationale, headline, source, model, ts in rows:
|
|
conv_f = float(conv) if conv is not None else None
|
|
is_dil_b = bool(is_dil)
|
|
out.append({
|
|
"symbol": (sym or "").upper(),
|
|
"catalyst_type": ctype,
|
|
"sentiment": round(float(sentiment), 3) if sentiment is not None else None,
|
|
"conviction": round(conv_f, 3) if conv_f is not None else None,
|
|
"is_dilutive": is_dil_b,
|
|
"vetoed": bool(is_dil_b and conv_f is not None and conv_f >= 0.6),
|
|
"rationale": rationale,
|
|
"headline": headline,
|
|
"source": source,
|
|
"model": model,
|
|
"enriched_at": str(ts) if ts is not None else None,
|
|
})
|
|
return out
|