#!/usr/bin/env python3 """ Build a print-ready HTML deck for the **stock-only + MA slope** fund book. Example:: cd /Users/robzingale/trading_bot && PYTHONUNBUFFERED=1 .venv/bin/python \\ RenTech/strategy_stack/generate_stock_only_ma_slope_presentation.py \\ --start 2011-06-20 --end 2026-06-18 \\ --fund-scale 1.5 \\ --portfolio-daily RenTech/data/logs/stock_only_ma_slope_2011_fund15_plus_stock_only_plus_sp500_dip_plus_sector_momentum_plus_tactical_aw_plus_tsmom_plus_johansen_etf_plus_ma_slope_topn_plus_ma_slope_inverse_plus_vol_edge_plus_fund_plus_nav_q_mtm_daily.csv \\ --out stock_only_ma_slope_presentation.html """ from __future__ import annotations import argparse import json import sys from dataclasses import dataclass from pathlib import Path import numpy as np 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.strategy_stack.data_loader import DataLoader from RenTech.strategy_stack.main import _compute_daily_backtest_features LOGS = _REPO / "RenTech" / "data" / "logs" FUND_WEIGHTS = { "tactical_aw": 0.385, "equity_dip": 0.17, "vol_edge": 0.11, "qs_actionable_etf": 0.04, "ma_slope_topn": 0.08, "johansen_etf": 0.0825, "tsmom": 0.0725, "ma_slope_inverse": 0.05, } DEFAULT_PORTFOLIO_DAILY = ( LOGS / "stock_only_ma_slope_fund15_2011_plus_stock_only_plus_sp500_dip_plus_tactical_aw_plus_tsmom_plus_johansen_etf_plus_ma_slope_topn_plus_ma_slope_inverse_plus_vol_edge_plus_fund_plus_nav_q_mtm_daily.csv" ) SLEEVES: list[dict] = [ { "key": "tactical_aw", "name": "Tactical All Weather", "token": "tactical_aw", "weight": FUND_WEIGHTS["tactical_aw"], "daily": LOGS / "tactical_aw_standard_daily.csv", "extra": LOGS / "tactical_aw_standard_allocations.csv", "blurb": ( "Macro ETF book (SPY 30%, TLT 40%, IEF 15%, GLD/DBC 7.5% each). " "Each sleeve moves to cash when price is below SMA(200) or 12-1 momentum " "is negative. Core equity-beta anchor of the stock book." ), "time_col": "total_invested_weight", "time_daily": LOGS / "tactical_aw_standard_daily.csv", }, { "key": "equity_dip", "name": "Equity Dip (CM SP100)", "token": "equity_dip", "weight": FUND_WEIGHTS["equity_dip"], "daily": LOGS / "cracking_markets_sp100_dip_daily.csv", "blurb": ( "CrackingMarkets dip on S&P 100: −3% day, SMA200 up, ATR filter, rank ATR/close. " "Limit entry next day at close−0.9×ATR; exits on profit target, prior high, or " ">10d hold. Max 10 concurrent names." ), }, { "key": "qs_actionable_etf", "name": "QS Actionable-4 (SPY)", "token": "qs_actionable_etf", "weight": FUND_WEIGHTS["qs_actionable_etf"], "daily": LOGS / "qs_actionable_4_standard_daily.csv", "blurb": ( "Equal-weight SPY calendar/overnight diversifiers: Turnaround Tuesday, " "overnight after 3 down days, first day of month, overnight at 10-day low." ), }, { "key": "vol_edge", "name": "Volatility Edge ETN", "token": "vol_edge", "weight": FUND_WEIGHTS["vol_edge"], "daily": LOGS / "volatility_edge_etn_evrp_boc_daily.csv", "blurb": ( "VIX ETN sleeve (SSRN 5316487 Strategy 3): trades VIX term-structure " "(EVRP) with long/short VIX ETN weights; complements equity with " "vol-specific edge." ), "time_expr": "w_short + w_long", }, { "key": "ma_slope_topn", "name": "MA Slope Top-10", "token": "ma_slope_topn", "weight": FUND_WEIGHTS["ma_slope_topn"], "daily": LOGS / "ma_slope_sp500_top10_top10_dual_product_monthly_atr2x_daily.csv", "rebalances": LOGS / "ma_slope_sp500_top10_top10_dual_product_monthly_atr2x_rebalances.csv", "blurb": ( "S&P 500 cross-sectional momentum: monthly top-10 by dual EMA slope " "(fast×slow), equal weight. Per-name 2× ATR chandelier stop between " "rebalances; cash when stopped." ), }, { "key": "johansen_etf", "name": "Johansen ETF Triplets", "token": "johansen_etf", "weight": FUND_WEIGHTS["johansen_etf"], "daily": LOGS / "johansen_triplet_etf_standard_daily.csv", "blurb": ( "Six ETF triplet stat-arb sleeves (Chan-style Johansen cointegration), " "equal-weight combine. Market-neutral-ish macro pairs; low correlation " "to single-factor equity." ), }, { "key": "tsmom", "name": "TSMOM / Managed Futures", "token": "tsmom", "weight": FUND_WEIGHTS["tsmom"], "daily": LOGS / "tsmom_managed_futures_daily.csv", "blurb": ( "8-asset time-series momentum (SPY, EFA, EEM, TLT, IEF, GLD, DBC, UUP); " "3/6/12-month signal blend, vol-normalized, monthly rebalance. " "Near-zero equity beta standalone." ), "time_expr": "gross_exposure", }, { "key": "ma_slope_inverse", "name": "MA Slope Inverse (SH)", "token": "ma_slope_inverse", "weight": FUND_WEIGHTS["ma_slope_inverse"], "daily": LOGS / "ma_slope_inverse_spy_balanced_bear_dual_or_sma200_spy_slope_or_sma200_invreq_top_n1_SH_daily.csv", "positions": LOGS / "ma_slope_inverse_spy_balanced_bear_dual_or_sma200_spy_slope_or_sma200_invreq_top_n1_SH_positions.csv", "blurb": ( "Bear hedge: long SH (−1× SPY) when SPY dual-slope is bearish or below " "SMA200, with inverse-ETF momentum confirmation. Stays hedged until SPY " "regime clears — not whipsawed on inverse bounce alone." ), }, ] @dataclass class SleeveStats: key: str name: str weight: float blurb: str n_days: int total_return_pct: float cagr_pct: float sharpe: float max_dd_pct: float daily_win_rate_pct: float avg_winner_day_pct: float avg_loser_day_pct: float time_in_position_pct: float trade_win_rate_pct: float | None avg_winner_trade_pct: float | None avg_loser_trade_pct: float | None avg_hold_days: float | None n_trades: int | None spy_total_return_pct: float spy_cagr_pct: float spy_sharpe: float spy_max_dd_pct: float beta_spy: float corr_spy: float def _load_daily(path: Path, start: pd.Timestamp, end: pd.Timestamp) -> pd.DataFrame: df = pd.read_csv(path, parse_dates=["date"]).sort_values("date") df["date"] = pd.to_datetime(df["date"]).dt.normalize() mask = (df["date"] >= start) & (df["date"] <= end) return df.loc[mask].copy() def _daily_win_rate_pct(r: pd.Series, *, in_market: pd.Series | None = None) -> float: """ Win rate on decisive days only: winners / (winners + losers). Flat (≈0) sessions are excluded. When *in_market* is provided, restrict to those days first (e.g. tactical AW invested weight, dip active days). """ r = r.dropna().astype(float) if in_market is not None: mask = in_market.reindex(r.index).fillna(False).astype(bool) r = r.loc[mask] n_win = int((r > 0).sum()) n_loss = int((r < 0).sum()) denom = n_win + n_loss if denom == 0: return float("nan") return 100.0 * n_win / denom def _in_market_mask( sleeve: dict, df: pd.DataFrame, start: pd.Timestamp, end: pd.Timestamp, ) -> pd.Series | None: """True on days the sleeve has material market exposure (when knowable).""" key = sleeve["key"] dates = pd.to_datetime(df["date"]).dt.normalize() if sleeve.get("time_col") and sleeve["time_col"] in df.columns: return df[sleeve["time_col"]].fillna(0.0).astype(float) > 0.01 if sleeve.get("time_expr") == "w_short + w_long" and {"w_short", "w_long"} <= set(df.columns): return (df["w_short"] + df["w_long"]).fillna(0.0).astype(float) > 0.01 if sleeve.get("time_expr") == "gross_exposure" and { "gross_long_notional_usd", "gross_short_notional_usd", } <= set(df.columns): gross = df["gross_long_notional_usd"].abs() + df["gross_short_notional_usd"].abs() return gross > 1.0 if sleeve.get("positions"): pos = pd.read_csv(sleeve["positions"], parse_dates=["date"]) pos["date"] = pd.to_datetime(pos["date"]).dt.normalize() pos = pos[(pos["date"] >= start) & (pos["date"] <= end)] if "weight" in pos.columns: daily_w = pos.groupby("date")["weight"].sum() return dates.isin(daily_w[daily_w > 0].index) if key == "tactical_aw": if "total_invested_weight" in df.columns: return df["total_invested_weight"].fillna(0.0).astype(float) > 0.01 if sleeve.get("time_daily"): ta = _load_daily(Path(sleeve["time_daily"]), start, end) merged = df.merge(ta[["date", "total_invested_weight"]], on="date", how="left") return merged["total_invested_weight"].fillna(0.0).astype(float) > 0.01 if key == "equity_dip": return df["daily_ret"].astype(float).abs() > 1e-6 # QS actionable, Johansen, etc.: no position log — caller uses decisive days only. return None def _metrics_from_returns( r: pd.Series, capital: float = 100_000.0, *, in_market: pd.Series | None = None, ) -> dict: r = r.dropna().astype(float) n = len(r) if n < 2: return {} eq = capital * (1.0 + r).cumprod() years = n / 252.0 total_ret = eq.iloc[-1] / capital - 1.0 cagr = (eq.iloc[-1] / capital) ** (1.0 / years) - 1.0 if years > 0 else np.nan dd = (eq / eq.cummax() - 1.0).min() sd = float(r.std(ddof=1)) sharpe = float(r.mean() / sd * np.sqrt(252.0)) if sd > 1e-12 else np.nan wins = r[r > 0] losses = r[r < 0] return { "n_days": n, "total_return_pct": total_ret * 100, "cagr_pct": cagr * 100, "sharpe": sharpe, "max_dd_pct": dd * 100, "daily_win_rate_pct": _daily_win_rate_pct(r, in_market=in_market), "avg_winner_day_pct": float(wins.mean() * 100) if len(wins) else 0.0, "avg_loser_day_pct": float(losses.mean() * 100) if len(losses) else 0.0, } def _episode_trades_from_flag(dates: pd.Series, flag: pd.Series, daily_ret: pd.Series) -> pd.DataFrame: """Contiguous episodes where flag is True; return per-episode compound return.""" rows: list[dict] = [] in_ep = False ep_start = None ep_rets: list[float] = [] for dt, f, ret in zip(dates, flag, daily_ret): if f and not in_ep: in_ep = True ep_start = dt ep_rets = [float(ret)] elif f and in_ep: ep_rets.append(float(ret)) elif not f and in_ep: compound = float(np.prod(1.0 + np.array(ep_rets)) - 1.0) rows.append( { "start": ep_start, "end": dt, "hold_days": len(ep_rets), "return_pct": compound * 100, } ) in_ep = False if in_ep and ep_rets: compound = float(np.prod(1.0 + np.array(ep_rets)) - 1.0) rows.append( { "start": ep_start, "end": dates.iloc[-1], "hold_days": len(ep_rets), "return_pct": compound * 100, } ) return pd.DataFrame(rows) def _trade_stats_from_episodes(ep: pd.DataFrame) -> dict: if ep is None or ep.empty: return { "trade_win_rate_pct": None, "avg_winner_trade_pct": None, "avg_loser_trade_pct": None, "avg_hold_days": None, "n_trades": None, } wins = ep.loc[ep["return_pct"] > 0, "return_pct"] losses = ep.loc[ep["return_pct"] < 0, "return_pct"] return { "trade_win_rate_pct": 100.0 * float((ep["return_pct"] > 0).mean()), "avg_winner_trade_pct": float(wins.mean()) if len(wins) else None, "avg_loser_trade_pct": float(losses.mean()) if len(losses) else None, "avg_hold_days": float(ep["hold_days"].mean()), "n_trades": int(len(ep)), } def _time_in_position(sleeve: dict, df: pd.DataFrame, start: pd.Timestamp, end: pd.Timestamp) -> float: key = sleeve["key"] if sleeve.get("time_col") and sleeve["time_col"] in df.columns: return 100.0 * float(df[sleeve["time_col"]].mean()) if sleeve.get("time_expr") == "w_short + w_long" and {"w_short", "w_long"} <= set(df.columns): return 100.0 * float((df["w_short"] + df["w_long"]).mean()) if sleeve.get("time_expr") == "gross_exposure" and { "gross_long_notional_usd", "gross_short_notional_usd", } <= set(df.columns): gross = df["gross_long_notional_usd"].abs() + df["gross_short_notional_usd"].abs() return 100.0 * float((gross / 100_000.0).mean()) if sleeve.get("positions"): pos = pd.read_csv(sleeve["positions"], parse_dates=["date"]) pos["date"] = pd.to_datetime(pos["date"]).dt.normalize() pos = pos[(pos["date"] >= start) & (pos["date"] <= end)] if "weight" in pos.columns: daily_w = pos.groupby("date")["weight"].sum().reindex(df["date"]).fillna(0.0) return 100.0 * float((daily_w > 0).mean()) if sleeve.get("rebalances") and key == "ma_slope_topn": # Approximate: invested when daily return volatility > 0 or always ~full when strategy active return 100.0 * float((df["daily_ret"].abs() > 1e-12).mean()) * 0.85 if key == "equity_dip": # Episodic: days with material PnL vs cash active = df["daily_ret"].abs() > 1e-6 return 100.0 * float(active.mean()) return 100.0 * float((df["daily_ret"].abs() > 1e-8).mean()) def _monthly_episodes(dates: pd.Series, daily_ret: pd.Series) -> pd.DataFrame: """One episode per calendar month (for monthly-rebalance sleeves).""" df = pd.DataFrame({"date": pd.to_datetime(dates), "r": daily_ret.astype(float)}) df["ym"] = df["date"].dt.to_period("M") rows: list[dict] = [] for _, g in df.groupby("ym", sort=True): rets = g["r"].values compound = float(np.prod(1.0 + rets) - 1.0) rows.append( { "start": g["date"].iloc[0], "end": g["date"].iloc[-1], "hold_days": len(g), "return_pct": compound * 100.0, } ) return pd.DataFrame(rows) def _episode_trades_for_sleeve(sleeve: dict, df: pd.DataFrame) -> pd.DataFrame | None: key = sleeve["key"] r = df["daily_ret"].astype(float) dates = df["date"] if sleeve.get("positions"): pos = pd.read_csv(sleeve["positions"], parse_dates=["date"]) pos["date"] = pd.to_datetime(pos["date"]).dt.normalize() pos = pos.merge(df[["date", "daily_ret"]], on="date", how="right").fillna({"weight": 0.0}) flag = pos["weight"].fillna(0.0) > 0 return _episode_trades_from_flag(dates, flag, r) if sleeve.get("rebalances") and key == "sector_momentum": reb = pd.read_csv(sleeve["rebalances"], parse_dates=["effective_date", "signal_date"]) reb["effective_date"] = pd.to_datetime(reb["effective_date"]).dt.normalize() held = reb.groupby("effective_date")["weight"].sum() flag = df["date"].isin(held.index) | df["daily_ret"].abs().gt(1e-8) return _episode_trades_from_flag(dates, flag, r) if key == "ma_slope_topn" and sleeve.get("rebalances"): return _monthly_episodes(dates, r) if key == "vol_edge" and "rebalanced" in df.columns: invested = (df["w_short"] + df["w_long"]) > 0.01 return _episode_trades_from_flag(dates, invested, r) if key == "tactical_aw": if "total_invested_weight" in df.columns: flag = df["total_invested_weight"].fillna(0.0) > 0.01 elif sleeve.get("time_daily"): ta = _load_daily( Path(sleeve["time_daily"]), pd.Timestamp(dates.min()), pd.Timestamp(dates.max()), ) merged = df.merge(ta[["date", "total_invested_weight"]], on="date", how="left") flag = merged["total_invested_weight"].fillna(0.0) > 0.01 else: flag = df["daily_ret"].abs() > 1e-8 return _episode_trades_from_flag(dates, flag, r) if key == "equity_dip": flag = df["daily_ret"].abs() > 1e-6 return _episode_trades_from_flag(dates, flag, r) if key in ("tsmom", "johansen_etf"): return _monthly_episodes(dates, r) return None def analyze_sleeve(sleeve: dict, spy_r: pd.Series, start: pd.Timestamp, end: pd.Timestamp) -> SleeveStats: df = _load_daily(sleeve["daily"], start, end) r = df["daily_ret"].astype(float) in_market = _in_market_mask(sleeve, df, start, end) m = _metrics_from_returns(r, in_market=in_market) aligned = pd.DataFrame({"s": r.values, "spy": spy_r.reindex(df["date"]).values}, index=df["date"]).dropna() beta = float(np.cov(aligned["s"], aligned["spy"])[0, 1] / np.var(aligned["spy"])) if len(aligned) > 2 else np.nan corr = float(aligned["s"].corr(aligned["spy"])) if len(aligned) > 2 else np.nan spy_m = _metrics_from_returns(spy_r.reindex(df["date"]).dropna()) ep = _episode_trades_for_sleeve(sleeve, df) t = _trade_stats_from_episodes(ep) if ep is not None else _trade_stats_from_episodes(pd.DataFrame()) tip = _time_in_position(sleeve, df, start, end) return SleeveStats( key=sleeve["key"], name=sleeve["name"], weight=sleeve["weight"], blurb=sleeve["blurb"], n_days=int(m.get("n_days", 0)), total_return_pct=float(m.get("total_return_pct", np.nan)), cagr_pct=float(m.get("cagr_pct", np.nan)), sharpe=float(m.get("sharpe", np.nan)), max_dd_pct=float(m.get("max_dd_pct", np.nan)), daily_win_rate_pct=float(m.get("daily_win_rate_pct", np.nan)), avg_winner_day_pct=float(m.get("avg_winner_day_pct", np.nan)), avg_loser_day_pct=float(m.get("avg_loser_day_pct", np.nan)), time_in_position_pct=tip, beta_spy=beta, corr_spy=corr, spy_total_return_pct=float(spy_m.get("total_return_pct", np.nan)), spy_cagr_pct=float(spy_m.get("cagr_pct", np.nan)), spy_sharpe=float(spy_m.get("sharpe", np.nan)), spy_max_dd_pct=float(spy_m.get("max_dd_pct", np.nan)), **t, ) def analyze_portfolio(path: Path, spy_r: pd.Series, start: pd.Timestamp, end: pd.Timestamp) -> dict: df = _load_daily(path, start, end) r = df["equity_mtm_usd"].astype(float).pct_change().fillna(0.0) m = _metrics_from_returns(r) aligned = pd.DataFrame({"p": r.values, "spy": spy_r.reindex(df["date"]).values}, index=df["date"]).dropna() beta = float(np.cov(aligned["p"], aligned["spy"])[0, 1] / np.var(aligned["spy"])) if len(aligned) > 2 else np.nan corr = float(aligned["p"].corr(aligned["spy"])) if len(aligned) > 2 else np.nan spy_m = _metrics_from_returns(spy_r.reindex(df["date"]).dropna()) yearly = [] eq = 100_000.0 for yr, g in df.groupby(df["date"].dt.year): end_eq = float(g["equity_mtm_usd"].iloc[-1]) ret = (end_eq / eq - 1) * 100 if eq > 0 else 0.0 intra = eq + (g["equity_mtm_usd"] - g["equity_mtm_usd"].iloc[0]) dd = (intra / intra.cummax() - 1).min() * 100 yearly.append({"year": int(yr), "return_pct": ret, "max_dd_pct": dd}) eq = end_eq return {**m, "beta_spy": beta, "corr_spy": corr, "spy": spy_m, "yearly": yearly, "end_equity": eq, "_daily_df": df} def _sleeve_invested_fraction_series( sleeve: dict, dates: pd.DatetimeIndex, start: pd.Timestamp, end: pd.Timestamp, ) -> pd.Series: """Fraction of sleeve notional deployed in market (0–1+, shorts may exceed 1 on TSMOM).""" key = sleeve["key"] if key == "tactical_aw": path = Path(sleeve.get("time_daily") or sleeve["daily"]) df = _load_daily(path, start, end) s = df.set_index("date")["total_invested_weight"].astype(float) return s.reindex(dates).fillna(0.0) if key == "vol_edge": df = _load_daily(Path(sleeve["daily"]), start, end) s = df.set_index("date")["w_short"].astype(float) + df.set_index("date")["w_long"].astype(float) return s.reindex(dates).fillna(0.0) if key == "tsmom": df = _load_daily(Path(sleeve["daily"]), start, end) gross = df["gross_long_notional_usd"].abs() + df["gross_short_notional_usd"].abs() return (gross / 100_000.0).reindex(dates).fillna(0.0) if key == "johansen_etf": return pd.Series(1.0, index=dates) if key == "ma_slope_topn" and sleeve.get("rebalances"): reb = pd.read_csv(sleeve["rebalances"], parse_dates=["effective_date"]) reb["effective_date"] = pd.to_datetime(reb["effective_date"]).dt.normalize() weights = ( reb.loc[reb["weight"] > 0] .groupby("effective_date")["weight"] .sum() .sort_index() ) if weights.empty: return pd.Series(0.0, index=dates) wdf = weights.reset_index().rename(columns={"effective_date": "date", "weight": "w"}) frame = pd.DataFrame({"date": dates}).sort_values("date") merged = pd.merge_asof(frame, wdf.sort_values("date"), on="date") return pd.Series(merged["w"].fillna(0.0).to_numpy(), index=dates) if key == "ma_slope_inverse" and sleeve.get("positions"): pos = pd.read_csv(sleeve["positions"], parse_dates=["date"]) pos["date"] = pd.to_datetime(pos["date"]).dt.normalize() s = pos.groupby("date")["weight"].sum().clip(0.0, 1.0) return s.reindex(dates).fillna(0.0) if key == "equity_dip": df = _load_daily(Path(sleeve["daily"]), start, end) active = (df.set_index("date")["daily_ret"].abs() > 1e-6).astype(float) # Episodic sleeve: when active, ~half the book on average (typical 4–6 names of 10). return (active * 0.5).reindex(dates).fillna(0.0) return pd.Series(1.0, index=dates) def compute_invested_capital( portfolio_df: pd.DataFrame, sleeves: list[dict], start: pd.Timestamp, end: pd.Timestamp, ) -> dict: """ Daily / monthly / yearly average dollars deployed across sleeves. *allocated* = sum of fund sleeve notionals (quarterly budgets). *invested* = notionals × per-sleeve market exposure fraction. *margin* = combined margin estimate from combine (when present). """ df = portfolio_df.copy() dates = pd.DatetimeIndex(df["date"]) notional_cols = [c for c in df.columns if c.startswith("notional_") and c.endswith("_usd")] df["allocated_usd"] = df[notional_cols].sum(axis=1).astype(float) invested = np.zeros(len(df), dtype=np.float64) for sleeve in sleeves: key = sleeve["key"] ncol = f"notional_{key}_usd" if ncol not in df.columns: continue frac = _sleeve_invested_fraction_series(sleeve, dates, start, end) invested += df[ncol].astype(float).to_numpy() * frac.reindex(dates).fillna(0.0).to_numpy() df["invested_usd"] = invested if "margin_combined_usd" in df.columns: df["margin_usd"] = df["margin_combined_usd"].astype(float) else: df["margin_usd"] = np.nan df["ym"] = df["date"].dt.to_period("M") monthly_agg: dict = { "avg_invested_usd": ("invested_usd", "mean"), "avg_allocated_usd": ("allocated_usd", "mean"), "avg_nav_usd": ("equity_mtm_usd", "mean"), "days": ("date", "count"), } if df["margin_usd"].notna().any(): monthly_agg["avg_margin_usd"] = ("margin_usd", "mean") monthly = df.groupby("ym", sort=True).agg(**monthly_agg).reset_index() monthly["ym"] = monthly["ym"].astype(str) yearly_agg: dict = { "avg_invested_usd": ("invested_usd", "mean"), "avg_allocated_usd": ("allocated_usd", "mean"), "avg_nav_usd": ("equity_mtm_usd", "mean"), } if df["margin_usd"].notna().any(): yearly_agg["avg_margin_usd"] = ("margin_usd", "mean") yearly_inv = df.groupby(df["date"].dt.year, sort=True).agg(**yearly_agg).reset_index() yearly_inv = yearly_inv.rename(columns={"date": "year"}) return { "monthly": monthly.to_dict("records"), "yearly_invested": yearly_inv.to_dict("records"), "overall_avg_invested_usd": float(df["invested_usd"].mean()), "overall_avg_allocated_usd": float(df["allocated_usd"].mean()), "overall_avg_margin_usd": float(df["margin_usd"].mean()) if df["margin_usd"].notna().any() else None, "peak_invested_usd": float(df["invested_usd"].max()), "peak_allocated_usd": float(df["allocated_usd"].max()), } def _fmt(x: float | None, pct: bool = True, digits: int = 1) -> str: if x is None or (isinstance(x, float) and not np.isfinite(x)): return "—" if pct: return f"{x:+.{digits}f}%" if digits else f"{x:+.0f}%" return f"{x:.{digits}f}" def _html_table(headers: list[str], rows: list[list[str]]) -> str: th = "".join(f"{h}" for h in headers) body = "" for row in rows: body += "" + "".join(f"{c}" for c in row) + "" return f"{th}{body}
" # Static execution rules (partner-facing playbook; snapshots appended at render time). EXECUTION_RULES: dict[str, dict[str, str]] = { "tactical_aw": { "instruments": "SPY, TLT, IEF, GLD, DBC (ETFs only)", "cadence": "Daily check; trade when a gate flips (close vs prior day).", "entry_exit": ( "Baseline targets: SPY 30%, TLT 40%, IEF 15%, GLD 7.5%, DBC 7.5%. " "Each ETF is active only if close > SMA(200) and " "12-minus-1-month momentum > 0; otherwise that sleeve goes to cash. " "Redistribute active weights proportionally to baseline (binary mode)." ), "sizing": "Fund slice = {weight:.0%} × fund_scale × quarter-open NAV. " "Inside sleeve, deploy invested % across active ETFs per allocation CSV.", "orders": "Market-on-close (MOC) or next-open market orders to match target weights. " "Use tactical_aw_standard_allocations.csv as the model book.", "runner": "run_tactical_all_weather_standard.py", }, "equity_dip": { "instruments": "S&P 100 single names (liquid US large caps)", "cadence": "Scan after each close; place limits for next session; manage exits daily.", "entry_exit": ( "Signal (day T): stock down ≥3% vs prior close, close > SMA(200), " "ATR(5)/close > 3%, rank candidates by ATR/close (highest vol first). " "Entry (day T+1): GTC/limit buy at signal_close − 0.9×ATR(5); cancel if unfilled. " "Exit (first hit): hold >10 sessions, close > prior session high, " "or close ≥ fill + 0.5×ATR(5). Max 10 concurrent positions; equal weight among held names." ), "sizing": "Fund slice = {weight:.0%} × fund_scale × quarter-open NAV. " "Per-name notional ≈ sleeve capital / number of open dips (≤10).", "orders": "Limit entry at computed price; exit with market or limit at profit/stop rules. " "No overlap on same name until flat.", "runner": "run_sp500_dip_standard.py --universe sp100 --execution-style cracking_markets", }, "vol_edge": { "instruments": "Short vol: SVXY (pre-2018) stitched to SVIX; Long vol: VIXY", "cadence": "Daily after VIX/VIX3M/SPY close; rebalance when target weight drift > 2%.", "entry_exit": ( "Compute eVRP = VIX − 30-day expected realized vol (SPY). " "Term structure: contango if VIX < VIX3M. " "evrp_boc rules: eVRP>0 & contango → 20% short vol; " "eVRP≤0 & contango → 10% short vol; eVRP≤0 & backwardation → 20% long vol (VIXY); " "else cash. Short leg uses SVIX (or SVXY pre-stitch)." ), "sizing": "Fund slice = {weight:.0%} × fund_scale × quarter-open NAV. " "Apply w_short / w_long as % of sleeve capital (not whole fund).", "orders": "ETN shares to match target weights; remainder cash. ~15% margin proxy on gross ETN exposure.", "runner": "run_volatility_edge_etn.py --variant evrp_boc", }, "ma_slope_topn": { "instruments": "S&P 500 cross-section (top 10 names each month)", "cadence": "Rebalance month-end signal → first session of new month; check ATR stops daily.", "entry_exit": ( "Rank universe: dual EMA slope score (fast EMA slope × slow EMA slope) with both slopes > 0 " "and price > EMA(10). Hold top 10 equal weight. " "Intra-month stop: 2×ATR(14) chandelier trail — exit name to cash until next rebalance." ), "sizing": "Fund slice = {weight:.0%} × fund_scale × quarter-open NAV. " "Each of 10 names ≈ 10% of sleeve capital when fully invested.", "orders": "Month-end: sell dropped names, buy new entries at open/MOC. " "Daily: exit stopped names; do not replace until next month.", "runner": "run_ma_slope_sp500_topn_standard.py (default: top10, dual_product, monthly, 2×ATR)", }, "johansen_etf": { "instruments": ( "Six ETF triplets (equal-weight combine): " "GDXJ–IAU–SIL; GLD–UNG–USO; XLB–XLI–XLP; COP–USO–XOP; DBC–PDBC–USO; EWA–EWC–IGE" ), "cadence": "Daily PnL from causal Johansen hedge ratios; refit eigenvector every ~63 sessions.", "entry_exit": ( "Per triplet: Johansen cointegration eigenvector → dollar-neutral spread; " "z-score of spread vs half-life lookback → scale long/short legs. " "No discrete entries — continuous weights on the three legs. " "Portfolio = equal weight across the six triplet sleeves." ), "sizing": "Fund slice = {weight:.0%} × fund_scale × quarter-open NAV. " "Each triplet gets ~⅙ of sleeve capital; legs sized by model weights (can be long and short).", "orders": "Requires triplet leg weights from runner output; rebalance when refit changes hedge ratios materially. " "Use margin-aware broker for short legs.", "runner": "run_johansen_triplet_etf_portfolio.py --fidelity causal", }, "tsmom": { "instruments": "SPY, EFA, EEM, TLT, IEF, GLD, DBC, UUP", "cadence": "Signal at month-end; execute next session (no lookahead).", "entry_exit": ( "Blended 3/6/12-month time-series momentum (skip recent week/month to avoid reversal). " "Sign = long if positive, short if negative. " "Vol-normalize each asset to ~15% annual vol; portfolio targets ~12% vol. " "Flip direction when monthly signal changes." ), "sizing": "Fund slice = {weight:.0%} × fund_scale × quarter-open NAV. " "Per-asset notional from vol-scaled weights (long and short).", "orders": "Monthly rebalance to model weights; shorts require margin/locate. " "Run script for exact day-of weights.", "runner": "run_tsmom_managed_futures.py", }, "ma_slope_inverse": { "instruments": "SH (−1× SPY inverse ETF)", "cadence": "Daily after SPY close; hold until SPY regime clears.", "entry_exit": ( "Enter long SH when SPY dual-EMA slope is bearish OR SPY < SMA(200), " "with inverse-ETF momentum confirmation (default). " "Exit when SPY regime turns off — do not exit on SH bounce alone." ), "sizing": "Fund slice = {weight:.0%} × fund_scale × quarter-open NAV. " "When active, ~100% of sleeve in SH; else cash.", "orders": "Buy SH when regime on; sell to cash when regime off. Single position.", "runner": "run_ma_slope_inverse_spy_standard.py --spy-regime bear_dual_or_sma200", }, } def _latest_row(path: Path, as_of: pd.Timestamp) -> pd.Series | None: if not path.is_file(): return None df = pd.read_csv(path, parse_dates=["date"]) if df.empty: return None df["date"] = pd.to_datetime(df["date"]).dt.normalize() sub = df[df["date"] <= as_of] if sub.empty: return None return sub.sort_values("date").iloc[-1] def _load_execution_snapshots(sleeves: list[dict], as_of: pd.Timestamp) -> dict[str, str]: """One-line model portfolio snapshot per sleeve (from log CSVs).""" out: dict[str, str] = {} for s in sleeves: key = s["key"] if key == "tactical_aw" and "extra" in s: path = Path(s["extra"]) if path.is_file(): df = pd.read_csv(path, parse_dates=["date"]) df["date"] = pd.to_datetime(df["date"]).dt.normalize() sub = df[df["date"] <= as_of] if not sub.empty: dlast = sub["date"].max() day = sub[sub["date"] == dlast] active = day[day["weight"] > 0] parts = [f"{r.ticker} {r.weight * 100:.1f}%" for r in active.itertuples()] inv = float(day["total_invested_weight"].iloc[0]) * 100 cash = float(day["cash_weight"].iloc[0]) * 100 out[key] = ( f"Model {dlast.date()}: {inv:.0f}% invested — " f"{', '.join(parts) if parts else 'all cash'}; {cash:.0f}% cash." ) elif key == "vol_edge": row = _latest_row(s["daily"], as_of) if row is not None: ts = float(row.get("target_w_short", 0) or 0) tl = float(row.get("target_w_long", 0) or 0) legs = [] if ts > 0.001: legs.append(f"short vol {ts * 100:.0f}% (SVIX/SVXY)") if tl > 0.001: legs.append(f"long vol {tl * 100:.0f}% (VIXY)") leg_txt = ", ".join(legs) if legs else "flat (cash)" out[key] = ( f"Model {row['date'].date()}: {leg_txt}. " f"VIX={float(row['vix']):.1f}, VIX3M={float(row['vix3m']):.1f}, " f"eVRP={float(row['evrp']):+.2f}." ) elif key == "ma_slope_topn" and "rebalances" in s: path = Path(s["rebalances"]) if path.is_file(): df = pd.read_csv(path, parse_dates=["effective_date", "signal_date"]) df["effective_date"] = pd.to_datetime(df["effective_date"]).dt.normalize() sub = df[(df["effective_date"] <= as_of) & (df["weight"] > 0)] if not sub.empty: eff = sub["effective_date"].max() names = sub[sub["effective_date"] == eff].sort_values("rank")["ticker"].tolist() out[key] = ( f"Model month starting {eff.date()}: " f"equal weight {len(names)} names — {', '.join(names)}." ) elif key == "ma_slope_inverse" and "positions" in s: path = Path(s["positions"]) daily_path = Path(s["daily"]) pos_date = None ticker = "SH" if path.is_file(): pdf = pd.read_csv(path, parse_dates=["date"]) pdf["date"] = pd.to_datetime(pdf["date"]).dt.normalize() psub = pdf[pdf["date"] <= as_of] if not psub.empty: pos_date = psub["date"].max() ticker = str(psub.iloc[-1].get("ticker", "SH")) daily_max = None if daily_path.is_file(): ddf = pd.read_csv(daily_path, parse_dates=["date"]) daily_max = pd.to_datetime(ddf["date"]).max() if pos_date is not None: held = float(psub.iloc[-1].get("weight", 1) or 0) > 0 stale = daily_max is not None and pos_date < daily_max - pd.Timedelta(days=5) note = " (refresh positions CSV)" if stale else "" state = f"long {ticker}" if held else "cash" out[key] = f"Model {pos_date.date()}: {state}.{note}" elif key == "equity_dip": out[key] = ( "Model: event-driven — check prior session for −3% dip signals; " "no standing allocation. Export open orders from dip runner if automating." ) elif key in ("tsmom", "johansen_etf"): row = _latest_row(s["daily"], as_of) if row is not None: if key == "tsmom" and "gross_exposure" in row.index: ge = float(row["gross_exposure"]) * 100 out[key] = ( f"Model {row['date'].date()}: gross exposure {ge:.0f}% — " f"run {EXECUTION_RULES[key]['runner']} for leg weights." ) else: out[key] = ( f"Model {row['date'].date()}: " f"continuous triplet weights — run {EXECUTION_RULES[key]['runner']} for legs." ) return out def _fmt_usd(x: float) -> str: return f"${x:,.0f}" def _fmt_usd_k(x: float | None) -> str: if x is None or (isinstance(x, float) and not np.isfinite(x)): return "—" if abs(x) >= 1_000_000: return f"${x / 1_000_000:.2f}M" if abs(x) >= 10_000: return f"${x / 1_000:.0f}k" return f"${x:,.0f}" def _merge_yearly_returns(portfolio: dict, invested: dict) -> list[dict]: inv_by_year = {int(r["year"]): r for r in invested.get("yearly_invested", [])} merged = [] for y in portfolio.get("yearly", []): yr = int(y["year"]) row = dict(y) inv = inv_by_year.get(yr, {}) row["avg_invested_usd"] = inv.get("avg_invested_usd") row["avg_allocated_usd"] = inv.get("avg_allocated_usd") row["avg_margin_usd"] = inv.get("avg_margin_usd") row["avg_nav_usd"] = inv.get("avg_nav_usd") merged.append(row) return merged def _monthly_invested_table_rows(monthly: list[dict], y0: int, y1: int) -> list[list[str]]: rows = [] for m in monthly: ym = str(m["ym"]) year = int(ym[:4]) if year < y0 or year > y1: continue rows.append( [ ym, _fmt_usd_k(float(m["avg_invested_usd"])), _fmt_usd_k(float(m["avg_allocated_usd"])), _fmt_usd_k(float(m.get("avg_nav_usd", np.nan))), ] ) return rows def _execution_playbook_html( key: str, weight: float, fund_scale: float, nav_usd: float, snapshot: str, ) -> str: rules = EXECUTION_RULES.get(key, {}) if not rules: return "" sleeve_cap = weight * fund_scale * nav_usd sizing = rules.get("sizing", "").format(weight=weight) items = [ f"
  • Instruments: {rules.get('instruments', '—')}
    • ", f"
  • When: {rules.get('cadence', '—')}
    • ", f"
  • Rules: {rules.get('entry_exit', '—')}
    • ", f"
  • Sizing: {sizing} " f"Example at latest NAV {_fmt_usd(nav_usd)} × {fund_scale:.1f}× → " f"{_fmt_usd(sleeve_cap)} sleeve budget this quarter.
    • ", f"
  • Orders: {rules.get('orders', '—')}
    • ", f"
  • Refresh: {rules.get('runner', '—')}
    • ", ] snap = f'

    {snapshot}

    ' if snapshot else "" return f'

    Execution playbook

    {snap}
    ' def _execution_summary_table(sleeves: list[SleeveStats], fund_scale: float) -> str: rows = [] for s in sleeves: r = EXECUTION_RULES.get(s.key, {}) rows.append( [ f"{s.name} ({s.weight * 100:.0f}%)", r.get("cadence", "—")[:60] + ("…" if len(r.get("cadence", "")) > 60 else ""), r.get("instruments", "—")[:55] + ("…" if len(r.get("instruments", "")) > 55 else ""), r.get("runner", "—"), ] ) return _html_table(["Sleeve", "Cadence", "Instruments", "Runner"], rows) def _slide(title: str, body: str, *, subtitle: str = "", slide_class: str = "") -> str: sub = f'

    {subtitle}

    ' if subtitle else "" return f"""

    {title}

    {sub}
    {body}
    """ def render_html( *, start: str, end: str, portfolio: dict, sleeves: list[SleeveStats], invested: dict, out_path: Path, fund_scale: float = 1.5, ) -> None: spy = portfolio["spy"] p_rows = [ ["Total return", _fmt(portfolio["total_return_pct"]), _fmt(spy["total_return_pct"])], ["CAGR", _fmt(portfolio["cagr_pct"]), _fmt(spy["cagr_pct"])], ["Sharpe (daily)", _fmt(portfolio["sharpe"], pct=False, digits=2), _fmt(spy["sharpe"], pct=False, digits=2)], ["Max drawdown", _fmt(portfolio["max_dd_pct"]), _fmt(spy["max_dd_pct"])], ["Daily win rate", _fmt(portfolio["daily_win_rate_pct"]), _fmt(spy["daily_win_rate_pct"])], ["Avg up day", _fmt(portfolio["avg_winner_day_pct"]), _fmt(spy["avg_winner_day_pct"])], ["Avg down day", _fmt(portfolio["avg_loser_day_pct"]), _fmt(spy["avg_loser_day_pct"])], ["β vs SPY", _fmt(portfolio["beta_spy"], pct=False, digits=2), "1.00"], ["ρ vs SPY", _fmt(portfolio["corr_spy"], pct=False, digits=2), "—"], ["End equity ($100k start)", f"${portfolio['end_equity']:,.0f}", "—"], ] summary_rows = [] for s in sleeves: summary_rows.append( [ f"{s.name} ({s.weight*100:.0f}%)", _fmt(s.total_return_pct), _fmt(s.cagr_pct), _fmt(s.sharpe, pct=False, digits=2), _fmt(s.max_dd_pct), _fmt(s.daily_win_rate_pct), _fmt(s.time_in_position_pct), _fmt(s.corr_spy, pct=False, digits=2), ] ) summary_rows.append( [ "Combined fund", _fmt(portfolio["total_return_pct"]), _fmt(portfolio["cagr_pct"]), _fmt(portfolio["sharpe"], pct=False, digits=2), _fmt(portfolio["max_dd_pct"]), _fmt(portfolio["daily_win_rate_pct"]), "100%", _fmt(portfolio["corr_spy"], pct=False, digits=2), ] ) summary_rows.append( [ "SPY buy & hold", _fmt(spy["total_return_pct"]), _fmt(spy["cagr_pct"]), _fmt(spy["sharpe"], pct=False, digits=2), _fmt(spy["max_dd_pct"]), _fmt(spy["daily_win_rate_pct"]), "100%", "1.00", ] ) yearly_merged = _merge_yearly_returns(portfolio, invested) yearly_rows = [ [ str(y["year"]), _fmt(y["return_pct"]), _fmt(y["max_dd_pct"]), _fmt_usd_k(y.get("avg_nav_usd")), _fmt_usd_k(y.get("avg_invested_usd")), _fmt_usd_k(y.get("avg_allocated_usd")), ] for y in yearly_merged ] mid = (len(yearly_rows) + 1) // 2 yearly_hdr = ["Year", "Return", "Max DD", "Avg NAV", "Avg invested", "Avg allocated"] yearly_html_1 = _html_table(yearly_hdr, yearly_rows[:mid]) yearly_html_2 = _html_table(yearly_hdr, yearly_rows[mid:]) monthly = invested.get("monthly", []) m_rows_1 = _monthly_invested_table_rows(monthly, 2011, 2017) m_rows_2 = _monthly_invested_table_rows(monthly, 2018, 2022) m_rows_3 = _monthly_invested_table_rows(monthly, 2023, 2030) monthly_hdr = ["Month", "Avg invested", "Avg allocated", "Avg NAV"] m1_mid = (len(m_rows_1) + 1) // 2 m2_mid = (len(m_rows_2) + 1) // 2 m3_mid = (len(m_rows_3) + 1) // 2 inv_summary = ( f"Full-window daily mean: invested {_fmt_usd(invested['overall_avg_invested_usd'])}, " f"allocated {_fmt_usd(invested['overall_avg_allocated_usd'])}, " f"peak invested {_fmt_usd(invested['peak_invested_usd'])}." ) if invested.get("overall_avg_margin_usd") is not None: inv_summary += f" Avg margin {_fmt_usd(invested['overall_avg_margin_usd'])}." weight_spans = "".join( f'{s.name.split()[0]} {s.weight * 100:.0f}%' for s in sleeves ) n_sleeves = len(sleeves) as_of = pd.Timestamp(end) snapshots = _load_execution_snapshots(SLEEVES, as_of) nav_usd = float(portfolio.get("end_equity", 100_000.0)) sizing_rows = [] for s in sleeves: cap = s.weight * fund_scale * nav_usd sizing_rows.append( [ s.name.split()[0] if len(s.name.split()[0]) <= 12 else s.name[:14], f"{s.weight * 100:.1f}%", f"{s.weight * fund_scale * 100:.1f}%", _fmt_usd(cap), ] ) sleeve_slides = "" for s in sleeves: ep_html = f"""
    Trade win rate{_fmt(s.trade_win_rate_pct) if s.trade_win_rate_pct is not None else "—"}
    Avg win episode{_fmt(s.avg_winner_trade_pct) if s.avg_winner_trade_pct is not None else "—"}
    Avg loss episode{_fmt(s.avg_loser_trade_pct) if s.avg_loser_trade_pct is not None else "—"}
    Avg hold (days){f"{s.avg_hold_days:.1f}" if s.avg_hold_days is not None else "—"}
    Episodes{s.n_trades if s.n_trades is not None else "—"}
    """ exec_html = _execution_playbook_html( s.key, s.weight, fund_scale, nav_usd, snapshots.get(s.key, "") ) body = f"""

    {s.blurb}

    Standalone $100k · fund weight {s.weight*100:.0f}%

    {ep_html}
    {_html_table( ["Metric", "Sleeve", "SPY"], [ ["Total return", _fmt(s.total_return_pct), _fmt(s.spy_total_return_pct)], ["CAGR", _fmt(s.cagr_pct), _fmt(s.spy_cagr_pct)], ["Sharpe", _fmt(s.sharpe, pct=False, digits=2), _fmt(s.spy_sharpe, pct=False, digits=2)], ["Max DD", _fmt(s.max_dd_pct), _fmt(s.spy_max_dd_pct)], ["Daily win rate", _fmt(s.daily_win_rate_pct), _fmt(spy["daily_win_rate_pct"])], ["Avg up day", _fmt(s.avg_winner_day_pct), _fmt(spy["avg_winner_day_pct"])], ["Avg down day", _fmt(s.avg_loser_day_pct), _fmt(spy["avg_loser_day_pct"])], ["Time invested", _fmt(s.time_in_position_pct), "—"], ["β vs SPY", _fmt(s.beta_spy, pct=False, digits=2), "1.00"], ["ρ vs SPY", _fmt(s.corr_spy, pct=False, digits=2), "—"], ], )}
    {exec_html}""" sleeve_slides += _slide(s.name, body, slide_class="slide-dense slide-exec") slides = [ f"""

    Research backtest · stock / ETF book

    Stock-Only Multi-Strategy Fund

    Tactical AW · CrackingMarkets dip · Vol Edge · MA slope top-N + SH hedge · TSMOM · Johansen ETF

    {start} → {end} $100,000 start fund_scale {fund_scale:.1f}× quarterly NAV sizing · no sector momentum (high SPY ρ)
    {_fmt(portfolio['total_return_pct'])}Total return
    {_fmt(portfolio['cagr_pct'])}CAGR
    {_fmt(portfolio['sharpe'], pct=False, digits=2)}Sharpe
    {_fmt(portfolio['max_dd_pct'])}Max DD
    {_fmt(spy['cagr_pct'])}SPY CAGR
    """, _slide( "How the portfolio works", """
    1. {n_sleeves} sleeves backtested standalone at $100k (daily CSV per strategy).
    2. Fund combine blends sleeve returns with fixed weights → 100%.
    3. Quarterly sizing: notional = weight × fund_scale × prior NAV each quarter open.
    4. No options — ETFs/equities only (theta, VRP, VXX, macro options excluded).
    5. Sector momentum omitted — standalone ρ(SPY)≈0.89; redundant with dip + MA top-N.
    {weight_spans}
    """, ), _slide( "Live execution — fund operations", f"""

    Quarterly checklist (first trading day of Jan / Apr / Jul / Oct)

    1. Record prior-day fund NAV (combined equity).
    2. For each sleeve: budget = weight × {fund_scale:.1f}× × NAV.
    3. Run sleeve refresh scripts (see next slide) → target weights / signals.
    4. Rebalance each sleeve to its budget; sum margin ≈ peak ~$270k at 1.5× on $100k start (grows with NAV).
    5. Daily: vol-edge drift check; dip limits & exits; MA slope ATR stops; tactical gates.

    Data: Yahoo Finance daily bars (same as backtest). MOC or next-open fills assumed.

    Sleeve budgets at latest NAV {_fmt_usd(nav_usd)} · fund_scale {fund_scale:.1f}×

    {_html_table(["Sleeve", "Weight", "Gross wt", "≈ Budget"], sizing_rows)}

    Gross wt = weight × fund_scale. Budget uses latest NAV as illustration; live uses quarter-open NAV.

    """, slide_class="slide-dense", ), _slide( "Execution summary (all sleeves)", _execution_summary_table(sleeves, fund_scale) + """

    Each sleeve slide includes full entry/exit rules, order type, and latest model snapshot from log CSVs.

    """, slide_class="slide-dense", ), _slide( "Combined fund vs SPY", _html_table(["Metric", "Fund", "SPY"], p_rows), subtitle="Same calendar window · fund uses quarterly-sized NAV compounding", ), _slide( "Sleeve summary (standalone)", _html_table( ["Sleeve", "Total", "CAGR", "Sharpe", "Max DD", "Win%", "Invested", "ρ"], summary_rows, ) + '

    Standalone ≠ weighted sum of fund (quarterly sizing + compounding).

    ', slide_class="slide-dense", ), _slide( "Fund returns by year", f'
    {yearly_html_1}{yearly_html_2}
    ' + f'

    Chained NAV return = PnL / start-of-year equity. ' f'Avg NAV = mean daily fund equity. ' f'Avg invested = mean daily gross exposure (sleeve notional × in-market fraction). ' f'Avg allocated = mean sum of quarterly sleeve budgets. {inv_summary}

    ', ), _slide( "Average capital deployed — monthly (2011–2017)", f'
    {_html_table(monthly_hdr, m_rows_1[:m1_mid])}
    ' f'
    {_html_table(monthly_hdr, m_rows_1[m1_mid:])}
    ' + '

    Calendar-month mean of daily invested / allocated / NAV.

    ', slide_class="slide-dense slide-monthly", ), _slide( "Average capital deployed — monthly (2018–2022)", f'
    {_html_table(monthly_hdr, m_rows_2[:m2_mid])}
    ' f'
    {_html_table(monthly_hdr, m_rows_2[m2_mid:])}
    ', slide_class="slide-dense slide-monthly", ), _slide( "Average capital deployed — monthly (2023–2026)", f'
    {_html_table(monthly_hdr, m_rows_3[:m3_mid])}
    ' f'
    {_html_table(monthly_hdr, m_rows_3[m3_mid:])}
    ', slide_class="slide-dense slide-monthly", ), ] slides.append(sleeve_slides) slides.append( _slide( "Presenter notes", """ """, ) ) deck_body = "\n".join(slides) html = f""" Stock-Only MA Slope Fund — 16:9 Deck
    {deck_body}
    """ out_path.write_text(html, encoding="utf-8") def main() -> None: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--start", default="2011-06-20") ap.add_argument("--end", default="2026-06-18") ap.add_argument( "--portfolio-daily", type=Path, default=DEFAULT_PORTFOLIO_DAILY, ) ap.add_argument("--fund-scale", type=float, default=1.5, help="Shown on title slide (must match combine run)") ap.add_argument("--out", type=Path, default=_REPO / "stock_only_ma_slope_presentation.html") args = ap.parse_args() start = pd.Timestamp(args.start) end = pd.Timestamp(args.end) spy_df = _compute_daily_backtest_features(DataLoader().fetch_daily("SPY", period="max")) spy_df.index = pd.to_datetime(spy_df.index).tz_localize(None) spy_r = spy_df["ret"].astype(float) spy_r.index = spy_r.index.normalize() sleeves_stats = [analyze_sleeve(s, spy_r, start, end) for s in SLEEVES] portfolio = analyze_portfolio(args.portfolio_daily.expanduser().resolve(), spy_r, start, end) daily_df = portfolio.pop("_daily_df") invested = compute_invested_capital(daily_df, SLEEVES, start, end) yearly_merged = _merge_yearly_returns(portfolio, invested) out = args.out.expanduser().resolve() render_html( start=str(start.date()), end=str(end.date()), portfolio=portfolio, sleeves=sleeves_stats, invested=invested, out_path=out, fund_scale=float(args.fund_scale), ) json_path = out.with_suffix(".json") payload = { "window": {"start": str(start.date()), "end": str(end.date())}, "portfolio": {k: v for k, v in portfolio.items() if k != "yearly"}, "portfolio_yearly": yearly_merged, "invested_capital": { k: v for k, v in invested.items() if k != "monthly" }, "invested_monthly": invested.get("monthly", []), "sleeves": [s.__dict__ for s in sleeves_stats], } json_path.write_text(json.dumps(payload, indent=2) + "\n") print(f"Wrote {out}") print(f"Wrote {json_path}") print(f"\nFund: return {portfolio['total_return_pct']:+.1f}% CAGR {portfolio['cagr_pct']:+.1f}% " f"Sharpe {portfolio['sharpe']:.2f} MaxDD {portfolio['max_dd_pct']:.1f}%") if __name__ == "__main__": main()