#!/usr/bin/env python3 """ Build a single JSON snapshot for the Best Ideas + IBKR command center UI. Reads local artifacts (fund daily, recommendations, rebalance logs) and optionally pulls live account + positions from TWS / IB Gateway. Example:: cd /Users/robzingale/trading_bot PYTHONUNBUFFERED=1 .venv/bin/python -m RenTech.monitor.build_command_center_snapshot --ibkr # Offline (no TWS): still loads recommendations + model targets PYTHONUNBUFFERED=1 .venv/bin/python -m RenTech.monitor.build_command_center_snapshot Output: ``RenTech/data/logs/command_center_snapshot.json`` """ from __future__ import annotations import argparse import asyncio import json import sys from datetime import datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo import pandas as pd _REPO = Path(__file__).resolve().parents[2] if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO)) from RenTech.monitor.signal_explain import build_signal_explanations # noqa: E402 from RenTech.monitor.strategy_catalog import build_strategy_summaries # noqa: E402 from RenTech.strategy_stack.combine_best_ideas_stack import ( # noqa: E402 FUND_WEIGHT_TABLE_TACTICAL_TSMOM, ) from RenTech.strategy_stack.multi_strategy_manager import SPDR_SECTOR_TICKERS # noqa: E402 NY = ZoneInfo("America/New_York") LOGS = _REPO / "RenTech" / "data" / "logs" DEFAULT_OUT = LOGS / "command_center_snapshot.json" SLEEVE_LABELS: dict[str, str] = { "spy_theta": "SPY Theta (lit4+VRP)", "vxx_regime": "VXX Regime Stack", "vxx_long_call": "VXX Long Call", "macro_aw": "Macro AW Options", "tactical_aw": "Tactical All Weather", "tsmom": "TSMOM / Managed Futures", "equity_dip": "Equity Dip", "sector_momentum": "Sector Momentum", } FUND_DAILY_DEFAULT = ( LOGS / "best_ideas_stack_10dd_plus_vxx_long_call_plus_macro_aw_plus_sector_momentum_plus_tactical_aw_plus_tsmom_plus_fund_plus_nav_q_mtm_daily.csv" ) FUND_META_DEFAULT = ( LOGS / "best_ideas_stack_10dd_plus_vxx_long_call_plus_macro_aw_plus_sector_momentum_plus_tactical_aw_plus_tsmom_plus_fund_plus_nav_q_mtm_meta.json" ) FUND_YEARLY_DEFAULT = ( LOGS / "best_ideas_stack_10dd_plus_vxx_long_call_plus_macro_aw_plus_sector_momentum_plus_tactical_aw_plus_tsmom_plus_fund_plus_nav_q_mtm_yearly.csv" ) # Legacy fund daily (older combine without quarterly sizing) FUND_DAILY_LEGACY = ( LOGS / "best_ideas_stack_plus_equity_dip_plus_vxx_long_call_plus_macro_aw_plus_sector_momentum_plus_tactical_aw_plus_fund_mtm_daily.csv" ) RECOMMENDATION_PATHS = ( LOGS / "ibkr_trade_recommendation.json", LOGS / "ibkr_trade_recommendation_r2a_diagonal.json", LOGS / "ibkr_trade_recommendation_r2b_spread.json", ) MACRO_TICKERS = ("SPY", "TLT", "IEF", "GLD", "DBC") LIVE_FUND_SIGNALS_PATH = LOGS / "live_fund_signals.json" def _read_json(path: Path) -> Any | None: if not path.is_file(): return None try: return json.loads(path.read_text()) except json.JSONDecodeError: return None def _latest_platform_run(log_dir: Path) -> dict[str, Any] | None: if not log_dir.is_dir(): return None files = sorted(log_dir.glob("platform_run_*.json"), key=lambda p: p.stat().st_mtime, reverse=True) if not files: return None data = _read_json(files[0]) if isinstance(data, dict): data["_source_file"] = files[0].name return data async def _fetch_ibkr_portfolio() -> tuple[dict[str, Any] | None, str | None]: try: from RenTech.live.ibkr_session import ( connect_ib, disconnect_ib, ensure_positions_loaded, ) from RenTech.live.platform_config import load_platform_config from RenTech.live.portfolio import fetch_portfolio_snapshot except ImportError as e: return None, f"IB modules unavailable: {e}" cfg = load_platform_config() ib = None try: ib = await connect_ib(cfg.broker) await ensure_positions_loaded(ib) snap = await fetch_portfolio_snapshot(ib) positions = [ { "symbol": p.symbol, "sec_type": p.sec_type, "expiry": p.expiry, "strike": p.strike, "right": p.right, "position": p.position, "avg_cost": p.avg_cost, "market_value": p.market_value, "unrealized_pnl": p.unrealized_pnl, "con_id": p.con_id, } for p in snap.positions ] payload = { "as_of": snap.as_of.isoformat(), "net_liquidation_usd": snap.net_liquidation_usd, "available_funds_usd": snap.available_funds_usd, "excess_liquidity_usd": snap.excess_liquidity_usd, "maintenance_margin_usd": snap.maintenance_margin_usd, "unrealized_pnl_usd": snap.unrealized_pnl_usd, "realized_pnl_today_usd": snap.realized_pnl_today_usd, "margin_utilization": snap.margin_utilization, "positions": positions, "positions_by_symbol": { sym: [x for x in positions if x["symbol"] == sym] for sym in sorted({p["symbol"] for p in positions}) }, } return payload, None except Exception as e: return None, str(e) finally: if ib is not None: await disconnect_ib(ib) def _etf_targets_from_rebalances( path: Path, *, date_col: str = "effective_date", weight_col: str = "weight", ticker_col: str = "ticker", ) -> tuple[dict[str, float], dict[str, Any]]: meta: dict[str, Any] = {"path": str(path), "status": "missing"} if not path.is_file(): return {}, meta df = pd.read_csv(path) if df.empty: meta["status"] = "empty" return {}, meta df[date_col] = pd.to_datetime(df[date_col]).dt.normalize() today = pd.Timestamp.now(tz=NY).normalize().tz_localize(None) future = df[df[date_col] > today] if len(future): eff = future[date_col].min() mode = "upcoming" else: eff = df[df[date_col] <= today][date_col].max() mode = "current" grp = df[(df[date_col] == eff) & (df[weight_col].astype(float) > 0)] targets = { str(r[ticker_col]): float(r[weight_col]) for _, r in grp.iterrows() } entries = grp.loc[grp.get("is_new_entry", False) == True, ticker_col].astype(str).tolist() # noqa: E712 exits = df[(df[date_col] == eff) & (df.get("is_exit_from_prior", False) == True)][ # noqa: E712 ticker_col ].astype(str).tolist() rankings: list[dict[str, Any]] = [] if "aqr_mom" in df.columns and "rank" in df.columns: snap_rows = df[df[date_col] == eff].sort_values("rank") for _, r in snap_rows.iterrows(): rankings.append( { "ticker": str(r[ticker_col]), "rank": int(r["rank"]), "aqr_mom": float(r["aqr_mom"]) if pd.notna(r["aqr_mom"]) else None, "held": float(r[weight_col]) > 0, } ) meta.update( { "status": "ok", "mode": mode, "effective_date": str(pd.Timestamp(eff).date()), "period_end_date": str(grp["period_end_date"].iloc[0]) if "period_end_date" in grp else None, "new_entries": entries, "exits_from_prior": exits, "tickers": list(targets.keys()), "rankings": rankings, "top_k": int(grp["top_k"].iloc[0]) if "top_k" in grp.columns and len(grp) else 3, } ) return targets, meta def _tactical_targets(path: Path) -> tuple[dict[str, float], dict[str, Any]]: meta: dict[str, Any] = {"path": str(path), "status": "missing"} if not path.is_file(): return {}, meta df = pd.read_csv(path) if df.empty: meta["status"] = "empty" return {}, meta df["date"] = pd.to_datetime(df["date"]).dt.normalize() today = pd.Timestamp.now(tz=NY).normalize().tz_localize(None) d = df[df["date"] <= today]["date"].max() if (df["date"] <= today).any() else df["date"].max() grp = df[df["date"] == d] targets = { str(r["ticker"]): float(r["weight"]) for _, r in grp.iterrows() if float(r["weight"]) > 0 } cash_w = float(grp["cash_weight"].iloc[0]) if len(grp) else 0.0 meta.update( { "status": "ok", "as_of": str(pd.Timestamp(d).date()), "cash_weight": cash_w, "tickers": list(targets.keys()), } ) return targets, meta def _macro_upcoming(path: Path) -> list[dict[str, Any]]: if not path.is_file(): return [] df = pd.read_csv(path) if df.empty or "entry_date" not in df.columns: return [] df["entry_date"] = pd.to_datetime(df["entry_date"]).dt.normalize() today = pd.Timestamp.now(tz=NY).normalize().tz_localize(None) # Open structures: exit in future if "exit_date" in df.columns: df["exit_date"] = pd.to_datetime(df["exit_date"]).dt.normalize() open_mask = df["exit_date"] >= today else: open_mask = pd.Series(True, index=df.index) rows = [] for _, r in df.loc[open_mask].tail(20).iterrows(): rows.append( { "sleeve": str(r.get("sleeve", r.get("structure", ""))), "entry_date": str(r["entry_date"].date()), "exit_date": str(r["exit_date"].date()) if "exit_date" in r else "", "underlying": str(r.get("underlying", r.get("ticker", ""))), } ) return rows def _ib_stock_mv(portfolio: dict[str, Any] | None) -> dict[str, float]: if not portfolio: return {} out: dict[str, float] = {} for p in portfolio.get("positions", []): if str(p.get("sec_type", "")).upper() not in ("STK", "ETF"): continue sym = str(p.get("symbol", "")) if not sym: continue out[sym] = out.get(sym, 0.0) + float(p.get("market_value", 0) or 0) return out def _ib_option_legs(portfolio: dict[str, Any] | None) -> list[dict[str, Any]]: if not portfolio: return [] legs = [] for p in portfolio.get("positions", []): if str(p.get("sec_type", "")).upper() != "OPT": continue legs.append(p) return legs def _compare_etf_book( *, book: str, targets: dict[str, float], ib_mv: dict[str, float], fund_nav: float, fund_weight: float, meta: dict[str, Any], ) -> list[dict[str, Any]]: """Suggest ADD/REMOVE/HOLD for ETF sleeves vs IB stock market values.""" actions: list[dict[str, Any]] = [] sleeve_nav = fund_nav * fund_weight if fund_nav > 0 else 0.0 universe = set(targets.keys()) held = {s: mv for s, mv in ib_mv.items() if abs(mv) > 50} for sym, w in sorted(targets.items(), key=lambda x: -x[1]): target_usd = sleeve_nav * w have = held.get(sym, 0.0) if have < target_usd * 0.5: actions.append( { "priority": 20, "action": "ADD", "book": book, "symbol": sym, "summary": f"Increase {sym} toward model weight {w:.1%}", "detail": { "target_weight": w, "target_notional_usd": round(target_usd, 2), "ib_market_value_usd": round(have, 2), **meta, }, } ) else: actions.append( { "priority": 80, "action": "HOLD", "book": book, "symbol": sym, "summary": f"Hold {sym} (~{w:.1%} of {book} sleeve)", "detail": { "target_weight": w, "ib_market_value_usd": round(have, 2), }, } ) for sym, mv in held.items(): if sym in universe: continue if sym in SPDR_SECTOR_TICKERS or sym in MACRO_TICKERS: actions.append( { "priority": 10, "action": "REMOVE", "book": book, "symbol": sym, "summary": f"Flat {sym} — not in current {book} model", "detail": {"ib_market_value_usd": round(mv, 2), **meta}, } ) return actions def _actions_from_recommendations(recs: list[dict[str, Any]]) -> list[dict[str, Any]]: actions: list[dict[str, Any]] = [] for rec in recs: decision = str(rec.get("decision", rec.get("entry_action", ""))).lower() regime = str(rec.get("regime", "")) qty = rec.get("proposed_qty") action_type = "WATCH" if decision in ("enter", "buy", "open", "recommend_enter"): action_type = "ADD" elif decision in ("exit", "close", "sell"): action_type = "REMOVE" legs = rec.get("legs") or [] leg_txt = "; ".join( f"{l.get('open_action','')} {l.get('ratio',1)}× {l.get('symbol','')}" f" {l.get('expiry','')} {l.get('right','')} {l.get('strike','')}" for l in legs[:6] ) actions.append( { "priority": 1 if action_type == "ADD" else 5, "action": action_type, "book": "spy_theta", "symbol": "SPY", "summary": f"VRP {regime.upper()}: {decision} qty={qty}", "detail": { "created_at": rec.get("created_at"), "entry_limit_per_share": rec.get("entry_limit_per_share"), "risk_budget_usd": rec.get("risk_budget_usd"), "legs": legs, "legs_text": leg_txt, "source": rec.get("_source", "recommendation"), }, } ) return actions def _resolve_fund_paths( daily: Path | None, meta: Path | None, yearly: Path | None, ) -> tuple[Path, Path | None, Path | None]: """Pick canonical fund artifacts; fall back to legacy daily if needed.""" d = daily or FUND_DAILY_DEFAULT if not d.is_file() and FUND_DAILY_LEGACY.is_file(): d = FUND_DAILY_LEGACY m = meta or FUND_META_DEFAULT if not m.is_file(): m = None y = yearly or FUND_YEARLY_DEFAULT if not y.is_file(): y = None return d, m, y def _load_fund_portfolio_summary( daily_path: Path, meta_path: Path | None, yearly_path: Path | None, ) -> dict[str, Any]: """Load Best Ideas fund-mode portfolio summary for the command center UI.""" if not daily_path.is_file(): return {"status": "missing", "path": str(daily_path)} df = pd.read_csv(daily_path, parse_dates=["date"]) if df.empty: return {"status": "empty", "path": str(daily_path)} df = df.sort_values("date") meta = _read_json(meta_path) if meta_path else {} meta = meta if isinstance(meta, dict) else {} start_cap = float(meta.get("capital_usd", 100_000.0)) row = df.iloc[-1] eq = df["equity_mtm_usd"].astype(float) total_pnl = float(eq.iloc[-1] - start_cap) years = max((df["date"].iloc[-1] - df["date"].iloc[0]).days / 365.25, 1e-9) cagr = (float(eq.iloc[-1]) / start_cap) ** (1.0 / years) - 1.0 if eq.iloc[-1] > 0 else float("nan") dd = (eq / eq.cummax() - 1.0).min() rets = df["daily_return_mtm"].astype(float) if "daily_return_mtm" in df.columns else eq.pct_change().fillna(0.0) sd = float(rets.std(ddof=1)) sharpe = float(rets.mean() / sd * (252.0**0.5)) if sd > 1e-12 else float("nan") fund_weights = meta.get("fund_weights") or {} pnl_cols = [c for c in df.columns if c.startswith("pnl_fund_")] sleeves: dict[str, Any] = {} for col in pnl_cols: key = col.replace("pnl_fund_", "") pnl_s = df[col].fillna(0.0).astype(float) total_sleeve_pnl = float(pnl_s.sum()) notional_col = f"notional_{key}_usd" notional = float(row[notional_col]) if notional_col in row else None mean_notional = ( float(df[notional_col].mean()) if notional_col in df.columns else None ) w = fund_weights.get(key) if w is None: w = FUND_WEIGHT_TABLE_TACTICAL_TSMOM.get(key) sleeves[key] = { "label": SLEEVE_LABELS.get(key, key), "weight": float(w) if w is not None else None, "notional_usd": notional, "mean_notional_usd": mean_notional, "pnl_today_usd": float(row[col]) if col in row else 0.0, "total_pnl_usd": total_sleeve_pnl, "return_on_start_capital_pct": total_sleeve_pnl / start_cap * 100.0, "pct_of_fund_pnl": total_sleeve_pnl / total_pnl * 100.0 if total_pnl else 0.0, } yearly_rows: list[dict[str, Any]] = [] if yearly_path and yearly_path.is_file(): yr = pd.read_csv(yearly_path) for _, r in yr.iterrows(): yearly_rows.append( { "year": int(r["year"]), "return_pct_chained": float(r.get("return_pct_chained", 0)), "return_pct_constant": float(r.get("return_pct_constant", 0)), "pnl_usd": float(r.get("pnl_usd", 0)), "end_equity_chained": float(r.get("end_equity_chained", 0)), "max_dd_pct_chained": float(r.get("max_dd_pct_chained", 0)), } ) else: fund_eq = eq.values dates = pd.to_datetime(df["date"]) for y in sorted(dates.dt.year.unique()): mask = dates.dt.year == y g = df.loc[mask] start_eq = float(eq.shift(1).loc[g.index[0]]) if pd.isna(start_eq): start_eq = start_cap end_eq = float(eq.loc[g.index[-1]]) yearly_rows.append( { "year": int(y), "return_pct_chained": (end_eq / start_eq - 1.0) * 100.0, "pnl_usd": float(g["pnl_best_ideas_mtm"].sum()) if "pnl_best_ideas_mtm" in g.columns else end_eq - start_eq, "end_equity_chained": end_eq, } ) yearly_by_sleeve: list[dict[str, Any]] = [] for col in pnl_cols: key = col.replace("pnl_fund_", "") pnl_s = df[col].fillna(0.0).astype(float) for y in sorted(df["date"].dt.year.unique()): mask = df["date"].dt.year == y p = float(pnl_s.loc[mask].sum()) start_nav = float(eq.shift(1).loc[mask].iloc[0]) if pd.isna(start_nav) or start_nav <= 0: start_nav = start_cap yearly_by_sleeve.append( { "year": int(y), "sleeve": key, "label": SLEEVE_LABELS.get(key, key), "pnl_usd": p, "return_pct_of_fund_nav": p / start_nav * 100.0, } ) return { "status": "ok", "path": str(daily_path), "meta_path": str(meta_path) if meta_path else None, "combine_mode": meta.get("combine_mode", "fund_nav_quarterly_sized"), "fund_nav_rebalance": meta.get("fund_nav_rebalance", "quarterly"), "fund_scale": meta.get("fund_scale") or meta.get("fund_weights"), "start_date": str(pd.Timestamp(df["date"].iloc[0]).date()), "end_date": str(pd.Timestamp(row["date"]).date()), "date": str(pd.Timestamp(row["date"]).date()), "capital_usd": start_cap, "fund_nav_usd": float(row.get("equity_mtm_usd", 0)), "nav_ref_usd": float(row.get("nav_ref_usd", row.get("equity_mtm_usd", 0))), "fund_daily_return": float(row.get("daily_return_mtm", 0)), "total_return_pct": float(eq.iloc[-1] / start_cap - 1.0) * 100.0, "cagr_pct": float(cagr) * 100.0, "sharpe": sharpe, "max_drawdown_pct": float(dd) * 100.0, "total_pnl_usd": total_pnl, "margin_combined_usd": float(row.get("margin_combined_usd", 0)), "margin_combined_mean_usd": float(df["margin_combined_usd"].mean()) if "margin_combined_usd" in df.columns else None, "margin_combined_peak_usd": float(df["margin_combined_usd"].max()) if "margin_combined_usd" in df.columns else None, "min_account_size_usd": meta.get("min_account_size_usd"), "bp_utilization_pct": float(row.get("bp_utilization_pct", 0)), "fund_weights": fund_weights, "sleeves": sleeves, "yearly": yearly_rows, "yearly_by_sleeve": yearly_by_sleeve, } def _load_fund_latest(summary: dict[str, Any]) -> dict[str, Any]: """Backward-compatible slice of portfolio summary for legacy UI fields.""" if summary.get("status") != "ok": return summary sleeves_out: dict[str, Any] = {} for key, sl in (summary.get("sleeves") or {}).items(): sleeves_out[key] = { "weight": sl.get("weight"), "allocated_usd": sl.get("notional_usd"), "pnl_today_usd": sl.get("pnl_today_usd"), "total_pnl_usd": sl.get("total_pnl_usd"), } return { "status": "ok", "date": summary.get("date"), "fund_nav_usd": summary.get("fund_nav_usd"), "fund_daily_return": summary.get("fund_daily_return"), "sleeves": sleeves_out, } def _merge_live_fund_signals(snap: dict[str, Any], live: dict[str, Any]) -> dict[str, Any]: """Overlay IBKR-derived targets/actions; keep backtest fund performance as reference.""" out = dict(snap) out["data_source"] = "ibkr_live_merged" ibp = (live.get("ibkr") or {}).get("portfolio") if ibp: out["ibkr"] = { "connected": True, "error": None, "portfolio": ibp, "option_positions": _ib_option_legs(ibp), } sig = dict(out.get("signals") or {}) for key in ("sector_momentum", "tactical_aw", "spy_gates", "vrp_cycle", "recommendations"): if key in (live.get("signals") or {}): sig[key] = live["signals"][key] sig["data_source"] = live.get("data_source", "ibkr_live") sig["live_fund_signals_at"] = live.get("generated_at") sig["bars_fetched"] = live.get("bars_fetched") sig["bars_missing"] = live.get("bars_missing") out["signals"] = sig live_actions = list(live.get("actions") or []) keep_books = {"equity_dip", "macro_aw"} model_extra = [a for a in out.get("actions", []) if a.get("book") in keep_books] out["actions"] = sorted( live_actions + model_extra, key=lambda a: (a["priority"], a["action"], a.get("book", "")), ) fund = dict(out.get("fund") or {}) model = dict(fund.get("model") or {}) model["fund_nav_usd_live_ibkr"] = live.get("fund_nav_usd") model["live_weights"] = live.get("fund_weights") fund["model"] = model fund["reference_note"] = ( "Returns/Sharpe from backtest CSV; targets and ADD/REMOVE from IBKR live_fund_signals" ) out["fund"] = fund return out def build_snapshot( *, use_ibkr: bool, live_signals: bool = False, live_signals_from_file: bool = False, live_signals_path: Path | None = None, fund_daily: Path | None = None, fund_meta: Path | None = None, fund_yearly: Path | None = None, ) -> dict[str, Any]: now = datetime.now(NY) live_payload: dict[str, Any] | None = None if live_signals or live_signals_from_file: # noqa: SIM102 — load or fetch live payload path = live_signals_path or LIVE_FUND_SIGNALS_PATH if live_signals_from_file: data = _read_json(path) if not isinstance(data, dict): raise FileNotFoundError(f"Missing live fund signals: {path}") live_payload = data else: from RenTech.live.fund_signals import run_live_fund_signals_async, write_live_fund_signals live_payload = asyncio.run(run_live_fund_signals_async()) write_live_fund_signals(path, live_payload) ib_portfolio, ib_err = (None, None) if live_payload and (live_payload.get("ibkr") or {}).get("portfolio"): ib_portfolio = live_payload["ibkr"]["portfolio"] elif use_ibkr: ib_portfolio, ib_err = asyncio.run(_fetch_ibkr_portfolio()) daily_path, meta_path, yearly_path = _resolve_fund_paths(fund_daily, fund_meta, fund_yearly) combine_meta = _read_json(meta_path) if meta_path else {} combine_meta = combine_meta if isinstance(combine_meta, dict) else {} portfolio = _load_fund_portfolio_summary(daily_path, meta_path, yearly_path) portfolio["strategies_summary"] = build_strategy_summaries(portfolio, combine_meta) fund = _load_fund_latest(portfolio) fund_nav = float(portfolio.get("fund_nav_usd", 100_000.0)) weights = portfolio.get("fund_weights") or { k: float(FUND_WEIGHT_TABLE_TACTICAL_TSMOM[k]) for k in FUND_WEIGHT_TABLE_TACTICAL_TSMOM } recs: list[dict[str, Any]] = [] for p in RECOMMENDATION_PATHS: data = _read_json(p) if isinstance(data, dict): data = {**data, "_source": p.name} recs.append(data) signal_state = _read_json(LOGS / "live_signal_state.json") vrp_state = _read_json(_REPO / "RenTech/strategy_stack/live_vrp_state.json") platform_run = _latest_platform_run(LOGS / "live_platform") sector_targets, sector_meta = _etf_targets_from_rebalances( LOGS / "sector_momentum_standard_rebalances.csv" ) tactical_targets, tactical_meta = _tactical_targets( LOGS / "tactical_aw_standard_allocations.csv" ) macro_open = _macro_upcoming(LOGS / "macro_aw_options_portfolio_eq_trades.csv") ib_mv = _ib_stock_mv(ib_portfolio) actions: list[dict[str, Any]] = [] actions.extend(_actions_from_recommendations(recs)) if weights.get("sector_momentum", 0.0) > 0: actions.extend( _compare_etf_book( book="sector_momentum", targets=sector_targets, ib_mv=ib_mv, fund_nav=fund_nav, fund_weight=weights.get("sector_momentum", 0.0), meta=sector_meta, ) ) actions.extend( _compare_etf_book( book="tactical_aw", targets=tactical_targets, ib_mv=ib_mv, fund_nav=fund_nav, fund_weight=weights.get("tactical_aw", 0.0), meta=tactical_meta, ) ) actions.append( { "priority": 50, "action": "WATCH", "book": "equity_dip", "symbol": "", "summary": "Buy-the-dip: enter on ≥3% daily drop (top 5 SPX / top 10 R3k names)", "detail": { "sp500_top_n": 5, "russell3000_top_n": 10, "hold_days": 10, "note": "Not in default stack; optional sleeve", }, } ) if macro_open: actions.append( { "priority": 40, "action": "HOLD", "book": "macro_aw", "symbol": "", "summary": f"{len(macro_open)} open Macro AW option structures (model)", "detail": {"open_structures": macro_open[:8]}, } ) actions.sort(key=lambda a: (a["priority"], a["action"], a.get("book", ""))) kill_switch = (_REPO / "RenTech/live/config/KILL_SWITCH").is_file() snap = { "generated_at": now.isoformat(), "ibkr": { "connected": ib_portfolio is not None, "error": ib_err, "portfolio": ib_portfolio, "option_positions": _ib_option_legs(ib_portfolio), }, "risk": { "kill_switch_active": kill_switch, "platform_run_source": (platform_run or {}).get("_source_file"), "platform_risk": (platform_run or {}).get("risk"), }, "signals": { "live_signal_state": signal_state, "vrp_state": vrp_state, "sector_momentum": {"targets": sector_targets, **sector_meta}, "tactical_aw": {"targets": tactical_targets, **tactical_meta}, "macro_aw": {"open_structures": macro_open}, "recommendations": recs, }, "fund": { "model": fund, "portfolio": portfolio, "target_weights": weights, "daily_csv": str(daily_path), "meta_csv": str(meta_path) if meta_path else None, }, "actions": actions, } if live_payload is not None: snap = _merge_live_fund_signals(snap, live_payload) snap["signal_explain"] = build_signal_explanations(snap) return snap def main() -> None: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument( "--ibkr", action="store_true", help="Connect to TWS/IB Gateway (requires ib_insync + running gateway)", ) ap.add_argument( "--out", type=Path, default=DEFAULT_OUT, help="Output JSON path", ) ap.add_argument( "--fund-daily", type=Path, default=None, help="Override fund MTM daily CSV (default: best_ideas_stack_10dd fund+nav_q)", ) ap.add_argument( "--fund-meta", type=Path, default=None, help="Override fund MTM meta JSON", ) ap.add_argument( "--live-signals", action="store_true", help="Use IBKR historical bars for tactical/sector targets (+ VRP recommend). " "Loads live_fund_signals.json if present; otherwise connects to IB and writes it.", ) ap.add_argument( "--live-signals-path", type=Path, default=LIVE_FUND_SIGNALS_PATH, help="Path for live fund signals JSON (read or write)", ) ap.add_argument( "--live-signals-from-file", action="store_true", help="Merge tactical/sector/VRP from existing live_fund_signals.json (no IB bar fetch)", ) args = ap.parse_args() snap = build_snapshot( use_ibkr=bool(args.ibkr) or bool(args.live_signals), live_signals=bool(args.live_signals), live_signals_from_file=bool(args.live_signals_from_file), live_signals_path=args.live_signals_path, fund_daily=args.fund_daily, fund_meta=args.fund_meta, ) out = args.out.expanduser().resolve() out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(snap, indent=2, default=str) + "\n") print(f"Wrote {out}") print( f" IBKR connected: {snap['ibkr']['connected']} " f"actions: {len(snap['actions'])} " f"positions: {len((snap['ibkr'].get('portfolio') or {}).get('positions', []))}" ) if snap["ibkr"].get("error"): print(f" IBKR note: {snap['ibkr']['error']}") if snap.get("data_source") == "ibkr_live_merged": print(f" Live signals: {snap.get('signals', {}).get('live_fund_signals_at', '')}") if __name__ == "__main__": main()