from __future__ import annotations from dataclasses import asdict, dataclass from datetime import timedelta from pathlib import Path import pandas as pd from .vol_backtest import max_drawdown REQUIRED_QUOTE_COLUMNS = { "date", "underlying_symbol", "underlying_price", "contract_symbol", "option_type", "expiration", "strike", "bid", "ask", } @dataclass class OptionBacktestTrade: entry_date: str exit_date: str contract_symbol: str option_type: str strike: float expiration: str quantity: int entry_price: float exit_price: float fees: float pnl: float def to_dict(self) -> dict: return asdict(self) def validate_quote_frame(quotes: pd.DataFrame) -> None: missing = REQUIRED_QUOTE_COLUMNS - set(quotes.columns) if missing: raise ValueError(f"Historical option quotes missing required columns: {sorted(missing)}") def prepare_quotes(quotes: pd.DataFrame) -> pd.DataFrame: validate_quote_frame(quotes) frame = quotes.copy() frame["date"] = pd.to_datetime(frame["date"]).dt.normalize() frame["expiration"] = pd.to_datetime(frame["expiration"]).dt.normalize() frame["option_type"] = frame["option_type"].str.lower() quoted_mid = (frame["bid"] + frame["ask"]) / 2 if "mid" not in frame.columns: frame["mid"] = quoted_mid else: frame["mid"] = frame["mid"].where(frame["mid"].notna(), quoted_mid) frame["dte"] = (frame["expiration"] - frame["date"]).dt.days frame = frame[(frame["bid"] >= 0) & (frame["ask"] > 0) & (frame["dte"] >= 0)] return frame.sort_values(["date", "expiration", "strike", "option_type"]).reset_index(drop=True) def load_option_quotes_csv(path: str | Path) -> pd.DataFrame: return prepare_quotes(pd.read_csv(path)) def available_exit_date( quotes: pd.DataFrame, entry_date: pd.Timestamp, target_exit_date: pd.Timestamp, contract_symbol: str, ) -> pd.Timestamp | None: contract_quotes = quotes[ (quotes["contract_symbol"] == contract_symbol) & (quotes["date"] >= target_exit_date) ] if contract_quotes.empty: contract_quotes = quotes[quotes["contract_symbol"] == contract_symbol] contract_quotes = contract_quotes[ (contract_quotes["date"] > entry_date) & (contract_quotes["date"] < target_exit_date) ] if contract_quotes.empty: return None return contract_quotes["date"].max() if contract_quotes.empty: return None return contract_quotes["date"].min() def quote_price(row: pd.Series, side: str, price_field: str) -> float: if price_field == "mid": return float(row["mid"]) if price_field != "trade": raise ValueError("price_field must be 'trade' or 'mid'.") if side == "buy": return float(row["ask"]) return float(row["bid"]) def select_expiration_slice(day_quotes: pd.DataFrame, target_dte: int) -> pd.DataFrame: candidates = day_quotes[day_quotes["dte"] > 0] if candidates.empty: return candidates expiration = candidates.assign(dte_error=(candidates["dte"] - target_dte).abs()).sort_values("dte_error").iloc[0]["expiration"] return candidates[candidates["expiration"] == expiration] def select_atm_contract(expiration_slice: pd.DataFrame, option_type: str) -> pd.Series | None: contracts = expiration_slice[expiration_slice["option_type"] == option_type] if contracts.empty: return None spot = float(expiration_slice["underlying_price"].iloc[0]) return contracts.assign(strike_error=(contracts["strike"] - spot).abs()).sort_values("strike_error").iloc[0] def backtest_long_straddle_from_quotes( quotes: pd.DataFrame, symbol: str, target_dte: int = 30, holding_days: int = 5, entry_every_days: int = 5, contract_multiplier: int = 100, fee_per_contract: float = 0.65, price_field: str = "trade", ) -> dict: frame = prepare_quotes(quotes) frame = frame[frame["underlying_symbol"].str.upper() == symbol.upper()] if frame.empty: raise ValueError(f"No historical option quotes found for {symbol}.") trades: list[OptionBacktestTrade] = [] trade_groups = [] equity = [0.0] dates = sorted(frame["date"].unique()) next_entry_date = dates[0] for entry_date in dates: entry_date = pd.Timestamp(entry_date) if entry_date < next_entry_date: continue day_quotes = frame[frame["date"] == entry_date] expiration_slice = select_expiration_slice(day_quotes, target_dte) if expiration_slice.empty: continue call = select_atm_contract(expiration_slice, "call") put = select_atm_contract(expiration_slice, "put") if call is None or put is None: continue target_exit_date = entry_date + timedelta(days=holding_days) pending_group_trades = [] group_pnl = 0.0 for leg in [call, put]: exit_date = available_exit_date(frame, entry_date, target_exit_date, str(leg["contract_symbol"])) if exit_date is None: continue exit_quote = frame[ (frame["date"] == exit_date) & (frame["contract_symbol"] == leg["contract_symbol"]) ].iloc[0] entry_price = quote_price(leg, "buy", price_field) exit_price = quote_price(exit_quote, "sell", price_field) fees = fee_per_contract * 2 pnl = (exit_price - entry_price) * contract_multiplier - fees trade = OptionBacktestTrade( entry_date=str(entry_date.date()), exit_date=str(pd.Timestamp(exit_date).date()), contract_symbol=str(leg["contract_symbol"]), option_type=str(leg["option_type"]), strike=float(leg["strike"]), expiration=str(pd.Timestamp(leg["expiration"]).date()), quantity=1, entry_price=round(entry_price, 4), exit_price=round(exit_price, 4), fees=round(fees, 2), pnl=round(pnl, 2), ) pending_group_trades.append(trade) group_pnl += pnl if len(pending_group_trades) == 2: trades.extend(pending_group_trades) equity.append(equity[-1] + group_pnl) trade_groups.append( { "entry_date": str(entry_date.date()), "exit_date": pending_group_trades[0].exit_date, "strategy": "long_straddle", "pnl": round(group_pnl, 2), "legs": [trade.to_dict() for trade in pending_group_trades], } ) next_entry_date = entry_date + timedelta(days=entry_every_days) equity_series = pd.Series(equity) group_pnls = [group["pnl"] for group in trade_groups] wins = [pnl for pnl in group_pnls if pnl > 0] losses = [pnl for pnl in group_pnls if pnl <= 0] return { "strategy": "long_straddle", "symbol": symbol.upper(), "target_dte": target_dte, "holding_days": holding_days, "entry_every_days": entry_every_days, "contract_multiplier": contract_multiplier, "fee_per_contract": fee_per_contract, "price_field": price_field, "trade_count": len(trade_groups), "leg_trade_count": len(trades), "total_pnl": round(float(equity_series.iloc[-1]), 2) if not equity_series.empty else 0.0, "max_drawdown": round(max_drawdown(equity_series + 100000), 6), "win_rate": len(wins) / len(group_pnls) if group_pnls else 0.0, "avg_win": round(sum(wins) / len(wins), 2) if wins else 0.0, "avg_loss": round(sum(losses) / len(losses), 2) if losses else 0.0, "trades": trade_groups[:200], "data_requirements": [ "Historical option quotes with date, expiration, strike, bid, ask, and underlying_price.", "For production-grade backtests, include deltas, IV, volume, open interest, and corporate action adjusted symbols.", ], "limitations": [ "No early assignment model yet.", "No margin model yet.", "No intraday fills; entry and exit use the daily quote row.", "Results are only as good as the historical option quote data supplied.", ], }