"""Option-contract mapping + option-candle handling for the OB Breaker backtest. STAGE 2 — options backtest layer. This module is READ-ONLY with respect to the broker: it only * reads the NFO instrument master (a local CSV) via option_utils.load_nfo_instruments * (optionally) pulls historical option candles via history_utils.get_option_candles It NEVER imports or calls any order-placement / GTT / buy-sell function. It does not create paper trades. It just turns an underlying directional signal into a concrete option contract and provides that contract's historical OHLCV. Design choices (correctness first): * Strike selection is ARITHMETIC (ATM = round(ref/step)*step, offsets by step), so it works identically for CSV-only backtests and for broker-master lookups. * The instrument master is used only for: auto strike-step inference, expiry listing, lot size, and the option tradingsymbol needed for broker fetch. * Everything degrades gracefully when the master is unavailable (CSV mode). """ from __future__ import annotations import glob import os import sys from dataclasses import dataclass, field from datetime import date, datetime from pathlib import Path from typing import Any, Dict, List, Optional import numpy as np import pandas as pd REPO_ROOT = Path(__file__).resolve().parent.parent if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- @dataclass class OptionBacktestConfig: # contract selection expiry_selection: str = "nearest_weekly" # nearest_weekly | monthly | manual manual_expiry: Optional[str] = None # YYYY-MM-DD (used if manual) strike_selection: str = "ATM" # ATM | ITM1 | ITM2 | OTM1 strike_step_mode: str = "auto" # auto | manual manual_strike_step: Optional[float] = None # sizing lots: int = 1 lot_size_override_by_symbol: Dict[str, int] = field(default_factory=dict) default_lot_size: int = 1 # last-resort fallback # data source option_data_source: str = "csv" # csv | broker option_csv_dir: str = "data/options" option_timeframe: str = "5minute" missing_option_candle_policy: str = "skip" # skip | next_available # exit behaviour option_exit_mode: str = "underlying_exit" # underlying_exit | option_premium_exit underlying_exit_price: str = "close" # close | open (for underlying_exit mode) option_stop_loss_pct: float = 25.0 option_target_pct: float = 40.0 same_candle_priority: str = "conservative" # conservative -> SL first # cost model (simple, configurable placeholders; percentages are of value) brokerage_per_order: float = 20.0 stt_sell_percent: float = 0.0625 # STT on option SELL premium exchange_charges_percent: float = 0.03503 # NSE txn charge on turnover sebi_charges_percent: float = 0.0001 # SEBI on turnover stamp_duty_percent: float = 0.003 # stamp duty on BUY value gst_percent_on_charges: float = 18.0 # GST on (brokerage+exch+sebi) option_slippage_points: float = 0.0 # premium points, adverse fills @classmethod def from_dict(cls, d: Dict[str, Any]) -> "OptionBacktestConfig": known = {f.name for f in cls.__dataclass_fields__.values()} # type: ignore[attr-defined] return cls(**{k: v for k, v in (d or {}).items() if k in known}) @dataclass class OptionContract: underlying_symbol: str option_type: str # CE | PE strike: float expiry: Optional[date] option_symbol: str # tradingsymbol (or synthetic label for CSV mode) lot_size: int # --------------------------------------------------------------------------- # Instrument master (read-only) — lazy, degrades to None if unavailable # --------------------------------------------------------------------------- _MASTER_CACHE: Optional[pd.DataFrame] = None _MASTER_TRIED = False def _normalize_master(df: pd.DataFrame) -> pd.DataFrame: """Normalize zerodha/dhan master shapes to columns: underlying, option_symbol, expiry(date), strike(float), lot_size, option_type. """ out = pd.DataFrame() cols = {c.lower(): c for c in df.columns} # underlying name if "name" in cols: out["underlying"] = df[cols["name"]].astype(str).str.upper() elif "underlying_symbol" in cols: out["underlying"] = df[cols["underlying_symbol"]].astype(str).str.upper() else: out["underlying"] = "" out["option_symbol"] = df[cols.get("tradingsymbol", list(df.columns)[0])].astype(str).str.upper() # option type if "instrument_type" in cols: out["option_type"] = df[cols["instrument_type"]].astype(str).str.upper() elif "option_type" in cols: out["option_type"] = df[cols["option_type"]].astype(str).str.upper() else: out["option_type"] = "" out["expiry"] = pd.to_datetime(df[cols["expiry"]], errors="coerce").dt.date if "expiry" in cols else None out["strike"] = pd.to_numeric(df[cols["strike"]], errors="coerce") if "strike" in cols else np.nan out["lot_size"] = pd.to_numeric(df[cols["lot_size"]], errors="coerce") if "lot_size" in cols else np.nan return out[out["option_type"].isin(["CE", "PE"])].copy() def load_master() -> Optional[pd.DataFrame]: """Return normalized NFO option master, or None if unavailable (CSV-only mode).""" global _MASTER_CACHE, _MASTER_TRIED if _MASTER_TRIED: return _MASTER_CACHE _MASTER_TRIED = True try: from option_utils import load_nfo_instruments # read-only master loader _MASTER_CACHE = _normalize_master(load_nfo_instruments()) except Exception as exc: # pragma: no cover - depends on local files/creds print(f"[option_mapping] instrument master unavailable ({exc}); " f"using CSV/manual mode.") _MASTER_CACHE = None return _MASTER_CACHE # --------------------------------------------------------------------------- # Strike step / strike / expiry / lot size # --------------------------------------------------------------------------- def infer_strike_step(symbol: str, cfg: OptionBacktestConfig, master: Optional[pd.DataFrame], ref_price: float) -> float: """Determine the strike interval for a symbol.""" if cfg.strike_step_mode == "manual" and cfg.manual_strike_step: return float(cfg.manual_strike_step) # auto: infer from the distinct strikes present in the master for this symbol if master is not None: sub = master[master["underlying"] == symbol.upper()]["strike"].dropna().unique() strikes = np.sort(np.unique(sub)) if len(strikes) >= 2: diffs = np.diff(strikes) diffs = diffs[diffs > 0] if len(diffs): return float(np.median(diffs)) if cfg.manual_strike_step: return float(cfg.manual_strike_step) # heuristic fallback by price magnitude (documented approximation) for hi, step in [(200, 2.5), (500, 5), (1000, 10), (2500, 20), (5000, 50)]: if ref_price < hi: return step return 100.0 def atm_strike(ref_price: float, step: float) -> float: return round(ref_price / step) * step def select_strike(ref_price: float, step: float, selection: str, cepe: str) -> float: """ATM/ITM1/ITM2/OTM1 with the direction-aware definitions from the spec. CE: ITM = below ATM, OTM = above ATM. PE: ITM = above ATM, OTM = below ATM. """ atm = atm_strike(ref_price, step) sel = selection.upper() if sel == "ATM": return atm n = {"ITM1": 1, "ITM2": 2, "OTM1": 1}.get(sel) if n is None: raise ValueError(f"Unknown strike_selection: {selection}") if cepe == "CE": direction = -1 if sel.startswith("ITM") else +1 # ITM below, OTM above else: # PE direction = +1 if sel.startswith("ITM") else -1 # ITM above, OTM below return atm + direction * n * step def select_expiry(symbol: str, signal_date: date, cfg: OptionBacktestConfig, master: Optional[pd.DataFrame]) -> Optional[date]: if cfg.expiry_selection == "manual": return pd.to_datetime(cfg.manual_expiry).date() if cfg.manual_expiry else None if master is None: # no master to enumerate expiries -> require manual return pd.to_datetime(cfg.manual_expiry).date() if cfg.manual_expiry else None exps = sorted({e for e in master[master["underlying"] == symbol.upper()]["expiry"].dropna() if e >= signal_date}) if not exps: return None if cfg.expiry_selection == "nearest_weekly": return exps[0] if cfg.expiry_selection == "monthly": # monthly = last expiry of a calendar month; pick the earliest such >= signal_date all_exps = sorted({e for e in master[master["underlying"] == symbol.upper()]["expiry"].dropna()}) by_month: Dict[tuple, date] = {} for e in all_exps: by_month[(e.year, e.month)] = max(by_month.get((e.year, e.month), e), e) monthlies = sorted(v for v in by_month.values() if v >= signal_date) return monthlies[0] if monthlies else exps[-1] return exps[0] def get_lot_size(symbol: str, strike: float, expiry: Optional[date], cepe: str, cfg: OptionBacktestConfig, master: Optional[pd.DataFrame]) -> int: ov = {k.upper(): v for k, v in (cfg.lot_size_override_by_symbol or {}).items()} if symbol.upper() in ov: return int(ov[symbol.upper()]) if master is not None: sub = master[(master["underlying"] == symbol.upper()) & (master["option_type"] == cepe)] if expiry is not None: e = sub[sub["expiry"] == expiry] if not e.empty and pd.notna(e["lot_size"].iloc[0]): return int(e["lot_size"].iloc[0]) vals = sub["lot_size"].dropna() if len(vals): return int(vals.mode().iloc[0]) return int(cfg.default_lot_size) def find_option_symbol(symbol: str, strike: float, expiry: Optional[date], cepe: str, master: Optional[pd.DataFrame]) -> Optional[str]: """Exact tradingsymbol lookup for broker fetch (read-only).""" if master is None: return None sub = master[(master["underlying"] == symbol.upper()) & (master["option_type"] == cepe) & (np.isclose(master["strike"], strike))] if expiry is not None: sub = sub[sub["expiry"] == expiry] if sub.empty: return None return str(sub["option_symbol"].iloc[0]) def build_contract(symbol: str, direction: str, ref_price: float, signal_dt, cfg: OptionBacktestConfig, master: Optional[pd.DataFrame]) -> Optional[OptionContract]: """Turn a directional signal into a concrete option contract. direction: "long" -> CE, "short" -> PE. Returns None if no expiry can be resolved. """ cepe = "CE" if direction == "long" else "PE" signal_date = pd.Timestamp(signal_dt).date() expiry = select_expiry(symbol, signal_date, cfg, master) step = infer_strike_step(symbol, cfg, master, ref_price) strike = select_strike(ref_price, step, cfg.strike_selection, cepe) lot_size = get_lot_size(symbol, strike, expiry, cepe, cfg, master) opt_symbol = find_option_symbol(symbol, strike, expiry, cepe, master) if opt_symbol is None: # synthetic label (used for CSV lookup / logging when master lacks the row) exp_tag = expiry.strftime("%Y%m%d") if expiry else "NA" opt_symbol = f"{symbol.upper()}_{exp_tag}_{int(round(strike))}_{cepe}" return OptionContract(underlying_symbol=symbol.upper(), option_type=cepe, strike=float(strike), expiry=expiry, option_symbol=opt_symbol, lot_size=int(lot_size)) # --------------------------------------------------------------------------- # Option candles (CSV or broker), read-only # --------------------------------------------------------------------------- def _normalize_option_df(df: pd.DataFrame) -> pd.DataFrame: ren = {} for c in df.columns: k = str(c).strip().lower() if k in ("datetime", "date", "time", "timestamp"): ren[c] = "timestamp" elif k in ("open", "high", "low", "close", "volume"): ren[c] = k df = df.rename(columns=ren) if "volume" not in df.columns: df["volume"] = 0.0 df = df[["timestamp", "open", "high", "low", "close", "volume"]].copy() df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce") df = df.dropna(subset=["timestamp"]) if df["timestamp"].dt.tz is None: df["timestamp"] = df["timestamp"].dt.tz_localize("Asia/Kolkata") else: df["timestamp"] = df["timestamp"].dt.tz_convert("Asia/Kolkata") for c in ("open", "high", "low", "close", "volume"): df[c] = pd.to_numeric(df[c], errors="coerce") return df.dropna(subset=["open", "high", "low", "close"]).sort_values("timestamp").reset_index(drop=True) def load_option_candles_csv(contract: OptionContract, cfg: OptionBacktestConfig) -> Optional[pd.DataFrame]: """Find a CSV under option_csv_dir matching this contract. Naming convention (flexible glob): must contain SYMBOL, STRIKE and CE/PE. e.g. data/options/AXISBANK_2026-06-26_1180_CE_5minute.csv """ d = Path(cfg.option_csv_dir) if not d.exists(): return None sym = contract.underlying_symbol.upper() strike_tag = str(int(round(contract.strike))) cepe = contract.option_type # try a few globs from most to least specific patterns = [ f"*{sym}*{strike_tag}*{cepe}*.csv", f"*{sym}*{cepe}*{strike_tag}*.csv", f"{contract.option_symbol}*.csv", ] for pat in patterns: for f in sorted(d.glob(pat)): name = f.name.upper() if sym in name and strike_tag in name and cepe in name: try: return _normalize_option_df(pd.read_csv(f)) except Exception: continue return None def fetch_option_candles_broker(contract: OptionContract, from_dt: datetime, to_dt: datetime, interval: str) -> Optional[pd.DataFrame]: """Read-only historical option candles via the repo's existing helper. NOTE: broker historical APIs frequently lack data for EXPIRED contracts, so this path is best-effort. CSV mode is the reliable route for old expiries. """ try: from history_utils import get_option_candles # read-only NFO history df = get_option_candles(contract.option_symbol, from_dt, to_dt, interval=interval) if df is None or df.empty: return None return _normalize_option_df(df) except Exception as exc: # pragma: no cover print(f"[option_mapping] broker option fetch failed for " f"{contract.option_symbol}: {exc}") return None def load_option_candles(contract: OptionContract, cfg: OptionBacktestConfig, from_dt=None, to_dt=None) -> Optional[pd.DataFrame]: if cfg.option_data_source == "broker": return fetch_option_candles_broker(contract, from_dt, to_dt, cfg.option_timeframe) return load_option_candles_csv(contract, cfg) def find_entry_index(opt_df: pd.DataFrame, entry_ts, policy: str) -> Optional[int]: """Locate the option candle for the underlying entry timestamp. Exact timestamp match preferred. If missing: 'skip' -> None, 'next_available' -> first candle at/after entry_ts (same day only). """ entry_ts = pd.Timestamp(entry_ts) ts = opt_df["timestamp"] exact = np.where((ts == entry_ts).to_numpy())[0] if len(exact): return int(exact[0]) if policy == "next_available": after = ts[(ts >= entry_ts) & (ts.dt.date == entry_ts.date())] if len(after): return int(after.index[0]) return None