#!/usr/bin/env python3 """ Combine the **VRP regime** book (+ IV overlays) with the **VXX contango** sleeve (90 % bear call credit spread + 10 % long OTM call) and report portfolio-level metrics with vs. without the VXX sleeve. **Percentage allocation “backtest”** (no Theta re-run): load realized trades from JSONLs and merge daily PnL with ``--allocation-json`` pointing at ``optimize_overlay_risk_fracs.py`` output (``portfolio_vrp_plus_vxx_cli_snippet`` fractions × ``--total-capital``). That is the same merge model the optimizer uses; sleeves are **fractions of the account**, then converted to USD risk budgets internally for scaling vs each JSONL’s ref risk. **One account, one starting equity** (``--total-capital``, default 100k): every equity curve starts at that number. **IV overlays and VXX** are scaled by **broker risk budget** (max loss for credit structures, debit paid for long premium): ``sleeve_pnl × (your_risk_usd / ref_risk_usd)``. Reference risk per sleeve defaults to the **median** ``broker_risk_usd`` in each JSONL (or inferred for legacy rows). Override with ``--put-risk-ref`` / ``--overlay-ref`` / ``--vxx-ref``. VRP still uses ``capital_vrp / vrp_ref``. Per-trade VRP risk lives inside ``VRPBacktester``. Active sleeves: VRP main, OTM put, straddle, risk reversal, VXX bear call, VXX long call. Calendar spread is **excluded** — drop-one-out analysis (Apr 2026) showed it lost -$10.5K with Sharpe -5.61 and removing it improved portfolio Sharpe by +0.069. Do not re-add without fresh evidence of profitability. Example:: python RenTech/strategy_stack/portfolio_vrp_plus_vxx.py \\ --total-capital 100000 --capital-vxx-pct 0.024 --vxx-bear-pct 90 --vxx-call-pct 10 # Same as --capital-put 2000 on a 100k book: python RenTech/strategy_stack/portfolio_vrp_plus_vxx.py \\ --total-capital 100000 --capital-put-pct 0.02 Or from Python / after ``vrp_backtest_theta.py``:: from RenTech.strategy_stack.portfolio_vrp_plus_vxx import execute_portfolio_merge execute_portfolio_merge( vrp_trades=Path("RenTech/data/logs/vrp_low_dd_ov2_vxxbundle_vrp_trades.jsonl"), total_portfolio_capital=100_000, ) Overlay JSONLs (defaults under ``RenTech/data/logs/``): ``stress_longvol_otm_put.jsonl``, ``stress_longvol_straddle.jsonl``, ``risk_reversal.jsonl`` from ``backtest_iv_stress_long_vol.py`` / ``backtest_iv_calendar_risk_reversal.py``; VXX: ``vxx_portfolio_bear_call.jsonl`` and ``vxx_portfolio_long_call.jsonl`` from ``optimize_vxx_portfolio.py`` / ``backtest_vxx_bear_call_contango.py`` and ``explore_vxx_decay_strategies.py`` (see those scripts for exact flags). **Optimize sleeve risk budgets:** :func:`optimize_sleeve_notionals` (or CLI ``--optimize-sleeves``) searches overlay + VXX **broker risk budgets (USD)** with an optional sum cap and a **soft** max-drawdown target, maximizing ``Calmar + sharpe_weight * Sharpe`` with VRP scale fixed to the account. **Sharpe-only allocation:** :func:`optimize_portfolio_sharpe` (or CLI ``--optimize-sharpe``) searches put/straddle/RR/VXX as **fractions of** ``--total-capital`` to maximize **Sharpe** on the merged book, with optional sum cap (``--sharpe-sum-cap-pct``) and optional **hard** max drawdown cap (``--sharpe-max-dd-pct``, default 10%% peak-to-trough; use ``0`` to disable). Mutually exclusive with ``--optimize-sleeves``. """ from __future__ import annotations import argparse, json, math, sys from dataclasses import dataclass from pathlib import Path from typing import Any import numpy as np import pandas as pd _REPO = Path(__file__).resolve().parents[2] LOGS = _REPO / "RenTech" / "data" / "logs" # Latest low-DD overlap + vxxbundle export (see ``run_vrp_low_dd_vxx_bundle.py``); override with ``--vrp-trades``. DEFAULT_VRP_TRADES_JSONL = LOGS / "vrp_low_dd_ov2_vxxbundle_vrp_trades.jsonl" if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO)) from RenTech.strategy_stack.iv_mispricing_complement import ( _load_jsonl, _pnl_series_from_trades, _require_jsonl, ) def _load_pnl(path: Path, pnl_key: str = "pnl_total") -> pd.Series: trades = _load_jsonl(path) return _pnl_series_from_trades(trades, exit_key="exit_date", pnl_key=pnl_key) def _infer_broker_risk_overlay(trade: dict) -> float: """Max loss / debit (USD) for one overlay position; legacy JSONL when ``broker_risk_usd`` absent.""" v = trade.get("broker_risk_usd") if v is not None: x = float(v) if math.isfinite(x) and x > 0: return x st = str(trade.get("structure", "")) if st in ("otm_put", "straddle"): return max(abs(float(trade.get("entry_premium", 0))), 1.0) en = float(trade.get("entry_net_premium", 0)) spy = float(trade.get("spy_entry", 400)) cm = 100.0 if en >= 0: return max(en, 1.0) return max(abs(en) + 0.20 * spy * cm, 1.0) def _infer_broker_risk_vxx(trade: dict) -> float: v = trade.get("broker_risk_usd") if v is not None: x = float(v) if math.isfinite(x) and x > 0: return x s = str(trade.get("strategy", "")) if s == "bear_call" and trade.get("max_loss") is not None: return max(float(trade["max_loss"]), 1.0) ec = float(trade.get("entry_credit_or_debit", 0)) if s == "long_call" or ec < 0: return max(abs(ec) if ec != 0 else 150.0, 1.0) if s == "bear_call": return 320.0 return max(abs(ec), 200.0) def _median_risk_usd(path: Path, infer_fn) -> float: if not path.is_file(): return 1.0 trades = _load_jsonl(path) if not trades: return 1.0 vals = [infer_fn(t) for t in trades] vals = [v for v in vals if math.isfinite(v) and v > 0] return float(np.median(vals)) if vals else 1.0 def _overlay_sleeve_risk_reference_usd(path: Path, infer_fn) -> float: """ Reference USD for scaling an overlay/VXX sleeve to ``capital_*``. If **every** row has ``broker_risk_usd > 0``, use **sum** of those values (total capital deployed in the log — correct when contracts vary per trade). Otherwise fall back to the **median** of inferred per-trade risk (legacy JSONL). """ if not path.is_file(): return 1.0 trades = _load_jsonl(path) if not trades: return 1.0 documented = all( t.get("broker_risk_usd") is not None and math.isfinite(float(t["broker_risk_usd"])) and float(t["broker_risk_usd"]) > 0 for t in trades ) vals: list[float] = [] for t in trades: v = t.get("broker_risk_usd") if v is not None and math.isfinite(float(v)) and float(v) > 0: vals.append(float(v)) else: x = float(infer_fn(t)) if math.isfinite(x) and x > 0: vals.append(x) if not vals: return 1.0 if documented: return max(float(sum(vals)), 1.0) return max(float(np.median(vals)), 1.0) def _equity_from_pnl(pnl: pd.Series, capital: float) -> pd.Series: return capital + pnl.cumsum() @dataclass(frozen=True) class PortfolioPnLBasis: """VRP uses ``B_v = pnl/vrp_ref``. Overlays/VXX use **raw** daily $ PnL scaled by ``risk_budget/risk_ref``.""" index: pd.DatetimeIndex eq0: float vrp_ref: float put_risk_ref: float straddle_risk_ref: float rr_risk_ref: float vxx_bear_risk_ref: float vxx_call_risk_ref: float B_v: pd.Series B_p: pd.Series B_s: pd.Series B_r: pd.Series B_vb: pd.Series B_vc: pd.Series def load_portfolio_pnl_basis( *, vrp_trades: Path, vrp_ref: float = 100_000.0, put_risk_ref: float | None = None, straddle_risk_ref: float | None = None, rr_risk_ref: float | None = None, overlay_ref: float | None = None, vxx_bear_risk_ref: float | None = None, vxx_call_risk_ref: float | None = None, vxx_ref: float | None = None, put_trades: Path | None = None, straddle_trades: Path | None = None, risk_reversal_trades: Path | None = None, vxx_bear_trades: Path | None = None, vxx_call_trades: Path | None = None, total_portfolio_capital: float = 100_000.0, ) -> PortfolioPnLBasis: """ Load JSONLs once; overlay/VXX daily PnL is **raw dollars** from the reference backtest. Risk refs default to **sum** of ``broker_risk_usd`` per file when every row documents it (dynamic contracts / 1:1 sizing); otherwise **median** of inferred per-trade risk (legacy). ``overlay_ref`` (if set) overrides put, straddle, and RR refs. ``vxx_ref`` overrides both VXX legs. """ put_trades = put_trades or (LOGS / "stress_longvol_otm_put.jsonl") straddle_trades = straddle_trades or (LOGS / "stress_longvol_straddle.jsonl") risk_reversal_trades = risk_reversal_trades or (LOGS / "risk_reversal.jsonl") vxx_bear_trades = vxx_bear_trades or (LOGS / "vxx_portfolio_bear_call.jsonl") vxx_call_trades = vxx_call_trades or (LOGS / "vxx_portfolio_long_call.jsonl") vrp_path = vrp_trades.expanduser().resolve() if not vrp_path.is_file(): raise FileNotFoundError(f"VRP trades JSONL not found: {vrp_path}") m_put = _overlay_sleeve_risk_reference_usd(put_trades, _infer_broker_risk_overlay) m_str = _overlay_sleeve_risk_reference_usd(straddle_trades, _infer_broker_risk_overlay) m_rr = _overlay_sleeve_risk_reference_usd(risk_reversal_trades, _infer_broker_risk_overlay) m_vb = _overlay_sleeve_risk_reference_usd(vxx_bear_trades, _infer_broker_risk_vxx) m_vc = _overlay_sleeve_risk_reference_usd(vxx_call_trades, _infer_broker_risk_vxx) def _or_ov(explicit: float | None, med: float) -> float: if explicit is not None: return max(float(explicit), 1.0) if overlay_ref is not None: return max(float(overlay_ref), 1.0) return max(float(med), 1.0) pr = _or_ov(put_risk_ref, m_put) sr = _or_ov(straddle_risk_ref, m_str) rr = _or_ov(rr_risk_ref, m_rr) def _vxx_leg(explicit: float | None, med: float) -> float: if explicit is not None: return max(float(explicit), 1.0) if vxx_ref is not None: return max(float(vxx_ref), 1.0) return max(float(med), 1.0) vbr = _vxx_leg(vxx_bear_risk_ref, m_vb) vcr = _vxx_leg(vxx_call_risk_ref, m_vc) vrp_pnl = _load_pnl(vrp_path, pnl_key="pnl_usd") put_pnl = _load_pnl(put_trades) if put_trades.is_file() else pd.Series(dtype=float) str_pnl = _load_pnl(straddle_trades) if straddle_trades.is_file() else pd.Series(dtype=float) rr_pnl = _load_pnl(risk_reversal_trades) if risk_reversal_trades.is_file() else pd.Series(dtype=float) vxx_bear_pnl = _load_pnl(vxx_bear_trades) if vxx_bear_trades.is_file() else pd.Series(dtype=float) vxx_call_pnl = _load_pnl(vxx_call_trades) if vxx_call_trades.is_file() else pd.Series(dtype=float) all_idx = vrp_pnl.index for s in [put_pnl, str_pnl, rr_pnl, vxx_bear_pnl, vxx_call_pnl]: if not s.empty: all_idx = all_idx.union(s.index) all_idx = all_idx.sort_values() if len(all_idx) > 0: t_pad = all_idx[0] - pd.Timedelta(days=1) if t_pad not in all_idx: all_idx = all_idx.insert(0, t_pad) rf = max(float(vrp_ref), 1.0) B_v = vrp_pnl.reindex(all_idx, fill_value=0) / rf B_p = put_pnl.reindex(all_idx, fill_value=0) B_s = str_pnl.reindex(all_idx, fill_value=0) B_r = rr_pnl.reindex(all_idx, fill_value=0) B_vb = vxx_bear_pnl.reindex(all_idx, fill_value=0) B_vc = vxx_call_pnl.reindex(all_idx, fill_value=0) return PortfolioPnLBasis( index=all_idx, eq0=float(total_portfolio_capital), vrp_ref=float(vrp_ref), put_risk_ref=float(pr), straddle_risk_ref=float(sr), rr_risk_ref=float(rr), vxx_bear_risk_ref=float(vbr), vxx_call_risk_ref=float(vcr), B_v=B_v, B_p=B_p, B_s=B_s, B_r=B_r, B_vb=B_vb, B_vc=B_vc, ) def merged_full_equity( basis: PortfolioPnLBasis, *, capital_vrp: float | None, capital_put: float, capital_straddle: float, capital_risk_reversal: float, capital_vxx: float, vxx_bear_pct: float = 90.0, vxx_call_pct: float = 10.0, ) -> pd.Series: """Full book: VRP + IV overlays + VXX, starting at ``basis.eq0``.""" vrp_num = float(capital_vrp) if capital_vrp is not None else basis.eq0 cp, cs, cr = float(capital_put), float(capital_straddle), float(capital_risk_reversal) cvxx = float(capital_vxx) bear = cvxx * (vxx_bear_pct / 100.0) call = cvxx * (vxx_call_pct / 100.0) pr, sr, rr = max(basis.put_risk_ref, 1e-12), max(basis.straddle_risk_ref, 1e-12), max(basis.rr_risk_ref, 1e-12) vbr, vcr = max(basis.vxx_bear_risk_ref, 1e-12), max(basis.vxx_call_risk_ref, 1e-12) daily = ( vrp_num * basis.B_v + (cp / pr) * basis.B_p + (cs / sr) * basis.B_s + (cr / rr) * basis.B_r + (bear / vbr) * basis.B_vb + (call / vcr) * basis.B_vc ) return basis.eq0 + daily.cumsum() def max_drawdown_pct_magnitude(metrics: dict) -> float: """Positive number, e.g. 8.91 for 8.91 % peak-to-trough (``_metrics_block`` stores negative max_dd_pct).""" return float(-metrics["max_dd_pct"]) def optimize_sleeve_notionals( vrp_trades: Path, *, total_portfolio_capital: float = 100_000.0, capital_vrp: float | None = None, max_dd_limit_pct: float = 10.0, additional_notionals_cap: float | None = 100_000.0, max_single_sleeve: float = 100_000.0, sharpe_weight: float = 0.25, vxx_bear_pct: float = 90.0, vxx_call_pct: float = 10.0, vrp_ref: float = 100_000.0, overlay_ref: float | None = None, put_risk_ref: float | None = None, straddle_risk_ref: float | None = None, rr_risk_ref: float | None = None, vxx_ref: float | None = None, vxx_bear_risk_ref: float | None = None, vxx_call_risk_ref: float | None = None, put_trades: Path | None = None, straddle_trades: Path | None = None, risk_reversal_trades: Path | None = None, vxx_bear_trades: Path | None = None, vxx_call_trades: Path | None = None, maxiter: int = 80, seed: int = 0, polish: bool = False, ) -> dict[str, Any]: """ Search non‑negative **risk budgets** (USD) for **put / straddle / risk‑reversal / VXX** (VXX split by bear/call %) while **VRP scale** stays fixed at ``capital_vrp`` or ``total_portfolio_capital``. **Capital:** starting equity is always ``total_portfolio_capital`` (same as merge). Optional ``additional_notionals_cap`` requires ``capital_put + capital_straddle + capital_risk_reversal + capital_vxx <= cap`` — a combined **risk budget** (max loss / debit USD), not VRP. **Objective:** maximize ``Calmar + sharpe_weight * Sharpe`` on the **full** merged equity curve. **Feasibility:** solutions with max drawdown **above** ``max_dd_limit_pct`` (peak‑to‑trough %) are heavily penalized (soft constraint); the best reported point may be infeasible if the limit is too tight — check ``feasible`` and ``max_drawdown_pct`` in the result. Uses ``scipy.optimize.differential_evolution`` (non‑smooth OK). Install scipy if missing. """ from scipy.optimize import LinearConstraint, differential_evolution basis = load_portfolio_pnl_basis( vrp_trades=vrp_trades, vrp_ref=vrp_ref, put_risk_ref=put_risk_ref, straddle_risk_ref=straddle_risk_ref, rr_risk_ref=rr_risk_ref, overlay_ref=overlay_ref, vxx_bear_risk_ref=vxx_bear_risk_ref, vxx_call_risk_ref=vxx_call_risk_ref, vxx_ref=vxx_ref, put_trades=put_trades, straddle_trades=straddle_trades, risk_reversal_trades=risk_reversal_trades, vxx_bear_trades=vxx_bear_trades, vxx_call_trades=vxx_call_trades, total_portfolio_capital=total_portfolio_capital, ) vrp_num = float(capital_vrp) if capital_vrp is not None else basis.eq0 hi = float(max_single_sleeve) bounds = [(0.0, hi), (0.0, hi), (0.0, hi), (0.0, hi)] constraints = () if additional_notionals_cap is not None: cap = float(additional_notionals_cap) constraints = ( LinearConstraint(np.ones((1, 4)), 0.0, cap), ) rng = np.random.default_rng(seed) def objective(x: np.ndarray) -> float: cp, cs, cr, cvxx = float(x[0]), float(x[1]), float(x[2]), float(x[3]) eq = merged_full_equity( basis, capital_vrp=vrp_num, capital_put=cp, capital_straddle=cs, capital_risk_reversal=cr, capital_vxx=cvxx, vxx_bear_pct=vxx_bear_pct, vxx_call_pct=vxx_call_pct, ) m = _metrics_block(eq, "") dd_mag = max_drawdown_pct_magnitude(m) calmar, sharpe = float(m["calmar"]), float(m["sharpe"]) if dd_mag > max_dd_limit_pct + 0.05: return 1e6 + (dd_mag - max_dd_limit_pct) ** 2 * 1e4 score = calmar + sharpe_weight * sharpe return -score res = differential_evolution( objective, bounds=bounds, constraints=constraints, seed=rng, maxiter=maxiter, polish=polish, workers=1, updating="deferred", ) cp, cs, cr, cvxx = (float(res.x[0]), float(res.x[1]), float(res.x[2]), float(res.x[3])) eq_opt = merged_full_equity( basis, capital_vrp=vrp_num, capital_put=cp, capital_straddle=cs, capital_risk_reversal=cr, capital_vxx=cvxx, vxx_bear_pct=vxx_bear_pct, vxx_call_pct=vxx_call_pct, ) m_opt = _metrics_block(eq_opt, "optimized") dd_mag = max_drawdown_pct_magnitude(m_opt) feasible = dd_mag <= max_dd_limit_pct + 0.05 return { "feasible": feasible, "capital_vrp_scale": vrp_num, "capital_put": cp, "capital_straddle": cs, "capital_risk_reversal": cr, "capital_vxx": cvxx, "sum_overlay_vxx": cp + cs + cr + cvxx, "max_drawdown_pct": dd_mag, "calmar": m_opt["calmar"], "sharpe": m_opt["sharpe"], "cagr_pct": m_opt["cagr_pct"], "return_pct": m_opt["return_pct"], "end_equity": m_opt["end_equity"], "equity": eq_opt, "objective_score": m_opt["calmar"] + sharpe_weight * m_opt["sharpe"], "scipy_success": bool(res.success), "scipy_message": str(res.message), "nit": int(res.nit), } def optimize_portfolio_sharpe( vrp_trades: Path, *, total_portfolio_capital: float = 100_000.0, capital_vrp: float | None = None, sum_risk_budget_pct: float | None = 0.35, max_sleeve_pct: float = 0.25, max_dd_limit_pct: float | None = None, fixed_vxx_frac: float | None = None, fixed_straddle_frac: float | None = None, vxx_bear_pct: float = 90.0, vxx_call_pct: float = 10.0, vrp_ref: float = 100_000.0, overlay_ref: float | None = None, put_risk_ref: float | None = None, straddle_risk_ref: float | None = None, rr_risk_ref: float | None = None, vxx_ref: float | None = None, vxx_bear_risk_ref: float | None = None, vxx_call_risk_ref: float | None = None, put_trades: Path | None = None, straddle_trades: Path | None = None, risk_reversal_trades: Path | None = None, vxx_bear_trades: Path | None = None, vxx_call_trades: Path | None = None, maxiter: int = 120, seed: int = 0, polish: bool = False, ) -> dict[str, Any]: """ Search **allocation fractions** ``put, straddle, RR, VXX`` (each in ``[0, max_sleeve_pct]``) of ``total_portfolio_capital`` to **maximize Sharpe** on the full merged equity curve (VRP scale fixed). Risk budgets in USD are ``fraction × total_portfolio_capital``. Optional ``sum_risk_budget_pct`` caps ``f_put + f_straddle + f_rr + f_vxx`` (e.g. 0.35 = 35%% of account). If ``fixed_vxx_frac`` is set (e.g. ``0.01`` for 1%% of book to VXX), the search is only over put / straddle / RR; VXX is held fixed and excluded from ``sum_risk_budget_pct`` unless you subtract ``fixed_vxx_frac`` from the cap yourself. If ``fixed_straddle_frac`` is set (often ``0.0`` when the straddle JSONL is empty / disabled), straddle allocation is fixed and the search is over the remaining sleeves only. If ``max_dd_limit_pct`` is set, the solver enforces **peak-to-trough max DD** ≤ that value (hard inequality via ``NonlinearConstraint``, with a small numerical tolerance on the bound). """ from scipy.optimize import LinearConstraint, NonlinearConstraint, differential_evolution tc = float(total_portfolio_capital) fv_fixed: float | None = None if fixed_vxx_frac is not None: fv_fixed = float(fixed_vxx_frac) if not (0.0 <= fv_fixed <= 1.0): raise ValueError("fixed_vxx_frac must be in [0, 1]") fs_fixed: float | None = None if fixed_straddle_frac is not None: fs_fixed = float(fixed_straddle_frac) if not (0.0 <= fs_fixed <= 1.0): raise ValueError("fixed_straddle_frac must be in [0, 1]") basis = load_portfolio_pnl_basis( vrp_trades=vrp_trades, vrp_ref=vrp_ref, put_risk_ref=put_risk_ref, straddle_risk_ref=straddle_risk_ref, rr_risk_ref=rr_risk_ref, overlay_ref=overlay_ref, vxx_bear_risk_ref=vxx_bear_risk_ref, vxx_call_risk_ref=vxx_call_risk_ref, vxx_ref=vxx_ref, put_trades=put_trades, straddle_trades=straddle_trades, risk_reversal_trades=risk_reversal_trades, vxx_bear_trades=vxx_bear_trades, vxx_call_trades=vxx_call_trades, total_portfolio_capital=tc, ) vrp_num = float(capital_vrp) if capital_vrp is not None else basis.eq0 hi = float(max_sleeve_pct) def _sum_cap_remainder(cap_sum: float) -> float: rem = cap_sum if fs_fixed is not None: rem -= fs_fixed if fv_fixed is not None: rem -= fv_fixed if rem < 0: raise ValueError( "sum_risk_budget_pct must be >= fixed_straddle_frac + fixed_vxx_frac " f"(cap={cap_sum}, fs_fixed={fs_fixed}, fv_fixed={fv_fixed})" ) return rem # Search dimension: (put, straddle, rr, vxx) with optional fixed straddle and/or VXX. if fv_fixed is not None and fs_fixed is not None: bounds = [(0.0, hi), (0.0, hi)] def _unpack(x: np.ndarray) -> tuple[float, float, float, float]: return float(x[0]), fs_fixed, float(x[1]), fv_fixed elif fv_fixed is not None: bounds = [(0.0, hi), (0.0, hi), (0.0, hi)] def _unpack(x: np.ndarray) -> tuple[float, float, float, float]: return float(x[0]), float(x[1]), float(x[2]), fv_fixed elif fs_fixed is not None: bounds = [(0.0, hi), (0.0, hi), (0.0, hi)] def _unpack(x: np.ndarray) -> tuple[float, float, float, float]: return float(x[0]), fs_fixed, float(x[1]), float(x[2]) else: bounds = [(0.0, hi), (0.0, hi), (0.0, hi), (0.0, hi)] def _unpack(x: np.ndarray) -> tuple[float, float, float, float]: return float(x[0]), float(x[1]), float(x[2]), float(x[3]) cons_list: list[Any] = [] if sum_risk_budget_pct is not None: cap_sum = float(sum_risk_budget_pct) rem = _sum_cap_remainder(cap_sum) n_dim = len(bounds) cons_list.append(LinearConstraint(np.ones((1, n_dim)), 0.0, rem)) rng = np.random.default_rng(seed) dd_lim = max_dd_limit_pct dd_tol = 0.05 # bound slack / feasibility check (same order as ``_metrics_block`` rounding) def _metrics_from_fracs(x: np.ndarray) -> dict: fp, fs, fr, fv = _unpack(x) eq = merged_full_equity( basis, capital_vrp=vrp_num, capital_put=fp * tc, capital_straddle=fs * tc, capital_risk_reversal=fr * tc, capital_vxx=fv * tc, vxx_bear_pct=vxx_bear_pct, vxx_call_pct=vxx_call_pct, ) return _metrics_block(eq, "") if dd_lim is not None: def _max_dd_pct_mag(x: np.ndarray) -> float: m = _metrics_from_fracs(x) d = max_drawdown_pct_magnitude(m) return float(d) if math.isfinite(d) else 1e9 cons_list.append( NonlinearConstraint(_max_dd_pct_mag, -np.inf, float(dd_lim) + dd_tol), ) constraints = tuple(cons_list) def objective(x: np.ndarray) -> float: m = _metrics_from_fracs(x) sharpe = float(m["sharpe"]) if not math.isfinite(sharpe): return 1e9 return -sharpe res = differential_evolution( objective, bounds=bounds, constraints=constraints, seed=rng, maxiter=maxiter, polish=polish, workers=1, updating="deferred", ) fp, fs, fr, fv = _unpack(res.x) cp, cs, cr, cvxx = fp * tc, fs * tc, fr * tc, fv * tc eq_opt = merged_full_equity( basis, capital_vrp=vrp_num, capital_put=cp, capital_straddle=cs, capital_risk_reversal=cr, capital_vxx=cvxx, vxx_bear_pct=vxx_bear_pct, vxx_call_pct=vxx_call_pct, ) m_opt = _metrics_block(eq_opt, "sharpe_opt") dd_mag = max_drawdown_pct_magnitude(m_opt) feasible_dd = True if max_dd_limit_pct is not None: feasible_dd = dd_mag <= float(max_dd_limit_pct) + dd_tol return { "feasible_dd": feasible_dd, "max_drawdown_pct": dd_mag, "put_frac": fp, "straddle_frac": fs, "rr_frac": fr, "vxx_frac": fv, "sum_frac": fp + fs + fr + fv, "capital_vrp_scale": vrp_num, "capital_put": cp, "capital_straddle": cs, "capital_risk_reversal": cr, "capital_vxx": cvxx, "sharpe": m_opt["sharpe"], "calmar": m_opt["calmar"], "cagr_pct": m_opt["cagr_pct"], "return_pct": m_opt["return_pct"], "end_equity": m_opt["end_equity"], "equity": eq_opt, "scipy_success": bool(res.success), "scipy_message": str(res.message), "nit": int(res.nit), } def execute_portfolio_merge( *, vrp_trades: Path, total_portfolio_capital: float = 100_000.0, capital_vrp: float | None = None, capital_put: float = 10_000.0, capital_straddle: float = 10_000.0, capital_risk_reversal: float = 10_000.0, capital_vxx: float = 15_000.0, vxx_bear_pct: float = 90.0, vxx_call_pct: float = 10.0, vrp_ref: float = 100_000.0, overlay_ref: float | None = None, put_risk_ref: float | None = None, straddle_risk_ref: float | None = None, rr_risk_ref: float | None = None, vxx_ref: float | None = None, vxx_bear_risk_ref: float | None = None, vxx_call_risk_ref: float | None = None, put_trades: Path | None = None, straddle_trades: Path | None = None, risk_reversal_trades: Path | None = None, vxx_bear_trades: Path | None = None, vxx_call_trades: Path | None = None, out_csv: Path | None = None, print_report: bool = True, print_vxx_sweep: bool = True, ) -> pd.DataFrame: """ Merge **VRP** (``pnl_usd`` on exit dates) with optional IV overlay JSONLs and VXX sleeve JSONLs. Missing overlay/VXX files contribute **zero** PnL (same as CLI defaults). All equity series start at ``total_portfolio_capital``. Overlay/VXX PnL scales as ``risk_budget / risk_ref`` (median ``broker_risk_usd`` per JSONL unless refs overridden). If ``capital_vrp`` is None, VRP PnL is scaled as ``total_portfolio_capital / vrp_ref``. Used by ``portfolio_vrp_plus_vxx.py`` CLI and ``vrp_backtest_theta.py --full-portfolio-report``. """ vrp_num = float(capital_vrp) if capital_vrp is not None else float(total_portfolio_capital) put_trades = put_trades or (LOGS / "stress_longvol_otm_put.jsonl") straddle_trades = straddle_trades or (LOGS / "stress_longvol_straddle.jsonl") risk_reversal_trades = risk_reversal_trades or (LOGS / "risk_reversal.jsonl") vxx_bear_trades = vxx_bear_trades or (LOGS / "vxx_portfolio_bear_call.jsonl") vxx_call_trades = vxx_call_trades or (LOGS / "vxx_portfolio_long_call.jsonl") if out_csv is None: out_csv = LOGS / "portfolio_vrp_vxx_equity.csv" vrp_path = vrp_trades.expanduser().resolve() if print_report: print("Loading trade logs …") for label, p in [ ("OTM put overlay", put_trades), ("Stress straddle", straddle_trades), ("Risk reversal", risk_reversal_trades), ("VXX bear call", vxx_bear_trades), ("VXX long call", vxx_call_trades), ]: if not p.is_file(): print(f" [skip] {label}: file not found → {p}") basis = load_portfolio_pnl_basis( vrp_trades=vrp_path, vrp_ref=vrp_ref, put_risk_ref=put_risk_ref, straddle_risk_ref=straddle_risk_ref, rr_risk_ref=rr_risk_ref, overlay_ref=overlay_ref, vxx_bear_risk_ref=vxx_bear_risk_ref, vxx_call_risk_ref=vxx_call_risk_ref, vxx_ref=vxx_ref, put_trades=put_trades, straddle_trades=straddle_trades, risk_reversal_trades=risk_reversal_trades, vxx_bear_trades=vxx_bear_trades, vxx_call_trades=vxx_call_trades, total_portfolio_capital=total_portfolio_capital, ) all_idx = basis.index eq0 = basis.eq0 pr, sr, rr = max(basis.put_risk_ref, 1e-12), max(basis.straddle_risk_ref, 1e-12), max(basis.rr_risk_ref, 1e-12) vbr, vcr = max(basis.vxx_bear_risk_ref, 1e-12), max(basis.vxx_call_risk_ref, 1e-12) V = vrp_num * basis.B_v P = (float(capital_put) / pr) * basis.B_p S = (float(capital_straddle) / sr) * basis.B_s R = (float(capital_risk_reversal) / rr) * basis.B_r cvxx = float(capital_vxx) cap_vxx_bear = cvxx * (vxx_bear_pct / 100.0) cap_vxx_call = cvxx * (vxx_call_pct / 100.0) VB = (cap_vxx_bear / vbr) * basis.B_vb VC = (cap_vxx_call / vcr) * basis.B_vc eq_vrp_only = eq0 + V.cumsum() eq_no_vxx = eq0 + (V + P + S + R).cumsum() eq_vxx_sleeve = eq0 + (VB + VC).cumsum() eq_with_vxx = eq0 + (V + P + S + R + VB + VC).cumsum() frame = pd.DataFrame( { "eq_vrp_only": eq_vrp_only, "eq_vrp_plus_overlays": eq_no_vxx, "eq_vxx_sleeve": eq_vxx_sleeve, "eq_full_portfolio": eq_with_vxx, "pnl_vrp": V, "pnl_overlays": P + S + R, "pnl_vxx_bear": VB, "pnl_vxx_call": VC, }, index=all_idx, ) if print_report: m_vrp = _metrics_block(eq_vrp_only, "VRP only") m_no_vxx = _metrics_block(eq_no_vxx, "VRP + overlays") m_vxx = _metrics_block(eq_vxx_sleeve, "VXX sleeve only") m_full = _metrics_block(eq_with_vxx, "FULL (VRP + overlays + VXX)") corr_data = {"VRP": V, "IV_overlays": P + S + R, "VXX_bear": VB, "VXX_call": VC} corr_df = pd.DataFrame(corr_data).corr() print(f"\n{'='*95}") print("PORTFOLIO COMPARISON: VRP+Overlays vs VRP+Overlays+VXX") print(f"{'='*95}") print( f"\nAccount start (all curves): ${eq0:,.0f} | " f"VRP PnL scale numerator ${vrp_num:,.0f} vs ref ${vrp_ref:,.0f} | " f"IV overlay **risk budgets** (max loss / debit): put ${capital_put:,.0f} / ref ${pr:,.0f}, " f"straddle ${capital_straddle:,.0f} / ref ${sr:,.0f}, " f"RR ${capital_risk_reversal:,.0f} / ref ${rr:,.0f} | " f"VXX ${capital_vxx:,.0f} ({vxx_bear_pct:.0f}% bear / {vxx_call_pct:.0f}% call) " f"→ bear ${cap_vxx_bear:,.0f} / ref ${vbr:,.0f}, call ${cap_vxx_call:,.0f} / ref ${vcr:,.0f}" ) if eq0 > 0: print( " (same sleeves as **fraction of account** — what you set with " "*-pct flags or merge JSON: " f"put {100.0 * capital_put / eq0:.3f}%, straddle {100.0 * capital_straddle / eq0:.3f}%, " f"RR {100.0 * capital_risk_reversal / eq0:.3f}%, VXX {100.0 * capital_vxx / eq0:.3f}%; " "dollar line above is that fraction × account, used to scale each overlay JSONL vs its ref risk)" ) print( f"Date range: {all_idx[0].date()} → {all_idx[-1].date()} " f"({len(all_idx)} days; ~{m_vrp['years']:.2f} yr CAGR horizon)" ) print(f"\n{'─'*105}") hdr = ( f"{'Portfolio':<30} {'Start':>10} {'End':>10} {'Return':>9} {'Ret%':>7} {'CAGR':>8} " f"{'Sharpe':>7} {'MaxDD$':>9} {'MaxDD%':>8} {'Calmar':>7}" ) print(hdr) print(f"{'─'*105}") for m in [m_vrp, m_no_vxx, m_vxx, m_full]: cg = m["cagr_pct"] cag_s = f"{cg:>6.2f}%" if isinstance(cg, (int, float)) and math.isfinite(cg) else f"{'n/a':>8}" print( f"{m['label']:<30} {m['start_capital']:>10,.0f} {m['end_equity']:>10,.0f} " f"{m['total_return']:>+9,.0f} {m['return_pct']:>6.1f}% {cag_s} {m['sharpe']:>7.2f} " f"{m['max_dd_usd']:>9,.0f} {m['max_dd_pct']:>7.2f}% {m['calmar']:>7.2f}" ) print(f"{'─'*105}") print("\n── IMPROVEMENT FROM ADDING VXX SLEEVE ──") sharpe_delta = m_full["sharpe"] - m_no_vxx["sharpe"] calmar_delta = m_full["calmar"] - m_no_vxx["calmar"] dd_delta = m_full["max_dd_pct"] - m_no_vxx["max_dd_pct"] ret_delta = m_full["return_pct"] - m_no_vxx["return_pct"] if math.isfinite(m_full["cagr_pct"]) and math.isfinite(m_no_vxx["cagr_pct"]): cagr_d = m_full["cagr_pct"] - m_no_vxx["cagr_pct"] print( f" CAGR%: {m_no_vxx['cagr_pct']:.2f}% → {m_full['cagr_pct']:.2f}% ({cagr_d:+.2f}pp)" ) print(f" Sharpe: {m_no_vxx['sharpe']:.2f} → {m_full['sharpe']:.2f} ({sharpe_delta:+.3f})") print(f" Calmar: {m_no_vxx['calmar']:.2f} → {m_full['calmar']:.2f} ({calmar_delta:+.2f})") print(f" MaxDD%: {m_no_vxx['max_dd_pct']:.2f}% → {m_full['max_dd_pct']:.2f}% ({dd_delta:+.2f}pp)") print(f" Return%: {m_no_vxx['return_pct']:.1f}% → {m_full['return_pct']:.1f}% ({ret_delta:+.1f}pp)") print("\n── DAILY PNL CORRELATIONS ──") print(corr_df.to_string(float_format=lambda x: f"{x:+.3f}")) if print_report and print_vxx_sweep: print(f"\n{'='*95}") print( "VXX RISK-BUDGET SENSITIVITY — same account start; VRP + IV overlays unchanged; " "total VXX risk budget = fraction × account (split bear/call by %)" ) print(f"{'='*95}") print(f"{'VXX$':>8} {'VXX%':>6} {'Sharpe':>7} {'MaxDD%':>8} {'Calmar':>7} {'RetPct':>7} Notes") print(f"{'─'*60}") best_sharpe_alloc = None best_sharpe_val = -999.0 for vxx_frac in [0, 0.02, 0.05, 0.08, 0.10, 0.12, 0.15, 0.20, 0.25, 0.30]: vxx_cap = eq0 * vxx_frac bear_cap = vxx_cap * (vxx_bear_pct / 100.0) call_cap = vxx_cap * (vxx_call_pct / 100.0) vxx_b = basis.B_vb * (bear_cap / vbr) vxx_c = basis.B_vc * (call_cap / vcr) combo_pnl = V + P + S + R + vxx_b + vxx_c eq_combo = eq0 + combo_pnl.cumsum() mc = _metrics_block(eq_combo, f"VXX={vxx_frac:.0%}") note = "" if mc["sharpe"] > best_sharpe_val: best_sharpe_val = mc["sharpe"] best_sharpe_alloc = vxx_frac if vxx_frac == 0: note = "← baseline (no VXX)" print( f"{vxx_cap:>8,.0f} {vxx_frac:>5.0%} {mc['sharpe']:>7.2f} " f"{mc['max_dd_pct']:>7.2f}% {mc['calmar']:>7.2f} {mc['return_pct']:>6.1f}% {note}" ) if best_sharpe_alloc is not None and best_sharpe_alloc > 0: print(f"\n ★ Best Sharpe at {best_sharpe_alloc:.0%} VXX allocation") print(f"{'='*95}") out_csv = out_csv.expanduser() out_csv.parent.mkdir(parents=True, exist_ok=True) frame.to_csv(out_csv, date_format="%Y-%m-%d") if print_report: print(f"\nDaily equity CSV → {out_csv}") return frame def _metrics_block(eq: pd.Series, label: str) -> dict: daily = eq.diff().fillna(0) total_ret = float(eq.iloc[-1] - eq.iloc[0]) s0, s1 = float(eq.iloc[0]), float(eq.iloc[-1]) pct_ret = total_ret / s0 * 100 if s0 > 0 else 0 mn = float(daily.mean()) sd = float(daily.std()) if daily.std() > 0 else 1e-9 sharpe = (mn / sd) * np.sqrt(252) peak = eq.cummax() dd_abs = float((eq - peak).min()) dd_pct = float(((eq - peak) / peak.replace(0, np.nan)).fillna(0).min()) * 100 calmar = abs(total_ret / dd_abs) if dd_abs != 0 else 0 idx = eq.index cagr_pct = float("nan") years_sample = float("nan") if len(idx) >= 2: t0, t1 = pd.Timestamp(idx[0]), pd.Timestamp(idx[-1]) years_sample = max((t1 - t0).days / 365.25, 1e-9) if s0 > 0 and s1 > 0: cagr_pct = ((s1 / s0) ** (1.0 / years_sample) - 1.0) * 100.0 return { "label": label, "start_capital": round(s0, 0), "end_equity": round(s1, 0), "total_return": round(total_ret, 0), "return_pct": round(pct_ret, 1), "cagr_pct": round(cagr_pct, 2) if math.isfinite(cagr_pct) else float("nan"), "years": round(years_sample, 2) if math.isfinite(years_sample) else float("nan"), "sharpe": round(sharpe, 3), "max_dd_usd": round(dd_abs, 0), "max_dd_pct": round(dd_pct, 2), "calmar": round(calmar, 2), "daily_vol": round(sd, 2), } def _risk_budget_from_pct_or_usd(*, usd: float, pct: float | None, total_capital: float) -> float: """If ``pct`` is set, return ``total_capital * pct`` (``pct`` is a fraction, e.g. 0.02 = 2%%).""" if pct is not None: return float(total_capital) * float(pct) return float(usd) def merge_capital_from_allocation_json(path: Path | str, *, total_capital: float) -> dict[str, float]: """ Load ``optimize_overlay_risk_fracs.py`` output and return kwargs for :func:`execute_portfolio_merge`. Reads **fractions of account** from ``portfolio_vrp_plus_vxx_cli_snippet`` or ``overlay_risk_fractions`` (maps to ``capital_put``, … as ``total_capital * frac``). If only ``overlay_risk_usd`` exists (``calmar-usd`` mode), scales those USD budgets by ``total_capital / json['total_capital']``. Optional ``capital_vrp_pct`` in the snippet sets ``capital_vrp`` (VRP PnL scale numerator). """ cap = float(total_capital) if not math.isfinite(cap) or cap <= 0: raise ValueError("total_capital must be a positive finite float") p = Path(path).expanduser().resolve() raw = json.loads(p.read_text(encoding="utf-8")) tc0 = float(raw.get("total_capital", cap) or cap) scale = cap / tc0 if tc0 > 0 else 1.0 snip = raw.get("portfolio_vrp_plus_vxx_cli_snippet") snip_d = snip if isinstance(snip, dict) else {} if snip_d.get("capital_put_pct") is not None: put_p = float(snip_d.get("capital_put_pct", 0.0) or 0.0) str_p = float(snip_d.get("capital_straddle_pct", 0.0) or 0.0) rr_p = float(snip_d.get("capital_risk_reversal_pct", 0.0) or 0.0) vxx_p = float(snip_d.get("capital_vxx_pct", 0.0) or 0.0) bear = float(snip_d.get("vxx_bear_pct", 90.0) or 90.0) call = float(snip_d.get("vxx_call_pct", 10.0) or 10.0) out: dict[str, float] = { "capital_put": cap * put_p, "capital_straddle": cap * str_p, "capital_risk_reversal": cap * rr_p, "capital_vxx": cap * vxx_p, "vxx_bear_pct": bear, "vxx_call_pct": call, } elif isinstance(raw.get("overlay_risk_fractions"), dict) and raw["overlay_risk_fractions"]: ov = raw["overlay_risk_fractions"] def _f(k: str) -> float: v = ov.get(k) return float(v) if v is not None else 0.0 out = { "capital_put": cap * _f("stress_longvol_otm_put"), "capital_straddle": cap * _f("stress_longvol_straddle"), "capital_risk_reversal": cap * _f("risk_reversal"), "capital_vxx": cap * _f("vxx_sleeve"), "vxx_bear_pct": float(raw.get("vxx_bear_pct", 90.0) or 90.0), "vxx_call_pct": float(raw.get("vxx_call_pct", 10.0) or 10.0), } elif isinstance(raw.get("overlay_risk_usd"), dict) and raw["overlay_risk_usd"]: usd = raw["overlay_risk_usd"] def _u(k: str) -> float: v = usd.get(k) return float(v) if v is not None else 0.0 out = { "capital_put": _u("stress_longvol_otm_put") * scale, "capital_straddle": _u("stress_longvol_straddle") * scale, "capital_risk_reversal": _u("risk_reversal") * scale, "capital_vxx": _u("vxx_sleeve") * scale, "vxx_bear_pct": float(raw.get("vxx_bear_pct", 90.0) or 90.0), "vxx_call_pct": float(raw.get("vxx_call_pct", 10.0) or 10.0), } else: raise ValueError( f"{p}: expected 'portfolio_vrp_plus_vxx_cli_snippet' with capital_put_pct, " "or 'overlay_risk_fractions', or 'overlay_risk_usd'" ) vrp_pct = snip_d.get("capital_vrp_pct") if vrp_pct is not None: out["capital_vrp"] = cap * float(vrp_pct) return out def main(): ap = argparse.ArgumentParser(description="VRP + overlays + VXX contango portfolio") # Capital allocation # NOTE: Calendar spread sleeve deliberately excluded. Drop-one-out analysis # showed it lost -$10.5K (Sharpe -5.61) and *improved* portfolio Sharpe by # +0.069 when removed. Do not re-add without fresh evidence of profitability. ap.add_argument( "--total-capital", type=float, default=100_000.0, help="Starting equity for every reported curve (single account; default 100k)", ) ap.add_argument( "--allocation-json", type=Path, default=None, help="Run merged portfolio from optimize_overlay_risk_fracs.py output: sleeve fractions × " "--total-capital (incompatible with --capital-*-pct / --optimize-*).", ) ap.add_argument( "--capital-vrp", type=float, default=None, help="VRP PnL scale numerator vs --vrp-ref (default: same as --total-capital)", ) ap.add_argument( "--capital-vrp-pct", type=float, default=None, metavar="FRAC", help="VRP scale numerator = FRAC × --total-capital (e.g. 1.0); overrides --capital-vrp when set", ) ap.add_argument( "--capital-put", type=float, default=10_000, help="Put overlay broker risk budget USD (max loss / debit); scaled vs ref (median broker_risk_usd in JSONL)", ) ap.add_argument( "--capital-put-pct", type=float, default=None, metavar="FRAC", help="Put risk budget = FRAC × --total-capital (e.g. 0.02 for 2%%); overrides --capital-put when set", ) ap.add_argument("--capital-straddle", type=float, default=10_000, help="Straddle sleeve risk budget USD") ap.add_argument( "--capital-straddle-pct", type=float, default=None, metavar="FRAC", help="Straddle risk budget = FRAC × --total-capital; overrides --capital-straddle when set", ) ap.add_argument("--capital-risk-reversal", type=float, default=10_000, help="Risk-reversal sleeve risk budget USD") ap.add_argument( "--capital-risk-reversal-pct", type=float, default=None, metavar="FRAC", help="RR risk budget = FRAC × --total-capital; overrides --capital-risk-reversal when set", ) ap.add_argument( "--capital-vxx", type=float, default=15_000, help="Total VXX sleeve broker risk budget USD (split bear/call by %% flags)", ) ap.add_argument( "--capital-vxx-pct", type=float, default=None, metavar="FRAC", help="VXX total risk budget = FRAC × --total-capital; overrides --capital-vxx when set", ) ap.add_argument("--vxx-bear-pct", type=float, default=90, help="Percent of VXX capital to bear call (rest to long call)") ap.add_argument("--vxx-call-pct", type=float, default=10, help="Percent of VXX capital to long call") # Reference risk / scale (None = auto median broker_risk_usd per JSONL, or inferred legacy rows) ap.add_argument("--vrp-ref", type=float, default=100_000) ap.add_argument( "--overlay-ref", type=float, default=None, help="Override risk ref USD for all three IV overlays (put, straddle, RR)", ) ap.add_argument("--put-risk-ref", type=float, default=None, help="Override put overlay risk ref USD") ap.add_argument("--straddle-risk-ref", type=float, default=None) ap.add_argument("--rr-risk-ref", type=float, default=None) ap.add_argument( "--vxx-ref", type=float, default=None, help="Override risk ref USD for both VXX JSONLs (bear call + long call)", ) ap.add_argument("--vxx-bear-risk-ref", type=float, default=None) ap.add_argument("--vxx-call-risk-ref", type=float, default=None) # Paths (defaults to standard log locations) ap.add_argument( "--vrp-trades", type=Path, default=DEFAULT_VRP_TRADES_JSONL, help=f"VRP closed trades JSONL (default: {DEFAULT_VRP_TRADES_JSONL.name})", ) ap.add_argument("--put-trades", type=Path, default=LOGS / "stress_longvol_otm_put.jsonl") ap.add_argument("--straddle-trades", type=Path, default=LOGS / "stress_longvol_straddle.jsonl") ap.add_argument("--risk-reversal-trades", type=Path, default=LOGS / "risk_reversal.jsonl") ap.add_argument("--vxx-bear-trades", type=Path, default=LOGS / "vxx_portfolio_bear_call.jsonl") ap.add_argument("--vxx-call-trades", type=Path, default=LOGS / "vxx_portfolio_long_call.jsonl") ap.add_argument("--out-csv", type=Path, default=LOGS / "portfolio_vrp_vxx_equity.csv") ap.add_argument( "--no-vxx-sweep", action="store_true", help="Skip the VXX notional sensitivity table after the main comparison", ) _opt_grp = ap.add_mutually_exclusive_group() _opt_grp.add_argument( "--optimize-sleeves", action="store_true", help="Search put/straddle/RR/VXX risk budgets (USD) to maximize Calmar + w*Sharpe with a max-DD soft target", ) _opt_grp.add_argument( "--optimize-sharpe", action="store_true", help="Search sleeve allocation fractions (of --total-capital) to maximize Sharpe on the full merged book", ) ap.add_argument( "--sharpe-sum-cap-pct", type=float, default=0.35, metavar="FRAC", help="With --optimize-sharpe, max sum of put+straddle+RR+VXX fractions (default 0.35); use 0 for no sum cap", ) ap.add_argument( "--sharpe-max-sleeve-pct", type=float, default=0.25, metavar="FRAC", help="With --optimize-sharpe, upper bound per sleeve fraction (default 0.25 of total capital)", ) ap.add_argument( "--sharpe-max-dd-pct", type=float, default=10.0, metavar="PCT", help="With --optimize-sharpe, hard cap on peak-to-trough max DD %% (default 10); use 0 to disable", ) ap.add_argument( "--sharpe-fixed-vxx-pct", type=float, default=None, metavar="FRAC", help="With --optimize-sharpe, hold VXX sleeve at FRAC of --total-capital and optimize only IV sleeves " "(put/straddle/RR). --sharpe-sum-cap-pct then limits put+straddle+RR only (add this FRAC separately for total overlay+VXX).", ) ap.add_argument( "--sharpe-fixed-straddle-pct", type=float, default=None, metavar="FRAC", help="With --optimize-sharpe, hold straddle sleeve at FRAC (use 0 when straddle JSONL is disabled / zero PnL).", ) ap.add_argument( "--max-dd-pct", type=float, default=10.0, help="With --optimize-sleeves, penalize paths whose max drawdown exceeds this %% (default 10)", ) ap.add_argument( "--notionals-budget", type=float, default=100_000.0, help="With --optimize-sleeves, cap sum of risk budgets put+straddle+RR+VXX (default 100k); 0 = no cap", ) ap.add_argument( "--notionals-budget-pct", type=float, default=None, metavar="FRAC", help="With --optimize-sleeves, cap sum = FRAC × --total-capital; overrides --notionals-budget when set (0 still = no cap)", ) ap.add_argument("--sharpe-weight", type=float, default=0.25, help="With --optimize-sleeves, objective += this * Sharpe") ap.add_argument( "--opt-max-sleeve", type=float, default=100_000.0, help="With --optimize-sleeves, upper bound USD per searched sleeve (default 100k)", ) ap.add_argument( "--opt-max-sleeve-pct", type=float, default=None, metavar="FRAC", help="With --optimize-sleeves, per-sleeve upper bound = FRAC × --total-capital; overrides --opt-max-sleeve when set", ) ap.add_argument("--opt-maxiter", type=int, default=80, help="DE maxiter for --optimize-sleeves / --optimize-sharpe") ap.add_argument("--opt-seed", type=int, default=0, help="RNG seed for --optimize-sleeves / --optimize-sharpe") args = ap.parse_args() tc = float(args.total_capital) if args.allocation_json is not None: bad = [] if args.capital_put_pct is not None: bad.append("--capital-put-pct") if args.capital_straddle_pct is not None: bad.append("--capital-straddle-pct") if args.capital_risk_reversal_pct is not None: bad.append("--capital-risk-reversal-pct") if args.capital_vxx_pct is not None: bad.append("--capital-vxx-pct") if args.capital_vrp_pct is not None: bad.append("--capital-vrp-pct") if args.optimize_sharpe or args.optimize_sleeves: bad.append("--optimize-sharpe / --optimize-sleeves") if bad: print( "ERROR: --allocation-json cannot be combined with: " + ", ".join(bad), file=sys.stderr, ) sys.exit(1) cap_vrp_resolved = args.capital_vrp if args.capital_vrp_pct is not None and args.allocation_json is None: cap_vrp_resolved = tc * float(args.capital_vrp_pct) if args.allocation_json is not None: aj = args.allocation_json.expanduser() if not aj.is_file(): print(f"ERROR: --allocation-json not found: {aj.resolve()}", file=sys.stderr) sys.exit(1) try: alloc = merge_capital_from_allocation_json(aj, total_capital=tc) except (json.JSONDecodeError, OSError, ValueError) as e: print(f"ERROR: --allocation-json {aj}: {e}", file=sys.stderr) sys.exit(1) cap_put = float(alloc["capital_put"]) cap_str = float(alloc["capital_straddle"]) cap_rr = float(alloc["capital_risk_reversal"]) cap_vxx = float(alloc["capital_vxx"]) vxx_bear_eff = float(alloc["vxx_bear_pct"]) vxx_call_eff = float(alloc["vxx_call_pct"]) if "capital_vrp" in alloc: cap_vrp_resolved = float(alloc["capital_vrp"]) print("=" * 72) print("PORTFOLIO BACKTEST (merge model): sleeve sizes from % allocations in JSON") print("=" * 72) print(f" JSON: {aj.resolve()}") print( f" put {100.0 * cap_put / tc:.4f}% straddle {100.0 * cap_str / tc:.4f}% " f"RR {100.0 * cap_rr / tc:.4f}% VXX {100.0 * cap_vxx / tc:.4f}% of ${tc:,.0f} " f"| VXX split {vxx_bear_eff:.0f}/{vxx_call_eff:.0f} bear/call" ) if "capital_vrp" in alloc: print(f" VRP PnL scale numerator ${float(cap_vrp_resolved):,.0f} (capital_vrp_pct from JSON)") else: print(f" VRP PnL scale numerator ${tc:,.0f} (default: full --total-capital vs --vrp-ref)") print("=" * 72 + "\n") else: vxx_bear_eff = float(args.vxx_bear_pct) vxx_call_eff = float(args.vxx_call_pct) if args.allocation_json is None: cap_put = _risk_budget_from_pct_or_usd( usd=args.capital_put, pct=args.capital_put_pct, total_capital=tc ) cap_str = _risk_budget_from_pct_or_usd( usd=args.capital_straddle, pct=args.capital_straddle_pct, total_capital=tc ) cap_rr = _risk_budget_from_pct_or_usd( usd=args.capital_risk_reversal, pct=args.capital_risk_reversal_pct, total_capital=tc ) cap_vxx = _risk_budget_from_pct_or_usd( usd=args.capital_vxx, pct=args.capital_vxx_pct, total_capital=tc ) if args.notionals_budget_pct is not None: p = float(args.notionals_budget_pct) budget_cap = None if p <= 0 else tc * p elif args.notionals_budget <= 0: budget_cap = None else: budget_cap = float(args.notionals_budget) opt_max_sleeve = float(args.opt_max_sleeve) if args.opt_max_sleeve_pct is not None: opt_max_sleeve = tc * float(args.opt_max_sleeve_pct) _risk_ref_kw = dict( overlay_ref=args.overlay_ref, put_risk_ref=args.put_risk_ref, straddle_risk_ref=args.straddle_risk_ref, rr_risk_ref=args.rr_risk_ref, vxx_ref=args.vxx_ref, vxx_bear_risk_ref=args.vxx_bear_risk_ref, vxx_call_risk_ref=args.vxx_call_risk_ref, ) if args.optimize_sharpe: sharpe_sum_cap = None if float(args.sharpe_sum_cap_pct) <= 0 else float(args.sharpe_sum_cap_pct) sharpe_dd_cap = None if float(args.sharpe_max_dd_pct) <= 0 else float(args.sharpe_max_dd_pct) r = optimize_portfolio_sharpe( args.vrp_trades, total_portfolio_capital=args.total_capital, capital_vrp=cap_vrp_resolved, sum_risk_budget_pct=sharpe_sum_cap, max_sleeve_pct=float(args.sharpe_max_sleeve_pct), max_dd_limit_pct=sharpe_dd_cap, fixed_vxx_frac=args.sharpe_fixed_vxx_pct, fixed_straddle_frac=args.sharpe_fixed_straddle_pct, vxx_bear_pct=args.vxx_bear_pct, vxx_call_pct=args.vxx_call_pct, vrp_ref=args.vrp_ref, put_trades=args.put_trades, straddle_trades=args.straddle_trades, risk_reversal_trades=args.risk_reversal_trades, vxx_bear_trades=args.vxx_bear_trades, vxx_call_trades=args.vxx_call_trades, maxiter=int(args.opt_maxiter), seed=int(args.opt_seed), **_risk_ref_kw, ) tc2 = float(args.total_capital) print("\n" + "=" * 72) print("SHARPE-OPTIMIZED ALLOCATION (fractions of --total-capital; VRP scale fixed)") print("=" * 72) dd_note = "disabled" if sharpe_dd_cap is None else f"{sharpe_dd_cap:.1f}%" feas_dd = "n/a" if sharpe_dd_cap is None else r["feasible_dd"] fv_note = "" if args.sharpe_fixed_vxx_pct is not None: fv_note += f" | fixed VXX frac: {float(args.sharpe_fixed_vxx_pct):.6f}" if args.sharpe_fixed_straddle_pct is not None: fv_note += f" | fixed straddle frac: {float(args.sharpe_fixed_straddle_pct):.6f}" print(f" DD cap (hard): {dd_note} | feasible_dd: {feas_dd}{fv_note}") print(f" put_frac={r['put_frac']:.6f} straddle_frac={r['straddle_frac']:.6f} " f"rr_frac={r['rr_frac']:.6f} vxx_frac={r['vxx_frac']:.6f} sum={r['sum_frac']:.6f}") print(f" USD @ ${tc2:,.0f}: put=${r['capital_put']:,.0f} straddle=${r['capital_straddle']:,.0f} " f"RR=${r['capital_risk_reversal']:,.0f} VXX=${r['capital_vxx']:,.0f}") _cg = r["cagr_pct"] _cgs = f"{_cg:.2f}%" if math.isfinite(_cg) else "n/a" print(f" Sharpe={r['sharpe']:.3f} Calmar={r['calmar']:.2f} max_dd%={r['max_drawdown_pct']:.2f} " f"CAGR={_cgs} ret%={r['return_pct']:.1f}% end=${r['end_equity']:,.0f}") print(f" [{r['scipy_message']}]") print("\n Re-run with:\n") print( f" .venv/bin/python RenTech/strategy_stack/portfolio_vrp_plus_vxx.py \\\n" f" --total-capital {tc2:.0f} \\\n" f" --capital-put-pct {r['put_frac']:.8f} \\\n" f" --capital-straddle-pct {r['straddle_frac']:.8f} \\\n" f" --capital-risk-reversal-pct {r['rr_frac']:.8f} \\\n" f" --capital-vxx-pct {r['vxx_frac']:.8f} \\\n" f" --vxx-bear-pct {args.vxx_bear_pct:g} --vxx-call-pct {args.vxx_call_pct:g} \\\n" f" --no-vxx-sweep" ) print("=" * 72 + "\n") execute_portfolio_merge( vrp_trades=args.vrp_trades, total_portfolio_capital=args.total_capital, capital_vrp=cap_vrp_resolved, capital_put=r["capital_put"], capital_straddle=r["capital_straddle"], capital_risk_reversal=r["capital_risk_reversal"], capital_vxx=r["capital_vxx"], vxx_bear_pct=args.vxx_bear_pct, vxx_call_pct=args.vxx_call_pct, vrp_ref=args.vrp_ref, put_trades=args.put_trades, straddle_trades=args.straddle_trades, risk_reversal_trades=args.risk_reversal_trades, vxx_bear_trades=args.vxx_bear_trades, vxx_call_trades=args.vxx_call_trades, out_csv=args.out_csv, print_report=True, print_vxx_sweep=False, **_risk_ref_kw, ) elif args.optimize_sleeves: r = optimize_sleeve_notionals( args.vrp_trades, total_portfolio_capital=args.total_capital, capital_vrp=cap_vrp_resolved, max_dd_limit_pct=float(args.max_dd_pct), additional_notionals_cap=budget_cap, max_single_sleeve=opt_max_sleeve, sharpe_weight=float(args.sharpe_weight), vxx_bear_pct=args.vxx_bear_pct, vxx_call_pct=args.vxx_call_pct, vrp_ref=args.vrp_ref, put_trades=args.put_trades, straddle_trades=args.straddle_trades, risk_reversal_trades=args.risk_reversal_trades, vxx_bear_trades=args.vxx_bear_trades, vxx_call_trades=args.vxx_call_trades, maxiter=int(args.opt_maxiter), seed=int(args.opt_seed), **_risk_ref_kw, ) print("\n" + "=" * 72) print("OPTIMIZED SLEEVE RISK BUDGETS (VRP scale fixed; single account start)") print("=" * 72) print(f" Feasible (max DD ≤ {args.max_dd_pct:.1f}%): {r['feasible']}") print(f" capital_put={r['capital_put']:,.0f} capital_straddle={r['capital_straddle']:,.0f} " f"capital_risk_reversal={r['capital_risk_reversal']:,.0f} capital_vxx={r['capital_vxx']:,.0f}") print(f" sum(risk budgets overlay+VXX)={r['sum_overlay_vxx']:,.0f} | VRP scale numerator={r['capital_vrp_scale']:,.0f}") _cg = r["cagr_pct"] _cgs = f"{_cg:.2f}%" if math.isfinite(_cg) else "n/a" print(f" max_dd_pct={r['max_drawdown_pct']:.2f}% Calmar={r['calmar']:.2f} Sharpe={r['sharpe']:.2f} " f"CAGR={_cgs} ret%={r['return_pct']:.1f}% end=${r['end_equity']:,.0f}") print(f" objective (Calmar + {args.sharpe_weight}*Sharpe)={r['objective_score']:.3f}") print(f" [{r['scipy_message']}]") print("=" * 72 + "\n") execute_portfolio_merge( vrp_trades=args.vrp_trades, total_portfolio_capital=args.total_capital, capital_vrp=cap_vrp_resolved, capital_put=r["capital_put"], capital_straddle=r["capital_straddle"], capital_risk_reversal=r["capital_risk_reversal"], capital_vxx=r["capital_vxx"], vxx_bear_pct=args.vxx_bear_pct, vxx_call_pct=args.vxx_call_pct, vrp_ref=args.vrp_ref, put_trades=args.put_trades, straddle_trades=args.straddle_trades, risk_reversal_trades=args.risk_reversal_trades, vxx_bear_trades=args.vxx_bear_trades, vxx_call_trades=args.vxx_call_trades, out_csv=args.out_csv, print_report=True, print_vxx_sweep=False, **_risk_ref_kw, ) else: execute_portfolio_merge( vrp_trades=args.vrp_trades, total_portfolio_capital=args.total_capital, capital_vrp=cap_vrp_resolved, capital_put=cap_put, capital_straddle=cap_str, capital_risk_reversal=cap_rr, capital_vxx=cap_vxx, vxx_bear_pct=vxx_bear_eff, vxx_call_pct=vxx_call_eff, vrp_ref=args.vrp_ref, put_trades=args.put_trades, straddle_trades=args.straddle_trades, risk_reversal_trades=args.risk_reversal_trades, vxx_bear_trades=args.vxx_bear_trades, vxx_call_trades=args.vxx_call_trades, out_csv=args.out_csv, print_report=True, print_vxx_sweep=not args.no_vxx_sweep, **_risk_ref_kw, ) if __name__ == "__main__": main()