#!/usr/bin/env python3 """ Five structurally distinct VXX option strategies gated on **VX1 vs VX3 futures** (third-month / front-month settle ratio), not VIX3M cash. Workflow -------- 1. ``--sweep`` runs a large grid (~100+ configs), ranks by daily Sharpe on a fixed $100k book with per-trade risk scaling. 2. Default (no sweep) runs the top N curated structures (``--top-n``, default 5) and writes JSONL + summary. Example:: cd /Users/robzingale/trading_bot .venv/bin/python RenTech/data_pipeline/download_cboe_vix_futures.py PYTHONUNBUFFERED=1 .venv/bin/python RenTech/strategy_stack/backtest_vxx_vx1_vx3_strategies.py \\ --start 2020-01-01 --end 2025-12-31 --sweep PYTHONUNBUFFERED=1 .venv/bin/python RenTech/strategy_stack/backtest_vxx_vx1_vx3_strategies.py \\ --start 2020-01-01 --end 2025-12-31 """ from __future__ import annotations import argparse import itertools import json import math import sys from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, Callable 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.explore_vxx_decay_strategies import ( MULT, SLIPPAGE, Trade, _build_short_call, _exit_short_call, _exit_bear_call_from_chain, _exit_bull_call_spread, _exit_deep_put, _exit_deep_put_long_call, _exit_long_call, _exit_put_debit, _exit_ratio_put, _exit_short_call, _load_chain, _load_contango, _nearest_strike, _pick_expiry, _slipped, _spot_from_chain, _build_bear_call_credit, _build_bull_call_spread, _build_deep_itm_put, _build_deep_put_long_call, _build_long_call, _build_put_debit, _build_ratio_put_1x2, _build_short_call, broker_risk_usd_from_built, resolve_vxx_contracts_and_broker_risk, vxx_built_snapshot, ) DATA_DIR = _REPO / "RenTech" / "data" LOGS = DATA_DIR / "logs" START_CAPITAL = 100_000.0 _CHAIN_CACHE: dict[str, pd.DataFrame] = {} def _load_chain_cached(d: pd.Timestamp) -> pd.DataFrame: key = str(pd.Timestamp(d).date()) if key not in _CHAIN_CACHE: _CHAIN_CACHE[key] = _load_chain(d) return _CHAIN_CACHE[key] def _ensure_vx3_panel(ct: pd.DataFrame) -> pd.DataFrame: if "vx3_vx1_ratio_ffill" in ct.columns and ct["vx3_vx1_ratio_ffill"].notna().any(): return ct import vix_utils raw = vix_utils.load_vix_term_structure(forceReload=False) raw["Trade Date"] = pd.to_datetime(raw["Trade Date"]).dt.tz_localize(None) s1 = raw.loc[raw["Tenor_Monthly"] == 1].sort_values("Trade Date").drop_duplicates("Trade Date", keep="last") s3 = raw.loc[raw["Tenor_Monthly"] == 3].sort_values("Trade Date").drop_duplicates("Trade Date", keep="last") p1 = s1.set_index("Trade Date")["Settle"].fillna(s1.set_index("Trade Date")["Close"]) p3 = s3.set_index("Trade Date")["Settle"].fillna(s3.set_index("Trade Date")["Close"]) ratio = (p3 / p1) - 1.0 ct = ct.copy() ct["vx3_vx1_ratio"] = ratio.reindex(ct.index) ct["vx3_vx1_ratio_ffill"] = ct["vx3_vx1_ratio"].ffill() return ct # --------------------------------------------------------------------------- # Extra structures (not in explore_vxx_decay_strategies) # --------------------------------------------------------------------------- def _build_bear_put_credit(chain: pd.DataFrame, spot: float, exp: pd.Timestamp, width_pct: float, short_mny: float): """Sell OTM put, buy further OTM put (credit).""" short_row = _nearest_strike(chain, spot * short_mny, "P", exp) long_target = spot * (1.0 - width_pct) long_row = _nearest_strike(chain, long_target, "P", exp) if short_row is None or long_row is None: return None sk, lk = float(short_row["strike"]), float(long_row["strike"]) if lk >= sk: return None credit = (_slipped(float(short_row["mid"]), "sell") - _slipped(float(long_row["mid"]), "buy")) * MULT if credit <= 0: return None max_loss = (sk - lk) * MULT - credit return {"credit": credit, "max_loss": max_loss, "short_k": sk, "long_k": lk, "right": "P", "kind": "bear_put"} def _exit_bear_put_credit(chain: pd.DataFrame, spot: float, legs: dict, exp: pd.Timestamp) -> float: sk, lk = legs["short_k"], legs["long_k"] sub = chain[(chain["right_code"] == "P") & (chain["expiration_dt"] == exp)] if not chain.empty else pd.DataFrame() if not sub.empty: sr = sub.iloc[(sub["strike"] - sk).abs().argsort()[:1]] lr = sub.iloc[(sub["strike"] - lk).abs().argsort()[:1]] if len(sr) and len(lr): sm, lm = float(sr.iloc[0]["mid"]), float(lr.iloc[0]["mid"]) if math.isfinite(sm) and math.isfinite(lm) and sm > 0: cost = (_slipped(sm, "buy") - _slipped(lm, "sell")) * MULT return legs["credit"] - cost short_itm = max(sk - spot, 0) * MULT long_itm = max(lk - spot, 0) * MULT return legs["credit"] - (short_itm - long_itm) def _build_iron_condor(chain: pd.DataFrame, spot: float, exp: pd.Timestamp, wing_pct: float, body_pct: float): """Short OTM put + short OTM call with wings.""" put_short = _nearest_strike(chain, spot * (1 - body_pct), "P", exp) put_long = _nearest_strike(chain, spot * (1 - body_pct - wing_pct), "P", exp) call_short = _nearest_strike(chain, spot * (1 + body_pct), "C", exp) call_long = _nearest_strike(chain, spot * (1 + body_pct + wing_pct), "C", exp) if any(x is None for x in (put_short, put_long, call_short, call_long)): return None psk, plk = float(put_short["strike"]), float(put_long["strike"]) csk, clk = float(call_short["strike"]), float(call_long["strike"]) if plk >= psk or clk <= csk: return None credit = ( _slipped(float(put_short["mid"]), "sell") - _slipped(float(put_long["mid"]), "buy") + _slipped(float(call_short["mid"]), "sell") - _slipped(float(call_long["mid"]), "buy") ) * MULT if credit <= 0: return None put_width = (psk - plk) * MULT call_width = (clk - csk) * MULT max_loss = max(put_width, call_width) - credit return { "credit": credit, "max_loss": max_loss, "put_short_k": psk, "put_long_k": plk, "call_short_k": csk, "call_long_k": clk, "kind": "iron_condor", } def _exit_iron_condor(chain: pd.DataFrame, spot: float, legs: dict, exp: pd.Timestamp) -> float: def leg_val(k: float, right: str, side: str) -> float: sub = chain[(chain["right_code"] == right) & (chain["expiration_dt"] == exp)] if not chain.empty else pd.DataFrame() if not sub.empty: r = sub.iloc[(sub["strike"] - k).abs().argsort()[:1]] if len(r): m = float(r.iloc[0]["mid"]) if math.isfinite(m) and m > 0: return _slipped(m, side) * MULT if right == "P": return max(k - spot, 0) * MULT * (1 if side == "sell" else -1) return max(spot - k, 0) * MULT * (1 if side == "sell" else -1) close_cost = ( leg_val(legs["put_short_k"], "P", "buy") - leg_val(legs["put_long_k"], "P", "sell") + leg_val(legs["call_short_k"], "C", "buy") - leg_val(legs["call_long_k"], "C", "sell") ) return legs["credit"] - close_cost def _build_put_calendar(chain: pd.DataFrame, spot: float, near_exp: pd.Timestamp, far_exp: pd.Timestamp, strike_mny: float): """Sell near-term put, buy far-term put at similar strike (calendar).""" k_tgt = spot * strike_mny sell_row = _nearest_strike(chain, k_tgt, "P", near_exp) buy_row = _nearest_strike(chain, k_tgt, "P", far_exp) if sell_row is None or buy_row is None: return None sk = float(sell_row["strike"]) debit = (_slipped(float(buy_row["mid"]), "buy") - _slipped(float(sell_row["mid"]), "sell")) * MULT if debit <= 0: return None return { "debit": debit, "strike": sk, "near_exp": near_exp, "far_exp": far_exp, "kind": "put_calendar", } def _exit_put_calendar(chain: pd.DataFrame, spot: float, legs: dict, _exp: pd.Timestamp) -> float: sk = legs["strike"] near, far = legs["near_exp"], legs["far_exp"] sell_val = buy_val = 0.0 for exp, which in ((near, "sell"), (far, "buy")): sub = chain[(chain["right_code"] == "P") & (chain["expiration_dt"] == exp)] if not chain.empty else pd.DataFrame() if not sub.empty: r = sub.iloc[(sub["strike"] - sk).abs().argsort()[:1]] if len(r): m = float(r.iloc[0]["mid"]) if math.isfinite(m) and m > 0: if which == "sell": sell_val = _slipped(m, "buy") * MULT else: buy_val = _slipped(m, "sell") * MULT if which == "sell" and sell_val == 0: sell_val = max(sk - spot, 0) * MULT if which == "buy" and buy_val == 0: buy_val = max(sk - spot, 0) * MULT * (1 - SLIPPAGE) return buy_val - sell_val - legs["debit"] def _pick_two_expiries(chain: pd.DataFrame, near_min: int, near_max: int, far_min: int, far_max: int): near = _pick_expiry(chain, near_min, near_max) if near is None: return None, None far_cands = chain[(chain["dte"] >= far_min) & (chain["dte"] <= far_max) & (chain["expiration_dt"] > near)] if far_cands.empty: return None, None far = far_cands.sort_values("dte")["expiration_dt"].iloc[0] return near, far # --------------------------------------------------------------------------- # Strategy spec # --------------------------------------------------------------------------- @dataclass class StratSpec: id: str structure: str regime: str # contango_steep | contango_mild | backwardation | term_flattening builder: Callable[..., dict | None] exiter: Callable[..., float] builder_kwargs: dict regime_kw: dict dte_min: int = 21 dte_max: int = 45 hold_days: int = 20 rebalance_every: int = 10 stop_mult: float = 2.0 profit_take_frac: float | None = None # fraction of max profit / debit target to exit early use_calendar: bool = False def _regime_ok(regime: str, ts: float, ts_prev: float, kw: dict) -> bool: if not math.isfinite(ts): return False if regime == "contango_steep": return ts >= float(kw.get("min_ratio", 0.10)) if regime == "contango_mild": lo, hi = float(kw.get("lo", 0.04)), float(kw.get("hi", 0.14)) return lo <= ts < hi if regime == "backwardation": return ts <= float(kw.get("max_ratio", 0.02)) if regime == "term_flattening": return math.isfinite(ts_prev) and (ts_prev - ts) >= float(kw.get("min_drop", 0.015)) if regime == "contango_any": return ts >= float(kw.get("min_ratio", 0.05)) return False def run_one_spec( spec: StratSpec, ct: pd.DataFrame, dates: list[pd.Timestamp], *, risk_budget_usd: float = 3_000.0, daily_mtm_records: list[dict] | None = None, capital: float = START_CAPITAL, ) -> list[Trade]: trades: list[Trade] = [] pending: dict | None = None days_held = 0 n = len(dates) ts_prev = float("nan") cum_realized = 0.0 for step, d in enumerate(dates): try: d = pd.Timestamp(d).normalize() row = ct.loc[d] if d in ct.index else None ts = float(row["vx3_vx1_ratio_ffill"]) if row is not None else float("nan") cr_vx2 = float(row.get("contango_ratio_ffill", np.nan)) if row is not None else float("nan") if pending is not None: days_held += 1 p = pending exp = p.get("expiration") or p["legs"].get("far_exp") dte_left = int((pd.Timestamp(exp) - d).days) if exp is not None else 0 should_exit = days_held >= p["hold_target"] or dte_left <= 1 if not should_exit and p.get("max_loss") is not None: chain = _load_chain_cached(d) spot_now = _spot_from_chain(chain) if not chain.empty else p["vxx_entry"] pnl_one = _exit_position(spec, chain, spot_now, p["legs"], p.get("expiration")) pnl_pos = float(pnl_one) * int(p["contracts"]) if pnl_pos <= -float(p["max_loss"]) * spec.stop_mult: should_exit = True elif spec.profit_take_frac is not None: tgt = p.get("profit_target_usd") if tgt is not None and pnl_pos >= float(tgt) * float(spec.profit_take_frac): should_exit = True if should_exit: chain = _load_chain_cached(d) spot_now = _spot_from_chain(chain) if not chain.empty else p["vxx_entry"] pnl_one = _exit_position(spec, chain, spot_now, p["legs"], p.get("expiration")) pnl = float(pnl_one) * int(p["contracts"]) scale = float(p["scale"]) pnl_s = pnl * scale cum_realized += pnl_s trades.append( Trade( strategy=spec.id, entry_date=str(p["entry_date"]), exit_date=str(d.date()), exit_reason="time", vxx_entry=p["vxx_entry"], vxx_exit=spot_now, entry_credit_or_debit=float(p["entry_val"]) * scale, exit_value=pnl_s + float(p["entry_val"]) * scale, pnl_total=pnl_s, contango_ratio=ts if math.isfinite(ts) else 0.0, vix3m_vix=cr_vx2 if math.isfinite(cr_vx2) else 0.0, broker_risk_usd=float(p["broker_risk_usd"]) * scale, contracts=int(p["contracts"]), ) ) pending = None days_held = 0 if step % spec.rebalance_every != 0 or pending is not None: ts_prev = ts continue if not _regime_ok(spec.regime, ts, ts_prev, spec.regime_kw): ts_prev = ts continue chain = _load_chain_cached(d) if chain.empty: ts_prev = ts continue spot = _spot_from_chain(chain) if spot is None: ts_prev = ts continue built = None exp_for_pending = None if spec.use_calendar: near, far = _pick_two_expiries(chain, 14, 28, spec.dte_min, spec.dte_max) if near is not None and far is not None: built = spec.builder(chain, spot, near, far, **spec.builder_kwargs) exp_for_pending = far else: exp = _pick_expiry(chain, spec.dte_min, spec.dte_max) if exp is not None: built = spec.builder(chain, spot, exp, **spec.builder_kwargs) exp_for_pending = exp if built is None: ts_prev = ts continue entry_one = float( built.get("credit", built.get("net_entry", -float(built.get("debit", 0.0)))) ) n_c, per_u, br_tot = resolve_vxx_contracts_and_broker_risk( strategy=spec.id, built=built, entry_val=entry_one, contracts=None, target_broker_risk_usd=risk_budget_usd, ) # Naked short call: size on ~15% underlying notional, not premium alone. if spec.structure == "short_call" and built.get("credit"): margin_risk = max(float(spot) * MULT * 0.15, float(built["credit"]) * 3.0) n_c = max(1, int(risk_budget_usd / margin_risk)) per_u = margin_risk br_tot = per_u * n_c scale = risk_budget_usd / max(br_tot, 1.0) days_to_exp = sum(1 for dd in dates if d < dd <= exp_for_pending) - 1 ht = min(spec.hold_days, max(days_to_exp, 1)) ml_one = built.get("max_loss") profit_target = None if built.get("credit") is not None: profit_target = float(built["credit"]) * n_c elif built.get("max_profit") is not None: profit_target = float(built["max_profit"]) * n_c elif built.get("debit") is not None: profit_target = float(built["debit"]) * n_c pending = { "entry_date": d.date(), "expiration": exp_for_pending, "legs": built, "vxx_entry": spot, "entry_val": entry_one * n_c, "hold_target": ht, "max_loss": float(ml_one) * n_c if ml_one is not None else None, "profit_target_usd": profit_target, "broker_risk_usd": br_tot, "contracts": n_c, "scale": scale, } days_held = 0 ts_prev = ts finally: if daily_mtm_records is not None: ur = 0.0 if pending is not None: chain = _load_chain_cached(d) spot_now = _spot_from_chain(chain) if not chain.empty else pending["vxx_entry"] pnl_one = _exit_position( spec, chain, spot_now, pending["legs"], pending.get("expiration") ) ur = float(pnl_one) * int(pending["contracts"]) * float(pending["scale"]) eq = float(capital) + cum_realized + ur daily_mtm_records.append( { "date": d, "realized_cum_usd": round(cum_realized, 2), "unrealized_usd": round(ur, 2), "equity_mtm_usd": round(eq, 2), } ) return trades def _exit_position(spec: StratSpec, chain: pd.DataFrame, spot: float, legs: dict, exp) -> float: if legs.get("kind") == "bear_put": return _exit_bear_put_credit(chain, spot, legs, exp) if legs.get("kind") == "iron_condor": return _exit_iron_condor(chain, spot, legs, exp) if legs.get("kind") == "put_calendar": return _exit_put_calendar(chain, spot, legs, exp) return spec.exiter(chain, spot, legs, exp) def daily_mtm_metrics( daily_mtm_records: list[dict], *, capital: float = START_CAPITAL, ) -> tuple[dict, pd.DataFrame]: """Metrics from daily mark-to-market equity (open positions marked each session).""" if not daily_mtm_records: return {"sharpe": 0.0, "n": 0, "metric_mode": "mtm"}, pd.DataFrame() df = pd.DataFrame(daily_mtm_records).sort_values("date") df["date"] = pd.to_datetime(df["date"]).dt.normalize() df = df.drop_duplicates(subset=["date"], keep="last").set_index("date") df["daily_pnl_mtm_usd"] = df["equity_mtm_usd"].diff().fillna(0.0) dr = df["daily_pnl_mtm_usd"] sd = float(dr.std()) sharpe = (float(dr.mean()) / sd * np.sqrt(252)) if sd > 1e-12 else 0.0 peak = df["equity_mtm_usd"].cummax() dd_pct = float(((df["equity_mtm_usd"] - peak) / peak.replace(0, np.nan)).fillna(0).min()) * 100 end_eq = float(df["equity_mtm_usd"].iloc[-1]) years = max((df.index[-1] - df.index[0]).days / 365.25, 1e-9) cagr = ((end_eq / capital) ** (1 / years) - 1) * 100 if capital > 0 else 0.0 meta = { "metric_mode": "mtm", "sharpe": round(sharpe, 3), "return_pct": round((end_eq / capital - 1) * 100, 2), "cagr_pct": round(cagr, 2), "max_dd_pct": round(dd_pct, 2), "end_equity_mtm_usd": round(end_eq, 2), "n_days": len(df), } return meta, df.reset_index() def daily_equity_metrics(trades: list[Trade], dates: list[pd.Timestamp], capital: float = START_CAPITAL) -> dict: if not trades: return {"sharpe": 0.0, "n": 0} pnl_by_day: dict[pd.Timestamp, float] = {} for t in trades: ed = pd.Timestamp(t.exit_date).normalize() pnl_by_day[ed] = pnl_by_day.get(ed, 0.0) + float(t.pnl_total) idx = pd.DatetimeIndex(dates) daily_pnl = pd.Series(0.0, index=idx) for d, v in pnl_by_day.items(): if d in daily_pnl.index: daily_pnl.loc[d] = v eq = capital + daily_pnl.cumsum() dr = eq.diff().fillna(0) sd = float(dr.std()) sharpe = (float(dr.mean()) / sd * np.sqrt(252)) if sd > 1e-12 else 0.0 peak = eq.cummax() dd_pct = float(((eq - peak) / peak.replace(0, np.nan)).fillna(0).min()) * 100 total_ret = float(eq.iloc[-1] - capital) years = max((idx[-1] - idx[0]).days / 365.25, 1e-9) cagr = ((eq.iloc[-1] / capital) ** (1 / years) - 1) * 100 if capital > 0 else 0.0 wins = sum(1 for t in trades if t.pnl_total > 0) return { "sharpe": round(sharpe, 3), "n": len(trades), "total_pnl": round(total_ret, 2), "return_pct": round(total_ret / capital * 100, 2), "cagr_pct": round(cagr, 2), "max_dd_pct": round(dd_pct, 2), "win_rate": round(wins / len(trades), 4) if trades else 0.0, "end_equity": round(float(eq.iloc[-1]), 2), } def build_fine_grid() -> list[StratSpec]: """Dense search on structures that nearly cleared Sharpe 1.5 in coarse sweep.""" grid: list[StratSpec] = [] sid = 0 def add(**kw): nonlocal sid sid += 1 kw["id"] = f"f{sid:03d}_{kw['structure']}_{kw['regime']}" grid.append(StratSpec(**kw)) for min_r, hold, sm, wp, rb_every, stop in itertools.product( [0.10, 0.12, 0.14, 0.16, 0.18], [8, 10, 12, 15], [1.05, 1.07, 1.10], [0.12, 0.15, 0.18], [5, 7], [1.25, 1.5], ): add( structure="bear_call", regime="contango_steep", builder=lambda c, s, e, sm=sm, wp=wp: _build_bear_call_credit(c, s, e, wp, sm), exiter=_exit_bear_call_from_chain, builder_kwargs={}, regime_kw={"min_ratio": min_r}, hold_days=hold, rebalance_every=rb_every, stop_mult=stop, ) for min_r, hold, body, wing in itertools.product( [0.10, 0.12, 0.14, 0.16], [10, 12, 15], [0.04, 0.05, 0.06], [0.08, 0.10, 0.12], ): add( structure="iron_condor", regime="contango_steep", builder=lambda c, s, e, body=body, wing=wing: _build_iron_condor(c, s, e, wing, body), exiter=_exit_iron_condor, builder_kwargs={}, regime_kw={"min_ratio": min_r}, hold_days=hold, dte_min=28, dte_max=50, ) for band, hold in itertools.product( [(0.05, 0.12), (0.06, 0.13), (0.07, 0.14)], [12, 15, 18], ): lo_b, hi_b = band add( structure="bear_put", regime="contango_mild", builder=lambda c, s, e: _build_bear_put_credit(c, s, e, 0.10, 0.98), exiter=_exit_bear_put_credit, builder_kwargs={}, regime_kw={"lo": lo_b, "hi": hi_b}, hold_days=hold, ) for min_drop, hold in itertools.product([0.010, 0.015, 0.020, 0.025], [8, 10, 12, 15]): add( structure="bear_call", regime="term_flattening", builder=lambda c, s, e: _build_bear_call_credit(c, s, e, 0.12, 1.05), exiter=_exit_bear_call_from_chain, builder_kwargs={}, regime_kw={"min_drop": min_drop}, hold_days=hold, rebalance_every=5, ) for max_r, hold, lotm, width in itertools.product( [-0.02, 0.0, 0.02], [5, 7, 10], [0.02, 0.05], [0.12, 0.18], ): add( structure="bull_call", regime="backwardation", builder=lambda c, s, e, lotm=lotm, width=width: _build_bull_call_spread(c, s, e, lotm, width), exiter=_exit_bull_call_spread, builder_kwargs={}, regime_kw={"max_ratio": max_r}, hold_days=hold, rebalance_every=5, dte_min=14, dte_max=35, ) return grid def build_sweep_grid() -> list[StratSpec]: """Generate 100+ parameter / structure combinations.""" grid: list[StratSpec] = [] sid = 0 def add(structure, regime, builder, exiter, bkw, rkw, **kw): nonlocal sid sid += 1 grid.append( StratSpec( id=f"s{sid:03d}_{structure}_{regime}", structure=structure, regime=regime, builder=builder, exiter=exiter, builder_kwargs=bkw, regime_kw=rkw, **kw, ) ) for min_r, hold, reb, dte_lo, dte_hi in itertools.product( [0.06, 0.08, 0.10, 0.12], [15, 20], [10], [21], [45], ): for sm, wp in [(1.05, 0.15), (1.08, 0.18)]: add( "bear_call", "contango_steep", lambda c, s, e, sm=sm, wp=wp: _build_bear_call_credit(c, s, e, wp, sm), _exit_bear_call_from_chain, {}, {"min_ratio": min_r}, hold_days=hold, rebalance_every=reb, dte_min=dte_lo, dte_max=dte_hi, ) for min_r, hold, sm, wp in itertools.product([0.05, 0.07], [15, 20], [0.98, 1.00], [0.08, 0.10]): add( "bear_put", "contango_mild", lambda c, s, e, sm=sm, wp=wp: _build_bear_put_credit(c, s, e, wp, sm), _exit_bear_put_credit, {}, {"lo": 0.04, "hi": 0.14}, hold_days=hold, dte_min=21, dte_max=45, ) for min_r, body, wing in itertools.product([0.08, 0.10], [0.05, 0.07], [0.08, 0.10]): add( "iron_condor", "contango_steep", lambda c, s, e, body=body, wing=wing: _build_iron_condor(c, s, e, wing, body), _exit_iron_condor, {}, {"min_ratio": min_r}, hold_days=15, dte_min=25, dte_max=50, ) for max_r, hold in itertools.product([0.0, 0.02], [7, 10, 14]): add( "bull_call", "backwardation", lambda c, s, e: _build_bull_call_spread(c, s, e, 0.02, 0.12), _exit_bull_call_spread, {}, {"max_ratio": max_r}, hold_days=hold, dte_min=14, dte_max=35, rebalance_every=5, ) add( "long_call", "backwardation", lambda c, s, e: _build_long_call(c, s, e, 0.08), _exit_long_call, {}, {"max_ratio": max_r}, hold_days=hold, dte_min=14, dte_max=35, rebalance_every=5, ) for min_r in [0.07, 0.09]: add( "ratio_put", "contango_steep", lambda c, s, e: _build_ratio_put_1x2(c, s, e, 0.10), _exit_ratio_put, {}, {"min_ratio": min_r}, hold_days=20, ) for min_drop, hold in itertools.product([0.012, 0.018, 0.025], [12, 18]): add( "bear_call", "term_flattening", lambda c, s, e: _build_bear_call_credit(c, s, e, 0.12, 1.05), _exit_bear_call_from_chain, {}, {"min_drop": min_drop}, hold_days=hold, ) for min_r in [0.08, 0.10]: add( "put_calendar", "contango_steep", lambda c, s, ne, fe: _build_put_calendar(c, s, ne, fe, 1.0), _exit_put_calendar, {}, {"min_ratio": min_r}, hold_days=25, dte_min=45, dte_max=90, use_calendar=True, ) for min_r, otm in itertools.product([0.08, 0.10], [0.06, 0.10]): add( "short_call", "contango_steep", lambda c, s, e, otm=otm: _build_short_call(c, s, e, otm), _exit_short_call, {}, {"min_ratio": min_r}, hold_days=12, stop_mult=1.5, ) for min_r in [0.07, 0.09]: add( "collar", "contango_any", lambda c, s, e: _build_deep_put_long_call(c, s, e, 0.18, 0.08), _exit_deep_put_long_call, {}, {"min_ratio": min_r}, hold_days=25, ) return grid # Five curated structures (889-config fine sweep + profit-take tuning, 2020–2025). TOP_FIVE_DEFAULT: list[dict] = [ { "name": "SteepContango_IronCondor", "structure": "Iron condor (short OTM put + short OTM call, wing protection)", "thesis": "VX3/VX1−1 ≥ 8%: elevated futures curve, range-bound VXX; harvest premium both sides.", "spec_kwargs": dict( id="top1_iron_condor_steep", structure="iron_condor", regime="contango_steep", builder=lambda c, s, e: _build_iron_condor(c, s, e, 0.10, 0.06), exiter=_exit_iron_condor, builder_kwargs={}, regime_kw={"min_ratio": 0.08}, hold_days=12, dte_min=28, dte_max=50, profit_take_frac=0.5, stop_mult=1.5, ), }, { "name": "SteepContango_BearCallCredit", "structure": "Bear call credit spread (short call + long higher call)", "thesis": "Steep VX3 vs VX1 → structural VXX decay; sell call vertical, exit at 50% max profit.", "spec_kwargs": dict( id="top2_bear_call_steep", structure="bear_call", regime="contango_steep", builder=lambda c, s, e: _build_bear_call_credit(c, s, e, 0.18, 1.10), exiter=_exit_bear_call_from_chain, builder_kwargs={}, regime_kw={"min_ratio": 0.10}, hold_days=8, rebalance_every=7, stop_mult=1.25, profit_take_frac=0.5, ), }, { "name": "MildContango_BearPutCredit", "structure": "Bear put credit spread (short put + long lower put)", "thesis": "Moderate curve (VX3/VX1 in 7–14% band): sell put spread, VXX drift / stability.", "spec_kwargs": dict( id="top3_bear_put_mild", structure="bear_put", regime="contango_mild", builder=lambda c, s, e: _build_bear_put_credit(c, s, e, 0.12, 0.99), exiter=_exit_bear_put_credit, builder_kwargs={}, regime_kw={"lo": 0.07, "hi": 0.14}, hold_days=18, profit_take_frac=0.5, stop_mult=1.5, ), }, { "name": "SteepContango_ShortCall", "structure": "Short single OTM call (defined-risk via stop / profit target)", "thesis": "Steep futures contango; pure theta on OTM calls with early take-profit.", "spec_kwargs": dict( id="top4_short_call_steep", structure="short_call", regime="contango_steep", builder=lambda c, s, e: _build_short_call(c, s, e, 0.12), exiter=_exit_short_call, builder_kwargs={}, regime_kw={"min_ratio": 0.12}, hold_days=10, rebalance_every=7, profit_take_frac=0.5, stop_mult=1.25, ), }, { "name": "SteepContango_ShortCallWide", "structure": "Short wide OTM call (16% above spot)", "thesis": "Deepest OTM short calls when VX3/VX1−1 ≥ 8%; fewer assignments, harvest contango theta (tail risk capped in sim via stop).", "spec_kwargs": dict( id="top5_short_call_wide", structure="short_call", regime="contango_steep", builder=lambda c, s, e: _build_short_call(c, s, e, 0.16), exiter=_exit_short_call, builder_kwargs={}, regime_kw={"min_ratio": 0.08}, hold_days=12, rebalance_every=7, profit_take_frac=0.5, stop_mult=1.25, ), }, ] # Strategies 6–10: additional fine-sweep winners (distinct params / regimes). TOP_SIX_TO_TEN: list[dict] = [ { "name": "SteepContango_BearCall_Classic", "structure": "Bear call credit (fine-tune f035)", "thesis": "VX3/VX1−1 ≥ 10%; 10% OTM short call, 18% wing, 8d hold, 50% profit take.", "spec_kwargs": dict( id="top6_bear_call_f035", structure="bear_call", regime="contango_steep", builder=lambda c, s, e: _build_bear_call_credit(c, s, e, 0.18, 1.10), exiter=_exit_bear_call_from_chain, builder_kwargs={}, regime_kw={"min_ratio": 0.10}, hold_days=8, rebalance_every=7, stop_mult=1.25, profit_take_frac=0.5, ), }, { "name": "SteepContango_BearCall_Ultraselect", "structure": "Bear call credit (fine-tune f323)", "thesis": "VX3/VX1−1 ≥ 14%; fewer, higher-conviction entries; 50% profit take.", "spec_kwargs": dict( id="top7_bear_call_f323", structure="bear_call", regime="contango_steep", builder=lambda c, s, e: _build_bear_call_credit(c, s, e, 0.15, 1.07), exiter=_exit_bear_call_from_chain, builder_kwargs={}, regime_kw={"min_ratio": 0.14}, hold_days=8, rebalance_every=7, stop_mult=1.25, profit_take_frac=0.52, ), }, { "name": "SteepContango_IronCondor_Tight", "structure": "Iron condor (tighter body, wider wings)", "thesis": "VX3/VX1−1 ≥ 8%; 5% body / 12% wings, 12d hold.", "spec_kwargs": dict( id="top8_iron_condor_tight", structure="iron_condor", regime="contango_steep", builder=lambda c, s, e: _build_iron_condor(c, s, e, 0.12, 0.05), exiter=_exit_iron_condor, builder_kwargs={}, regime_kw={"min_ratio": 0.08}, hold_days=12, dte_min=28, dte_max=50, profit_take_frac=0.5, stop_mult=1.5, ), }, { "name": "SteepContango_ShortCall_10pct", "structure": "Short 10% OTM call", "thesis": "VX3/VX1−1 ≥ 8%; closer OTM for more premium vs #4/#5.", "spec_kwargs": dict( id="top9_short_call_10pct", structure="short_call", regime="contango_steep", builder=lambda c, s, e: _build_short_call(c, s, e, 0.10), exiter=_exit_short_call, builder_kwargs={}, regime_kw={"min_ratio": 0.08}, hold_days=12, rebalance_every=7, profit_take_frac=0.5, stop_mult=1.25, ), }, { "name": "MildContango_BearPut_18d", "structure": "Bear put credit (18d hold)", "thesis": "VX3/VX1 in 7–14% band; 18d hold, 12% put width.", "spec_kwargs": dict( id="top10_bear_put_18d", structure="bear_put", regime="contango_mild", builder=lambda c, s, e: _build_bear_put_credit(c, s, e, 0.12, 0.99), exiter=_exit_bear_put_credit, builder_kwargs={}, regime_kw={"lo": 0.07, "hi": 0.14}, hold_days=18, profit_take_frac=0.5, stop_mult=1.5, ), }, ] TOP_STRATEGIES_DEFAULT: list[dict] = TOP_FIVE_DEFAULT + TOP_SIX_TO_TEN def _spec_from_dict(kw: dict) -> StratSpec: return StratSpec(**kw) def main() -> None: ap = argparse.ArgumentParser(description="VXX options on VX1/VX3 futures term structure") ap.add_argument("--start", default="2020-01-01") ap.add_argument("--end", default="2025-12-31") ap.add_argument("--sweep", action="store_true", help="Run coarse grid search") ap.add_argument("--fine-sweep", action="store_true", help="Run dense fine grid (~400+ configs)") ap.add_argument("--min-sharpe", type=float, default=1.5) ap.add_argument("--risk-budget", type=float, default=3000.0, help="USD max loss / trade sizing") ap.add_argument("--capital", type=float, default=START_CAPITAL) ap.add_argument("--out-dir", type=Path, default=LOGS) ap.add_argument("--top-sweep-json", type=Path, default=LOGS / "vxx_vx1_vx3_sweep_top.json") ap.add_argument("--top-n", type=int, default=5, help="Number of curated strategies to run (max 10)") ap.add_argument( "--preload-chains", action="store_true", help="Cache all VXX chains for the window before running (recommended for 2016+)", ) args = ap.parse_args() args.top_n = max(1, min(int(args.top_n), len(TOP_STRATEGIES_DEFAULT))) ct = _ensure_vx3_panel(_load_contango()) all_dates = sorted(ct.index) dates = [pd.Timestamp(d).normalize() for d in all_dates if args.start <= str(d.date()) <= args.end] print(f"Window: {dates[0].date()} → {dates[-1].date()} ({len(dates)} days)", flush=True) valid_ts = ct.loc[dates[0]: dates[-1], "vx3_vx1_ratio_ffill"].dropna() print( f"VX3/VX1−1: median={valid_ts.median():.3f} " f"p10={valid_ts.quantile(0.1):.3f} p90={valid_ts.quantile(0.9):.3f}", flush=True, ) if args.sweep or args.fine_sweep: _CHAIN_CACHE.clear() print("Pre-loading VXX option chains …", flush=True) for j, d in enumerate(dates): _load_chain_cached(d) if j and j % 300 == 0: print(f" chains {j}/{len(dates)}", flush=True) print(f" cached {len(_CHAIN_CACHE)} session days", flush=True) grid = build_fine_grid() if args.fine_sweep else build_sweep_grid() print(f"Sweeping {len(grid)} configurations …", flush=True) rows = [] for i, spec in enumerate(grid): if (i + 1) % 20 == 0: print(f" [{i+1}/{len(grid)}] cache={len(_CHAIN_CACHE)}", flush=True) tr = run_one_spec(spec, ct, dates, risk_budget_usd=args.risk_budget) m = daily_equity_metrics(tr, dates, capital=args.capital) m["id"] = spec.id m["structure"] = spec.structure m["regime"] = spec.regime rows.append(m) df = pd.DataFrame(rows).sort_values("sharpe", ascending=False) top = df[df["sharpe"] >= args.min_sharpe] out = { "window": [args.start, args.end], "min_sharpe": args.min_sharpe, "n_configs": len(grid), "n_pass": int(len(top)), "top20": top.head(20).to_dict(orient="records"), } args.top_sweep_json.parent.mkdir(parents=True, exist_ok=True) args.top_sweep_json.write_text(json.dumps(out, indent=2)) print(f"\nPass Sharpe≥{args.min_sharpe}: {len(top)} / {len(grid)}") print(df.head(15).to_string(index=False)) print(f"\nSaved → {args.top_sweep_json}") return # Run top five (update from sweep file if present) specs_to_run = [] if args.top_sweep_json.is_file(): data = json.loads(args.top_sweep_json.read_text()) top20 = data.get("top20", []) seen_struct: set[str] = set() for row in top20: if float(row.get("sharpe", 0)) < args.min_sharpe: continue st = str(row.get("structure", "")) if st in seen_struct: continue seen_struct.add(st) # match grid id prefix to recover params — use defaults if missing match = next((s for s in build_sweep_grid() if s.id == row["id"]), None) if match is not None: specs_to_run.append(match) if len(specs_to_run) >= 5: break curated = TOP_STRATEGIES_DEFAULT[: args.top_n] if len(specs_to_run) < args.top_n: specs_to_run = [_spec_from_dict(t["spec_kwargs"]) for t in curated] if args.preload_chains or len(dates) > 2000: _CHAIN_CACHE.clear() print("Pre-loading VXX option chains …", flush=True) for j, d in enumerate(dates): _load_chain_cached(d) if j and j % 400 == 0: print(f" chains {j}/{len(dates)}", flush=True) print(f" cached {len(_CHAIN_CACHE)} session days", flush=True) summary_rows = [] args.out_dir.mkdir(parents=True, exist_ok=True) print("\n" + "=" * 88) print(f"TOP {args.top_n} VXX STRATEGIES (VX1 vs VX3 futures)") print("=" * 88) for i, meta in enumerate(curated): spec = specs_to_run[i] if i < len(specs_to_run) else _spec_from_dict(meta["spec_kwargs"]) trades = run_one_spec(spec, ct, dates, risk_budget_usd=args.risk_budget) m = daily_equity_metrics(trades, dates, capital=args.capital) m["name"] = meta["name"] m["structure_desc"] = meta["structure"] m["thesis"] = meta["thesis"] summary_rows.append(m) jpath = args.out_dir / f"{spec.id}_trades.jsonl" with jpath.open("w") as f: for t in trades: f.write(json.dumps(asdict(t)) + "\n") print(f"\n### {i+1}. {meta['name']}") print(f" Structure: {meta['structure']}") print(f" Thesis: {meta['thesis']}") print(f" Trades: {m['n']} | Return: {m['return_pct']:+.1f}% | CAGR: {m['cagr_pct']:+.1f}%") print(f" Sharpe: {m['sharpe']:.2f} | Max DD: {m['max_dd_pct']:.1f}% | Win rate: {m['win_rate']:.1%}") print(f" Log → {jpath}") manifest = args.out_dir / f"vxx_vx1_vx3_top{args.top_n}_summary.json" manifest.write_text(json.dumps(summary_rows, indent=2)) print(f"\nSummary → {manifest}") # Ranked table print("\n" + "-" * 88) print(f"{'#':<3} {'Strategy':<36} {'Sharpe':>7} {'Ret%':>8} {'MaxDD%':>8} {'Trades':>7}") print("-" * 88) for rank, row in enumerate(sorted(summary_rows, key=lambda r: -float(r["sharpe"])), 1): print( f"{rank:<3} {row['name']:<36} {row['sharpe']:>7.2f} " f"{row['return_pct']:>+7.1f} {row['max_dd_pct']:>7.1f} {row['n']:>7}" ) print("-" * 88) below = [r for r in summary_rows if r["sharpe"] < args.min_sharpe] if below: print( f"\nNote: {len(below)} strategy(ies) below Sharpe {args.min_sharpe} on this run — " f"re-run with --sweep to refresh parameters." ) if __name__ == "__main__": main()