#!/usr/bin/env python3 """ Full-book sizing search (VRP base + IV overlays + VXX): 1) Optional grids: - VXX internal per-entry risk as %% of NAV (regenerates ``optimize_vxx_portfolio.py`` logs). - IV overlay internal per-entry risk as %% of NAV for: * stress OTM put JSONL * stress straddle JSONL * risk reversal JSONL (regenerates ``backtest_iv_stress_long_vol.py`` / ``backtest_iv_calendar_risk_reversal.py`` outputs). 2) For each (VXX internal pct × IV internal pct tuple) trial, optimizes **sleeve allocation fractions** (put/straddle/RR/VXX) for merged-book Sharpe under optional DD/sum constraints (``portfolio_vrp_plus_vxx.optimize_portfolio_sharpe``). Outputs: - trials CSV (one row per completed trial) - best JSON summary Notes: - IV regeneration requires a trained ``vol_mispricing_xgb.joblib`` on disk (see script defaults). - If you omit all IV grids, behavior matches the older VXX-only outer loop. """ from __future__ import annotations import argparse import csv import itertools import json import math import re import shutil import subprocess import sys from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path _REPO = Path(__file__).resolve().parents[2] if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO)) from RenTech.strategy_stack.portfolio_vrp_plus_vxx import optimize_portfolio_sharpe LOGS = _REPO / "RenTech" / "data" / "logs" DEFAULT_IV_ARTIFACT = _REPO / "RenTech" / "data" / "models" / "vol_mispricing_xgb.joblib" DEFAULT_STRESS_SCRIPT = _REPO / "RenTech" / "strategy_stack" / "backtest_iv_stress_long_vol.py" DEFAULT_CAL_SCRIPT = _REPO / "RenTech" / "strategy_stack" / "backtest_iv_calendar_risk_reversal.py" @dataclass class TrialResult: vxx_internal_risk_pct: float iv_put_internal_risk_pct: float | None iv_straddle_internal_risk_pct: float | None iv_rr_internal_risk_pct: float | None status: str sharpe: float calmar: float cagr_pct: float return_pct: float max_drawdown_pct: float feasible_dd: bool put_frac: float straddle_frac: float rr_frac: float vxx_frac: float sum_frac: float detail: str = "" def _parse_vxx_pct_grid(s: str) -> list[float]: vals: list[float] = [] for tok in s.split(","): t = tok.strip() if not t: continue v = float(t) if not (0.0 < v <= 1.0): raise ValueError(f"invalid VXX pct '{t}' (must be in (0,1])") vals.append(v) if not vals: raise ValueError("empty --vxx-risk-pct-grid") return vals def _parse_iv_pct_tokens(s: str) -> list[float | None]: """ Comma-separated floats in (0,1], or the token ``none`` / empty token for a dimension. Empty string => [] (meaning: do not drive IV regeneration from this flag). """ out: list[float | None] = [] s = (s or "").strip() if not s: return out for tok in s.split(","): t = tok.strip() if t == "" or t.lower() == "none": out.append(None) continue v = float(t) if not (0.0 < v <= 1.0): raise ValueError(f"invalid IV pct '{t}' (must be in (0,1] or 'none')") out.append(v) return out def _iv_dims_from_args(args: argparse.Namespace) -> tuple[list[float | None], list[float | None], list[float | None]]: """ Returns (put_list, straddle_list, rr_list) where each list is non-empty. Precedence: - If ``--iv-risk-pct-grid`` is non-empty: use it for all three dimensions. - Else use per-dimension grids; any empty list becomes ``[None]`` (no regeneration for that dim). """ uni = (args.iv_risk_pct_grid or "").strip() if uni: xs = _parse_iv_pct_tokens(uni) if not xs: raise ValueError("empty --iv-risk-pct-grid after parsing") return xs, list(xs), list(xs) put = _parse_iv_pct_tokens(args.iv_put_risk_pct_grid or "") st = _parse_iv_pct_tokens(args.iv_straddle_risk_pct_grid or "") rr = _parse_iv_pct_tokens(args.iv_rr_risk_pct_grid or "") # If you only specify one dimension, mirror it to the other sleeves (common intent: # "run all IV overlays at the same internal risk pct"). nonempty = [g for g in (put, st, rr) if g] if len(nonempty) == 1: only = nonempty[0] put, st, rr = list(only), list(only), list(only) if not put: put = [None] if not st: st = [None] if not rr: rr = [None] return put, st, rr def _iv_enabled(dims: tuple[list[float | None], list[float | None], list[float | None]]) -> bool: put, st, rr = dims return any(v is not None for v in put) or any(v is not None for v in st) or any(v is not None for v in rr) def _fmt_iv_token(v: float | None) -> str: if v is None: return "none" s = f"{float(v):.6f}".replace(".", "p") return s def _iv_combo_tag(p: float | None, s: float | None, r: float | None) -> str: return f"iv_put_{_fmt_iv_token(p)}__straddle_{_fmt_iv_token(s)}__rr_{_fmt_iv_token(r)}" def _run(cmd: list[str]) -> tuple[bool, str]: p = subprocess.run(cmd, capture_output=True, text=True) if p.returncode != 0: detail = (p.stderr or p.stdout or "").strip() return False, detail[:1200] return True, "ok" def _run_vxx_backtest_for_pct( args: argparse.Namespace, pct: float, *, audit_dir: Path | None, audit_tag_suffix: str, ) -> tuple[bool, str]: cmd = [ str(args.python_bin), str(args.vxx_optimizer_script), "--broker-risk-pct-of-portfolio", str(pct), "--portfolio-capital", str(args.total_capital), "--start", args.start, "--end", args.end, "--contango-mode", args.contango_mode, "--contango-threshold", str(args.contango_threshold), "--vix3m-threshold", str(args.vix3m_threshold), "--hold-days", str(args.hold_days), "--rebalance-every", str(args.rebalance_every), "--dte-min", str(args.dte_min), "--dte-max", str(args.dte_max), "--step-pct", str(args.step_pct), ] if audit_dir is not None: tag = f"vxx_internal_{pct:.6f}".replace(".", "p") if audit_tag_suffix: tag = f"{tag}__{audit_tag_suffix}" cmd += ["--trade-audit-dir", str(audit_dir), "--trade-audit-tag", tag] return _run(cmd) def _stress_cmd( *, python_bin: Path, script: Path, artifact: Path, theta_dir: Path, start: str, end: str, structure: str, out_trades: Path, portfolio_capital: float, broker_risk_pct: float | None, min_vix: float, hold_days: int, rebalance_every: int, dte_min: int, dte_max: int, ) -> list[str]: cmd = [ str(python_bin), str(script), "--artifact", str(artifact), "--theta-dir", str(theta_dir), "--start", start, "--end", end, "--structure", structure, "--min-vix", str(min_vix), "--hold-days", str(hold_days), "--rebalance-every", str(rebalance_every), "--dte-min", str(dte_min), "--dte-max", str(dte_max), "--portfolio-capital", str(portfolio_capital), "--out-trades", str(out_trades), ] if broker_risk_pct is not None: cmd += ["--broker-risk-pct-of-portfolio", str(broker_risk_pct)] return cmd def _rr_cmd( *, python_bin: Path, script: Path, artifact: Path, theta_dir: Path, start: str, end: str, out_trades: Path, portfolio_capital: float, broker_risk_pct: float | None, min_vix: float, hold_days: int, rebalance_every: int, dte_min: int, dte_max: int, min_dte_gap: int, mny_band: float, min_pred_diff: float, put_mny_max: float, call_mny_min: float, min_pred_spread: float, ) -> list[str]: cmd = [ str(python_bin), str(script), "--artifact", str(artifact), "--theta-dir", str(theta_dir), "--start", start, "--end", end, "--mode", "risk_reversal", "--min-vix", str(min_vix), "--hold-days", str(hold_days), "--rebalance-every", str(rebalance_every), "--dte-min", str(dte_min), "--dte-max", str(dte_max), "--min-dte-gap", str(min_dte_gap), "--moneyness-band", str(mny_band), "--min-pred-diff", str(min_pred_diff), "--put-moneyness-max", str(put_mny_max), "--call-moneyness-min", str(call_mny_min), "--min-pred-spread", str(min_pred_spread), "--portfolio-capital", str(portfolio_capital), "--out-trades", str(out_trades), ] if broker_risk_pct is not None: cmd += ["--broker-risk-pct-of-portfolio", str(broker_risk_pct)] return cmd def _regenerate_iv_overlays( args: argparse.Namespace, *, iv_put: float | None, iv_straddle: float | None, iv_rr: float | None, ) -> tuple[bool, str]: art = args.iv_artifact.expanduser() if not art.is_file(): return False, f"missing IV artifact: {art}" # Writes canonical merge paths under RenTech/data/logs/ ok, msg = _run( _stress_cmd( python_bin=args.python_bin, script=args.iv_stress_script, artifact=art, theta_dir=args.theta_dir, start=args.iv_start, end=args.iv_end, structure="otm_put", out_trades=LOGS / "stress_longvol_otm_put.jsonl", portfolio_capital=float(args.total_capital), broker_risk_pct=iv_put, min_vix=float(args.iv_min_vix_stress), hold_days=int(args.iv_hold_days), rebalance_every=int(args.iv_rebalance_every), dte_min=int(args.iv_dte_min), dte_max=int(args.iv_dte_max), ) ) if not ok: return False, f"stress otm_put failed: {msg}" ok, msg = _run( _stress_cmd( python_bin=args.python_bin, script=args.iv_stress_script, artifact=art, theta_dir=args.theta_dir, start=args.iv_start, end=args.iv_end, structure="straddle", out_trades=LOGS / "stress_longvol_straddle.jsonl", portfolio_capital=float(args.total_capital), broker_risk_pct=iv_straddle, min_vix=float(args.iv_min_vix_stress), hold_days=int(args.iv_hold_days), rebalance_every=int(args.iv_rebalance_every), dte_min=int(args.iv_dte_min), dte_max=int(args.iv_dte_max), ) ) if not ok: return False, f"stress straddle failed: {msg}" ok, msg = _run( _rr_cmd( python_bin=args.python_bin, script=args.iv_calendar_script, artifact=art, theta_dir=args.theta_dir, start=args.iv_start, end=args.iv_end, out_trades=LOGS / "risk_reversal.jsonl", portfolio_capital=float(args.total_capital), broker_risk_pct=iv_rr, min_vix=float(args.iv_min_vix_rr), hold_days=int(args.iv_hold_days), rebalance_every=int(args.iv_rebalance_every), dte_min=int(args.iv_dte_min), dte_max=int(args.iv_dte_max), min_dte_gap=int(args.iv_min_dte_gap), mny_band=float(args.iv_mny_band_calendar), min_pred_diff=float(args.iv_min_pred_diff), put_mny_max=float(args.iv_put_moneyness_max), call_mny_min=float(args.iv_call_moneyness_min), min_pred_spread=float(args.iv_min_pred_spread), ) ) if not ok: return False, f"risk_reversal failed: {msg}" return True, "ok" def _copy_trial_audit_bundle( *, audit_run_dir: Path, tag: str, paths: list[tuple[str, Path]], ) -> None: d = audit_run_dir / tag d.mkdir(parents=True, exist_ok=True) for label, src in paths: if not src.is_file(): continue dst = d / f"{label}{src.suffix}" shutil.copy2(src, dst) def _safe_float(v: object, fallback: float = float("nan")) -> float: try: x = float(v) except Exception: return fallback return x if math.isfinite(x) else fallback def main() -> None: ap = argparse.ArgumentParser(description="Optimize overall sizing (VXX + optional IV overlays + sleeve fractions).") ap.add_argument("--vrp-trades", type=Path, default=_REPO / "RenTech" / "data" / "logs" / "vrp_trades.jsonl") ap.add_argument("--total-capital", type=float, default=100_000.0) 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", "both"], default="futures") ap.add_argument("--contango-threshold", type=float, default=0.03) ap.add_argument("--vix3m-threshold", type=float, default=1.05) ap.add_argument("--hold-days", type=int, default=20) 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("--step-pct", type=int, default=5) ap.add_argument( "--vxx-risk-pct-grid", type=str, default="0.005,0.01,0.015,0.02,0.025,0.03,0.035,0.04,0.045,0.05", help="Comma-separated VXX internal risk pct values; e.g. 0.01,0.02,0.03.", ) # IV overlay regeneration (optional). If all IV grids are empty / none-only, IV logs are not regenerated. ap.add_argument( "--iv-risk-pct-grid", type=str, default="", help="If set, applies the same comma list to OTM put + straddle + RR internal %% sizing.", ) ap.add_argument("--iv-put-risk-pct-grid", type=str, default="", help="Comma list or ``none`` tokens; empty = unused unless --iv-risk-pct-grid set.") ap.add_argument("--iv-straddle-risk-pct-grid", type=str, default="", help="Comma list or ``none`` tokens; empty = unused unless --iv-risk-pct-grid set.") ap.add_argument("--iv-rr-risk-pct-grid", type=str, default="", help="Comma list or ``none`` tokens; empty = unused unless --iv-risk-pct-grid set.") ap.add_argument("--iv-artifact", type=Path, default=DEFAULT_IV_ARTIFACT) ap.add_argument("--theta-dir", type=Path, default=_REPO / "RenTech" / "data" / "theta_chunks") ap.add_argument("--iv-start", type=str, default="", help="Defaults to --start when empty.") ap.add_argument("--iv-end", type=str, default="", help="Defaults to --end when empty.") ap.add_argument("--iv-stress-script", type=Path, default=DEFAULT_STRESS_SCRIPT) ap.add_argument("--iv-calendar-script", type=Path, default=DEFAULT_CAL_SCRIPT) ap.add_argument("--iv-min-vix-stress", type=float, default=12.0) ap.add_argument("--iv-min-vix-rr", type=float, default=0.0) ap.add_argument("--iv-hold-days", type=int, default=5) ap.add_argument("--iv-rebalance-every", type=int, default=5) ap.add_argument("--iv-dte-min", type=int, default=7) ap.add_argument("--iv-dte-max", type=int, default=60) ap.add_argument("--iv-min-dte-gap", type=int, default=7) ap.add_argument("--iv-mny-band-calendar", type=float, default=0.15) ap.add_argument("--iv-min-pred-diff", type=float, default=0.0) ap.add_argument("--iv-put-moneyness-max", type=float, default=0.98) ap.add_argument("--iv-call-moneyness-min", type=float, default=1.02) ap.add_argument("--iv-min-pred-spread", type=float, default=0.0) ap.add_argument("--sharpe-sum-cap-pct", type=float, default=0.35) ap.add_argument("--sharpe-max-sleeve-pct", type=float, default=0.25) ap.add_argument("--sharpe-max-dd-pct", type=float, default=10.0, help="0 disables DD cap") ap.add_argument("--opt-maxiter", type=int, default=80) ap.add_argument("--opt-seed", type=int, default=0) ap.add_argument("--vxx-bear-pct", type=float, default=90.0) ap.add_argument("--vxx-call-pct", type=float, default=10.0) ap.add_argument( "--vxx-optimizer-script", type=Path, default=_REPO / "RenTech" / "strategy_stack" / "optimize_vxx_portfolio.py", ) ap.add_argument("--python-bin", type=Path, default=_REPO / ".venv" / "bin" / "python") ap.add_argument( "--out-csv", type=Path, default=_REPO / "RenTech" / "data" / "logs" / "overall_sizing_trials.csv", ) ap.add_argument( "--out-best-json", type=Path, default=_REPO / "RenTech" / "strategy_stack" / "overall_sizing_best.json", ) ap.add_argument( "--trade-audit-root", type=Path, default=_REPO / "RenTech" / "data" / "logs" / "overall_portfolio_trade_audits", help="Copies per-trial logs here (set empty string to disable).", ) ap.add_argument( "--trade-audit-run-id", type=str, default="", help="Optional folder name under --trade-audit-root; default is UTC timestamp.", ) args = ap.parse_args() args_snapshot = {k: (str(v) if isinstance(v, Path) else v) for k, v in vars(args).items()} if not args.vrp_trades.expanduser().is_file(): print(f"ERROR: vrp trades not found: {args.vrp_trades}", file=sys.stderr) sys.exit(1) if not args.vxx_optimizer_script.expanduser().is_file(): print(f"ERROR: optimizer script not found: {args.vxx_optimizer_script}", file=sys.stderr) sys.exit(1) if not args.python_bin.expanduser().is_file(): print(f"ERROR: python bin not found: {args.python_bin}", file=sys.stderr) sys.exit(1) vxx_pcts = _parse_vxx_pct_grid(args.vxx_risk_pct_grid) iv_dims = _iv_dims_from_args(args) iv_active = _iv_enabled(iv_dims) if iv_active: if not args.iv_stress_script.is_file() or not args.iv_calendar_script.is_file(): print("ERROR: IV backtest script path invalid.", file=sys.stderr) sys.exit(1) if not args.iv_artifact.expanduser().is_file(): print( "ERROR: IV artifact not found.\n" f" looked for: {args.iv_artifact}\n" " train with:\n" " cd /Users/robzingale/trading_bot && .venv/bin/python RenTech/strategy_stack/train_vol_mispricing_xgb.py\n", file=sys.stderr, ) sys.exit(1) iv_start = args.iv_start.strip() or args.start iv_end = args.iv_end.strip() or args.end # stash resolved dates back onto args for manifests / outputs args.iv_start = iv_start args.iv_end = iv_end sum_cap = None if float(args.sharpe_sum_cap_pct) <= 0 else float(args.sharpe_sum_cap_pct) dd_cap = None if float(args.sharpe_max_dd_pct) <= 0 else float(args.sharpe_max_dd_pct) trials: list[TrialResult] = [] best: TrialResult | None = None best_raw: dict | None = None audit_root: Path | None = None audit_run_dir: Path | None = None tar = str(args.trade_audit_root or "").strip() if tar: audit_root = Path(tar).expanduser().resolve() run_id = (args.trade_audit_run_id or "").strip() if not run_id: run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") run_id = re.sub(r"[^A-Za-z0-9._-]+", "_", run_id).strip("_") or "run" audit_run_dir = audit_root / run_id audit_run_dir.mkdir(parents=True, exist_ok=True) (audit_run_dir / "overall_run_manifest.json").write_text( json.dumps( { "created_utc": datetime.now(timezone.utc).isoformat(), "script": "optimize_overall_portfolio_sizing.py", "args": {k: args_snapshot[k] for k in sorted(args_snapshot)}, }, indent=2, ), encoding="utf-8", ) put_list, st_list, rr_list = iv_dims iv_iter = itertools.product(put_list, st_list, rr_list) for iv_put, iv_str, iv_rr in iv_iter: if iv_active: print( f"[iv] regenerating overlays put={iv_put} straddle={iv_str} rr={iv_rr} " f"(dates {iv_start} → {iv_end})", flush=True, ) ok_iv, msg_iv = _regenerate_iv_overlays(args, iv_put=iv_put, iv_straddle=iv_str, iv_rr=iv_rr) if not ok_iv: tr = TrialResult( vxx_internal_risk_pct=float("nan"), iv_put_internal_risk_pct=iv_put, iv_straddle_internal_risk_pct=iv_str, iv_rr_internal_risk_pct=iv_rr, status="iv_backtest_failed", sharpe=float("nan"), calmar=float("nan"), cagr_pct=float("nan"), return_pct=float("nan"), max_drawdown_pct=float("nan"), feasible_dd=False, put_frac=float("nan"), straddle_frac=float("nan"), rr_frac=float("nan"), vxx_frac=float("nan"), sum_frac=float("nan"), detail=msg_iv, ) trials.append(tr) print(f" failed: {msg_iv}", flush=True) continue iv_tag = _iv_combo_tag(iv_put, iv_str, iv_rr) for pct in vxx_pcts: print(f"[trial] vxx internal risk pct = {pct:.4f} | {iv_tag}", flush=True) ok, msg = _run_vxx_backtest_for_pct(args, pct, audit_dir=audit_run_dir, audit_tag_suffix=iv_tag) if not ok: tr = TrialResult( vxx_internal_risk_pct=pct, iv_put_internal_risk_pct=iv_put, iv_straddle_internal_risk_pct=iv_str, iv_rr_internal_risk_pct=iv_rr, status="vxx_backtest_failed", sharpe=float("nan"), calmar=float("nan"), cagr_pct=float("nan"), return_pct=float("nan"), max_drawdown_pct=float("nan"), feasible_dd=False, put_frac=float("nan"), straddle_frac=float("nan"), rr_frac=float("nan"), vxx_frac=float("nan"), sum_frac=float("nan"), detail=msg, ) trials.append(tr) print(f" failed: {msg}", flush=True) continue try: r = optimize_portfolio_sharpe( args.vrp_trades.expanduser(), total_portfolio_capital=float(args.total_capital), sum_risk_budget_pct=sum_cap, max_sleeve_pct=float(args.sharpe_max_sleeve_pct), max_dd_limit_pct=dd_cap, vxx_bear_pct=float(args.vxx_bear_pct), vxx_call_pct=float(args.vxx_call_pct), put_trades=LOGS / "stress_longvol_otm_put.jsonl", straddle_trades=LOGS / "stress_longvol_straddle.jsonl", risk_reversal_trades=LOGS / "risk_reversal.jsonl", maxiter=int(args.opt_maxiter), seed=int(args.opt_seed), polish=False, ) tr = TrialResult( vxx_internal_risk_pct=pct, iv_put_internal_risk_pct=iv_put, iv_straddle_internal_risk_pct=iv_str, iv_rr_internal_risk_pct=iv_rr, status="ok", sharpe=_safe_float(r.get("sharpe")), calmar=_safe_float(r.get("calmar")), cagr_pct=_safe_float(r.get("cagr_pct")), return_pct=_safe_float(r.get("return_pct")), max_drawdown_pct=_safe_float(r.get("max_drawdown_pct")), feasible_dd=bool(r.get("feasible_dd", True)), put_frac=_safe_float(r.get("put_frac")), straddle_frac=_safe_float(r.get("straddle_frac")), rr_frac=_safe_float(r.get("rr_frac")), vxx_frac=_safe_float(r.get("vxx_frac")), sum_frac=_safe_float(r.get("sum_frac")), detail=str(r.get("scipy_message", "")), ) except Exception as e: tr = TrialResult( vxx_internal_risk_pct=pct, iv_put_internal_risk_pct=iv_put, iv_straddle_internal_risk_pct=iv_str, iv_rr_internal_risk_pct=iv_rr, status="sleeve_opt_failed", sharpe=float("nan"), calmar=float("nan"), cagr_pct=float("nan"), return_pct=float("nan"), max_drawdown_pct=float("nan"), feasible_dd=False, put_frac=float("nan"), straddle_frac=float("nan"), rr_frac=float("nan"), vxx_frac=float("nan"), sum_frac=float("nan"), detail=str(e), ) r = None trials.append(tr) if tr.status == "ok": print( f" Sharpe={tr.sharpe:.3f} DD%={tr.max_drawdown_pct:.2f} " f"put={tr.put_frac:.4f} str={tr.straddle_frac:.4f} rr={tr.rr_frac:.4f} vxx={tr.vxx_frac:.4f}", flush=True, ) if best is None or tr.sharpe > best.sharpe: best = tr best_raw = r if audit_run_dir is not None: trial_tag = f"{iv_tag}__vxx_internal_{pct:.6f}".replace(".", "p") _copy_trial_audit_bundle( audit_run_dir=audit_run_dir, tag=trial_tag, paths=[ ("stress_longvol_otm_put", LOGS / "stress_longvol_otm_put.jsonl"), ("stress_longvol_straddle", LOGS / "stress_longvol_straddle.jsonl"), ("risk_reversal", LOGS / "risk_reversal.jsonl"), ("vxx_portfolio_bear_call", LOGS / "vxx_portfolio_bear_call.jsonl"), ("vxx_portfolio_deep_put", LOGS / "vxx_portfolio_deep_put.jsonl"), ("vxx_portfolio_long_call", LOGS / "vxx_portfolio_long_call.jsonl"), ], ) else: print(f" failed: {tr.detail}", flush=True) args.out_csv.parent.mkdir(parents=True, exist_ok=True) with args.out_csv.open("w", newline="", encoding="utf-8") as f: w = csv.DictWriter( f, fieldnames=[ "vxx_internal_risk_pct", "iv_put_internal_risk_pct", "iv_straddle_internal_risk_pct", "iv_rr_internal_risk_pct", "status", "sharpe", "calmar", "cagr_pct", "return_pct", "max_drawdown_pct", "feasible_dd", "put_frac", "straddle_frac", "rr_frac", "vxx_frac", "sum_frac", "detail", ], ) w.writeheader() for t in trials: w.writerow(asdict(t)) print(f"\nWrote trials CSV -> {args.out_csv}") if best is None or best_raw is None: print("ERROR: no successful trials.", file=sys.stderr) sys.exit(2) best_doc = { "mode": "overall-sizing-full-book", "vrp_trades": str(args.vrp_trades.expanduser()), "total_capital": float(args.total_capital), "date_window": {"start": args.start, "end": args.end}, "iv_date_window": {"start": iv_start, "end": iv_end}, "trade_audit": { "enabled": bool(audit_run_dir), "audit_root": str(audit_root) if audit_root is not None else None, "audit_run_dir": str(audit_run_dir) if audit_run_dir is not None else None, }, "constraints": { "sharpe_sum_cap_pct": sum_cap, "sharpe_max_sleeve_pct": float(args.sharpe_max_sleeve_pct), "sharpe_max_dd_pct": dd_cap, }, "vxx_internal_risk_pct_grid": vxx_pcts, "iv_internal_risk_dims": { "put": put_list, "straddle": st_list, "risk_reversal": rr_list, "active": bool(iv_active), }, "best_vxx_internal_risk_pct": float(best.vxx_internal_risk_pct), "best_iv_internal_risk_pct": { "put": best.iv_put_internal_risk_pct, "straddle": best.iv_straddle_internal_risk_pct, "risk_reversal": best.iv_rr_internal_risk_pct, }, "best_trial": asdict(best), "best_overlay_risk_fractions": { "stress_longvol_otm_put": float(best.put_frac), "stress_longvol_straddle": float(best.straddle_frac), "risk_reversal": float(best.rr_frac), "vxx_sleeve": float(best.vxx_frac), }, "best_full_metrics": { "sharpe": _safe_float(best_raw.get("sharpe")), "calmar": _safe_float(best_raw.get("calmar")), "cagr_pct": _safe_float(best_raw.get("cagr_pct")), "return_pct": _safe_float(best_raw.get("return_pct")), "max_drawdown_pct": _safe_float(best_raw.get("max_drawdown_pct")), "end_equity": _safe_float(best_raw.get("end_equity")), "feasible_dd": bool(best_raw.get("feasible_dd", True)), "scipy_success": bool(best_raw.get("scipy_success", False)), "scipy_message": str(best_raw.get("scipy_message", "")), "nit": int(best_raw.get("nit", 0)), }, } args.out_best_json.parent.mkdir(parents=True, exist_ok=True) args.out_best_json.write_text(json.dumps(best_doc, indent=2), encoding="utf-8") print(f"Wrote best JSON -> {args.out_best_json}") print( "Best: " f"vxx_internal={best.vxx_internal_risk_pct:.4f}, " f"iv_put={best.iv_put_internal_risk_pct}, iv_straddle={best.iv_straddle_internal_risk_pct}, iv_rr={best.iv_rr_internal_risk_pct}, " f"Sharpe={best.sharpe:.3f}, " f"put={best.put_frac:.4f}, str={best.straddle_frac:.4f}, " f"rr={best.rr_frac:.4f}, vxx={best.vxx_frac:.4f}" ) if __name__ == "__main__": main()