#!/usr/bin/env python3 """ **VXX Bear-Put Debit Spread** — buy a near-ATM VXX put, sell a further-OTM VXX put (same expiry) — filtered by **VIX term-structure contango** from CBOE. Thesis: VXX structurally decays when VIX futures are in contango because VXX rolls from higher-priced back-month futures toward lower front-month/spot. Put spreads capture the decay with defined risk and no margin requirement. Data ---- * VXX 15:45 ET option snapshots (``vxx_1545_YYYY_MM.parquet``) * VIX futures contango panel (``vix_futures_cboe.parquet``, built by ``download_cboe_vix_futures.py``) * VXX underlying close from yfinance (or inferred from option chain) Entry ----- * ``contango_flag_ffill == 1`` **or** ``vix3m_vix_ratio > --vix3m-threshold`` * Pick the expiry in [``--dte-min``, ``--dte-max``] range * Long put: strike closest to ``VXX_close * --long-put-moneyness`` * Short put: strike closest to ``long_strike − --width`` (further OTM) Exit ---- * Hold for ``--hold-days`` trading days **or** 1 day before expiry, whichever is earlier. Mid-price mark-to-market on exit day. * Optional ``--take-profit-pct`` (close if spread reaches X% of max profit) * Optional ``--stop-loss-pct`` (close if spread loses X% of max loss) Example:: python RenTech/strategy_stack/backtest_vxx_put_spread_contango.py \\ --start 2018-06-01 --end 2025-12-31 \\ --out-trades RenTech/data/logs/vxx_put_spread_contango.jsonl python RenTech/strategy_stack/backtest_vxx_put_spread_contango.py \\ --start 2018-06-01 --end 2025-12-31 \\ --contango-mode vix3m --vix3m-threshold 1.05 \\ --hold-days 10 --width 3.0 \\ --out-trades RenTech/data/logs/vxx_put_spread_contango.jsonl """ from __future__ import annotations import argparse import json import math import sys from dataclasses import asdict, dataclass, field from pathlib import Path import numpy as np import pandas as pd _REPO = Path(__file__).resolve().parents[2] DATA_DIR = _REPO / "RenTech" / "data" THETA_DIR = DATA_DIR / "theta_chunks" CONTANGO_PATH = DATA_DIR / "vix_futures_cboe.parquet" SLIPPAGE_PCT = 0.005 # --------------------------------------------------------------------------- # Data helpers # --------------------------------------------------------------------------- def _load_contango() -> pd.DataFrame: if not CONTANGO_PATH.is_file(): print( f"ERROR: {CONTANGO_PATH} not found. Run download_cboe_vix_futures.py first.", file=sys.stderr, ) sys.exit(1) ct = pd.read_parquet(CONTANGO_PATH) ct.index = pd.to_datetime(ct.index) return ct def _estimate_spot_from_chain(chain: pd.DataFrame) -> float | None: """Infer VXX spot from near-term ATM put-call parity (avoids split-adjusted yfinance).""" if chain.empty: return None near = chain.copy() qt0 = pd.Timestamp(near["quote_datetime"].iloc[0]) sess = qt0.tz_localize(None).normalize() if qt0.tzinfo is None else qt0.tz_convert(None).normalize() exp_naive = near["expiration_dt"].dt.tz_localize(None) near["dte"] = (exp_naive - sess).dt.days near = near[(near["dte"] >= 7) & (near["dte"] <= 60)] if near.empty: return None # Use the nearest expiry min_exp = near.sort_values("dte")["expiration_dt"].iloc[0] atm_set = near[near["expiration_dt"] == min_exp] calls = atm_set[atm_set["right_code"] == "C"] puts = atm_set[atm_set["right_code"] == "P"] if calls.empty or puts.empty: return None # Merge on strike, compute synthetic spot = K + call_mid - put_mid merged = calls[["strike", "mid"]].merge( puts[["strike", "mid"]], on="strike", suffixes=("_c", "_p"), ) if merged.empty: return None merged["synth_spot"] = merged["strike"] + merged["mid_c"] - merged["mid_p"] merged["spread"] = (merged["mid_c"] - merged["mid_p"]).abs() best = merged.sort_values("spread").iloc[0] spot = float(best["synth_spot"]) if spot <= 0 or not math.isfinite(spot): return None return spot def _load_vxx_prices(start: str, end: str) -> pd.Series: """VXX daily close from yfinance (timezone-naive DatetimeIndex) — used only for dates.""" import yfinance as yf raw = yf.download("VXX", start=start, end=end, progress=False) if raw.empty: print("ERROR: no VXX price data from yfinance", file=sys.stderr) sys.exit(1) close = raw["Close"].squeeze() close.index = pd.to_datetime(close.index).tz_localize(None) close.name = "vxx_close" return close def _session_date(qt: pd.Series) -> pd.Series: qd = pd.to_datetime(qt, utc=False) if qd.dt.tz is None: qd = qd.dt.tz_localize("America/New_York", ambiguous="NaT", nonexistent="shift_forward") else: qd = qd.dt.tz_convert("America/New_York") return pd.to_datetime(qd.dt.date) def _load_vxx_chain(d: pd.Timestamp) -> pd.DataFrame: """Load VXX option rows for a single session date from monthly parquet.""" y, m = d.year, d.month path = THETA_DIR / f"vxx_1545_{y:04d}_{m:02d}.parquet" if not path.is_file(): return pd.DataFrame() df = pd.read_parquet(path) if df.empty: return df sess = _session_date(df["quote_datetime"]) df = df.loc[sess == pd.Timestamp(d.date())].copy() if df.empty: return df # Scale strikes: raw values are /10 (e.g. 1.4 → $14) strike = pd.to_numeric(df["strike"], errors="coerce") if float(strike.max(skipna=True)) < 150: strike = strike * 10.0 df["strike"] = strike df["mid"] = 0.5 * (pd.to_numeric(df["bid"], errors="coerce") + pd.to_numeric(df["ask"], errors="coerce")) df["right_code"] = df["right"].astype(str).str.upper().str.strip().str[0] df["expiration_dt"] = pd.to_datetime(df["expiration"]).dt.normalize() return df # --------------------------------------------------------------------------- # Trade record # --------------------------------------------------------------------------- @dataclass class VxxPutSpreadTrade: entry_date: str exit_date: str exit_reason: str expiration: str strike_long: float strike_short: float vxx_entry: float vxx_exit: float entry_debit: float exit_credit: float pnl_total: float max_profit: float max_loss: float contango_ratio: float vix3m_vix: float hold_days_actual: int # --------------------------------------------------------------------------- # Spread selection # --------------------------------------------------------------------------- def _pick_put_spread( chain: pd.DataFrame, vxx_px: float, *, dte_min: int, dte_max: int, long_moneyness: float, width: float | None, width_pct: float | None, ) -> dict | None: """Select best put spread: long near-ATM put, short further-OTM put.""" puts = chain[chain["right_code"] == "P"].copy() if puts.empty: return None qt0 = pd.Timestamp(puts["quote_datetime"].iloc[0]) sess = qt0.tz_localize(None).normalize() if qt0.tzinfo is None else qt0.tz_convert(None).normalize() exp_naive = puts["expiration_dt"].dt.tz_localize(None) puts["dte"] = (exp_naive - sess).dt.days eligible = puts[(puts["dte"] >= dte_min) & (puts["dte"] <= dte_max)] if eligible.empty: return None # Prefer the closest expiry in the DTE band target_exp = eligible.sort_values("dte")["expiration_dt"].iloc[0] exp_puts = eligible[eligible["expiration_dt"] == target_exp].copy() if len(exp_puts) < 2: return None dte = int(exp_puts["dte"].iloc[0]) target_long_k = vxx_px * long_moneyness exp_puts["dist_long"] = (exp_puts["strike"] - target_long_k).abs() long_row = exp_puts.sort_values("dist_long").iloc[0] long_k = float(long_row["strike"]) long_mid = float(long_row["mid"]) if not (math.isfinite(long_mid) and long_mid > 0): return None # Resolve width: percentage of VXX price, or fixed dollar effective_width = width_pct * vxx_px if width_pct else (width or 5.0) effective_width = max(effective_width, 1.0) target_short_k = long_k - effective_width candidates = exp_puts[exp_puts["strike"] < long_k - 0.01] if candidates.empty: return None candidates = candidates.copy() candidates["dist_short"] = (candidates["strike"] - target_short_k).abs() short_row = candidates.sort_values("dist_short").iloc[0] short_k = float(short_row["strike"]) short_mid = float(short_row["mid"]) if not (math.isfinite(short_mid) and short_mid > 0): return None if short_k >= long_k: return None debit = (long_mid - short_mid) * 100 # per contract spread_width = long_k - short_k max_profit = (spread_width * 100) - debit max_loss = debit if debit <= 0 or max_profit <= 0: return None return { "expiration": target_exp, "dte": dte, "long_k": long_k, "short_k": short_k, "long_mid_entry": long_mid, "short_mid_entry": short_mid, "debit": debit, "max_profit": max_profit, "max_loss": max_loss, } def _mark_spread(chain: pd.DataFrame, long_k: float, short_k: float, exp: pd.Timestamp) -> float | None: """Mark the spread to mid on exit day; return credit (per contract).""" puts = chain[(chain["right_code"] == "P") & (chain["expiration_dt"] == exp)] if puts.empty: return None long_rows = puts[(puts["strike"] - long_k).abs() < 0.01] short_rows = puts[(puts["strike"] - short_k).abs() < 0.01] if long_rows.empty or short_rows.empty: return None l_mid = float(long_rows.iloc[0]["mid"]) s_mid = float(short_rows.iloc[0]["mid"]) if not (math.isfinite(l_mid) and math.isfinite(s_mid)): return None credit = (l_mid - s_mid) * 100 return credit # --------------------------------------------------------------------------- # Main backtest loop # --------------------------------------------------------------------------- def run_backtest( *, contango_df: pd.DataFrame, vxx_close: pd.Series, start: str, end: str, contango_mode: str, vix3m_threshold: float, min_contango_ratio: float, max_vix: float, dte_min: int, dte_max: int, long_moneyness: float, width: float | None, width_pct: float | None, hold_days: int, rebalance_every: int, take_profit_pct: float, stop_loss_pct: float, slippage: float, ) -> list[VxxPutSpreadTrade]: # Use trading days from the contango panel (VIX spot has full daily coverage) all_dates = sorted(contango_df.index) dates = [d for d in all_dates if str(start) <= str(d.date()) <= str(end)] if not dates: print("ERROR: no overlapping dates", file=sys.stderr) return [] trades: list[VxxPutSpreadTrade] = [] pending: dict | None = None days_since_entry = 0 n_dates = len(dates) log_every = max(1, n_dates // 20) for step, d in enumerate(dates): d = pd.Timestamp(d).normalize() if step % log_every == 0: cum_pnl = sum(t.pnl_total for t in trades) print( f" [{step:>5}/{n_dates}] {d.date()} trades={len(trades)} cum_pnl=${cum_pnl:+,.0f}", flush=True, ) ct_row = contango_df.loc[d] # ---- check for exit of pending position ---- if pending is not None: days_since_entry += 1 dte_left = int((pending["expiration"] - d).days) reason = "" chain = _load_vxx_chain(d) vxx_px_exit = _estimate_spot_from_chain(chain) if not chain.empty else None # Try marking to market for early exits if days_since_entry < pending["hold_target"] and dte_left > 1 and not chain.empty: credit = _mark_spread(chain, pending["long_k"], pending["short_k"], pending["expiration"]) if credit is not None: pnl = credit - pending["debit"] if take_profit_pct > 0 and pnl >= pending["max_profit"] * take_profit_pct: reason = "take_profit" if not reason and stop_loss_pct > 0 and pnl <= -pending["max_loss"] * stop_loss_pct: reason = "stop_loss" # Time exit if not reason and (days_since_entry >= pending["hold_target"] or dte_left <= 1): reason = "time" if reason: credit = None if not chain.empty: credit = _mark_spread(chain, pending["long_k"], pending["short_k"], pending["expiration"]) if credit is None and vxx_px_exit is not None: long_intrinsic = max(pending["long_k"] - vxx_px_exit, 0.0) short_intrinsic = max(pending["short_k"] - vxx_px_exit, 0.0) credit = (long_intrinsic - short_intrinsic) * 100 elif credit is None: credit = 0.0 adj = credit * (1 - slippage) pnl = adj - pending["debit"] trades.append(VxxPutSpreadTrade( entry_date=str(pending["entry_date"]), exit_date=str(d.date()), exit_reason=reason, expiration=str(pending["expiration"].date()), strike_long=pending["long_k"], strike_short=pending["short_k"], vxx_entry=pending["vxx_entry"], vxx_exit=vxx_px_exit or 0.0, entry_debit=pending["debit"], exit_credit=adj, pnl_total=pnl, max_profit=pending["max_profit"], max_loss=pending["max_loss"], contango_ratio=pending["contango_ratio"], vix3m_vix=pending["vix3m_vix"], hold_days_actual=days_since_entry, )) pending = None days_since_entry = 0 continue # ---- entry logic ---- if step % rebalance_every != 0: continue # Check contango filter in_contango = False cr_val = float(ct_row.get("contango_ratio_ffill", np.nan)) v3v_val = float(ct_row.get("vix3m_vix_ratio", np.nan)) vix_spot = float(ct_row.get("vix_spot", np.nan)) if contango_mode == "futures": in_contango = math.isfinite(cr_val) and cr_val >= min_contango_ratio elif contango_mode == "vix3m": in_contango = math.isfinite(v3v_val) and v3v_val >= vix3m_threshold elif contango_mode == "either": f_ok = math.isfinite(cr_val) and cr_val >= min_contango_ratio v_ok = math.isfinite(v3v_val) and v3v_val >= vix3m_threshold in_contango = f_ok or v_ok if not in_contango: continue # Skip during VIX spikes (vol likely to expand further, hurting puts) if max_vix > 0 and math.isfinite(vix_spot) and vix_spot > max_vix: continue chain = _load_vxx_chain(d) if chain.empty: continue # Use chain-derived spot (actual traded price, not yfinance split-adjusted) vxx_px = _estimate_spot_from_chain(chain) if vxx_px is None: continue spread = _pick_put_spread( chain, vxx_px, dte_min=dte_min, dte_max=dte_max, long_moneyness=long_moneyness, width=width, width_pct=width_pct, ) if spread is None: continue # Apply slippage to entry debit (pay more) adj_debit = spread["debit"] * (1 + slippage) # How many trading days to hold exp = spread["expiration"] days_to_exp = sum(1 for dd in dates if d < dd <= exp) - 1 hold_target = min(hold_days, max(days_to_exp, 1)) pending = { "entry_date": d.date(), "expiration": exp, "long_k": spread["long_k"], "short_k": spread["short_k"], "debit": adj_debit, "max_profit": (spread["long_k"] - spread["short_k"]) * 100 - adj_debit, "max_loss": adj_debit, "vxx_entry": vxx_px, "contango_ratio": cr_val if math.isfinite(cr_val) else 0.0, "vix3m_vix": v3v_val if math.isfinite(v3v_val) else 0.0, "hold_target": hold_target, } days_since_entry = 0 return trades def main() -> None: ap = argparse.ArgumentParser( description="VXX bear-put debit spread, contango-filtered (CBOE VIX futures)." ) ap.add_argument("--start", type=str, default="2018-06-01") ap.add_argument("--end", type=str, default="2025-12-31") ap.add_argument( "--contango-mode", choices=("futures", "vix3m", "either"), default="either", help="Which contango signal: VX1/VX2 futures, VIX3M/VIX ratio, or either (default).", ) ap.add_argument("--min-contango-ratio", type=float, default=0.02, help="Min VX2/VX1 − 1 for futures mode (default 0.02 = 2%%).") ap.add_argument("--vix3m-threshold", type=float, default=1.05, help="Min VIX3M/VIX for vix3m mode (default 1.05).") ap.add_argument("--max-vix", type=float, default=25.0, help="Skip entry if VIX spot > this (0 = no filter). Default 25.") ap.add_argument("--dte-min", type=int, default=14) ap.add_argument("--dte-max", type=int, default=45) ap.add_argument("--long-put-moneyness", type=float, default=0.97, help="Long put strike / VXX close (default 0.97 = ~3%% OTM).") ap.add_argument("--width", type=float, default=None, help="Fixed dollar gap between strikes (overrides --width-pct).") ap.add_argument("--width-pct", type=float, default=0.10, help="Width as fraction of VXX price (default 0.10 = 10%%).") ap.add_argument("--hold-days", type=int, default=10) ap.add_argument("--rebalance-every", type=int, default=5, help="Only attempt entry every N trading days (default 5).") ap.add_argument("--take-profit-pct", type=float, default=0.50, help="Close if PnL >= X%% of max profit (0 = disabled).") ap.add_argument("--stop-loss-pct", type=float, default=0.80, help="Close if loss >= X%% of max loss (0 = disabled).") ap.add_argument("--slippage", type=float, default=SLIPPAGE_PCT) ap.add_argument("--out-trades", type=Path, default=None) ap.add_argument("--log-every", type=int, default=50, help="Print progress every N steps.") args = ap.parse_args() print("Loading contango panel …", flush=True) contango_df = _load_contango() vxx_close = pd.Series(dtype=float) # not used for pricing; kept for API compat w_desc = f"${args.width:.1f}" if args.width else f"{args.width_pct:.0%} of VXX" print( f"Running backtest {args.start} → {args.end} " f"contango_mode={args.contango_mode} hold_days={args.hold_days} " f"width={w_desc} moneyness={args.long_put_moneyness}", flush=True, ) trades = run_backtest( contango_df=contango_df, vxx_close=vxx_close, start=args.start, end=args.end, contango_mode=args.contango_mode, vix3m_threshold=args.vix3m_threshold, min_contango_ratio=args.min_contango_ratio, max_vix=args.max_vix, dte_min=args.dte_min, dte_max=args.dte_max, long_moneyness=args.long_put_moneyness, width=args.width, width_pct=args.width_pct if args.width is None else None, hold_days=args.hold_days, rebalance_every=args.rebalance_every, take_profit_pct=args.take_profit_pct, stop_loss_pct=args.stop_loss_pct, slippage=args.slippage, ) pnls = [t.pnl_total for t in trades] total = sum(pnls) if pnls else 0.0 wins = sum(1 for p in pnls if p > 0) losses = sum(1 for p in pnls if p <= 0) avg = float(np.mean(pnls)) if pnls else 0.0 median = float(np.median(pnls)) if pnls else 0.0 best = max(pnls) if pnls else 0.0 worst = min(pnls) if pnls else 0.0 exit_reasons: dict[str, int] = {} for t in trades: exit_reasons[t.exit_reason] = exit_reasons.get(t.exit_reason, 0) + 1 summary = { "strategy": "vxx_put_spread_contango", "period": f"{args.start} → {args.end}", "n_trades": len(trades), "total_pnl": round(total, 2), "avg_pnl": round(avg, 2), "median_pnl": round(median, 2), "win_rate": round(wins / len(trades), 4) if trades else 0.0, "wins": wins, "losses": losses, "best_trade": round(best, 2), "worst_trade": round(worst, 2), "exit_reasons": exit_reasons, "params": { "contango_mode": args.contango_mode, "min_contango_ratio": args.min_contango_ratio, "vix3m_threshold": args.vix3m_threshold, "dte_range": [args.dte_min, args.dte_max], "long_moneyness": args.long_put_moneyness, "width": args.width, "width_pct": args.width_pct, "hold_days": args.hold_days, "rebalance_every": args.rebalance_every, "take_profit_pct": args.take_profit_pct, "stop_loss_pct": args.stop_loss_pct, }, } print(json.dumps(summary, indent=2)) if args.out_trades: p = args.out_trades.expanduser() p.parent.mkdir(parents=True, exist_ok=True) with p.open("w", encoding="utf-8") as f: for t in trades: f.write(json.dumps(asdict(t)) + "\n") print(f"\nWrote {len(trades)} trades → {p}") if __name__ == "__main__": main()