mathidot111/First_agent_template

1from __future__ import annotations2 3from dataclasses import asdict, dataclass4from datetime import timedelta5from pathlib import Path6 7import pandas as pd8 9from .vol_backtest import max_drawdown10 11 12REQUIRED_QUOTE_COLUMNS = {13    "date",14    "underlying_symbol",15    "underlying_price",16    "contract_symbol",17    "option_type",18    "expiration",19    "strike",20    "bid",21    "ask",22}23 24 25@dataclass26class OptionBacktestTrade:27    entry_date: str28    exit_date: str29    contract_symbol: str30    option_type: str31    strike: float32    expiration: str33    quantity: int34    entry_price: float35    exit_price: float36    fees: float37    pnl: float38 39    def to_dict(self) -> dict:40        return asdict(self)41 42 43def validate_quote_frame(quotes: pd.DataFrame) -> None:44    missing = REQUIRED_QUOTE_COLUMNS - set(quotes.columns)45    if missing:46        raise ValueError(f"Historical option quotes missing required columns: {sorted(missing)}")47 48 49def prepare_quotes(quotes: pd.DataFrame) -> pd.DataFrame:50    validate_quote_frame(quotes)51    frame = quotes.copy()52    frame["date"] = pd.to_datetime(frame["date"]).dt.normalize()53    frame["expiration"] = pd.to_datetime(frame["expiration"]).dt.normalize()54    frame["option_type"] = frame["option_type"].str.lower()55    quoted_mid = (frame["bid"] + frame["ask"]) / 256    if "mid" not in frame.columns:57        frame["mid"] = quoted_mid58    else:59        frame["mid"] = frame["mid"].where(frame["mid"].notna(), quoted_mid)60    frame["dte"] = (frame["expiration"] - frame["date"]).dt.days61    frame = frame[(frame["bid"] >= 0) & (frame["ask"] > 0) & (frame["dte"] >= 0)]62    return frame.sort_values(["date", "expiration", "strike", "option_type"]).reset_index(drop=True)63 64 65def load_option_quotes_csv(path: str | Path) -> pd.DataFrame:66    return prepare_quotes(pd.read_csv(path))67 68 69def available_exit_date(70    quotes: pd.DataFrame,71    entry_date: pd.Timestamp,72    target_exit_date: pd.Timestamp,73    contract_symbol: str,74) -> pd.Timestamp | None:75    contract_quotes = quotes[76        (quotes["contract_symbol"] == contract_symbol)77        & (quotes["date"] >= target_exit_date)78    ]79    if contract_quotes.empty:80        contract_quotes = quotes[quotes["contract_symbol"] == contract_symbol]81        contract_quotes = contract_quotes[82            (contract_quotes["date"] > entry_date)83            & (contract_quotes["date"] < target_exit_date)84        ]85        if contract_quotes.empty:86            return None87        return contract_quotes["date"].max()88    if contract_quotes.empty:89        return None90    return contract_quotes["date"].min()91 92 93def quote_price(row: pd.Series, side: str, price_field: str) -> float:94    if price_field == "mid":95        return float(row["mid"])96    if price_field != "trade":97        raise ValueError("price_field must be 'trade' or 'mid'.")98    if side == "buy":99        return float(row["ask"])100    return float(row["bid"])101 102 103def select_expiration_slice(day_quotes: pd.DataFrame, target_dte: int) -> pd.DataFrame:104    candidates = day_quotes[day_quotes["dte"] > 0]105    if candidates.empty:106        return candidates107    expiration = candidates.assign(dte_error=(candidates["dte"] - target_dte).abs()).sort_values("dte_error").iloc[0]["expiration"]108    return candidates[candidates["expiration"] == expiration]109 110 111def select_atm_contract(expiration_slice: pd.DataFrame, option_type: str) -> pd.Series | None:112    contracts = expiration_slice[expiration_slice["option_type"] == option_type]113    if contracts.empty:114        return None115    spot = float(expiration_slice["underlying_price"].iloc[0])116    return contracts.assign(strike_error=(contracts["strike"] - spot).abs()).sort_values("strike_error").iloc[0]117 118 119def backtest_long_straddle_from_quotes(120    quotes: pd.DataFrame,121    symbol: str,122    target_dte: int = 30,123    holding_days: int = 5,124    entry_every_days: int = 5,125    contract_multiplier: int = 100,126    fee_per_contract: float = 0.65,127    price_field: str = "trade",128) -> dict:129    frame = prepare_quotes(quotes)130    frame = frame[frame["underlying_symbol"].str.upper() == symbol.upper()]131    if frame.empty:132        raise ValueError(f"No historical option quotes found for {symbol}.")133 134    trades: list[OptionBacktestTrade] = []135    trade_groups = []136    equity = [0.0]137    dates = sorted(frame["date"].unique())138    next_entry_date = dates[0]139 140    for entry_date in dates:141        entry_date = pd.Timestamp(entry_date)142        if entry_date < next_entry_date:143            continue144 145        day_quotes = frame[frame["date"] == entry_date]146        expiration_slice = select_expiration_slice(day_quotes, target_dte)147        if expiration_slice.empty:148            continue149 150        call = select_atm_contract(expiration_slice, "call")151        put = select_atm_contract(expiration_slice, "put")152        if call is None or put is None:153            continue154 155        target_exit_date = entry_date + timedelta(days=holding_days)156        pending_group_trades = []157        group_pnl = 0.0158        for leg in [call, put]:159            exit_date = available_exit_date(frame, entry_date, target_exit_date, str(leg["contract_symbol"]))160            if exit_date is None:161                continue162            exit_quote = frame[163                (frame["date"] == exit_date)164                & (frame["contract_symbol"] == leg["contract_symbol"])165            ].iloc[0]166 167            entry_price = quote_price(leg, "buy", price_field)168            exit_price = quote_price(exit_quote, "sell", price_field)169            fees = fee_per_contract * 2170            pnl = (exit_price - entry_price) * contract_multiplier - fees171            trade = OptionBacktestTrade(172                entry_date=str(entry_date.date()),173                exit_date=str(pd.Timestamp(exit_date).date()),174                contract_symbol=str(leg["contract_symbol"]),175                option_type=str(leg["option_type"]),176                strike=float(leg["strike"]),177                expiration=str(pd.Timestamp(leg["expiration"]).date()),178                quantity=1,179                entry_price=round(entry_price, 4),180                exit_price=round(exit_price, 4),181                fees=round(fees, 2),182                pnl=round(pnl, 2),183            )184            pending_group_trades.append(trade)185            group_pnl += pnl186 187        if len(pending_group_trades) == 2:188            trades.extend(pending_group_trades)189            equity.append(equity[-1] + group_pnl)190            trade_groups.append(191                {192                    "entry_date": str(entry_date.date()),193                    "exit_date": pending_group_trades[0].exit_date,194                    "strategy": "long_straddle",195                    "pnl": round(group_pnl, 2),196                    "legs": [trade.to_dict() for trade in pending_group_trades],197                }198            )199            next_entry_date = entry_date + timedelta(days=entry_every_days)200 201    equity_series = pd.Series(equity)202    group_pnls = [group["pnl"] for group in trade_groups]203    wins = [pnl for pnl in group_pnls if pnl > 0]204    losses = [pnl for pnl in group_pnls if pnl <= 0]205 206    return {207        "strategy": "long_straddle",208        "symbol": symbol.upper(),209        "target_dte": target_dte,210        "holding_days": holding_days,211        "entry_every_days": entry_every_days,212        "contract_multiplier": contract_multiplier,213        "fee_per_contract": fee_per_contract,214        "price_field": price_field,215        "trade_count": len(trade_groups),216        "leg_trade_count": len(trades),217        "total_pnl": round(float(equity_series.iloc[-1]), 2) if not equity_series.empty else 0.0,218        "max_drawdown": round(max_drawdown(equity_series + 100000), 6),219        "win_rate": len(wins) / len(group_pnls) if group_pnls else 0.0,220        "avg_win": round(sum(wins) / len(wins), 2) if wins else 0.0,221        "avg_loss": round(sum(losses) / len(losses), 2) if losses else 0.0,222        "trades": trade_groups[:200],223        "data_requirements": [224            "Historical option quotes with date, expiration, strike, bid, ask, and underlying_price.",225            "For production-grade backtests, include deltas, IV, volume, open interest, and corporate action adjusted symbols.",226        ],227        "limitations": [228            "No early assignment model yet.",229            "No margin model yet.",230            "No intraday fills; entry and exit use the daily quote row.",231            "Results are only as good as the historical option quote data supplied.",232        ],233    }234