"""Multi-symbol, multi-day backtest for the MACD + Parabolic SAR strategy. Runs the exact entry/exit rules (see ../indicators_macd_psar.py, exit_mode=both) across a set of stocks over a date range, priced on the REAL ATM CE/PE option candles, and reports per-symbol and aggregate net P&L. Usage: python macd_psar/backtest_range.py 2026-07-01 2026-07-17 python macd_psar/backtest_range.py 2026-07-01 2026-07-17 AXISBANK,SBIN Defaults to the priority_rank==1 stocks in option_stock_universe.csv. """ import os import sys from pathlib import Path import pandas as pd ENTRY_MODE = os.getenv("MP_ENTRY_MODE", "state").strip().lower() # state | cross_zero sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from indicators_macd_psar import add_macd_psar_indicators, mp_exit_reason # noqa: E402 from macd_psar.backtest_macd_psar import ( # noqa: E402 TZ, SESSION_START, ENTRY_CUTOFF, EOD_EXIT, EXIT_MODE, WARMUP_DAYS, _get_kite, _instrument_token, _round_trip_charges, _option_close_at, ) # Preload the NFO chain once (the single-day helper re-reads it per call). _NFO = pd.read_csv(Path(__file__).resolve().parent.parent / "instruments_nfo.csv") _NFO["name"] = _NFO["name"].astype(str).str.upper() _NFO["expiry_d"] = pd.to_datetime(_NFO["expiry"], errors="coerce").dt.date def _priority1_symbols(): uni = pd.read_csv(Path(__file__).resolve().parent.parent / "option_stock_universe.csv") return uni[uni["priority_rank"] == 1.0]["symbol"].astype(str).str.upper().tolist() def _atm_contract(symbol, direction, ref_price, trade_day): opt_type = "CE" if direction == "CALL" else "PE" c = _NFO[(_NFO["name"] == symbol.upper()) & (_NFO["instrument_type"] == opt_type)] future = c[c["expiry_d"] >= trade_day] if future.empty: return None nearest_exp = min(future["expiry_d"].unique()) chain = future[future["expiry_d"] == nearest_exp].copy() chain["dist"] = (chain["strike"].astype(float) - float(ref_price)).abs() pick = chain.sort_values(["dist", "strike"]).iloc[0] return { "tradingsymbol": str(pick["tradingsymbol"]).upper(), "instrument_token": int(pick["instrument_token"]), "strike": int(pick["strike"]), "lot_size": int(pick["lot_size"]), } def _fetch_5m(kite, token, frm_dt, to_dt): candles = kite.historical_data(token, frm_dt.to_pydatetime(), to_dt.to_pydatetime(), interval="5minute") df = pd.DataFrame(candles) if df.empty: return df df["date"] = pd.to_datetime(df["date"]) df["date"] = (df["date"].dt.tz_localize(TZ) if df["date"].dt.tz is None else df["date"].dt.tz_convert(TZ)) return df.rename(columns={"date": "timestamp"}).reset_index(drop=True) def _replay_day(day_df, symbol, target, kite, opt_cache, span_frm, span_to): """Replay one day; return a list of priced trade dicts.""" trades = [] position = None for i in range(len(day_df)): row = day_df.iloc[i] ts = row["timestamp"] t = ts.time() if t < SESSION_START: continue if position is not None and ts > position["signal_time"]: reason = mp_exit_reason(row, position["direction"], EXIT_MODE) if reason or t >= EOD_EXIT: position["exit_time"] = ts position["exit_reason"] = reason or "EOD_SQUARE_OFF" trades.append(position) position = None if position is None and t < ENTRY_CUTOFF: direction = "CALL" if bool(row["macd_call_signal"]) else \ "PUT" if bool(row["macd_put_signal"]) else None if direction: px = float(row["close"]) contract = _atm_contract(symbol, direction, px, target) position = {"symbol": symbol, "direction": direction, "date": target, "signal_time": ts, "entry_time": ts, "entry_px": px, "contract": contract} if position is not None: last = day_df.iloc[-1] position["exit_time"] = last["timestamp"] position["exit_reason"] = "EOD_LAST_BAR" trades.append(position) # price on real option candles for tr in trades: c = tr["contract"] tr["net"] = None if not c: continue tsym = c["tradingsymbol"] if tsym not in opt_cache: try: opt_cache[tsym] = _fetch_5m(kite, c["instrument_token"], span_frm, span_to) except Exception: opt_cache[tsym] = pd.DataFrame() odf = opt_cache[tsym] ep = _option_close_at(odf, tr["entry_time"]) xp = _option_close_at(odf, tr["exit_time"]) if ep and xp: qty = c["lot_size"] gross = (xp - ep) * qty tr["entry_prem"], tr["exit_prem"] = ep, xp tr["net"] = round(gross - _round_trip_charges(ep, xp, qty), 2) return trades def run(start, end, symbols): kite = _get_kite() start_d = pd.Timestamp(start).date() end_d = pd.Timestamp(end).date() warm_frm = pd.Timestamp(start).tz_localize(TZ) - pd.Timedelta(days=WARMUP_DAYS) span_to = pd.Timestamp(end).tz_localize(TZ).replace(hour=15, minute=30) opt_frm = pd.Timestamp(start).tz_localize(TZ).replace(hour=9, minute=15) all_trades = [] per_symbol = {} opt_cache = {} for sym in symbols: try: token = _instrument_token(sym) udf = _fetch_5m(kite, token, warm_frm.replace(hour=9, minute=15), span_to) except Exception as e: print(f"[skip] {sym}: {e}") continue if udf.empty: print(f"[skip] {sym}: no underlying data") continue ind = add_macd_psar_indicators(udf, entry_mode=ENTRY_MODE).reset_index(drop=True) sym_trades = [] for d in sorted(set(ind["timestamp"].dt.date)): if d < start_d or d > end_d: continue day_df = ind[ind["timestamp"].dt.date == d].reset_index(drop=True) if day_df.empty: continue # Option candles are cached per contract over the FULL range so a # strike reused on multiple days is fetched once and priced correctly. sym_trades += _replay_day(day_df, sym, d, kite, opt_cache, opt_frm, span_to) priced = [t for t in sym_trades if t.get("net") is not None] net = round(sum(t["net"] for t in priced), 2) wins = sum(1 for t in priced if t["net"] > 0) per_symbol[sym] = {"trades": len(sym_trades), "priced": len(priced), "wins": wins, "net": net} all_trades += sym_trades print(f" {sym:12s} trades={len(sym_trades):3d} priced={len(priced):3d} " f"wins={wins:3d} net={net:+10.0f}") # Dump every priced trade so timing / other filters can be studied offline # without re-hitting the API. dump = [{ "symbol": t["symbol"], "date": str(t["date"]), "direction": t["direction"], "entry_time": t["entry_time"], "exit_time": t["exit_time"], "entry_hhmm": pd.Timestamp(t["entry_time"]).strftime("%H:%M"), "entry_prem": t.get("entry_prem"), "exit_prem": t.get("exit_prem"), "exit_reason": t.get("exit_reason"), "net": t.get("net"), } for t in all_trades if t.get("net") is not None] out_path = Path(__file__).resolve().parent / f"trades_{ENTRY_MODE}_{start}_{end}.csv" pd.DataFrame(dump).to_csv(out_path, index=False) print(f"\n[dump] {len(dump)} priced trades -> {out_path}") priced = [t for t in all_trades if t.get("net") is not None] total_net = round(sum(t["net"] for t in priced), 2) wins = sum(1 for t in priced if t["net"] > 0) losses = sum(1 for t in priced if t["net"] < 0) gross_win = sum(t["net"] for t in priced if t["net"] > 0) gross_loss = sum(t["net"] for t in priced if t["net"] < 0) print("\n" + "=" * 64) print(f"MACD+PSAR (exit=both, entry={ENTRY_MODE}) {start} -> {end} | {len(symbols)} symbols") print("=" * 64) print(f"Trades taken : {len(all_trades)} (priced on real options: {len(priced)})") print(f"Wins / Losses : {wins} / {losses} " f"(win rate {100*wins/len(priced):.1f}%)" if priced else "no priced trades") if priced: print(f"Gross winners : {gross_win:+.0f}") print(f"Gross losers : {gross_loss:+.0f}") print(f"Profit factor : {(-gross_win/gross_loss):.2f}" if gross_loss else "inf") print(f"Avg net / trade : {total_net/len(priced):+.0f}") print(f"\nNET P&L (1 lot/trade, incl. charges): {total_net:+,.0f}") print("VERDICT:", "PROFITABLE" if total_net > 0 else "NOT PROFITABLE") if __name__ == "__main__": start = sys.argv[1] if len(sys.argv) > 1 else "2026-07-01" end = sys.argv[2] if len(sys.argv) > 2 else "2026-07-17" syms = (sys.argv[3].split(",") if len(sys.argv) > 3 else _priority1_symbols()) print(f"Symbols ({len(syms)}): {', '.join(syms)}\n") run(start, end, [s.strip().upper() for s in syms])