From c35224af94dd630b568d0daddaea78e097f110b9 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Wed, 17 Jun 2026 17:33:22 -0400 Subject: [PATCH] Board: per-day trade history + accurate day P&L; manual symbol weights - momentum watch: replace 'open positions' with TODAY'S trade history (FIFO round-trips from the broker orders snapshot: time/symbol/qty/entry/exit/P&L + open lots), resets each day. Headline day P&L is broker-accurate (equity - last_equity, now carried on AccountSummaryUpdatedEvent), shown in the header and the panel title. sync_orders limit 50->500 so the log captures the whole day across restarts. - SYMBOL_WEIGHTS: optional manual per-symbol priority nudge applied to the armed-trigger rank score (small; doesn't override gap/volume logic). Co-Authored-By: Claude Opus 4.8 --- alpaca_paper/sync.py | 3 +- momentum_cli.py | 125 ++++++++++++++++++++++++++++++++-------- run_live_paper.py | 18 +++++- storage/event_schema.py | 3 + 4 files changed, 123 insertions(+), 26 deletions(-) diff --git a/alpaca_paper/sync.py b/alpaca_paper/sync.py index 0b86fe3..ff48e80 100644 --- a/alpaca_paper/sync.py +++ b/alpaca_paper/sync.py @@ -73,6 +73,7 @@ class AlpacaPaperSync: cash_balance=float(account.get("cash") or 0), buying_power=float(account.get("buying_power") or 0), net_liquidating_value=float(account.get("equity") or 0), + last_equity=float(account.get("last_equity") or 0), ) ) return account @@ -110,7 +111,7 @@ class AlpacaPaperSync: ) return positions - def sync_orders(self, status: str = "all", limit: int = 50) -> list[dict] | None: + def sync_orders(self, status: str = "all", limit: int = 500) -> list[dict] | None: try: raw = self.client.get_orders(status=status, limit=limit) except Exception as exc: diff --git a/momentum_cli.py b/momentum_cli.py index 492abaa..2b60690 100644 --- a/momentum_cli.py +++ b/momentum_cli.py @@ -448,6 +448,82 @@ _STATE_STYLE = { } +def _todays_trades(con): + """Reconstruct TODAY's round-trip trades (FIFO buy->sell per symbol) from the + latest broker orders snapshot, plus still-open lots. Resets each day (filters + to today's fills). Returns (trades, realized_pnl).""" + from collections import defaultdict, deque + from datetime import datetime as _dt + try: + from zoneinfo import ZoneInfo + today = _dt.now(ZoneInfo("America/New_York")).date().isoformat() + except Exception: # noqa: BLE001 + today = _dt.now().date().isoformat() + rows = con.execute("SELECT payload_json FROM events WHERE event_type='account_orders_updated' " + "ORDER BY timestamp DESC LIMIT 1").fetchall() + if not rows: + return [], 0.0 + orders = json.loads(rows[0][0]).get("orders") or [] + fills = [o for o in orders + if float(o.get("filled_quantity") or 0) > 0 + and str(o.get("submitted_at") or "")[:10] == today] + fills.sort(key=lambda o: o.get("submitted_at") or "") + lots: dict = defaultdict(deque) + trades = [] + for o in fills: + sym, side = o.get("symbol"), o.get("side") + qty = float(o.get("filled_quantity") or 0) + px = float(o.get("filled_avg_price") or 0) + tm = str(o.get("submitted_at") or "")[11:19] + if side == "buy": + lots[sym].append([qty, px, tm]) + elif side == "sell": + rem = qty + while rem > 1e-9 and lots[sym]: + lot = lots[sym][0] + m = min(rem, lot[0]) + trades.append({"symbol": sym, "qty": m, "entry": lot[1], "exit": px, + "pnl": (px - lot[1]) * m, "time": lot[2], "status": "closed"}) + lot[0] -= m + rem -= m + if lot[0] <= 1e-9: + lots[sym].popleft() + for sym, dq in lots.items(): + for qty, px, tm in dq: + trades.append({"symbol": sym, "qty": qty, "entry": px, "exit": None, + "pnl": None, "time": tm, "status": "open"}) + realized = sum(t["pnl"] for t in trades if t["status"] == "closed") + return trades, realized + + +def _day_pnl(con): + """Authoritative day P&L = latest equity - prior-session close (broker truth, + resets daily). Uses the broker's last_equity baseline; falls back to the + first equity recorded today for events emitted before that field existed.""" + latest = con.execute("SELECT payload_json FROM events WHERE event_type='account_summary_updated' " + "ORDER BY timestamp DESC LIMIT 1").fetchall() + if not latest: + return None + p = json.loads(latest[0][0]) + try: + e_now = float(p.get("total_equity") or 0) + last_eq = float(p.get("last_equity") or 0) + if e_now and last_eq: + return e_now - last_eq + except (TypeError, ValueError): + pass + first = con.execute("SELECT payload_json FROM events WHERE event_type='account_summary_updated' " + "AND timestamp::date = CURRENT_DATE ORDER BY timestamp ASC LIMIT 1").fetchall() + if not first: + return None + try: + e_now = float(p.get("total_equity") or 0) + e_start = float(json.loads(first[0][0]).get("total_equity") or 0) + return (e_now - e_start) if (e_now and e_start) else None + except (TypeError, ValueError): + return None + + def _render_board(con): """Build the live monitoring board renderable.""" from rich.console import Group @@ -503,33 +579,35 @@ def _render_board(con): pos_ev = con.execute("SELECT payload_json FROM events WHERE event_type='account_positions_updated' " "ORDER BY timestamp DESC LIMIT 1").fetchall() - pt = Table(title="open positions", box=box.SIMPLE) - for c in ("symbol", "qty", "entry", "last", "uPnL $", "uPnL %"): - pt.add_column(c, justify="left" if c == "symbol" else "right") + open_unreal = 0.0 if pos_ev: for p in (json.loads(pos_ev[0][0]).get("positions") or []): - # the sync stores it as 'unrealized_pnl'; fall back to 'unrealized_pl' - upl = p.get("unrealized_pnl") - if upl is None: - upl = p.get("unrealized_pl") - entry = p.get("avg_entry_price") - last = p.get("current_price") - pct = p.get("unrealized_pnl_pct") try: - upl_str = f"[{'green' if float(upl) >= 0 else 'red'}]{float(upl):+.0f}[/]" + open_unreal += float(p.get("unrealized_pnl") if p.get("unrealized_pnl") is not None + else p.get("unrealized_pl") or 0) except (TypeError, ValueError): - upl_str = "-" - try: - pct_str = f"[{'green' if float(pct) >= 0 else 'red'}]{float(pct) * 100:+.1f}%[/]" - except (TypeError, ValueError): - pct_str = "-" - pt.add_row( - str(p.get("symbol")), - str(p.get("quantity") or p.get("qty") or ""), - f"{float(entry):.2f}" if entry not in (None, "") else "-", - f"{float(last):.2f}" if last not in (None, "") else "-", - upl_str, pct_str, - ) + pass + trades, realized = _todays_trades(con) + day_pnl = _day_pnl(con) + if day_pnl is None: # fallback if equity snapshots missing + day_pnl = realized + open_unreal + dcol = "green" if day_pnl >= 0 else "red" + pt = Table(title=f"today's trades — day P&L [{dcol}]{day_pnl:+,.0f}[/] " + f"([dim]matched {realized:+,.0f} · open {open_unreal:+,.0f}[/])", + box=box.SIMPLE) + for c in ("time", "symbol", "qty", "entry", "exit", "P&L", ""): + pt.add_column(c, justify="left" if c in ("symbol", "") else "right") + for t in sorted(trades, key=lambda x: x["time"], reverse=True)[:14]: + pnl = t["pnl"] + if pnl is None: + pnl_str, tag = "[dim]—[/dim]", "[yellow]● open[/yellow]" + else: + pnl_str = f"[{'green' if pnl >= 0 else 'red'}]{pnl:+.0f}[/]" + tag = "closed" + pt.add_row(t["time"], str(t["symbol"]), f"{t['qty']:.0f}", f"{t['entry']:.2f}", + f"{t['exit']:.2f}" if t["exit"] else "—", pnl_str, tag) + if not trades: + pt.add_row("—", "[dim]no trades today yet[/dim]", "—", "—", "—", "—", "—") risk = con.execute("SELECT timestamp, message FROM events WHERE event_type='risk_rule_triggered' " "ORDER BY timestamp DESC LIMIT 4").fetchall() @@ -543,6 +621,7 @@ def _render_board(con): f"[bold]momentum live[/bold] " f"[{pcolor}]{phase}[/{pcolor}] [dim]{detail}[/dim] " f"equity [green]{eq}[/green] " + f"day P&L [bold {dcol}]{day_pnl:+,.0f}[/] " f"[{hb_color}]♥ {hb_text}[/{hb_color}] " f"[bold cyan]{armed_n} armed[/bold cyan] · {n_eval} scanning\n" f"[dim]hunting $1–20 gappers (≥ gap% on ≥ 2× volume); buy the break of the " diff --git a/run_live_paper.py b/run_live_paper.py index 28f3843..e0a8f31 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -469,6 +469,17 @@ def main(argv: list[str] | None = None) -> int: news_map = recent_news_map( rt["research_con"], lookback_hours=int(os.environ.get("NEWS_LOOKBACK_HOURS", "8"))) news_boost = float(os.environ.get("NEWS_RANK_BOOST", "2.0")) + # optional MANUAL per-symbol priority nudge (SYMBOL_WEIGHTS=ABCD:1.2,XYZ:1.1) + # — a small boost to names you want watched; keep modest so it doesn't + # override the gap/volume logic. + weights: dict[str, float] = {} + for part in os.environ.get("SYMBOL_WEIGHTS", "").split(","): + if ":" in part: + s, w = part.split(":", 1) + try: + weights[s.strip().upper()] = float(w) + except ValueError: + pass candidates: list[dict] = [] # rank ALL session symbols by gap*rvol (thresholds applied later, per # trigger, so the board still shows the field pre-gap) @@ -492,8 +503,11 @@ def main(argv: list[str] | None = None) -> int: "range_pct": ((hi - lo) / hi) if (hi and lo and hi > 0) else 0.0, "cum_volume": g.cumulative_volume, "catalyst": catalyst, - # a fresh catalyst is WHY a small-cap runs — boost it up the rank - "_score": g.gap_pct * max(g.relative_volume, 0.1) * (news_boost if catalyst else 1.0), + # a fresh catalyst is WHY a small-cap runs — boost it up the rank; + # times an optional manual per-symbol weight (default 1.0) + "_score": (g.gap_pct * max(g.relative_volume, 0.1) + * (news_boost if catalyst else 1.0) + * weights.get(g.symbol, 1.0)), "complete": complete, }) # pre-open / before any gappers form: surface the queued watchlist so the diff --git a/storage/event_schema.py b/storage/event_schema.py index ad4abb9..49676ec 100644 --- a/storage/event_schema.py +++ b/storage/event_schema.py @@ -232,6 +232,9 @@ class AccountSummaryUpdatedEvent(BaseEvent): cash_balance: float = Field(description="Available cash balance") buying_power: float = Field(description="Buying power") net_liquidating_value: float = Field(description="Net liquidation value") + last_equity: float = Field( + default=0.0, description="Prior-session close equity (day P&L baseline)" + ) class AccountPositionsUpdatedEvent(BaseEvent):