backtest/run_backtest.py

"""Backtest runner for the Order Block MTF strategy.

Extended run: ~4 months (2026-02/03 -> 2026-07-17), a wider symbol set, both
engines plus a BigBeluga ``obmode=Full`` variant, and the tightened filter set
(``require_bos_only`` / ``fvg_window_bars`` / ``gap_fill_check`` / ``itm_delta``).

Evaluation split (per the plan):
* **Underlying R** is computed for every trade over the whole 4-month window.
* **Option rupee P&L** is only computed for entries from ``option_pnl_start``
  (default 2026-07-01) -- the current expiry month -- because the 28-Jul-2026
  contract's history doesn't cover the earlier months. Earlier trades are
  ``underlying_only`` (R only). Stats therefore report R over all trades and
  rupee metrics over current-month option trades only.

Grid: indicator {luxalgo, bigbeluga (Length), bigbeluga_full}
      x entry {aggressive, conservative} x tp {swing, rr}
      x opening_filter {on, off}   (= 24 configs)

Signals are computed once per (symbol, engine) and reused across the strategy
variants. Outputs under ``results/``.
"""

from __future__ import annotations

import argparse
import itertools
import math
from datetime import datetime
from pathlib import Path

import numpy as np
import pandas as pd

from data import fetch_kite
from strategy.orderblock_mtf import (
    StrategyConfig, make_adapter, compute_signals, run_symbol,
)

HERE = Path(__file__).resolve().parent
ROOT = HERE.parent
RESULTS = ROOT / "results"
RESULTS.mkdir(exist_ok=True)

EXPIRY = "2026-07-28"
HOURLY_FROM = datetime(2026, 2, 1)     # 60m warm-up (ATR200 warm by ~mid-Mar)
FIVEMIN_FROM = datetime(2026, 3, 1)    # 5m history (Kite serves from ~2026-03-02)
TO = datetime(2026, 7, 17, 15, 30)
RISK_FREE = 0.065
ENGINES = ["luxalgo", "bigbeluga", "bigbeluga_full"]


# --------------------------------------------------------------------------- #
# Black-Scholes (erf-based; no scipy)
# --------------------------------------------------------------------------- #
def _ncdf(x):
    return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))


def bs_price(S, K, T, r, sigma, right):
    if T <= 0 or sigma <= 0:
        return max(0.0, (S - K) if right == "CE" else (K - S))
    d1 = (math.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * math.sqrt(T))
    d2 = d1 - sigma * math.sqrt(T)
    if right == "CE":
        return S * _ncdf(d1) - K * math.exp(-r * T) * _ncdf(d2)
    return K * math.exp(-r * T) * _ncdf(-d2) - S * _ncdf(-d1)


def bs_delta(S, K, T, r, sigma, right):
    if T <= 0 or sigma <= 0:
        if right == "CE":
            return 1.0 if S > K else 0.0
        return -1.0 if S < K else 0.0
    d1 = (math.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * math.sqrt(T))
    return _ncdf(d1) if right == "CE" else _ncdf(d1) - 1.0


# --------------------------------------------------------------------------- #
# Option loader
# --------------------------------------------------------------------------- #
class OptionLoader:
    def __init__(self, kite, nfo, df_5m_by_symbol):
        self.kite = kite
        self.nfo = nfo
        self.df_5m_by_symbol = df_5m_by_symbol
        self._chain_cache = {}
        self._series_cache = {}
        self._vol_cache = {}

    def _chain(self, symbol):
        if symbol not in self._chain_cache:
            self._chain_cache[symbol] = fetch_kite.option_chain(
                self.nfo, symbol, EXPIRY)
        return self._chain_cache[symbol]

    def _sigma(self, symbol):
        if symbol not in self._vol_cache:
            u = self.df_5m_by_symbol[symbol]
            rets = np.log(u["close"]).diff().dropna()
            s = float(rets.std() * math.sqrt(75 * 252)) if len(rets) > 5 else 0.3
            self._vol_cache[symbol] = min(max(s, 0.05), 2.0)
        return self._vol_cache[symbol]

    def _select(self, symbol, right, spot, entry_time, itm_delta):
        chain = self._chain(symbol)
        if chain.empty:
            return None
        side = chain[chain["instrument_type"] == right]
        if side.empty:
            return None
        if itm_delta is None:
            return side.loc[(side["strike"] - spot).abs().idxmin()]
        sigma = self._sigma(symbol)
        u = self.df_5m_by_symbol[symbol]
        expiry_ts = pd.Timestamp(EXPIRY + " 15:30", tz=u.index.tz)
        T = max((expiry_ts - entry_time).total_seconds() / (365 * 24 * 3600), 1e-6)
        d = side["strike"].apply(
            lambda K: abs(bs_delta(spot, float(K), T, RISK_FREE, sigma, right)))
        return side.loc[(d - itm_delta).abs().idxmin()]

    def __call__(self, symbol, right, spot, entry_time, itm_delta=None):
        row = self._select(symbol, right, spot, entry_time, itm_delta)
        if row is None:
            return None
        strike = float(row["strike"])
        key = (symbol, right, strike)
        if key in self._series_cache:
            return self._series_cache[key]
        odf = fetch_kite.fetch_candles(
            self.kite, int(row["instrument_token"]), FIVEMIN_FROM, TO,
            "5minute", label=row["tradingsymbol"])
        lot = int(row.get("lot_size", 1) or 1)
        if odf.empty:
            odf = self._proxy_series(symbol, strike, right)
            proxy = True
        else:
            proxy = False
        res = dict(df=odf, symbol=row["tradingsymbol"], strike=strike,
                   lot_size=lot, proxy=proxy)
        self._series_cache[key] = res
        return res

    def _proxy_series(self, symbol, strike, right):
        u = self.df_5m_by_symbol[symbol]
        sigma = self._sigma(symbol)
        expiry_ts = pd.Timestamp(EXPIRY + " 15:30", tz=u.index.tz)
        rows = {}
        for t, bar in u.iterrows():
            T = max((expiry_ts - t).total_seconds() / (365 * 24 * 3600), 1e-6)
            rows[t] = {c: bs_price(float(bar[c]), strike, T, RISK_FREE, sigma, right)
                       for c in ["open", "high", "low", "close"]}
        df = pd.DataFrame(rows).T
        df.index = u.index
        return df[["open", "high", "low", "close"]]


# --------------------------------------------------------------------------- #
# Stats (R over all trades; rupees over current-month option trades only)
# --------------------------------------------------------------------------- #
def summarize(trades: pd.DataFrame) -> dict:
    n = len(trades)
    if n == 0:
        return dict(n_trades=0, n_opt_trades=0, win_rate=np.nan, avg_r=np.nan,
                    median_r=np.nan, total_pnl=np.nan, profit_factor=np.nan,
                    max_dd_pnl=np.nan, avg_pnl=np.nan)
    win_rate = (trades["realized_r"] > 0).mean()
    avg_r = trades["realized_r"].mean()
    median_r = trades["realized_r"].median()
    opt = trades[~trades["underlying_only"]].dropna(subset=["pnl_rupees"])
    if opt.empty:
        return dict(n_trades=n, n_opt_trades=0, win_rate=win_rate, avg_r=avg_r,
                    median_r=median_r, total_pnl=np.nan, profit_factor=np.nan,
                    max_dd_pnl=np.nan, avg_pnl=np.nan)
    gains = opt.loc[opt["pnl_rupees"] > 0, "pnl_rupees"].sum()
    losses = -opt.loc[opt["pnl_rupees"] < 0, "pnl_rupees"].sum()
    pf = (gains / losses) if losses > 0 else np.inf
    eq = opt.sort_values("exit_time")["pnl_rupees"].cumsum()
    dd = float((eq - eq.cummax()).min()) if len(eq) else 0.0
    return dict(n_trades=n, n_opt_trades=len(opt), win_rate=win_rate, avg_r=avg_r,
                median_r=median_r, total_pnl=opt["pnl_rupees"].sum(),
                profit_factor=pf, max_dd_pnl=dd, avg_pnl=opt["pnl_rupees"].mean())


# --------------------------------------------------------------------------- #
# Data + symbols
# --------------------------------------------------------------------------- #
def pick_symbols(n_total=50):
    u = pd.read_csv(ROOT / "option_stock_universe.csv")
    rank1 = list(u[u["priority_rank"] == 1.0]["symbol"])
    rest = u[u["priority_rank"] != 1.0].sort_values("option_rows", ascending=False)
    fill = [s for s in rest["symbol"] if s not in rank1]
    out = rank1 + fill[: max(0, n_total - len(rank1))]
    return out


def load_underlyings(kite, nse, symbols):
    d1h, d5m, ok = {}, {}, []
    for s in symbols:
        try:
            token = fetch_kite.nse_token(nse, s)
        except KeyError:
            print(f"  skip {s}: no NSE token")
            continue
        h = fetch_kite.fetch_candles(kite, token, HOURLY_FROM, TO, "60minute", label=s)
        f = fetch_kite.fetch_candles(kite, token, FIVEMIN_FROM, TO, "5minute", label=s)
        if h.empty or f.empty:
            print(f"  skip {s}: empty candles")
            continue
        d1h[s], d5m[s] = h, f
        ok.append(s)
    return d1h, d5m, ok


# --------------------------------------------------------------------------- #
# Main
# --------------------------------------------------------------------------- #
def run(symbols):
    kite = fetch_kite.get_kite()
    nse = fetch_kite.load_instruments(kite, "NSE")
    nfo = fetch_kite.load_instruments(kite, "NFO")
    print(f"Loading {len(symbols)} underlyings...")
    d1h, d5m, symbols = load_underlyings(kite, nse, symbols)
    print(f"  {len(symbols)} usable symbols")
    loader = OptionLoader(kite, nfo, d5m)

    base = StrategyConfig()
    print("Precomputing engine signals (once per symbol x engine)...")
    sig_cache = {}
    for eng in ENGINES:
        for s in symbols:
            ad = make_adapter(eng)
            sig_cache[(eng, s)] = (ad, compute_signals(d1h[s], d5m[s], ad, base))

    grid = list(itertools.product(
        ENGINES, ["aggressive", "conservative"], ["swing", "rr"], [True, False]))

    all_summ, all_audit = [], []
    for eng, entry, tp, opf in grid:
        cfg = StrategyConfig(entry_mode=entry, tp_mode=tp, opening_filter=opf)
        cfg_id = f"{eng}_{entry}_{tp}_opf{int(opf)}"
        trades_all, audit_all, seen_audit = [], [], False
        for s in symbols:
            adapter, sigs = sig_cache[(eng, s)]
            tdf, adf = run_symbol(s, d1h[s], d5m[s], adapter, cfg, loader, signals=sigs)
            if not tdf.empty:
                trades_all.append(tdf)
            if not adf.empty and not seen_audit and entry == "aggressive" and tp == "swing" and opf:
                audit_all.append(adf.assign(config=cfg_id))
        if audit_all:
            all_audit.append(pd.concat(audit_all, ignore_index=True))
        trades = pd.concat(trades_all, ignore_index=True) if trades_all else pd.DataFrame()
        trades.to_csv(RESULTS / f"trades_{cfg_id}.csv", index=False)
        summ = summarize(trades)
        summ.update(dict(config=cfg_id, indicator=eng, entry_mode=entry,
                         tp_mode=tp, opening_filter=opf))
        all_summ.append(summ)
        pnl = summ["total_pnl"]
        print(f"{cfg_id:44s} n={summ['n_trades']:3d} opt={summ['n_opt_trades']:3d} "
              f"avgR={summ['avg_r']:+.2f} pnl={'' if pd.isna(pnl) else int(pnl)}")

    summ_df = pd.DataFrame(all_summ)[
        ["config", "indicator", "entry_mode", "tp_mode", "opening_filter",
         "n_trades", "n_opt_trades", "win_rate", "avg_r", "median_r",
         "profit_factor", "total_pnl", "max_dd_pnl", "avg_pnl"]]
    summ_df.to_csv(RESULTS / "summary_stats.csv", index=False)
    if all_audit:
        pd.concat(all_audit, ignore_index=True).to_csv(
            RESULTS / "signal_audit.csv", index=False)

    comp = summ_df.groupby("indicator").agg(
        configs=("config", "count"),
        total_trades=("n_trades", "sum"),
        opt_trades=("n_opt_trades", "sum"),
        mean_avg_r=("avg_r", "mean"),
        mean_win_rate=("win_rate", "mean"),
        total_pnl_optmonth=("total_pnl", "sum"),
    ).reset_index()
    comp.to_csv(RESULTS / "comparison_luxalgo_vs_bigbeluga.csv", index=False)

    _plot(symbols, grid)
    print("\n=== summary_stats.csv ===")
    with pd.option_context("display.width", 200, "display.max_columns", 20):
        print(summ_df.to_string(index=False))
    print("\n=== comparison (R over all trades; pnl = current-month options) ===")
    print(comp.to_string(index=False))
    total_r = summ_df["n_trades"].sum()
    print(f"\nTotal config-trades={total_r}; unique R-evaluated trades per config "
          f"range {summ_df['n_trades'].min()}-{summ_df['n_trades'].max()}.")
    return summ_df


def _plot(symbols, grid):
    try:
        import matplotlib
        matplotlib.use("Agg")
        import matplotlib.pyplot as plt
    except Exception:
        return
    # Equity in R (all trades) and rupees (option trades) -- one panel each
    fig, (axr, axp) = plt.subplots(1, 2, figsize=(15, 6))
    for eng, entry, tp, opf in grid:
        if not opf:
            continue  # de-clutter: plot opening-filter-on configs only
        cfg_id = f"{eng}_{entry}_{tp}_opf{int(opf)}"
        f = RESULTS / f"trades_{cfg_id}.csv"
        if not f.exists() or f.stat().st_size == 0:
            continue
        try:
            t = pd.read_csv(f)
        except pd.errors.EmptyDataError:
            continue
        if t.empty or "exit_time" not in t.columns:
            continue
        t = t.sort_values("exit_time")
        axr.plot(pd.to_datetime(t["exit_time"]), t["realized_r"].cumsum(),
                 marker=".", ms=3, label=cfg_id)
        opt = t[~t["underlying_only"]].dropna(subset=["pnl_rupees"])
        if not opt.empty:
            axp.plot(pd.to_datetime(opt["exit_time"]), opt["pnl_rupees"].cumsum(),
                     marker="o", ms=3, label=cfg_id)
    axr.set_title("Cumulative underlying R (all trades, 4 months)")
    axr.set_ylabel("cumulative R"); axr.grid(alpha=0.3); axr.legend(fontsize=5, ncol=2)
    axp.set_title("Cumulative option P&L (current-month trades)")
    axp.set_ylabel("Rs"); axp.grid(alpha=0.3); axp.legend(fontsize=5, ncol=2)
    fig.tight_layout()
    fig.savefig(RESULTS / "equity_curves.png", dpi=110)
    plt.close(fig)


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--symbols", nargs="*", default=None)
    ap.add_argument("--n", type=int, default=50)
    ap.add_argument("--quick", action="store_true", help="RELIANCE only")
    args = ap.parse_args()
    if args.quick:
        run(["RELIANCE"])
    else:
        run(args.symbols or pick_symbols(args.n))