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