""" Event-driven **wide iron condor** backtest: enter **T−1** before the **FOMC decision**, exit **T+1** after, using :class:`~RenTech.core.synthetic_data_loader.SyntheticLoader` (or Parquet chains) to harvest post-event **IV contraction** on a fixed horizon (no intratrade TP/SL). Default event dates are **FOMC decision days only** (:data:`~RenTech.strategy_stack.macro_calendar.FOMC_DECISION_DATES`). On **T−1**, trades are taken only if **VIX ≥ 15** ("fear gate"); otherwise the event is skipped. Override with ``event_dates=`` if needed. """ from __future__ import annotations import sys import warnings from dataclasses import dataclass from pathlib import Path from typing import Literal 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.core.options_backtest import find_contract_in_chain from RenTech.core.options_data_loader import IVolatilityLoader, OptionChain, OptionContract from RenTech.core.synthetic_data_loader import SyntheticLoader, SyntheticOptionChain from RenTech.strategy_stack.macro_calendar import FOMC_DECISION_DATES from RenTech.strategy_stack.vrp_backtester import ( CONTRACT_MULTIPLIER, _close_long_per_share, _close_short_per_share, _mid_ok, _norm_day, _open_buy_per_share, _open_sell_per_share, contract_key, load_spy_vix_from_yfinance, normalize_spy_df, put_intrinsic_per_share, ) LegAction = Literal["buy", "sell"] # --------------------------------------------------------------------------- # FOMC-only calendar + fear gate (see macro_calendar.py) # --------------------------------------------------------------------------- FEAR_GATE_VIX_MIN: float = 15.0 MOCK_EVENT_DATES: list[pd.Timestamp] = list(FOMC_DECISION_DATES) ContractKeyT = tuple[pd.Timestamp, float, str] def choose_target_dte_exp_after_event(as_of: pd.Timestamp, event_date: pd.Timestamp) -> int: """ Smallest calendar **DTE** such that ``as_of + DTE`` is **strictly after** ``event_date`` (normalized), so the listed expiry sits past the event. """ as_n = _norm_day(as_of) ev_n = _norm_day(event_date) for dte in range(1, 46): exp = as_n + pd.Timedelta(days=int(dte)) if _norm_day(exp) > ev_n: return int(dte) return 7 # (target_delta, option_type letter, action) — wider wings vs 15Δ shorts for gap risk IC_LEG_SPECS: list[tuple[float, str, LegAction]] = [ (-0.10, "P", "sell"), (-0.02, "P", "buy"), (0.10, "C", "sell"), (0.02, "C", "buy"), ] @dataclass class IronCondorPosition: """ Four legs, one expiration. ``initial_net_credit`` = net premium at open (contract dollars). ``max_margin`` ≈ max wing width × 100 − net credit (capital at risk proxy). """ entry_date: pd.Timestamp contract_legs: tuple[OptionContract, OptionContract, OptionContract, OptionContract] leg_actions: tuple[LegAction, LegAction, LegAction, LegAction] leg_entry_exec_per_share: tuple[float, float, float, float] initial_net_credit: float max_margin: float @property def legs(self) -> tuple[OptionContract, ...]: return self.contract_legs @staticmethod def from_contracts( entry_date: pd.Timestamp, legs: list[OptionContract], actions: list[LegAction], ) -> IronCondorPosition: if len(legs) != 4 or len(actions) != 4: raise ValueError("Iron condor requires 4 legs and 4 actions.") execs: list[float] = [] net = 0.0 for leg, act in zip(legs, actions): if act == "buy": px = _open_buy_per_share(leg) if px is None: raise ValueError(f"No buy quote for leg K={leg.strike} {leg.option_type}.") net -= px * CONTRACT_MULTIPLIER else: px = _open_sell_per_share(leg) if px is None: raise ValueError(f"No sell quote for leg K={leg.strike} {leg.option_type}.") net += px * CONTRACT_MULTIPLIER execs.append(float(px)) k_ps = next(float(legs[i].strike) for i, a in enumerate(actions) if a == "sell" and legs[i].option_type == "P") k_pb = next(float(legs[i].strike) for i, a in enumerate(actions) if a == "buy" and legs[i].option_type == "P") k_cs = next(float(legs[i].strike) for i, a in enumerate(actions) if a == "sell" and legs[i].option_type == "C") k_cb = next(float(legs[i].strike) for i, a in enumerate(actions) if a == "buy" and legs[i].option_type == "C") put_w = abs(k_ps - k_pb) * CONTRACT_MULTIPLIER call_w = abs(k_cs - k_cb) * CONTRACT_MULTIPLIER wing_max = max(put_w, call_w) credit = max(net, 0.0) max_m = max(wing_max - credit, 1e-6) return IronCondorPosition( entry_date=_norm_day(entry_date), contract_legs=(legs[0], legs[1], legs[2], legs[3]), leg_actions=(actions[0], actions[1], actions[2], actions[3]), leg_entry_exec_per_share=(execs[0], execs[1], execs[2], execs[3]), initial_net_credit=float(net), max_margin=float(max_m), ) def calculate_current_mtm( self, current_chain: OptionChain, spy_closing_price: float, as_of: pd.Timestamp, ffill_mid: dict[ContractKeyT, float], ) -> float: spy = float(spy_closing_price) day = _norm_day(as_of) total = 0.0 for leg, act, ent in zip(self.contract_legs, self.leg_actions, self.leg_entry_exec_per_share): exp_n = _norm_day(leg.expiration) expired = day >= exp_n k = contract_key(leg) live = find_contract_in_chain( current_chain, leg.expiration, leg.strike, leg.option_type, ) exit_px: float | None = None if live is not None: if act == "buy": exit_px = _close_long_per_share(live) else: exit_px = _close_short_per_share(live) if exit_px is None: if expired: if leg.option_type == "P": exit_px = put_intrinsic_per_share(float(leg.strike), spy) else: exit_px = max(0.0, spy - float(leg.strike)) else: exit_px = ffill_mid.get(k) if exit_px is None and _mid_ok(leg.mid): exit_px = float(leg.mid) if exit_px is None: exit_px = 0.0 if act == "buy": total += (exit_px - ent) * CONTRACT_MULTIPLIER else: total += (ent - exit_px) * CONTRACT_MULTIPLIER return float(total) def _trading_day_before(index: pd.DatetimeIndex, event: pd.Timestamp) -> pd.Timestamp | None: ev = _norm_day(event) before = index[index < ev] if len(before) == 0: return None return _norm_day(pd.Timestamp(before.max())) def _trading_day_after(index: pd.DatetimeIndex, event: pd.Timestamp) -> pd.Timestamp | None: ev = _norm_day(event) after = index[index > ev] if len(after) == 0: return None return _norm_day(pd.Timestamp(after.min())) def _refresh_ffill(chain: OptionChain, store: dict[ContractKeyT, float]) -> None: for c in chain.contracts: if _mid_ok(c.mid): store[contract_key(c)] = float(c.mid) def _seed_from_ic(pos: IronCondorPosition, store: dict[ContractKeyT, float]) -> None: for leg in pos.contract_legs: if _mid_ok(leg.mid): store[contract_key(leg)] = float(leg.mid) def open_iron_condor( chain: SyntheticOptionChain, as_of: pd.Timestamp, event_date: pd.Timestamp, ) -> IronCondorPosition | None: """Build four legs with shared ``target_dte`` from :meth:`SyntheticOptionChain.find_target_leg`.""" target_dte = choose_target_dte_exp_after_event(as_of, event_date) legs: list[OptionContract] = [] actions: list[LegAction] = [] for delta, ot, act in IC_LEG_SPECS: try: leg = chain.find_target_leg( target_dte=target_dte, target_delta=delta, option_type=ot, ) except ValueError: return None legs.append(leg) actions.append(act) exps = {_norm_day(c.expiration) for c in legs} if len(exps) != 1: warnings.warn(f"IC legs landed on multiple expiries {exps}; check synthetic chain.", stacklevel=2) try: return IronCondorPosition.from_contracts(as_of, legs, actions) except ValueError: return None @dataclass class EventTradeResult: event_date: pd.Timestamp t_minus_1: pd.Timestamp t_plus_1: pd.Timestamp pnl_usd: float max_margin: float vix_entry: float vix_exit: float initial_net_credit: float def run_event_backtest( loader: SyntheticLoader | IVolatilityLoader, spy_df: pd.DataFrame, event_dates: list[pd.Timestamp] | None = None, ) -> list[EventTradeResult]: """ Backtest FOMC iron condors (default: :data:`FOMC_DECISION_DATES`). Skips an event when T−1 VIX is below :data:`FEAR_GATE_VIX_MIN`. Returned results include only trades that passed the gate and opened successfully. """ sdf = normalize_spy_df(spy_df) idx = sdf.index events = event_dates if event_dates is not None else MOCK_EVENT_DATES results: list[EventTradeResult] = [] for event in events: ev = _norm_day(pd.Timestamp(event)) tm1 = _trading_day_before(idx, ev) tp1 = _trading_day_after(idx, ev) if tm1 is None or tp1 is None: warnings.warn(f"Skip event {ev.date()}: no T-1 or T+1 in panel.", stacklevel=2) continue if tm1 not in sdf.index or tp1 not in sdf.index: continue vix_entry = float(sdf.loc[tm1, "vix_close"]) if vix_entry < FEAR_GATE_VIX_MIN: continue chain_entry = loader.get_chain_for_date(tm1) if not isinstance(chain_entry, SyntheticOptionChain): warnings.warn("SyntheticOptionChain expected for find_target_leg IC.", stacklevel=2) continue pos = open_iron_condor(chain_entry, tm1, ev) if pos is None: warnings.warn(f"Could not open IC on {tm1.date()} for event {ev.date()}.", stacklevel=2) continue ffill_mid: dict[ContractKeyT, float] = {} _seed_from_ic(pos, ffill_mid) chain_exit = loader.get_chain_for_date(tp1) if isinstance(loader, SyntheticLoader): loader.attach_position_quotes(chain_exit, pos) if chain_exit.contracts: _refresh_ffill(chain_exit, ffill_mid) spy_exit = float(sdf.loc[tp1, "close"]) vix_exit = float(sdf.loc[tp1, "vix_close"]) pnl = pos.calculate_current_mtm( chain_exit, spy_closing_price=spy_exit, as_of=tp1, ffill_mid=ffill_mid, ) results.append( EventTradeResult( event_date=ev, t_minus_1=tm1, t_plus_1=tp1, pnl_usd=float(pnl), max_margin=pos.max_margin, vix_entry=vix_entry, vix_exit=vix_exit, initial_net_credit=pos.initial_net_credit, ) ) return results def print_event_metrics(results: list[EventTradeResult]) -> None: """Metrics use only executed trades (post fear-gate), including average VIX at T−1 and T+1.""" n = len(results) if n == 0: print("Event-driven IC: no completed trades.") return wins = sum(1 for r in results if r.pnl_usd > 0) total_pnl = sum(r.pnl_usd for r in results) # Exclude margin floor (1e-6) when credit ≥ wing width in IronCondorPosition. rom = [r.pnl_usd / r.max_margin for r in results if r.max_margin > 1.0] avg_rom = sum(rom) / len(rom) if rom else 0.0 vix_in = sum(r.vix_entry for r in results) / n vix_out = sum(r.vix_exit for r in results) / n print("Event-driven wide iron condor (T−1 entry → T+1 exit, 1× multiplier)") print(f" Total events traded: {n}") print(f" Win rate: {wins / n:.2%}") print(f" Total PnL (USD): {total_pnl:,.2f}") print(f" Avg return on margin: {avg_rom:.4f} (PnL / max_margin; max_margin > $1)") print(f" Avg VIX on entry (T−1): {vix_in:.2f}") print(f" Avg VIX on exit (T+1): {vix_out:.2f}") print(f" Avg VIX change (exit−entry): {vix_out - vix_in:+.2f} (negative ⇒ crush)") if __name__ == "__main__": # Panel must cover T+1 after the last FOMC in FOMC_DECISION_DATES (see macro_calendar.py). _panel = load_spy_vix_from_yfinance("2005-01-01", "2026-12-31") _ld = SyntheticLoader(_panel) _res = run_event_backtest(_ld, _panel, MOCK_EVENT_DATES) print_event_metrics(_res)