""" Fixed-expiry VIX calendar spread simulator. Signals come from Markov mispricing on constant-tenor spread levels, but **execution** locks specific ``near_expiry`` / ``far_expiry`` contracts at entry and holds until flat, sign-flip, or near-leg expiry roll. """ from __future__ import annotations from dataclasses import dataclass from pathlib import Path import numpy as np import pandas as pd SPREAD_TENOR_PAIRS: dict[str, tuple[int, int]] = { "M1_M2": (1, 2), "M2_M3": (2, 3), "M3_M4": (3, 4), "M4_M5": (4, 5), } @dataclass class CalendarLeg: """Open fixed-expiry calendar spread.""" spread: str near_expiry: pd.Timestamp far_expiry: pd.Timestamp weight: float entry_date: pd.Timestamp entry_spread_pts: float entry_edge: float entry_decision: str cum_pnl_pct: float = 0.0 @dataclass class VixContractStore: """Price lookup by (trade_date, expiry).""" contracts: pd.DataFrame def __post_init__(self) -> None: df = self.contracts.copy() df["trade_date"] = pd.to_datetime(df["trade_date"]).dt.normalize() df["expiry"] = pd.to_datetime(df["expiry"]).dt.normalize() df["price"] = df["price"].astype(np.float64) self._by_date_expiry = df.set_index(["trade_date", "expiry"])["price"] self._by_date_tenor = ( df.dropna(subset=["Tenor_Monthly"]) .sort_values(["trade_date", "Tenor_Monthly"]) .drop_duplicates(["trade_date", "Tenor_Monthly"], keep="last") .set_index(["trade_date", "Tenor_Monthly"]) ) def price(self, trade_date: pd.Timestamp, expiry: pd.Timestamp) -> float: key = (pd.Timestamp(trade_date).normalize(), pd.Timestamp(expiry).normalize()) try: return float(self._by_date_expiry.loc[key]) except KeyError: return float("nan") def tenor_contract( self, trade_date: pd.Timestamp, tenor: int ) -> tuple[pd.Timestamp, float] | None: td = pd.Timestamp(trade_date).normalize() try: row = self._by_date_tenor.loc[(td, float(tenor))] if isinstance(row, pd.DataFrame): row = row.iloc[-1] exp = pd.Timestamp(row["expiry"]).normalize() return exp, float(row["price"]) except KeyError: return None def spread_pts(self, trade_date: pd.Timestamp, near_exp: pd.Timestamp, far_exp: pd.Timestamp) -> float: pn = self.price(trade_date, near_exp) pf = self.price(trade_date, far_exp) if not np.isfinite(pn) or not np.isfinite(pf): return float("nan") return pn - pf def spread_return( self, date_prev: pd.Timestamp, date_curr: pd.Timestamp, near_exp: pd.Timestamp, far_exp: pd.Timestamp, ) -> float: """Daily return on long near / short far for locked expiries.""" p0n = self.price(date_prev, near_exp) p1n = self.price(date_curr, near_exp) p0f = self.price(date_prev, far_exp) p1f = self.price(date_curr, far_exp) if not all(np.isfinite(x) and x > 0 for x in (p0n, p1n, p0f, p1f)): return 0.0 return (p1n / p0n - 1.0) - (p1f / p0f - 1.0) def contract_return( self, date_prev: pd.Timestamp, date_curr: pd.Timestamp, expiry: pd.Timestamp, ) -> float: """Daily return on a locked outright VIX future.""" p0 = self.price(date_prev, expiry) p1 = self.price(date_curr, expiry) if not all(np.isfinite(x) and x > 0 for x in (p0, p1)): return 0.0 return p1 / p0 - 1.0 def needs_roll(self, trade_date: pd.Timestamp, expiry: pd.Timestamp) -> bool: """Roll when contract expired or no longer quoted.""" return self.needs_near_roll(trade_date, expiry) def needs_near_roll(self, trade_date: pd.Timestamp, near_exp: pd.Timestamp) -> bool: """Roll when near contract expired or no longer quoted.""" td = pd.Timestamp(trade_date).normalize() ne = pd.Timestamp(near_exp).normalize() if td >= ne: return True return not np.isfinite(self.price(td, ne)) def _close_trade( leg: CalendarLeg, exit_date: pd.Timestamp, store: VixContractStore, *, capital: float, exit_reason: str, exit_decision: str, exit_edge: float, trade_id: int, mark_date: pd.Timestamp | None = None, ) -> dict: md = mark_date if mark_date is not None else exit_date exit_spread = store.spread_pts(md, leg.near_expiry, leg.far_expiry) if not np.isfinite(exit_spread) and mark_date is not None: exit_spread = store.spread_pts(exit_date, leg.near_expiry, leg.far_expiry) if not np.isfinite(exit_spread): exit_spread = leg.entry_spread_pts hold = (exit_date - leg.entry_date).days if hold < 0: hold = 0 return { "trade_id": trade_id, "spread": leg.spread, "direction": "LONG_SPREAD" if leg.weight > 0 else "SHORT_SPREAD", "entry_date": leg.entry_date.strftime("%Y-%m-%d"), "exit_date": exit_date.strftime("%Y-%m-%d"), "holding_days": max(1, hold), "near_expiry": leg.near_expiry.strftime("%Y-%m-%d"), "far_expiry": leg.far_expiry.strftime("%Y-%m-%d"), "entry_spread_pts": leg.entry_spread_pts, "exit_spread_pts": exit_spread, "entry_edge": leg.entry_edge, "exit_edge": exit_edge, "entry_weight": leg.weight, "pnl_pct_simple": leg.cum_pnl_pct, "pnl_usd": capital * leg.cum_pnl_pct, "entry_decision": leg.entry_decision, "exit_decision": exit_decision, "exit_reason": exit_reason, "same_near_contract": True, "model_type": "fixed_expiry_calendar", } def simulate_fixed_calendar_portfolio( *, store: VixContractStore, dates: pd.DatetimeIndex, target_weights: pd.DataFrame, edge_df: pd.DataFrame, decision_df: pd.DataFrame, spreads: list[str], capital: float, cash_annual_yield: float, gross_cap: float, weight_eps: float = 0.005, ) -> tuple[pd.DataFrame, pd.DataFrame]: """ Walk dates applying fixed-expiry calendar legs. Returns (daily_portfolio_df, trade_log_df). """ daily_rf = cash_annual_yield / 252.0 open_legs: dict[str, CalendarLeg | None] = {s: None for s in spreads} trades: list[dict] = [] trade_id = 0 port_ret = pd.Series(0.0, index=dates) leg_pnl = pd.DataFrame(0.0, index=dates, columns=spreads) gross = pd.Series(0.0, index=dates) prev_date: pd.Timestamp | None = None for dt in dates: day_pnl = 0.0 day_gross = 0.0 for spread in spreads: if spread not in target_weights.columns: continue tw = float(target_weights.loc[dt, spread]) if pd.notna(target_weights.loc[dt, spread]) else 0.0 edge = float(edge_df.loc[dt, spread]) if spread in edge_df.columns and pd.notna(edge_df.loc[dt, spread]) else float("nan") dec = str(decision_df.loc[dt, spread]) if spread in decision_df.columns else "PASS" leg = open_legs[spread] def _open_new(w: float) -> None: nonlocal leg, trade_id pair = SPREAD_TENOR_PAIRS[spread] near_info = store.tenor_contract(dt, pair[0]) far_info = store.tenor_contract(dt, pair[1]) if near_info is None or far_info is None: return near_exp, _ = near_info far_exp, _ = far_info if near_exp >= far_exp or near_exp <= dt: return sp = store.spread_pts(dt, near_exp, far_exp) if not np.isfinite(sp): return leg = CalendarLeg( spread=spread, near_expiry=near_exp, far_expiry=far_exp, weight=w, entry_date=dt, entry_spread_pts=sp, entry_edge=edge if np.isfinite(edge) else float("nan"), entry_decision=dec, ) open_legs[spread] = leg def _close(reason: str, mark_date: pd.Timestamp | None = None) -> None: nonlocal leg, trade_id if leg is None: return trade_id += 1 trades.append( _close_trade( leg, dt, store, capital=capital, exit_reason=reason, exit_decision=dec, exit_edge=edge if np.isfinite(edge) else float("nan"), trade_id=trade_id, mark_date=mark_date, ) ) open_legs[spread] = None leg = None # Manage existing leg if leg is not None: roll = store.needs_near_roll(dt, leg.near_expiry) sign_flip = (tw * leg.weight) < 0 and abs(tw) >= weight_eps flat = abs(tw) < weight_eps or dec == "PASS" if roll: mark_dt = prev_date if prev_date is not None else dt if prev_date is not None: r = store.spread_return(prev_date, dt, leg.near_expiry, leg.far_expiry) contrib = leg.weight * r leg.cum_pnl_pct += contrib day_pnl += contrib leg_pnl.loc[dt, spread] += contrib _close("near_expiry_roll", mark_date=mark_dt) if abs(tw) >= weight_eps and dec in ("LONG", "SHORT") and not sign_flip: _open_new(tw) leg = open_legs[spread] elif flat or sign_flip: if prev_date is not None: r = store.spread_return(prev_date, dt, leg.near_expiry, leg.far_expiry) contrib = leg.weight * r leg.cum_pnl_pct += contrib day_pnl += contrib leg_pnl.loc[dt, spread] += contrib reason = "sign_flip" if sign_flip and not flat else "flat" _close(reason) if sign_flip and abs(tw) >= weight_eps and dec in ("LONG", "SHORT"): _open_new(tw) leg = open_legs[spread] else: if prev_date is not None: r = store.spread_return(prev_date, dt, leg.near_expiry, leg.far_expiry) w_use = tw if abs(tw) >= weight_eps else leg.weight contrib = w_use * r leg.cum_pnl_pct += contrib day_pnl += contrib leg_pnl.loc[dt, spread] += contrib leg.weight = w_use else: if abs(tw) >= weight_eps and dec in ("LONG", "SHORT"): _open_new(tw) leg = open_legs[spread] leg = open_legs[spread] if leg is not None: day_gross += abs(leg.weight) day_gross = min(day_gross, gross_cap) cash_w = max(0.0, 1.0 - day_gross) port_ret.loc[dt] = day_pnl + cash_w * daily_rf gross.loc[dt] = day_gross prev_date = dt # End of sample: close open legs if len(dates): last = dates[-1] for spread in spreads: leg = open_legs.get(spread) if leg is None: continue trade_id += 1 trades.append( _close_trade( leg, last, store, capital=capital, exit_reason="end_of_sample", exit_decision=str(decision_df.loc[last, spread]) if spread in decision_df.columns else "PASS", exit_edge=float(edge_df.loc[last, spread]) if spread in edge_df.columns else float("nan"), trade_id=trade_id, ) ) daily = pd.DataFrame( { "portfolio_bar_ret": port_ret, "gross_exposure": gross, "cash_weight": (1.0 - gross).clip(lower=0.0), }, index=dates, ) for spread in spreads: if spread in target_weights.columns: daily[f"weight_{spread}"] = target_weights[spread] daily[f"edge_{spread}"] = edge_df[spread] if spread in edge_df.columns else np.nan daily[f"decision_{spread}"] = decision_df[spread] if spread in decision_df.columns else "PASS" daily[f"leg_pnl_{spread}"] = leg_pnl[spread] trade_df = pd.DataFrame(trades) return daily, trade_df def load_contract_store(path: Path) -> VixContractStore: df = pd.read_parquet(path) return VixContractStore(df)