#!/usr/bin/env python3 """ Canonical **VXX long OTM call** tail hedge (10% OTM, contango-gated). Audit reference ``engine_vxx_1pct_* / vxx_portfolio_long_call.jsonl`` uses **1%** NAV risk per entry. **Portfolio default:** ``--book-pnl-frac 0.05`` → ~**5%** of Best Ideas stacked-book PnL (not 1% audit sizing). 15-day hold, futures contango >= 3%. Writes daily **MTM** PnL for ``combine_best_ideas_stack.py``. Example:: cd /Users/robzingale/trading_bot PYTHONUNBUFFERED=1 .venv/bin/python RenTech/strategy_stack/run_vxx_long_call_daily.py \\ --start 2016-01-04 --end 2026-04-02 """ from __future__ import annotations import argparse import json import math import sys from pathlib import Path 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.backtest_vxx_vx1_vx3_strategies import daily_mtm_metrics from RenTech.strategy_stack.explore_vxx_decay_strategies import ( Trade, _build_long_call, _exit_long_call, _load_chain, _load_contango, _pick_expiry, _spot_from_chain, leg_strikes_from_vxx_built, resolve_vxx_contracts_and_broker_risk, vxx_built_snapshot, ) from RenTech.strategy_stack.iv_mispricing_complement import _load_jsonl, _pnl_series_from_trades LOGS = _REPO / "RenTech" / "data" / "logs" DEFAULT_OUT_PREFIX = LOGS / "vxx_long_call_standard" DEFAULT_JSONL = ( LOGS / "overall_portfolio_trade_audits" / "engine_vxx_1pct_2016_2026" / "vxx_portfolio_long_call.jsonl" ) # At 1% NAV/entry, long-call MTM PnL was ~22.7% of full Best Ideas stack (2016–2026). AUDIT_BROKER_RISK_PCT = 0.01 BOOK_PNL_SHARE_AT_AUDIT = 0.227 DEFAULT_BOOK_PNL_FRAC = 0.05 def broker_risk_pct_for_book_frac(book_pnl_frac: float) -> float: """Linear scale vs audit 1% sizing to target share of combined-book PnL.""" if book_pnl_frac <= 0: raise ValueError("book_pnl_frac must be positive") return AUDIT_BROKER_RISK_PCT * book_pnl_frac / BOOK_PNL_SHARE_AT_AUDIT def run_long_call_mtm( dates: list[pd.Timestamp], ct: pd.DataFrame, *, capital: float, otm_pct: float, contango_threshold: float, dte_min: int, dte_max: int, hold_days: int, rebalance_every: int, broker_risk_pct: float, ) -> tuple[list[Trade], list[dict]]: """Long call with daily mark-to-market (open positions marked each session).""" trades: list[Trade] = [] pending: dict | None = None days_held = 0 cum_realized = 0.0 records: list[dict] = [] 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)) if pending is not None: days_held += 1 p = pending dte_left = int((p["expiration"] - d).days) should_exit = (days_held >= p["hold_target"]) or (dte_left <= 1) if should_exit: 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_one = _exit_long_call(chain, spot_now, p["legs"], p["expiration"]) pnl = float(pnl_one) * int(p["contracts"]) cum_realized += pnl exp_s = str(pd.Timestamp(p["expiration"]).date()) ss, ls, ps, cs = leg_strikes_from_vxx_built(p["legs"]) trades.append( Trade( strategy="long_call", 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"]), exit_value=pnl + float(p["entry_val"]), pnl_total=pnl, contango_ratio=p.get("cr", 0.0), vix3m_vix=p.get("v3v", 0.0), broker_risk_usd=float(p["broker_risk_usd"]), contracts=int(p["contracts"]), broker_risk_per_contract_usd=float(p["broker_risk_per_contract"]), underlying="VXX", expiration=exp_s, short_strike=ss, long_strike=ls, put_strike=ps, call_strike=cs, nav_at_entry_usd=float(p.get("nav_at_entry", capital)), risk_pct_of_portfolio=float(p.get("risk_pct_of_portfolio", broker_risk_pct)), dte_at_entry=int(p.get("dte_at_entry", 0)), dte_at_exit=int(dte_left), hold_target_days=int(p["hold_target"]), days_held=int(days_held), max_loss_one_usd=None, max_loss_total_usd=None, target_broker_risk_usd=p.get("target_broker_risk_usd"), contracts_requested=p.get("contracts_requested"), entry_legs_json=str(p.get("entry_legs_json", "{}")), ) ) pending = None days_held = 0 ur = 0.0 if pending is not None: chain = _load_chain(d) spot_now = _spot_from_chain(chain) if not chain.empty else pending["vxx_entry"] pnl_one = _exit_long_call( chain, spot_now, pending["legs"], pending["expiration"] ) ur = float(pnl_one) * int(pending["contracts"]) eq_mtm = float(capital) + cum_realized + ur records.append( { "date": d, "realized_cum_usd": round(cum_realized, 2), "unrealized_usd": round(ur, 2), "equity_mtm_usd": round(eq_mtm, 2), } ) if step % rebalance_every != 0: continue if pending is not None: continue if not (math.isfinite(cr_val) and cr_val >= contango_threshold): 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 = _build_long_call(chain, spot, exp, otm_pct) if built is None: continue entry_val_one = -float(built.get("debit", 0.0)) nav_at = max(float(capital) + cum_realized, 1.0) tgt = max(nav_at * float(broker_risk_pct), 1.0) n_c, per_risk, br_total = resolve_vxx_contracts_and_broker_risk( strategy="long_call", built=built, entry_val=entry_val_one, contracts=None, target_broker_risk_usd=tgt, ) days_to_exp = sum(1 for dd in dates if d < dd <= exp) - 1 ht = min(hold_days, max(days_to_exp, 1)) dte_entry = int((exp - d).days) pending = { "entry_date": d.date(), "expiration": exp, "legs": built, "vxx_entry": spot, "entry_val": entry_val_one * float(n_c), "hold_target": ht, "broker_risk_usd": br_total, "broker_risk_per_contract": per_risk, "contracts": n_c, "nav_at_entry": float(nav_at), "risk_pct_of_portfolio": float(broker_risk_pct), "dte_at_entry": int(dte_entry), "target_broker_risk_usd": float(tgt), "contracts_requested": None, "entry_legs_json": json.dumps(vxx_built_snapshot(built)), "cr": cr_val if math.isfinite(cr_val) else 0.0, "v3v": float(ct_row.get("vix3m_vix_ratio", np.nan)) if math.isfinite(float(ct_row.get("vix3m_vix_ratio", np.nan))) else 0.0, } days_held = 0 return trades, records def _daily_from_jsonl(path: Path, idx: pd.DatetimeIndex) -> pd.Series: rows = _load_jsonl(path) pnl = _pnl_series_from_trades(rows, exit_key="exit_date", pnl_key="pnl_total") return pnl.reindex(idx).fillna(0.0) def main() -> None: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--start", default="2016-01-04") ap.add_argument("--end", default="") ap.add_argument("--capital", type=float, default=100_000.0) ap.add_argument("--otm-pct", type=float, default=0.10, help="Long call OTM %% (default 10)") ap.add_argument( "--book-pnl-frac", type=float, default=DEFAULT_BOOK_PNL_FRAC, help=( "Target share of Best Ideas stacked-book PnL (default 0.05 = 5%%). " "Maps to --broker-risk-pct unless that flag is set." ), ) ap.add_argument( "--broker-risk-pct", type=float, default=None, metavar="FRAC", help="Override NAV risk per entry (e.g. 0.01 = engine_vxx_1pct audit)", ) ap.add_argument( "--audit-1pct", action="store_true", help="Shorthand for --broker-risk-pct 0.01 (audit reproduction)", ) ap.add_argument("--contango-threshold", type=float, default=0.03) ap.add_argument("--hold-days", type=int, default=15) ap.add_argument("--rebalance-every", type=int, default=10) ap.add_argument("--dte-min", type=int, default=21) ap.add_argument("--dte-max", type=int, default=45) ap.add_argument( "--from-jsonl", type=Path, default=None, help="Skip backtest; build exit-day daily PnL from trade JSONL (no MTM)", ) ap.add_argument("--out-prefix", type=Path, default=DEFAULT_OUT_PREFIX) args = ap.parse_args() if args.audit_1pct and args.broker_risk_pct is not None: raise SystemExit("Use only one of --audit-1pct and --broker-risk-pct") if args.audit_1pct: broker_risk_pct = AUDIT_BROKER_RISK_PCT book_pnl_frac = BOOK_PNL_SHARE_AT_AUDIT elif args.broker_risk_pct is not None: broker_risk_pct = float(args.broker_risk_pct) book_pnl_frac = broker_risk_pct * BOOK_PNL_SHARE_AT_AUDIT / AUDIT_BROKER_RISK_PCT else: book_pnl_frac = float(args.book_pnl_frac) broker_risk_pct = broker_risk_pct_for_book_frac(book_pnl_frac) ct = _load_contango() all_dates = sorted(ct.index) dates = [d for d in all_dates if str(d.date()) >= args.start] if args.end.strip(): dates = [d for d in dates if str(d.date()) <= args.end] if not dates: raise SystemExit("No trading dates in range") cap = float(args.capital) prefix = args.out_prefix.expanduser().resolve() prefix.parent.mkdir(parents=True, exist_ok=True) daily_path = Path(f"{prefix}_daily.csv") meta_path = Path(f"{prefix}_meta.json") if args.from_jsonl is not None: idx = pd.DatetimeIndex(dates) pnl = _daily_from_jsonl(args.from_jsonl, idx) scale = broker_risk_pct / AUDIT_BROKER_RISK_PCT pnl = pnl * scale eq = cap + pnl.cumsum() out = pd.DataFrame( { "date": idx.strftime("%Y-%m-%d"), "daily_pnl_usd": pnl.values, "equity_usd": eq.values, "metric_mode": "exit_day_jsonl", } ) out.to_csv(daily_path, index=False) meta = { "strategy": "vxx_long_call", "token": "vxx_long_call", "metric_mode": "exit_day_jsonl", "source_jsonl": str(args.from_jsonl.resolve()), "capital_usd": cap, "start": str(idx.min().date()), "end": str(idx.max().date()), "n_trading_days": int(len(idx)), "total_pnl_usd": round(float(pnl.sum()), 2), "daily_csv": str(daily_path), } meta_path.write_text(json.dumps(meta, indent=2) + "\n") print(f"Exit-day PnL from JSONL → {daily_path} Σpnl=${pnl.sum():+,.0f}", flush=True) return print( f"Running VXX long call MTM ({dates[0].date()} → {dates[-1].date()}, " f"{len(dates)} days, ~{book_pnl_frac:.0%} book PnL target, " f"{broker_risk_pct:.2%} NAV/entry) …", flush=True, ) trades, records = run_long_call_mtm( dates, ct, capital=cap, otm_pct=float(args.otm_pct), contango_threshold=float(args.contango_threshold), dte_min=int(args.dte_min), dte_max=int(args.dte_max), hold_days=int(args.hold_days), rebalance_every=int(args.rebalance_every), broker_risk_pct=broker_risk_pct, ) mtm_meta, mtm_df = daily_mtm_metrics(records, capital=cap) mtm_df["daily_pnl_usd"] = mtm_df["daily_pnl_mtm_usd"] out = mtm_df[ ["date", "daily_pnl_mtm_usd", "daily_pnl_usd", "equity_mtm_usd", "unrealized_usd"] ].copy() out["date"] = pd.to_datetime(out["date"]).dt.strftime("%Y-%m-%d") out.to_csv(daily_path, index=False) cmd = ( f"cd {_REPO} && PYTHONUNBUFFERED=1 .venv/bin/python " f"RenTech/strategy_stack/run_vxx_long_call_daily.py --start {args.start}" ) if args.end.strip(): cmd += f" --end {args.end}" meta = { "strategy": "vxx_long_call", "token": "vxx_long_call", "metric_mode": "mtm", "otm_pct": float(args.otm_pct), "book_pnl_frac_target": round(book_pnl_frac, 4), "broker_risk_pct_of_nav": round(broker_risk_pct, 6), "contango_threshold": float(args.contango_threshold), "n_trades": len(trades), "capital_usd": cap, "start": str(pd.Timestamp(out["date"].iloc[0]).date()), "end": str(pd.Timestamp(out["date"].iloc[-1]).date()), "n_trading_days": int(len(out)), "command": cmd, "daily_csv": str(daily_path), "audit_jsonl_reference": str(DEFAULT_JSONL), **mtm_meta, } meta_path.write_text(json.dumps(meta, indent=2) + "\n") print( f"Wrote {daily_path} trades={len(trades)} " f"return={mtm_meta.get('return_pct', 0):+.1f}% " f"Sharpe={mtm_meta.get('sharpe', 0):.2f} " f"maxDD={mtm_meta.get('max_dd_pct', 0):.1f}%", flush=True, ) if __name__ == "__main__": main()