backtest/option_backtest.py

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.",
        ],
    }