#!/usr/bin/env python3 """ VXX **bull call spread** entered during low contango / backwardation as a crisis hedge for the VRP book. When contango collapses (VX2/VX1 - 1 drops below threshold), VXX is likely spiking → buy call spreads to capture upside. Sweeps entry thresholds and hold periods, then shows correlation with VRP loss days and combined portfolio impact. Example:: python RenTech/strategy_stack/backtest_vxx_backwardation_hedge.py """ from __future__ import annotations import json, math, sys from dataclasses import dataclass, asdict from pathlib import Path import numpy as np import pandas as pd _REPO = Path(__file__).resolve().parents[2] sys.path.insert(0, str(_REPO / "RenTech" / "strategy_stack")) from explore_vxx_decay_strategies import ( _load_contango, _load_chain, _spot_from_chain, _pick_expiry, _build_bull_call_spread, _exit_bull_call_spread, _build_long_call, _exit_long_call, _slipped, MULT, SLIPPAGE, Trade, ) from iv_mispricing_complement import _load_jsonl, _pnl_series_from_trades LOGS = _REPO / "RenTech" / "data" / "logs" def _run_backwardation_strategy( name: str, builder, exiter, ct: pd.DataFrame, dates: list[pd.Timestamp], *, contango_upper: float, dte_min: int = 14, dte_max: int = 35, hold_days: int = 10, rebalance_every: int = 5, ) -> tuple[list[Trade], pd.Series]: trades: list[Trade] = [] pending = None days_held = 0 equity = 0.0 daily_eq = {} for step, d in enumerate(dates): d = pd.Timestamp(d).normalize() ct_row = ct.loc[d] cr_val = float(ct_row.get("contango_ratio_ffill", np.nan)) # ── exit ── if pending is not None: days_held += 1 p = pending dte_left = int((p["expiration"] - d).days) if days_held >= p["hold_target"] or dte_left <= 1: chain = _load_chain(d) spot_now = _spot_from_chain(chain) if not chain.empty else None if spot_now is None: spot_now = p["vxx_entry"] pnl = exiter(chain, spot_now, p["legs"], p["expiration"]) equity += pnl deb = float(p["legs"].get("debit", 0) or 0) trades.append(Trade( strategy=name, 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=-deb, exit_value=pnl, pnl_total=pnl, contango_ratio=p.get("cr", 0), vix3m_vix=0, broker_risk_usd=max(deb, 1.0), )) pending = None days_held = 0 daily_eq[d] = equity # ── entry: only when contango is LOW (approaching/in backwardation) ── if step % rebalance_every != 0: continue if pending is not None: continue if not math.isfinite(cr_val): continue if cr_val > contango_upper: continue chain = _load_chain(d) if chain.empty: continue spot = _spot_from_chain(chain) if spot is None: continue exp = _pick_expiry(chain, dte_min, dte_max) if exp is None: continue built = builder(chain, spot, exp) if built is None: continue days_to_exp = sum(1 for dd in dates if d < dd <= exp) - 1 ht = min(hold_days, max(days_to_exp, 1)) pending = { "entry_date": d.date(), "expiration": exp, "legs": built, "vxx_entry": spot, "hold_target": ht, "cr": cr_val, } days_held = 0 return trades, pd.Series(daily_eq, name=name).sort_index() def _metrics(trades): if not trades: return {} pnls = [t.pnl_total for t in trades] wins = sum(1 for p in pnls if p > 0) cum = np.cumsum(pnls) peak = np.maximum.accumulate(cum) dd = cum - peak return { "n": len(trades), "total_pnl": round(sum(pnls), 1), "avg_pnl": round(float(np.mean(pnls)), 1), "win_rate": round(wins / len(trades), 3), "best": round(max(pnls), 1), "worst": round(min(pnls), 1), "max_dd": round(float(dd.min()), 1), } def main(): ct = _load_contango() all_dates = sorted(ct.index) dates = [d for d in all_dates if "2018-06-01" <= str(d.date()) <= "2025-12-31"] print(f"Dates: {dates[0].date()} → {dates[-1].date()} ({len(dates)} days)\n") # ── sweep: threshold × hold_days × structure ── structures = { "bull_call_5_15": lambda c, s, e: _build_bull_call_spread(c, s, e, 0.05, 0.15), "bull_call_0_10": lambda c, s, e: _build_bull_call_spread(c, s, e, 0.00, 0.10), "bull_call_0_20": lambda c, s, e: _build_bull_call_spread(c, s, e, 0.00, 0.20), "long_call_5pct": lambda c, s, e: _build_long_call(c, s, e, 0.05), "long_call_10pct": lambda c, s, e: _build_long_call(c, s, e, 0.10), } exiters = { "bull_call_5_15": _exit_bull_call_spread, "bull_call_0_10": _exit_bull_call_spread, "bull_call_0_20": _exit_bull_call_spread, "long_call_5pct": _exit_long_call, "long_call_10pct": _exit_long_call, } thresholds = [0.02, 0.01, 0.00, -0.01, -0.02, -0.03] hold_options = [5, 10, 15] print(f"{'Structure':<20} {'Thresh':>7} {'Hold':>5} {'Trades':>6} {'TotalPnL':>9} {'AvgPnL':>7} {'WinRate':>8} {'Best':>7} {'Worst':>7} {'MaxDD':>7}") print("─" * 105) best_sharpe = -999 best_config = None all_results = [] for sname, builder in structures.items(): for thresh in thresholds: for hold in hold_options: trades, eq = _run_backwardation_strategy( sname, builder, exiters[sname], ct, dates, contango_upper=thresh, hold_days=hold, rebalance_every=5, ) m = _metrics(trades) if not m: continue pnls = [t.pnl_total for t in trades] sd = float(np.std(pnls)) if len(pnls) > 1 else 1e-9 sharpe = (float(np.mean(pnls)) / sd) * np.sqrt(26) if sd > 0 else 0 all_results.append({ "structure": sname, "threshold": thresh, "hold": hold, "sharpe": round(sharpe, 2), **m, }) if sharpe > best_sharpe: best_sharpe = sharpe best_config = (sname, thresh, hold, trades, eq) print( f"{sname:<20} {thresh:>+6.2f} {hold:>5} {m['n']:>6} " f"{m['total_pnl']:>+9.0f} {m['avg_pnl']:>+7.1f} " f"{m['win_rate']:>7.1%} {m['best']:>+7.0f} {m['worst']:>+7.0f} {m['max_dd']:>+7.0f}" ) if best_config is None: print("\nNo viable configs found.") return bname, bthresh, bhold, btrades, beq = best_config print(f"\n★ Best Sharpe ({best_sharpe:.2f}): {bname} threshold={bthresh:+.2f} hold={bhold}") # ── correlation with VRP loss days ── vrp = _load_jsonl(LOGS / "vrp_trades.jsonl") vrp_pnl = _pnl_series_from_trades(vrp, exit_key="exit_date", pnl_key="pnl_usd") idx = vrp_pnl.index.union(beq.index).sort_values() vrp_daily = vrp_pnl.reindex(idx, fill_value=0) hedge_daily = beq.diff().reindex(idx).fillna(0) corr = float(vrp_daily.corr(hedge_daily)) print(f"\nDaily PnL correlation with VRP: {corr:+.3f}") # PnL of hedge on VRP's worst days vrp_loss_days = vrp_daily[vrp_daily < -1000].index hedge_on_vrp_loss = hedge_daily.reindex(vrp_loss_days).fillna(0) print(f"Hedge PnL on VRP's worst days (VRP < -$1K, n={len(vrp_loss_days)}):") print(f" Mean: {hedge_on_vrp_loss.mean():+.1f}") print(f" Sum: {hedge_on_vrp_loss.sum():+.1f}") # Save best config trades out = LOGS / "vxx_backwardation_hedge.jsonl" out.parent.mkdir(parents=True, exist_ok=True) with out.open("w") as f: for t in btrades: f.write(json.dumps(asdict(t)) + "\n") print(f"\nSaved {len(btrades)} trades → {out}") if __name__ == "__main__": main()