#!/usr/bin/env python3 """ **VXX Bear-Call Credit Spread** — sell a slightly-OTM VXX call, buy a further-OTM VXX call (same expiry) — filtered by **VIX term-structure contango** from CBOE. Thesis: VXX structurally decays when VIX futures are in contango. Selling call credit spreads profits when VXX stays flat or falls — collecting theta while the structural roll cost grinds the underlying lower. Best parameters (from sweep over 2018-2025): short_moneyness=1.05 width_pct=0.15 hold_days=20 vix3m_threshold=1.08 dte 21-45 rebalance every 10d Results: +$1,361 total, 78% WR, 1.18 Sharpe, -$373 max DD, 74 trades. Data ---- * VXX 15:45 ET option snapshots (``vxx_1545_YYYY_MM.parquet``) * VIX futures contango panel (``vix_futures_cboe.parquet``) * VXX spot inferred from option chain put-call parity (no yfinance needed) Example:: python RenTech/strategy_stack/backtest_vxx_bear_call_contango.py \\ --start 2018-06-01 --end 2025-12-31 \\ --out-trades RenTech/data/logs/vxx_bear_call_contango.jsonl """ from __future__ import annotations import argparse import json import math import sys from dataclasses import asdict, dataclass 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 = 0.005 MULT = 100 # --------------------------------------------------------------------------- # 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 _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_chain(d: pd.Timestamp) -> pd.DataFrame: 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 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() qt0 = pd.Timestamp(df["quote_datetime"].iloc[0]) sess_ts = qt0.tz_localize(None).normalize() if qt0.tzinfo is None else qt0.tz_convert(None).normalize() exp_naive = df["expiration_dt"].dt.tz_localize(None) df["dte"] = (exp_naive - sess_ts).dt.days return df def _spot_from_chain(chain: pd.DataFrame) -> float | None: if chain.empty: return None near = chain[(chain["dte"] >= 7) & (chain["dte"] <= 60)] if near.empty: return None min_exp = near.sort_values("dte")["expiration_dt"].iloc[0] atm = near[near["expiration_dt"] == min_exp] calls = atm[atm["right_code"] == "C"][["strike", "mid"]] puts = atm[atm["right_code"] == "P"][["strike", "mid"]] if calls.empty or puts.empty: return None m = calls.merge(puts, on="strike", suffixes=("_c", "_p")) if m.empty: return None m["s"] = m["strike"] + m["mid_c"] - m["mid_p"] m["gap"] = (m["mid_c"] - m["mid_p"]).abs() spot = float(m.sort_values("gap").iloc[0]["s"]) return spot if (math.isfinite(spot) and spot > 0) else None def _nearest_strike(chain: pd.DataFrame, target: float, right: str, exp: pd.Timestamp) -> pd.Series | None: sub = chain[(chain["right_code"] == right) & (chain["expiration_dt"] == exp)] if sub.empty: return None idx = (sub["strike"] - target).abs().idxmin() row = sub.loc[idx] mid = float(row["mid"]) return row if (math.isfinite(mid) and mid > 0) else None # --------------------------------------------------------------------------- # Trade record # --------------------------------------------------------------------------- @dataclass class BearCallTrade: entry_date: str exit_date: str exit_reason: str expiration: str strike_short: float strike_long: float vxx_entry: float vxx_exit: float entry_credit: float exit_cost: float pnl_total: float max_profit: float max_loss: float contango_ratio: float vix3m_vix: float hold_days_actual: int # --------------------------------------------------------------------------- # Backtest # --------------------------------------------------------------------------- def run_backtest( *, start: str, end: str, short_moneyness: float, width_pct: float, hold_days: int, rebalance_every: int, contango_mode: str, contango_threshold: float, vix3m_threshold: float, dte_min: int, dte_max: int, take_profit_pct: float, stop_loss_pct: float, ) -> list[BearCallTrade]: ct = _load_contango() all_dates = sorted(ct.index) dates = [d for d in all_dates if start <= str(d.date()) <= end] if not dates: print("ERROR: no dates in range", file=sys.stderr) return [] trades: list[BearCallTrade] = [] pending: dict | None = None days_held = 0 n = len(dates) log_every = max(1, n // 20) for step, d in enumerate(dates): d = pd.Timestamp(d).normalize() v3v = float(ct.loc[d].get("vix3m_vix_ratio", np.nan)) cr_val = float(ct.loc[d].get("contango_ratio_ffill", np.nan)) if step % log_every == 0: cum = sum(t.pnl_total for t in trades) print(f" [{step:>5}/{n}] {d.date()} trades={len(trades)} cum_pnl=${cum:+,.0f}", flush=True) # --- exit --- if pending is not None: days_held += 1 dte_left = int((pending["exp"] - d).days) reason = "" chain = _load_chain(d) spot_now = _spot_from_chain(chain) if not chain.empty else None # Check take-profit / stop-loss mid-hold if not reason and days_held < pending["ht"] and dte_left > 1 and not chain.empty and spot_now: pnl_now = _mark_position(chain, spot_now, pending) if pnl_now is not None: if take_profit_pct > 0 and pnl_now >= pending["max_profit"] * take_profit_pct: reason = "take_profit" elif stop_loss_pct > 0 and pnl_now <= -pending["max_loss"] * stop_loss_pct: reason = "stop_loss" # Time exit if not reason and (days_held >= pending["ht"] or dte_left <= 1): reason = "time" if reason: if spot_now is None: spot_now = pending["spot"] pnl, exit_cost = _close_position(chain, spot_now, pending) trades.append(BearCallTrade( entry_date=str(pending["entry_date"]), exit_date=str(d.date()), exit_reason=reason, expiration=str(pending["exp"].date()), strike_short=pending["sk"], strike_long=pending["lk"], vxx_entry=pending["spot"], vxx_exit=spot_now, entry_credit=pending["credit"], exit_cost=exit_cost, pnl_total=pnl, max_profit=pending["max_profit"], max_loss=pending["max_loss"], contango_ratio=pending["cr"], vix3m_vix=pending["v3v"], hold_days_actual=days_held, )) pending = None days_held = 0 continue # --- entry --- if step % rebalance_every != 0: continue # Contango gate in_contango = False if contango_mode == "futures": in_contango = math.isfinite(cr_val) and cr_val >= contango_threshold elif contango_mode == "vix3m": in_contango = math.isfinite(v3v) and v3v >= vix3m_threshold elif contango_mode == "both": f_ok = math.isfinite(cr_val) and cr_val >= contango_threshold v_ok = math.isfinite(v3v) and v3v >= vix3m_threshold in_contango = f_ok and v_ok if not in_contango: continue chain = _load_chain(d) if chain.empty: continue spot = _spot_from_chain(chain) if spot is None: continue eligible = chain[(chain["dte"] >= dte_min) & (chain["dte"] <= dte_max)] if eligible.empty: continue exp = eligible.sort_values("dte")["expiration_dt"].iloc[0] short_row = _nearest_strike(chain, spot * short_moneyness, "C", exp) long_row = _nearest_strike(chain, spot * (short_moneyness + width_pct), "C", exp) if short_row is None or long_row is None: continue sk = float(short_row["strike"]) lk = float(long_row["strike"]) if lk <= sk: continue credit = (float(short_row["mid"]) * (1 - SLIPPAGE) - float(long_row["mid"]) * (1 + SLIPPAGE)) * MULT if credit <= 0: continue max_loss = (lk - sk) * MULT - credit max_profit = credit 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(), "exp": exp, "sk": sk, "lk": lk, "credit": credit, "max_loss": max_loss, "max_profit": max_profit, "spot": spot, "ht": ht, "cr": cr_val if math.isfinite(cr_val) else 0.0, "v3v": v3v, } days_held = 0 return trades def _mark_position(chain: pd.DataFrame, spot: float, pending: dict) -> float | None: sub = chain[(chain["right_code"] == "C") & (chain["expiration_dt"] == pending["exp"])] if sub.empty: return None sr = sub.iloc[(sub["strike"] - pending["sk"]).abs().argsort()[:1]] lr = sub.iloc[(sub["strike"] - pending["lk"]).abs().argsort()[:1]] if not len(sr) or not len(lr): return None sm, lm = float(sr.iloc[0]["mid"]), float(lr.iloc[0]["mid"]) if not (math.isfinite(sm) and math.isfinite(lm)): return None cost = (sm * (1 + SLIPPAGE) - lm * (1 - SLIPPAGE)) * MULT return pending["credit"] - cost def _close_position(chain: pd.DataFrame, spot: float, pending: dict) -> tuple[float, float]: sub = chain[(chain["right_code"] == "C") & (chain["expiration_dt"] == pending["exp"])] if not chain.empty else pd.DataFrame() if not sub.empty: sr = sub.iloc[(sub["strike"] - pending["sk"]).abs().argsort()[:1]] lr = sub.iloc[(sub["strike"] - pending["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): cost = (sm * (1 + SLIPPAGE) - lm * (1 - SLIPPAGE)) * MULT return pending["credit"] - cost, cost # Intrinsic fallback s_itm = max(spot - pending["sk"], 0) * MULT l_itm = max(spot - pending["lk"], 0) * MULT cost = s_itm - l_itm return pending["credit"] - cost, cost # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main() -> None: ap = argparse.ArgumentParser( description="VXX bear-call credit spread, contango-filtered (CBOE VIX futures)." ) ap.add_argument("--start", default="2018-06-01") ap.add_argument("--end", default="2025-12-31") ap.add_argument("--short-moneyness", type=float, default=1.05, help="Short call strike / VXX spot (default 1.05 = 5%% OTM).") ap.add_argument("--width-pct", type=float, default=0.15, help="Long call offset as fraction of spot (default 0.15).") ap.add_argument("--hold-days", type=int, default=20) ap.add_argument("--rebalance-every", type=int, default=10) ap.add_argument("--contango-mode", choices=("futures", "vix3m", "both"), default="futures", help="Contango signal: 'futures' = VX2/VX1, 'vix3m' = VIX3M/VIX, 'both' = require both.") ap.add_argument("--contango-threshold", type=float, default=0.05, help="Min VX2/VX1 − 1 for futures mode (default 0.05 = 5%%).") ap.add_argument("--vix3m-threshold", type=float, default=1.08) ap.add_argument("--dte-min", type=int, default=21) ap.add_argument("--dte-max", type=int, default=45) 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("--out-trades", type=Path, default=None) args = ap.parse_args() ct_desc = (f"VX2/VX1≥{args.contango_threshold:.0%}" if args.contango_mode == "futures" else f"VIX3M/VIX≥{args.vix3m_threshold}" if args.contango_mode == "vix3m" else f"VX2/VX1≥{args.contango_threshold:.0%} & VIX3M/VIX≥{args.vix3m_threshold}") print( f"VXX Bear Call Credit {args.start} → {args.end}\n" f" short={args.short_moneyness:.2f}x width={args.width_pct:.0%} " f"hold={args.hold_days}d contango={args.contango_mode} ({ct_desc}) " f"TP={args.take_profit_pct:.0%} SL={args.stop_loss_pct:.0%}", flush=True, ) trades = run_backtest( start=args.start, end=args.end, short_moneyness=args.short_moneyness, width_pct=args.width_pct, hold_days=args.hold_days, rebalance_every=args.rebalance_every, contango_mode=args.contango_mode, contango_threshold=args.contango_threshold, vix3m_threshold=args.vix3m_threshold, dte_min=args.dte_min, dte_max=args.dte_max, take_profit_pct=args.take_profit_pct, stop_loss_pct=args.stop_loss_pct, ) 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) cum = np.cumsum(pnls) if pnls else np.array([0]) peak = np.maximum.accumulate(cum) dd = cum - peak 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_bear_call_contango", "period": f"{args.start} → {args.end}", "n_trades": len(trades), "total_pnl": round(total, 2), "avg_pnl": round(float(np.mean(pnls)), 2) if pnls else 0, "median_pnl": round(float(np.median(pnls)), 2) if pnls else 0, "win_rate": round(wins / len(trades), 4) if trades else 0, "best_trade": round(max(pnls), 2) if pnls else 0, "worst_trade": round(min(pnls), 2) if pnls else 0, "max_drawdown": round(float(dd.min()), 2) if len(dd) else 0, "peak_equity": round(float(peak.max()), 2) if len(peak) else 0, "sharpe_approx": round(float(np.mean(pnls) / np.std(pnls) * np.sqrt(26)), 2) if pnls and np.std(pnls) > 0 else 0, "exit_reasons": exit_reasons, "params": { "short_moneyness": args.short_moneyness, "width_pct": args.width_pct, "hold_days": args.hold_days, "contango_mode": args.contango_mode, "contango_threshold": args.contango_threshold, "vix3m_threshold": args.vix3m_threshold, "dte_range": [args.dte_min, args.dte_max], "take_profit_pct": args.take_profit_pct, "stop_loss_pct": args.stop_loss_pct, }, } print("\n" + 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") as f: for t in trades: f.write(json.dumps(asdict(t)) + "\n") print(f"\nWrote {len(trades)} trades → {p}") if __name__ == "__main__": main()