backtest/option_mapping.py

"""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