""" Fixed-expiry VIX outright futures simulator (VX1 / VX2 / VX3). Markov signals use constant-tenor price percentiles, but **execution** locks a specific ``expiry`` at entry and rolls explicitly when that contract expires. """ from __future__ import annotations from dataclasses import dataclass import numpy as np import pandas as pd from RenTech.strategy_stack.vix_fixed_calendar_engine import VixContractStore OUTRIGHT_TENORS: dict[str, int] = { "VX1": 1, "VX2": 2, "VX3": 3, } @dataclass class OutrightLeg: """Open fixed-expiry outright VIX future.""" contract: str expiry: pd.Timestamp weight: float entry_date: pd.Timestamp entry_price: float entry_edge: float entry_decision: str cum_pnl_pct: float = 0.0 def _close_trade( leg: OutrightLeg, 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_price = store.price(md, leg.expiry) if not np.isfinite(exit_price) and mark_date is not None: exit_price = store.price(exit_date, leg.expiry) if not np.isfinite(exit_price): exit_price = leg.entry_price hold = (exit_date - leg.entry_date).days if hold < 0: hold = 0 return { "trade_id": trade_id, "contract": leg.contract, "direction": "LONG" if leg.weight > 0 else "SHORT", "entry_date": leg.entry_date.strftime("%Y-%m-%d"), "exit_date": exit_date.strftime("%Y-%m-%d"), "holding_days": max(1, hold), "expiry": leg.expiry.strftime("%Y-%m-%d"), "entry_price": leg.entry_price, "exit_price": exit_price, "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_contract": True, "model_type": "fixed_expiry_outright", } def simulate_fixed_outright_portfolio( *, store: VixContractStore, dates: pd.DatetimeIndex, target_weights: pd.DataFrame, edge_df: pd.DataFrame, decision_df: pd.DataFrame, contracts: list[str], capital: float, cash_annual_yield: float, gross_cap: float, weight_eps: float = 0.005, record_trades: bool = True, ) -> tuple[pd.DataFrame, pd.DataFrame]: """ Walk dates applying fixed-expiry outright legs per VX tenor slot. Returns (daily_portfolio_df, trade_log_df). """ daily_rf = cash_annual_yield / 252.0 open_legs: dict[str, OutrightLeg | None] = {c: None for c in contracts} trades: list[dict] = [] trade_id = 0 port_ret = pd.Series(0.0, index=dates) leg_pnl = pd.DataFrame(0.0, index=dates, columns=contracts) 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 contract in contracts: if contract not in target_weights.columns: continue tw = float(target_weights.loc[dt, contract]) if pd.notna(target_weights.loc[dt, contract]) else 0.0 edge = ( float(edge_df.loc[dt, contract]) if contract in edge_df.columns and pd.notna(edge_df.loc[dt, contract]) else float("nan") ) dec = str(decision_df.loc[dt, contract]) if contract in decision_df.columns else "PASS" tenor = OUTRIGHT_TENORS.get(contract) if tenor is None: continue leg = open_legs[contract] def _open_new(w: float) -> None: nonlocal leg, trade_id info = store.tenor_contract(dt, tenor) if info is None: return exp, px = info if exp <= dt or not np.isfinite(px) or px <= 0: return leg = OutrightLeg( contract=contract, expiry=exp, weight=w, entry_date=dt, entry_price=px, entry_edge=edge if np.isfinite(edge) else float("nan"), entry_decision=dec, ) open_legs[contract] = leg def _close(reason: str, mark_date: pd.Timestamp | None = None) -> None: nonlocal leg, trade_id if leg is None: return if record_trades: 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[contract] = None leg = None if leg is not None: roll = store.needs_roll(dt, leg.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.contract_return(prev_date, dt, leg.expiry) contrib = leg.weight * r leg.cum_pnl_pct += contrib day_pnl += contrib leg_pnl.loc[dt, contract] += contrib _close("contract_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[contract] elif flat or sign_flip: if prev_date is not None: r = store.contract_return(prev_date, dt, leg.expiry) contrib = leg.weight * r leg.cum_pnl_pct += contrib day_pnl += contrib leg_pnl.loc[dt, contract] += 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[contract] else: if prev_date is not None: r = store.contract_return(prev_date, dt, leg.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, contract] += contrib leg.weight = w_use else: if abs(tw) >= weight_eps and dec in ("LONG", "SHORT"): _open_new(tw) leg = open_legs[contract] leg = open_legs[contract] 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 if len(dates): last = dates[-1] for contract in contracts: leg = open_legs.get(contract) if leg is None: continue if record_trades: trade_id += 1 trades.append( _close_trade( leg, last, store, capital=capital, exit_reason="end_of_sample", exit_decision=( str(decision_df.loc[last, contract]) if contract in decision_df.columns else "PASS" ), exit_edge=( float(edge_df.loc[last, contract]) if contract 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 contract in contracts: if contract in target_weights.columns: daily[f"weight_{contract}"] = target_weights[contract] daily[f"edge_{contract}"] = ( edge_df[contract] if contract in edge_df.columns else np.nan ) daily[f"decision_{contract}"] = ( decision_df[contract] if contract in decision_df.columns else "PASS" ) daily[f"leg_pnl_{contract}"] = leg_pnl[contract] trade_df = pd.DataFrame(trades) return daily, trade_df