#!/usr/bin/env python3 """ Build a **shareable, multi-tenant-ready** snapshot for the Today Trades website (Stage A). Schema is designed so Stage B can add more ``books[]`` entries without rewriting the UI. Example:: cd /Users/robzingale/trading_bot && PYTHONUNBUFFERED=1 .venv/bin/python \\ -m RenTech.monitor.build_today_trades_snapshot \\ --nav 50000 --fund-scale 1.5 \\ --held AMZN,QCOM,AXP,INTC,LRCX,AMAT,MU,AMD,GEV \\ --out RenTech/data/logs/today_trades_snapshot.json Then serve:: .venv/bin/python -m RenTech.monitor.serve_today_trades --password 'change-me' """ from __future__ import annotations import argparse import asyncio import json import math import sys from datetime import datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo import pandas as pd import yfinance as yf _REPO = Path(__file__).resolve().parents[2] if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO)) from RenTech.live.stock_book_utils import ( # noqa: E402 DEFAULT_DIP_MAX_HOLDINGS, STOCK_ONLY_WEIGHTS, STOCK_SLEEVE_LABELS, TODAY_TRADES_CORE_SLEEVES, dip_per_name_budget_usd, sleeve_budget_usd, today_trades_core_weights, today_trades_strategy_guide, ) from RenTech.live.stock_only_signals import ( # noqa: E402 _dip_actions, _scan_cm_dip_signals, build_stock_only_signals, ) NY = ZoneInfo("America/New_York") LOGS = _REPO / "RenTech" / "data" / "logs" DEFAULT_OUT = LOGS / "today_trades_snapshot.json" DEFAULT_STATE = _REPO / "RenTech" / "data" / "live_state" / "cm_dip_positions.json" DEFAULT_HISTORY = _REPO / "RenTech" / "monitor" / "config" / "today_trades_history.json" SCHEMA_VERSION = 1 # Fallback signal/entry dates when state has none (recent live book). _DEFAULT_SIGNAL: dict[str, str] = { "AMZN": "2026-07-22", "QCOM": "2026-07-22", "AXP": "2026-07-23", "INTC": "2026-07-23", "LRCX": "2026-07-23", "AMAT": "2026-07-23", "MU": "2026-07-23", "AMD": "2026-07-23", "GEV": "2026-07-23", } _DEFAULT_ENTRY: dict[str, str] = { "AMZN": "2026-07-23", "QCOM": "2026-07-23", "AXP": "2026-07-24", "INTC": "2026-07-24", "LRCX": "2026-07-24", "AMAT": "2026-07-24", "MU": "2026-07-24", "AMD": "2026-07-27", "GEV": "2026-07-27", } def _today() -> pd.Timestamp: return pd.Timestamp.now(tz=NY).normalize().tz_localize(None) def _atr5(df: pd.DataFrame) -> pd.Series: c = df["close"].astype(float) h = df["high"].astype(float) l = df["low"].astype(float) prev = c.shift(1) tr = pd.concat([(h - l), (h - prev).abs(), (l - prev).abs()], axis=1).max(axis=1) return tr.rolling(5, min_periods=5).mean() def _load_state(path: Path) -> dict[str, Any]: if not path.is_file(): return {"positions": {}} try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {"positions": {}} return data if isinstance(data, dict) else {"positions": {}} def _held_from_args_and_state( held_csv: str | None, state: dict[str, Any], ) -> dict[str, dict[str, Any]]: """symbol -> {signal_date, entry_date, fill_price, atr_usd}.""" out: dict[str, dict[str, Any]] = {} for sym, meta in (state.get("positions") or {}).items(): if not isinstance(meta, dict): continue if meta.get("status") == "recommended": continue s = str(sym).upper() out[s] = { "signal_date": meta.get("signal_date") or _DEFAULT_SIGNAL.get(s), "entry_date": meta.get("entry_date") or _DEFAULT_ENTRY.get(s), "fill_price": meta.get("fill_price") or meta.get("limit_price"), "atr_usd": meta.get("atr_at_signal") or meta.get("atr_usd"), "shares": meta.get("shares"), "source": "state", } if held_csv and held_csv.strip(): for part in held_csv.split(","): s = part.strip().upper() if not s: continue out.setdefault( s, { "signal_date": _DEFAULT_SIGNAL.get(s), "entry_date": _DEFAULT_ENTRY.get(s), "fill_price": None, "atr_usd": None, "shares": None, "source": "cli", }, ) return out def _evaluate_open_dips( held: dict[str, dict[str, Any]], *, today: pd.Timestamp, profit_atr_mult: float = 0.5, hold_trading_days: int = 10, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Return (exits, holds) with levels and last price.""" if not held: return [], [] tickers = sorted(held) raw = yf.download( tickers, start=(today - pd.Timedelta(days=90)).strftime("%Y-%m-%d"), end=(today + pd.Timedelta(days=1)).strftime("%Y-%m-%d"), auto_adjust=True, progress=False, group_by="ticker", threads=True, ) try: px = yf.download(tickers, period="1d", interval="1m", auto_adjust=True, progress=False) if isinstance(px.columns, pd.MultiIndex): last = px["Close"].ffill().iloc[-1] asof = str(px.index[-1]) else: last = px["Close"].iloc[-1] asof = str(px.index[-1]) except Exception: last = None asof = None # Prior session = previous business day prior_sess = (today - pd.offsets.BDay(1)).normalize() exits: list[dict[str, Any]] = [] holds: list[dict[str, Any]] = [] for t in tickers: meta = held[t] try: if isinstance(raw.columns, pd.MultiIndex): df = raw[t].dropna(how="all").copy() else: df = raw.dropna(how="all").copy() df.columns = [str(c).lower() for c in df.columns] df.index = pd.to_datetime(df.index).tz_localize(None) atr = _atr5(df) sig_s = meta.get("signal_date") or _DEFAULT_SIGNAL.get(t) if not sig_s: # last resort: 5 sessions ago sig_ts = df.index[-3] if len(df) >= 3 else df.index[-1] else: sig_ts = pd.Timestamp(sig_s) row = df.loc[:sig_ts].iloc[-1] atr_v = float(meta.get("atr_usd") or atr.loc[:sig_ts].iloc[-1]) buy_limit = float(row["close"]) - 0.9 * atr_v fill = float(meta["fill_price"]) if meta.get("fill_price") else buy_limit profit = fill + float(profit_atr_mult) * atr_v if prior_sess in df.index: prior_hi = float(df.loc[prior_sess, "high"]) else: prior_hi = float(df.loc[:prior_sess].iloc[-1]["high"]) ent_s = meta.get("entry_date") or _DEFAULT_ENTRY.get(t) or str(sig_ts.date()) days = int(len(pd.bdate_range(pd.Timestamp(ent_s), today))) if last is not None and t in getattr(last, "index", []): last_px = float(last[t]) elif last is not None and hasattr(last, "get"): last_px = float(last.get(t, float("nan"))) else: last_px = float(df["close"].iloc[-1]) reasons: list[str] = [] if math.isfinite(last_px) and last_px >= profit: reasons.append("profit") if math.isfinite(last_px) and last_px > prior_hi: reasons.append("prior_high") if days > int(hold_trading_days): reasons.append("time_stop") trade = { "symbol": t, "side": "SELL", "sleeve": "equity_dip", "action": "EXIT" if reasons else "HOLD", "last_px": round(last_px, 2) if math.isfinite(last_px) else None, "profit_target": round(profit, 2), "prior_high": round(prior_hi, 2), "days_held": days, "fill_price": round(fill, 2), "atr_usd": round(atr_v, 4), "reasons": reasons, "order_type": "MOC" if reasons else None, "asof": asof, "summary": ( f"Exit {t} ({'+'.join(reasons)})" if reasons else f"Hold {t} — profit ≥ ${profit:.2f} or close > ${prior_hi:.2f}" ), } if reasons: exits.append(trade) else: holds.append(trade) except Exception as exc: holds.append( { "symbol": t, "side": "SELL", "sleeve": "equity_dip", "action": "HOLD", "summary": f"Hold {t} — could not price ({exc})", "error": str(exc), } ) exits.sort(key=lambda x: x["symbol"]) holds.sort(key=lambda x: x["symbol"]) return exits, holds def _entry_trades( *, nav: float, fund_scale: float, dip_weight: float, max_holdings: int, open_symbols: set[str], open_mv_usd: float, skip_entries: bool, ) -> tuple[list[dict[str, Any]], dict[str, Any], list[str]]: notes: list[str] = [] sleeve = sleeve_budget_usd(nav, dip_weight, fund_scale) per_name = dip_per_name_budget_usd(nav, dip_weight, fund_scale, max_holdings=max_holdings) meta: dict[str, Any] = { "sleeve_budget_usd": round(sleeve, 2), "per_name_budget_usd": round(per_name, 2), "max_holdings": max_holdings, "open_count": len(open_symbols), "open_mv_usd": round(open_mv_usd, 2), "over_budget": open_mv_usd >= sleeve * 0.95, } if skip_entries or meta["over_budget"] or len(open_symbols) >= max_holdings: reason = ( "over sleeve budget" if meta["over_budget"] else f"at max holdings ({len(open_symbols)}/{max_holdings})" ) notes.append(f"Skip new dip entries — {reason}.") sigs, scan_meta = _scan_cm_dip_signals() meta["scan"] = {k: scan_meta.get(k) for k in ("status", "n_candidates", "entry_date", "source")} # Still list candidates as watch-only watch = [] for s in sigs[:8]: watch.append( { "symbol": s["ticker"], "side": "BUY", "sleeve": "equity_dip", "action": "WATCH", "limit_price": s.get("limit_price"), "profit_target": s.get("profit_target"), "notional_usd": round(per_name, 2), "order_type": "LMT", "time_in_force": "DAY", "summary": f"Watch {s['ticker']} LMT ${s.get('limit_price')} (do not enter — {reason})", "drop_pct": s.get("drop_pct"), } ) meta["watch_candidates"] = watch return [], meta, notes sigs, scan_meta = _scan_cm_dip_signals() meta["scan"] = {k: scan_meta.get(k) for k in ("status", "n_candidates", "entry_date", "source")} actions = _dip_actions( sigs, fund_nav=nav, fund_weight=dip_weight, fund_scale=fund_scale, max_holdings=max_holdings, open_symbols=open_symbols, ) enters: list[dict[str, Any]] = [] for a in actions: if a.get("action") != "ADD": continue det = a.get("detail") or {} enters.append( { "symbol": a.get("symbol"), "side": "BUY", "sleeve": "equity_dip", "action": "ENTER", "limit_price": det.get("limit_price"), "profit_target": det.get("profit_target"), "notional_usd": det.get("target_notional_usd"), "order_type": "LMT", "time_in_force": "DAY", "shares": None, "drop_pct": det.get("drop_pct"), "atr_usd": det.get("atr_usd"), "summary": a.get("summary"), } ) if det.get("limit_price") and det.get("target_notional_usd"): lim = float(det["limit_price"]) if lim > 0: enters[-1]["shares"] = max(1, int(float(det["target_notional_usd"]) / lim)) return enters, meta, notes def _map_spy_proxy(sym: str, spy_proxy: str) -> str: s = str(sym).upper() if s == "SPY" and spy_proxy: return str(spy_proxy).upper() return s def _last_prices(symbols: list[str]) -> dict[str, float]: syms = sorted({s for s in symbols if s and s != "CASH"}) if not syms: return {} try: px = yf.download(syms, period="5d", auto_adjust=True, progress=False) if isinstance(px.columns, pd.MultiIndex): last = px["Close"].ffill().iloc[-1] else: last = px["Close"].iloc[-1] out: dict[str, float] = {} for s in syms: try: v = float(last[s]) if s in getattr(last, "index", []) else float(last.get(s, float("nan"))) if math.isfinite(v) and v > 0: out[s] = v except Exception: continue return out except Exception: return {} def _portfolio_from_stock_only_signals( payload: dict[str, Any], *, spy_proxy: str = "VOO", allowed_sleeves: frozenset[str] | None = None, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any]]: """ Build net portfolio + per-sleeve rows from offline ``build_stock_only_signals``. Returns (net_rows, by_sleeve_rows, sleeves_meta). """ allow = allowed_sleeves if allowed_sleeves is not None else TODAY_TRADES_CORE_SLEEVES nav = float(payload.get("fund_nav_usd") or 0.0) scale = float(payload.get("fund_scale") or 1.5) weights = dict(payload.get("fund_weights") or {}) signals = dict(payload.get("signals") or {}) budgets = dict(payload.get("sleeve_budgets_usd") or {}) by_sleeve: list[dict[str, Any]] = [] net: dict[str, float] = {} sleeves_meta: dict[str, Any] = {} for book, w in weights.items(): if float(w) <= 0: continue if book not in allow: continue if book == "equity_dip": # Dip handled separately as exit/enter/hold tickets continue sig = signals.get(book) or {} targets = dict(sig.get("targets") or {}) sleeve_budget = float(budgets.get(book) or sleeve_budget_usd(nav, float(w), scale)) sleeves_meta[book] = { "status": sig.get("status", "ok" if targets else "empty"), "weight": float(w), "budget_usd": round(sleeve_budget, 2), "label": STOCK_SLEEVE_LABELS.get(book, book), "targets": {k: round(float(v), 4) for k, v in targets.items()}, **{k: v for k, v in sig.items() if k not in ("targets",)}, } if not targets: continue for sym, tw in targets.items(): trade_sym = _map_spy_proxy(str(sym), spy_proxy) notional = sleeve_budget * float(tw) net[trade_sym] = net.get(trade_sym, 0.0) + notional by_sleeve.append( { "symbol": trade_sym, "model_symbol": str(sym).upper(), "side": "BUY" if notional >= 0 else "SELL", "sleeve": book, "sleeve_label": STOCK_SLEEVE_LABELS.get(book, book), "action": "HOLD", "target_weight": round(float(tw), 4), "notional_usd": round(notional, 2), "order_type": "MKT", "summary": ( f"{STOCK_SLEEVE_LABELS.get(book, book)}: " f"{'long' if notional >= 0 else 'short'} {trade_sym} " f"· {abs(float(tw)):.1%} of sleeve" + (" (VOO for SPY)" if trade_sym != str(sym).upper() else "") ), } ) # Prices + share counts prices = _last_prices(list(net.keys()) + [r["symbol"] for r in by_sleeve]) for r in by_sleeve: px = prices.get(r["symbol"]) r["last_px"] = round(px, 2) if px else None n = float(r.get("notional_usd") or 0.0) if px and px > 0 and abs(n) >= px: r["shares"] = int(round(abs(n) / px)) * (1 if n >= 0 else -1) else: r["shares"] = None net_rows: list[dict[str, Any]] = [] for sym, notional in sorted(net.items(), key=lambda x: -abs(x[1])): if abs(notional) < 25: continue px = prices.get(sym) shares = None if px and px > 0 and abs(notional) >= px * 0.5: shares = int(round(abs(notional) / px)) * (1 if notional >= 0 else -1) net_rows.append( { "symbol": sym, "side": "BUY" if notional >= 0 else "SELL", "sleeve": "net_book", "action": "HOLD", "notional_usd": round(notional, 2), "shares": shares, "last_px": round(px, 2) if px else None, "order_type": "MKT", "summary": ( f"Net stock-only {'long' if notional >= 0 else 'short'} {sym}" ), } ) return net_rows, by_sleeve, sleeves_meta def _load_history(path: Path | None = None) -> dict[str, Any]: """Curated chat-era tickets since Jul 23 (+ optional live appends later).""" p = path or DEFAULT_HISTORY if not p.is_absolute(): p = (_REPO / p).resolve() if not p.exists(): return {"start": "2026-07-23", "events": [], "note": "No history file."} try: raw = json.loads(p.read_text(encoding="utf-8")) except Exception as exc: return {"start": "2026-07-23", "events": [], "note": f"History load failed: {exc}"} events = list(raw.get("events") or []) events.sort(key=lambda e: (str(e.get("date") or ""), str(e.get("action") or ""), str(e.get("symbol") or ""))) # Newest days first for the UI by_date: dict[str, list[dict[str, Any]]] = {} for e in events: d = str(e.get("date") or "") by_date.setdefault(d, []).append(e) days = [ {"date": d, "events": by_date[d], "n": len(by_date[d])} for d in sorted(by_date.keys(), reverse=True) ] return { "start": raw.get("start", "2026-07-23"), "label": raw.get("label", "History"), "source": raw.get("source", "chat_recommendations"), "note": raw.get("note", ""), "n_events": len(events), "n_days": len(days), "days": days, "events": events, } # Attribute core ETF tickets to the live sleeves we show on the site. _CORE_SYMBOL_TO_SLEEVE: dict[str, str] = { "VOO": "tactical_aw", "SPY": "tactical_aw", "DBC": "tactical_aw", "SH": "ma_slope_inverse", "SVIX": "vol_edge", "VIXY": "vol_edge", "SVXY": "vol_edge", "UUP": "tsmom", "EFA": "tsmom", "EEM": "tsmom", "IEF": "tsmom", "TLT": "tsmom", "GLD": "tsmom", } def _close_panel(symbols: list[str], start: str) -> pd.DataFrame: """Daily close panel from Yahoo (auto-adjusted), columns = tickers.""" syms = sorted({str(s).upper() for s in symbols if s and str(s).upper() not in ("CASH", "BOOK")}) if not syms: return pd.DataFrame() try: raw = yf.download( syms, start=start, auto_adjust=True, progress=False, threads=True, ) except Exception: return pd.DataFrame() if raw is None or raw.empty: return pd.DataFrame() if isinstance(raw.columns, pd.MultiIndex): close = raw["Close"].copy() else: close = raw[["Close"]].copy() close.columns = [syms[0]] close.index = pd.to_datetime(close.index).tz_localize(None).normalize() close = close.sort_index().ffill() return close def _px_on(panel: pd.DataFrame, sym: str, day: pd.Timestamp) -> float | None: if panel.empty or sym not in panel.columns: return None day = pd.Timestamp(day).normalize() sub = panel.loc[:day, sym].dropna() if sub.empty: return None v = float(sub.iloc[-1]) return v if math.isfinite(v) and v > 0 else None def _resolve_sleeve(event: dict[str, Any]) -> str: sleeve = str(event.get("sleeve") or "").strip() sym = str(event.get("symbol") or "").upper() if sleeve == "equity_dip": return "equity_dip" if sleeve in TODAY_TRADES_CORE_SLEEVES: return sleeve if sleeve == "core_etf": return _CORE_SYMBOL_TO_SLEEVE.get(sym, "tactical_aw") return _CORE_SYMBOL_TO_SLEEVE.get(sym, sleeve or "other") def _sleeve_performance( history: dict[str, Any], *, budgets: dict[str, float], as_of: pd.Timestamp, spy_proxy: str = "VOO", ) -> dict[str, Any]: """ Illustrative PnL since history start from chat tickets (Yahoo marks). ENTER / REBALANCE set positions; EXIT realizes at that day's close. Open lots marked to latest close. Core ETF symbols are attributed to sleeves. """ start = str(history.get("start") or "2026-07-23") events = list(history.get("events") or []) if not events: return { "start": start, "as_of": str(as_of.date()), "method": "chat_tickets_yahoo_mtm", "rows": [], "total_pnl_usd": 0.0, "note": "No history events to mark.", } symbols: set[str] = set() for e in events: sym = str(e.get("symbol") or "").upper() if sym and sym not in ("CASH", "BOOK"): symbols.add(_map_spy_proxy(sym, spy_proxy)) panel = _close_panel(sorted(symbols), start=start) # lot state: symbol -> {shares, avg_cost, sleeve} lots: dict[str, dict[str, Any]] = {} realized: dict[str, float] = {} def _add_realized(sleeve: str, pnl: float) -> None: realized[sleeve] = realized.get(sleeve, 0.0) + float(pnl) chronological = sorted( events, key=lambda e: ( str(e.get("date") or ""), {"ENTER": 0, "REBALANCE": 1, "HOLD": 2, "EXIT": 3}.get(str(e.get("action") or "").upper(), 9), str(e.get("symbol") or ""), ), ) for e in chronological: action = str(e.get("action") or "").upper() sym_raw = str(e.get("symbol") or "").upper() if not sym_raw or sym_raw in ("CASH", "BOOK"): continue sym = _map_spy_proxy(sym_raw, spy_proxy) sleeve = _resolve_sleeve({**e, "symbol": sym_raw}) day = pd.Timestamp(str(e.get("date"))).normalize() side = str(e.get("side") or "BUY").upper() signed = 1.0 if side != "SELL" else -1.0 if action in ("ENTER", "REBALANCE"): shares_abs = e.get("shares") if shares_abs is None: continue shares = signed * abs(float(shares_abs)) px = e.get("limit_price") if px is None and e.get("notional_usd") and shares_abs: try: px = abs(float(e["notional_usd"])) / abs(float(shares_abs)) except Exception: px = None if px is None: px = _px_on(panel, sym, day) if px is None or not math.isfinite(float(px)) or float(px) <= 0: continue px = float(px) if action == "REBALANCE" or sym not in lots: lots[sym] = {"shares": shares, "avg_cost": px, "sleeve": sleeve} else: # Average in additional ENTER size prev = lots[sym] old_sh = float(prev["shares"]) new_sh = old_sh + shares if abs(new_sh) < 1e-9: lots.pop(sym, None) continue # Same-direction average; flip resets cost if old_sh * shares >= 0: prev["avg_cost"] = (old_sh * float(prev["avg_cost"]) + shares * px) / new_sh prev["shares"] = new_sh prev["sleeve"] = sleeve else: lots[sym] = {"shares": shares, "avg_cost": px, "sleeve": sleeve} elif action == "EXIT": lot = lots.get(sym) if not lot: continue sh = float(lot["shares"]) cost = float(lot["avg_cost"]) px = _px_on(panel, sym, day) if px is None and e.get("profit_target") is not None: px = float(e["profit_target"]) if px is None: continue pnl = sh * (float(px) - cost) _add_realized(str(lot.get("sleeve") or sleeve), pnl) lots.pop(sym, None) # Mark open lots unrealized: dict[str, float] = {} open_mv: dict[str, float] = {} last_day = pd.Timestamp(as_of).normalize() for sym, lot in lots.items(): sh = float(lot["shares"]) cost = float(lot["avg_cost"]) sleeve = str(lot.get("sleeve") or "other") px = _px_on(panel, sym, last_day) if px is None: continue pnl = sh * (float(px) - cost) unrealized[sleeve] = unrealized.get(sleeve, 0.0) + pnl open_mv[sleeve] = open_mv.get(sleeve, 0.0) + sh * float(px) sleeves = sorted(set(realized) | set(unrealized) | set(budgets) | set(TODAY_TRADES_CORE_SLEEVES)) rows: list[dict[str, Any]] = [] total_pnl = 0.0 for sleeve in sleeves: if sleeve not in TODAY_TRADES_CORE_SLEEVES and sleeve not in realized and sleeve not in unrealized: continue r = float(realized.get(sleeve, 0.0)) u = float(unrealized.get(sleeve, 0.0)) pnl = r + u total_pnl += pnl budget = float(budgets.get(sleeve) or 0.0) ret_pct = (pnl / budget * 100.0) if budget > 1 else None rows.append( { "sleeve": sleeve, "label": STOCK_SLEEVE_LABELS.get(sleeve, sleeve), "budget_usd": round(budget, 2) if budget else None, "realized_pnl_usd": round(r, 2), "unrealized_pnl_usd": round(u, 2), "pnl_usd": round(pnl, 2), "return_pct": round(ret_pct, 2) if ret_pct is not None else None, "open_mv_usd": round(float(open_mv.get(sleeve, 0.0)), 2), } ) rows.sort(key=lambda x: -abs(float(x.get("pnl_usd") or 0.0))) return { "start": start, "as_of": str(as_of.date()), "method": "chat_tickets_yahoo_mtm", "note": ( "Illustrative mark-to-market since live start using chat ENTER/EXIT/REBALANCE " "tickets and Yahoo closes. Not your IBKR fill ledger. " "VOO attributed to Tactical AW; dip sleeve uses ticket limits / exit-day closes." ), "total_pnl_usd": round(total_pnl, 2), "rows": rows, } def build_snapshot( *, account_id: str, label: str, nav: float, fund_scale: float, held: dict[str, dict[str, Any]], spy_proxy: str = "VOO", approx_per_name_mv: float | None = 2500.0, skip_new_entries: bool = False, ) -> dict[str, Any]: today = _today() dip_w = float(STOCK_ONLY_WEIGHTS.get("equity_dip", 0.17)) max_h = DEFAULT_DIP_MAX_HOLDINGS open_syms = set(held.keys()) # Prefer explicit approx MV (user oversized); else per_name * count sleeve = sleeve_budget_usd(nav, dip_w, fund_scale) per_name = dip_per_name_budget_usd(nav, dip_w, fund_scale, max_holdings=max_h) if approx_per_name_mv and approx_per_name_mv > 0: open_mv = float(approx_per_name_mv) * len(open_syms) else: open_mv = per_name * len(open_syms) exits, holds = _evaluate_open_dips(held, today=today) enters, dip_meta, notes = _entry_trades( nav=nav, fund_scale=fund_scale, dip_weight=dip_w, max_holdings=max_h, open_symbols=open_syms, open_mv_usd=open_mv, skip_entries=skip_new_entries, ) if open_mv >= sleeve * 0.95: notes.insert(0, f"Dip sleeve over budget (~${open_mv:,.0f} open vs ${sleeve:,.0f} target).") # Core live book only: tactical AW + TSMOM + vol edge + SH + QS + dips # (matches chat tickets; excludes Johansen / MA slope / Ride Rockets) core_w = today_trades_core_weights() try: so_payload = asyncio.run( build_stock_only_signals( None, fund_weights=core_w, fund_scale=fund_scale, fund_nav=nav, use_yahoo=True, renormalize_weights=False, ) ) net_rows, by_sleeve_rows, sleeves_meta = _portfolio_from_stock_only_signals( so_payload, spy_proxy=spy_proxy, allowed_sleeves=TODAY_TRADES_CORE_SLEEVES, ) except Exception as exc: notes.append(f"Stock-only book engine failed: {exc}") so_payload = {} net_rows, by_sleeve_rows, sleeves_meta = [], [], {} for book, meta in sleeves_meta.items(): if meta.get("status") in ("empty", "error") and book not in ("equity_dip",): err = meta.get("error") or meta.get("status") notes.append(f"{STOCK_SLEEVE_LABELS.get(book, book)}: no targets ({err}).") sleeves_out: dict[str, Any] = { "equity_dip": { "weight": dip_w, "budget_usd": round(sleeve, 2), "per_name_usd": round(per_name, 2), "max_holdings": max_h, "open_count": len(open_syms), "open_symbols": sorted(open_syms), "open_mv_usd": round(open_mv, 2), "over_budget": open_mv >= sleeve * 0.95, "label": STOCK_SLEEVE_LABELS.get("equity_dip", "Equity Dip"), **{k: dip_meta[k] for k in ("scan", "watch_candidates") if k in dip_meta}, }, **sleeves_meta, } history = _load_history() budgets = dict(so_payload.get("sleeve_budgets_usd") or {}) if "equity_dip" not in budgets: budgets["equity_dip"] = round(sleeve, 2) try: performance = _sleeve_performance( history, budgets=budgets, as_of=today, spy_proxy=spy_proxy, ) except Exception as exc: performance = { "start": history.get("start", "2026-07-23"), "as_of": str(today.date()), "method": "error", "note": f"Performance calc failed: {exc}", "total_pnl_usd": None, "rows": [], } book = { "account_id": account_id, "label": label, "nav_usd": round(nav, 2), "fund_scale": fund_scale, "spy_proxy": spy_proxy, "as_of_date": str(today.date()), "mode": "recommend_only", "fund_weights": so_payload.get("fund_weights") if so_payload else core_w, "sleeve_budgets_usd": budgets, "performance": performance, "strategy_guide": today_trades_strategy_guide(budgets), "sleeves": sleeves_out, "trades": { "portfolio": net_rows, "by_sleeve": by_sleeve_rows, "exit": exits, "enter": enters, "hold": holds, # Back-compat alias for older UI "all_weather": [r for r in by_sleeve_rows if r.get("sleeve") == "tactical_aw"], "history": history, }, "notes": notes, "disclaimer": ( "Research / education only. Not investment advice. " "Recommend-only — place orders yourself in your own brokerage account. " "Portfolio = core ETF book (AW + TSMOM + vol edge + SH + QS) + CM dips. " "History = chat tickets since 2026-07-23. " "Performance = illustrative Yahoo MTM on those tickets." ), } return { "schema_version": SCHEMA_VERSION, "generated_at": datetime.now(tz=NY).isoformat(timespec="seconds"), "product": "rentech_today_trades", "stage": "A_shareable_dashboard", "books": [book], } def main() -> None: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--out", type=Path, default=DEFAULT_OUT) ap.add_argument("--state", type=Path, default=DEFAULT_STATE) ap.add_argument("--account-id", default="demo") ap.add_argument("--label", default="Stock-only book") ap.add_argument("--nav", type=float, default=50_000.0) ap.add_argument("--fund-scale", type=float, default=1.5) ap.add_argument("--spy-proxy", default="VOO") ap.add_argument( "--held", default="", help="Comma tickers currently held as CM dips (merged with state file)", ) ap.add_argument( "--approx-per-name-mv", type=float, default=2500.0, help="Approx MV per open dip for budget gate (0 = use model per-name)", ) ap.add_argument( "--skip-new-entries", action="store_true", help="Force watch-only for new dip candidates", ) args = ap.parse_args() state = _load_state(args.state) held = _held_from_args_and_state(args.held or None, state) if not held and args.held == "": # Sensible demo defaults if nothing provided held = _held_from_args_and_state( "AMZN,QCOM,AXP,INTC,LRCX,AMAT,MU,AMD,GEV", {"positions": {}}, ) snap = build_snapshot( account_id=str(args.account_id), label=str(args.label), nav=float(args.nav), fund_scale=float(args.fund_scale), held=held, spy_proxy=str(args.spy_proxy), approx_per_name_mv=float(args.approx_per_name_mv) if args.approx_per_name_mv > 0 else None, skip_new_entries=bool(args.skip_new_entries), ) out = args.out if not out.is_absolute(): out = (_REPO / out).resolve() out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(snap, indent=2) + "\n", encoding="utf-8") book = snap["books"][0] print( json.dumps( { "ok": True, "out": str(out), "as_of_date": book["as_of_date"], "n_exit": len(book["trades"]["exit"]), "n_enter": len(book["trades"]["enter"]), "n_hold": len(book["trades"]["hold"]), "over_budget": book["sleeves"]["equity_dip"]["over_budget"], "notes": book["notes"], }, indent=2, ) ) if __name__ == "__main__": main()