#!/usr/bin/env python3 """ Run the legacy "winner" 4-sleeve portfolio with pinned definitions (not SID lookup). Pinned sleeves (from v3 catalog/results): - S063: short_rr gap=0.035, dte=40, putΔ=-0.18, callΔ=0.12, hold=5 - S067: short_rr gap=0.045, dte=40, putΔ=-0.18, callΔ=0.12, hold=5 - S060: putwrite SMA50, delta=-0.26, dte=35, hold=10 - S056: putwrite SMA200, delta=-0.26, dte=35, hold=10 Portfolio interpretation: - "4-lot book": each sleeve trades 1 lot; portfolio PnL is the sum of sleeve 1-lot PnL. """ from __future__ import annotations import argparse import json import math import sys import time from pathlib import Path import numpy as np import pandas as pd _REPO_ROOT = Path(__file__).resolve().parents[2] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from RenTech.strategy_stack import research_literature_theta_strategies as L from RenTech.strategy_stack.literature_search_agent import _compile_signal, _compile_trade from RenTech.strategy_stack.literature_strategy_catalog import StrategySpec _DEFAULT_THETA = _REPO_ROOT / "RenTech" / "data" / "theta_chunks" def legacy_specs() -> list[StrategySpec]: return [ StrategySpec( sid="S063_legacy", family="short_rr", description="IVput−IVcall>0.035 ΔP=-0.18 ΔC=0.12 DTE≈40 H=5d (legacy-pinned)", hold=5, sig_kind="rr", sig_params=(0.035, 40, -0.18, 0.12), trade_kind="rr", trade_params=(40, -0.18, 0.12), ), StrategySpec( sid="S067_legacy", family="short_rr", description="IVput−IVcall>0.045 ΔP=-0.18 ΔC=0.12 DTE≈40 H=5d (legacy-pinned)", hold=5, sig_kind="rr", sig_params=(0.045, 40, -0.18, 0.12), trade_kind="rr", trade_params=(40, -0.18, 0.12), ), StrategySpec( sid="S060_legacy", family="putwrite", description="SPY>SMA50 δ=-0.26 DTE≈35 H=10d (legacy-pinned)", hold=10, sig_kind="putwrite", sig_params=(-0.26, True, 22.0), trade_kind="put", trade_params=(35, -0.26), ), StrategySpec( sid="S056_legacy", family="putwrite", description="SPY>SMA200 δ=-0.26 DTE≈35 H=10d (legacy-pinned)", hold=10, sig_kind="putwrite", sig_params=(-0.26, False, 22.0), trade_kind="put", trade_params=(35, -0.26), ), ] def _repo_rel(p: Path) -> str: r = p.resolve() try: return str(r.relative_to(_REPO_ROOT)) except ValueError: return str(r) def main() -> None: ap = argparse.ArgumentParser(description="Legacy pinned 4-sleeve 1-lot-each portfolio run") ap.add_argument("--theta-dir", type=Path, default=_DEFAULT_THETA) ap.add_argument("--capital", type=float, default=1_000_000.0) ap.add_argument("--start", type=str, default="2016-01-04") ap.add_argument("--end", type=str, default="2026-04-30") ap.add_argument("--max-days", type=int, default=0) ap.add_argument( "--out-daily", type=Path, default=_REPO_ROOT / "RenTech" / "data" / "logs" / "literature_low_corr_portfolio_daily_pnl_legacy4_to_2026-04-30.csv", ) ap.add_argument( "--out-trades", type=Path, default=_REPO_ROOT / "RenTech" / "data" / "logs" / "literature_low_corr_portfolio_trade_log_legacy4_to_2026-04-30.csv", ) args = ap.parse_args() specs = legacy_specs() t0 = time.perf_counter() print(f"Building Theta context… ({len(specs)} sleeves)", flush=True) days, panel, get_chain, iv_atm, skew_put_minus_call_iv, n_contracts, spy_wide = ( L.prepare_theta_research_context( theta_dir=args.theta_dir, capital=float(args.capital), start=str(args.start), end=str(args.end), max_days=int(args.max_days), ) ) print(f"Context ready in {(time.perf_counter() - t0) / 60:.2f} min; {len(days)} sessions.", flush=True) idx = pd.DatetimeIndex([L._norm(d) for d in days]) daily_by_sid: dict[str, pd.Series] = {} rows: list[dict[str, object]] = [] for spec in specs: sig = _compile_signal(spec, panel, iv_atm, skew_put_minus_call_iv, n_contracts, spy_wide) tfn = _compile_trade(spec) t1 = time.perf_counter() trades = L.run_signal_backtest_trades( days, get_chain, panel, sig, int(spec.hold), tfn, spec.trade_params, spec.trade_kind, sid=spec.sid, family=spec.family, description=spec.description, ) print(f" {spec.sid} {spec.family}: {len(trades)} trades in {(time.perf_counter()-t1):.1f}s", flush=True) for t in trades: r = dict(t) r["qty_lots"] = 1 r["pnl_1x_usd"] = float(t["pnl_usd"]) r["pnl_sleeve_usd"] = float(t["pnl_usd"]) # 4-lot portfolio: no 1/N scaling r["pnl_portfolio_usd"] = float(t["pnl_usd"]) rows.append(r) ex = [pd.Timestamp(t["exit_date"]) for t in trades] pn = [float(t["pnl_usd"]) for t in trades] daily_by_sid[spec.sid] = L.daily_pnl_series(ex, pn, days) pnl = pd.Series(0.0, index=idx) for sid in daily_by_sid: pnl = pnl.add(daily_by_sid[sid], fill_value=0.0) cum = pnl.cumsum() eq = float(args.capital) + cum ret = eq.pct_change().replace([np.inf, -np.inf], np.nan).fillna(0.0) sh = L.sharpe_daily_returns(ret) dd = float((eq / eq.cummax() - 1.0).min()) if len(eq) else float("nan") out_d = args.out_daily.expanduser().resolve() out_t = args.out_trades.expanduser().resolve() out_d.parent.mkdir(parents=True, exist_ok=True) out_t.parent.mkdir(parents=True, exist_ok=True) daily_out = pd.DataFrame( { "pnl_portfolio_usd": pnl, "cumulative_pnl_usd": cum, "equity_usd": eq, "daily_return": ret, } ) for sid in daily_by_sid: daily_out[f"pnl_{sid}"] = daily_by_sid[sid] daily_out.to_csv(out_d) trades_df = pd.DataFrame(rows) if not trades_df.empty: trades_df = trades_df.sort_values(["entry_date", "sid"]).reset_index(drop=True) trades_df.to_csv(out_t, index=False) # import-ready by-lot view by_lot = out_t.with_name("literature_legacy4_trade_log_by_lot_to_2026-04-30.csv") if not trades_df.empty: b = trades_df.copy() b["strategy_group"] = "literature_legacy4_fixed_1lot" b["pnl_usd_per_lot"] = b["pnl_1x_usd"] cols = [ "strategy_group", "sid", "family", "description", "trade_kind", "entry_date", "exit_date", "hold_sessions", "calendar_days_in_trade", "entry_spy", "exit_spy", "vix_entry", "vix_exit", "qty_lots", "pnl_usd_per_lot", "pnl_1x_usd", "pnl_sleeve_usd", "legs_json", "n_legs", ] b = b[cols] b.to_csv(by_lot, index=False) meta = { "script": "literature_original4_legacy_run.py", "portfolio_mode": "4-lot book (1 lot per sleeve)", "legacy_specs": [s.sid for s in specs], "first_day": str(days[0].date()) if days else "", "last_day": str(days[-1].date()) if days else "", "sessions": len(days), "capital_start_usd": float(args.capital), "ending_equity_usd": float(eq.iloc[-1]) if len(eq) else float("nan"), "total_pnl_usd": float(cum.iloc[-1]) if len(cum) else float("nan"), "portfolio_sharpe_daily": float(sh) if math.isfinite(sh) else None, "max_drawdown_frac": dd if math.isfinite(dd) else None, "n_closed_trades": int(len(trades_df)), "entry_min": str(trades_df["entry_date"].min()) if len(trades_df) else "", "exit_max": str(trades_df["exit_date"].max()) if len(trades_df) else "", "out_daily_csv": _repo_rel(out_d), "out_trades_csv": _repo_rel(out_t), "out_by_lot_csv": _repo_rel(by_lot), } meta_path = out_d.parent / "literature_legacy4_meta_to_2026-04-30.json" meta_path.write_text(json.dumps(meta, indent=2), encoding="utf-8") print(json.dumps(meta, indent=2), flush=True) print(f"\nWrote {out_d}\nWrote {out_t}\nWrote {by_lot}\nWrote {meta_path}", flush=True) if __name__ == "__main__": main()