#!/usr/bin/env python3 """ Parameter sweep for VXX bear-call credit spread. Tests combinations of: - short_moneyness (where to sell the call relative to VXX spot) - width_pct (how far OTM the long protective call is) - hold_days - vix3m_threshold Reuses the chain loader / spot estimator from the exploration script. """ from __future__ import annotations import itertools import json import math import sys 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 def _load_contango(): ct = pd.read_parquet(CONTANGO_PATH) ct.index = pd.to_datetime(ct.index) return ct def _session_date(qt): 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): 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(chain): 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] c = atm[atm["right_code"] == "C"][["strike", "mid"]] p = atm[atm["right_code"] == "P"][["strike", "mid"]] if c.empty or p.empty: return None m = c.merge(p, 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() s = float(m.sort_values("gap").iloc[0]["s"]) return s if (math.isfinite(s) and s > 0) else None def _nearest(chain, target, right, exp): 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 def _run_single( dates, ct, chain_cache, short_mny, width_pct, hold_days, rebalance_every, vix3m_threshold, dte_min, dte_max, ): pnls = [] pending = None days_held = 0 for step, d in enumerate(dates): d = pd.Timestamp(d).normalize() v3v = float(ct.loc[d].get("vix3m_vix_ratio", np.nan)) # --- exit --- if pending is not None: days_held += 1 dte_left = int((pending["exp"] - d).days) if days_held >= pending["ht"] or dte_left <= 1: chain = chain_cache.get(d) if chain is None: chain = _load_chain(d) chain_cache[d] = chain spot_now = _spot(chain) if not chain.empty else pending["spot"] if spot_now is None: spot_now = pending["spot"] # Close: buy back short, sell long 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): cost = (float(sr.iloc[0]["mid"]) * (1 + SLIPPAGE) - float(lr.iloc[0]["mid"]) * (1 - SLIPPAGE)) * MULT pnl = pending["credit"] - cost pnls.append(pnl) pending = None days_held = 0 continue # Intrinsic fallback s_itm = max(spot_now - pending["sk"], 0) * MULT l_itm = max(spot_now - pending["lk"], 0) * MULT pnl = pending["credit"] - (s_itm - l_itm) pnls.append(pnl) pending = None days_held = 0 continue # --- entry --- if step % rebalance_every != 0: continue if not (math.isfinite(v3v) and v3v >= vix3m_threshold): continue chain = chain_cache.get(d) if chain is None: chain = _load_chain(d) chain_cache[d] = chain if chain.empty: continue s = _spot(chain) if s 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(chain, s * short_mny, "C", exp) long_row = _nearest(chain, s * (short_mny + width_pct), "C", exp) if short_row is None or long_row is None: continue sk, lk = float(short_row["strike"]), 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 days_to_exp = sum(1 for dd in dates if d < dd <= exp) - 1 ht = min(hold_days, max(days_to_exp, 1)) pending = {"exp": exp, "sk": sk, "lk": lk, "credit": credit, "spot": s, "ht": ht} days_held = 0 if not pnls: return None arr = np.array(pnls) cum = np.cumsum(arr) peak = np.maximum.accumulate(cum) dd = cum - peak wins = int((arr > 0).sum()) return { "n": len(pnls), "total": round(float(arr.sum()), 1), "avg": round(float(arr.mean()), 2), "median": round(float(np.median(arr)), 2), "wr": round(wins / len(pnls), 3), "sharpe": round(float(arr.mean() / arr.std()) * np.sqrt(26), 2) if arr.std() > 0 else 0, "maxdd": round(float(dd.min()), 1), "peak": round(float(peak.max()), 1), "worst": round(float(arr.min()), 1), } def main(): import argparse ap = argparse.ArgumentParser() ap.add_argument("--start", default="2018-06-01") ap.add_argument("--end", default="2025-12-31") args = ap.parse_args() ct = _load_contango() all_dates = sorted(ct.index) dates = [d for d in all_dates if args.start <= str(d.date()) <= args.end] print(f"Dates: {len(dates)}", flush=True) # Parameter grid short_mny_vals = [0.98, 1.00, 1.02, 1.05] width_pct_vals = [0.05, 0.08, 0.10, 0.15] hold_vals = [10, 15, 20, 30] vix3m_vals = [1.03, 1.05, 1.08] rebalance = 10 dte_min, dte_max = 21, 45 combos = list(itertools.product(short_mny_vals, width_pct_vals, hold_vals, vix3m_vals)) print(f"Sweeping {len(combos)} parameter combos …", flush=True) chain_cache: dict = {} results = [] for i, (sm, wp, hd, vt) in enumerate(combos): if i % 48 == 0: print(f" [{i}/{len(combos)}] …", flush=True) r = _run_single(dates, ct, chain_cache, sm, wp, hd, rebalance, vt, dte_min, dte_max) if r is None: continue r["short_mny"] = sm r["width_pct"] = wp r["hold_days"] = hd r["vix3m_thr"] = vt results.append(r) df = pd.DataFrame(results) if df.empty: print("No results!") return df = df.sort_values("total", ascending=False) print(f"\n{'='*110}") print(f"{'short_mny':>10} {'width%':>7} {'hold':>5} {'v3m_thr':>8} " f"{'trades':>6} {'total$':>8} {'avg$':>7} {'med$':>7} {'WR':>6} " f"{'sharpe':>7} {'maxDD':>7} {'peak$':>7} {'worst$':>7}") print("-" * 110) for _, row in df.head(25).iterrows(): print( f"{row['short_mny']:>10.2f} {row['width_pct']:>7.2f} {int(row['hold_days']):>5} " f"{row['vix3m_thr']:>8.2f} {int(row['n']):>6} {row['total']:>8.1f} " f"{row['avg']:>7.2f} {row['median']:>7.2f} {row['wr']:>6.1%} " f"{row['sharpe']:>7.2f} {row['maxdd']:>7.1f} {row['peak']:>7.1f} " f"{row['worst']:>7.1f}" ) print("=" * 110) # Save full results out = DATA_DIR / "logs" / "vxx_bear_call_sweep.csv" df.to_csv(out, index=False) print(f"\nFull sweep → {out}") if __name__ == "__main__": main()