macd_psar/backtest_range.py

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