backtest.py

#!/usr/bin/env python3
"""
Backtest engine — validates trading signal conditions against historical data.

Usage:
    python backtest.py --signal macd_cross_confirmed --universe quick --period 1y
    python backtest.py --signal macd_cross_raw --universe fno --period 2y
    python backtest.py --signal oversold_reversal --holding-days 20 --sl-pct 7 --target-pct 12
    python backtest.py --list-signals
    python backtest.py --compare macd_cross_confirmed macd_cross_raw --universe quick
"""

import argparse, json, sys, time, warnings
from datetime import datetime

import numpy as np
import talib
import yfinance as yf

warnings.filterwarnings("ignore")

# ── Stock universes ───────────────────────────────────────────────────────

QUICK_30 = [
    "RELIANCE", "TCS", "HDFCBANK", "INFY", "ICICIBANK", "BHARTIARTL", "SBIN",
    "HINDUNILVR", "ITC", "LT", "KOTAKBANK", "AXISBANK", "MARUTI", "SUNPHARMA",
    "TATASTEEL", "ONGC", "NTPC", "POWERGRID", "COALINDIA", "M&M", "TATAMOTORS",
    "WIPRO", "HCLTECH", "BAJFINANCE", "ASIANPAINT", "TITAN", "ULTRACEMCO",
    "TECHM", "INDUSINDBK", "DRREDDY",
]


def get_universe(name):
    if name == "quick":
        return QUICK_30
    elif name == "fno":
        try:
            sys.path.insert(0, "/Users/rudrakhunti/Desktop/Personal/Sharingan")
            from screener import get_fno_stocks
            stocks = get_fno_stocks()
            return stocks if stocks else QUICK_30
        except Exception:
            print("Could not fetch F&O list, using quick universe", file=sys.stderr)
            return QUICK_30
    elif name == "nse500":
        try:
            sys.path.insert(0, "/Users/rudrakhunti/Desktop/Personal/Sharingan")
            from nse500_stocks import NSE500
            return NSE500
        except ImportError:
            print("NSE 500 list not found, using quick universe", file=sys.stderr)
            return QUICK_30
    return QUICK_30


# ── Signal definitions ────────────────────────────────────────────────────

def compute_indicators(close, high, low):
    """Compute all indicators needed by signals."""
    n = len(close)
    if n < 60:
        return None

    rsi = talib.RSI(close, 14)
    macd, signal, hist = talib.MACD(close)
    adx = talib.ADX(high, low, close, 14)
    plus_di = talib.PLUS_DI(high, low, close, 14)
    minus_di = talib.MINUS_DI(high, low, close, 14)
    slowk, slowd = talib.STOCH(high, low, close)
    willr = talib.WILLR(high, low, close, 14)

    return {
        "close": close,
        "high": high,
        "low": low,
        "rsi": rsi,
        "macd": macd,
        "signal": signal,
        "hist": hist,
        "adx": adx,
        "plus_di": plus_di,
        "minus_di": minus_di,
        "stoch_k": slowk,
        "stoch_d": slowd,
        "willr": willr,
    }


def signal_macd_cross_raw(ind, i):
    """MACD crosses above signal line on day i."""
    if i < 1:
        return False
    return (ind["hist"][i] > 0 and ind["hist"][i - 1] <= 0
            and ind["adx"][i] > 20 and not np.isnan(ind["adx"][i]))


def signal_macd_cross_confirmed(ind, i):
    """MACD crossed yesterday AND today is green (Day 2 confirmation)."""
    if i < 2:
        return False
    cross_yesterday = ind["hist"][i - 1] > 0 and ind["hist"][i - 2] <= 0
    green_today = ind["close"][i] > ind["close"][i - 1]
    adx_ok = ind["adx"][i - 1] > 20 and not np.isnan(ind["adx"][i - 1])
    return cross_yesterday and green_today and adx_ok


def signal_oversold_reversal(ind, i):
    """Deeply oversold + bullish ADX direction."""
    if np.isnan(ind["rsi"][i]) or np.isnan(ind["adx"][i]):
        return False
    oversold = (ind["rsi"][i] < 30 or
                (ind["stoch_k"][i] < 20 and ind["willr"][i] < -90))
    bullish_dir = ind["plus_di"][i] > ind["minus_di"][i]
    strong_trend = ind["adx"][i] > 25
    return oversold and bullish_dir and strong_trend


def signal_macd_cross_bearish_adx(ind, i):
    """MACD cross in bearish ADX — the anti-pattern."""
    if i < 1:
        return False
    cross = ind["hist"][i] > 0 and ind["hist"][i - 1] <= 0
    bearish_adx = ind["minus_di"][i] > ind["plus_di"][i] and ind["adx"][i] > 30
    return cross and bearish_adx and not np.isnan(ind["adx"][i])


SIGNALS = {
    "macd_cross_raw": {
        "description": "MACD crosses above signal (enter same day). Baseline for comparison.",
        "func": signal_macd_cross_raw,
        "lesson": None,
    },
    "macd_cross_confirmed": {
        "description": "MACD crossed yesterday + green follow-through today (Lesson #1).",
        "func": signal_macd_cross_confirmed,
        "lesson": "Lesson #1: Buy the confirmation, not the cross itself.",
    },
    "oversold_reversal": {
        "description": "RSI<30 or (StochK<20 + WillR<-90) + bullish ADX>25 (Win Pattern #1).",
        "func": signal_oversold_reversal,
        "lesson": "Win Pattern #1: Deeply oversold + Strong ADX bullish + Macro catalyst.",
    },
    "macd_cross_bearish_adx": {
        "description": "MACD cross with bearish ADX>30 — should perform POORLY (Anti-pattern #1).",
        "func": signal_macd_cross_bearish_adx,
        "lesson": "Anti-pattern #1: Buying MACD cross in strong bearish ADX without confirmation.",
    },
}


# ── Backtest engine ───────────────────────────────────────────────────────

def run_backtest(symbol, df, signal_func, holding_days=20, sl_pct=7, target_pct=12):
    """Run backtest for a single stock. Returns list of trade results."""
    close = df["Close"].values.astype(float)
    high = df["High"].values.astype(float)
    low = df["Low"].values.astype(float)

    ind = compute_indicators(close, high, low)
    if ind is None:
        return []

    trades = []
    cooldown = 0  # prevent overlapping entries

    for i in range(33, len(close) - holding_days - 1):
        if cooldown > 0:
            cooldown -= 1
            continue

        if not signal_func(ind, i):
            continue

        entry_price = close[i]
        sl_price = entry_price * (1 - sl_pct / 100)
        target_price = entry_price * (1 + target_pct / 100)

        # Simulate the trade
        exit_price = None
        exit_day = None
        exit_reason = None
        max_favorable = 0
        max_adverse = 0

        for j in range(1, holding_days + 1):
            idx = i + j
            if idx >= len(close):
                break

            day_high = high[idx]
            day_low = low[idx]
            day_close = close[idx]

            # Track excursions
            fav = ((day_high - entry_price) / entry_price) * 100
            adv = ((day_low - entry_price) / entry_price) * 100
            max_favorable = max(max_favorable, fav)
            max_adverse = min(max_adverse, adv)

            # Check SL hit (intraday)
            if day_low <= sl_price:
                exit_price = sl_price
                exit_day = j
                exit_reason = "SL"
                break

            # Check target hit (intraday)
            if day_high >= target_price:
                exit_price = target_price
                exit_day = j
                exit_reason = "TARGET"
                break

        # If neither SL nor target, exit at holding_days close
        if exit_price is None:
            exit_idx = min(i + holding_days, len(close) - 1)
            exit_price = close[exit_idx]
            exit_day = holding_days
            exit_reason = "TIMEOUT"

        pnl_pct = ((exit_price - entry_price) / entry_price) * 100

        trades.append({
            "symbol": symbol,
            "entry_idx": i,
            "entry_price": round(entry_price, 2),
            "exit_price": round(exit_price, 2),
            "exit_day": exit_day,
            "exit_reason": exit_reason,
            "pnl_pct": round(pnl_pct, 2),
            "max_favorable_pct": round(max_favorable, 2),
            "max_adverse_pct": round(max_adverse, 2),
        })

        cooldown = 5  # wait 5 days before next entry on same stock

    return trades


def summarise(trades, signal_name):
    """Compute aggregate statistics from all trades."""
    if not trades:
        return {"signal": signal_name, "total_triggers": 0, "error": "No trades found"}

    n = len(trades)
    winners = [t for t in trades if t["pnl_pct"] > 0]
    losers = [t for t in trades if t["pnl_pct"] <= 0]
    sl_exits = [t for t in trades if t["exit_reason"] == "SL"]
    target_exits = [t for t in trades if t["exit_reason"] == "TARGET"]
    timeout_exits = [t for t in trades if t["exit_reason"] == "TIMEOUT"]

    pnls = [t["pnl_pct"] for t in trades]
    gross_wins = sum(t["pnl_pct"] for t in winners) if winners else 0
    gross_losses = abs(sum(t["pnl_pct"] for t in losers)) if losers else 0.01

    return {
        "signal": signal_name,
        "total_triggers": n,
        "win_rate": round(len(winners) / n, 3),
        "avg_return": round(np.mean(pnls), 2),
        "median_return": round(np.median(pnls), 2),
        "profit_factor": round(gross_wins / gross_losses, 2),
        "avg_winner": round(np.mean([t["pnl_pct"] for t in winners]), 2) if winners else 0,
        "avg_loser": round(np.mean([t["pnl_pct"] for t in losers]), 2) if losers else 0,
        "max_favorable_excursion_avg": round(np.mean([t["max_favorable_pct"] for t in trades]), 2),
        "max_adverse_excursion_avg": round(np.mean([t["max_adverse_pct"] for t in trades]), 2),
        "avg_hold_winners": round(np.mean([t["exit_day"] for t in winners]), 1) if winners else 0,
        "avg_hold_losers": round(np.mean([t["exit_day"] for t in losers]), 1) if losers else 0,
        "exit_breakdown": {
            "target": len(target_exits),
            "sl": len(sl_exits),
            "timeout": len(timeout_exits),
        },
        "best_trade": round(max(pnls), 2),
        "worst_trade": round(min(pnls), 2),
    }


# ── Main ──────────────────────────────────────────────────────────────────

BATCH_SIZE = 50


def download_universe(symbols, period):
    """Download OHLCV for all symbols, batched."""
    tickers = [f"{s}.NS" for s in symbols]
    all_data = {}

    for i in range(0, len(tickers), BATCH_SIZE):
        batch = tickers[i:i + BATCH_SIZE]
        batch_syms = symbols[i:i + BATCH_SIZE]
        print(f"  Downloading batch {i // BATCH_SIZE + 1}/{(len(tickers) - 1) // BATCH_SIZE + 1} ({len(batch)} stocks)...", file=sys.stderr)
        data = yf.download(batch, period=period, group_by="ticker", threads=True, progress=False)

        for sym, tick in zip(batch_syms, batch):
            try:
                if len(batch) == 1:
                    df = data
                else:
                    df = data[tick]
                df = df.dropna(subset=["Close"])
                if len(df) >= 60:
                    all_data[sym] = df
            except Exception:
                pass

        if i + BATCH_SIZE < len(tickers):
            time.sleep(1)

    return all_data


def main():
    parser = argparse.ArgumentParser(description="Trading Signal Backtester")
    parser.add_argument("--signal", help="Signal name to test")
    parser.add_argument("--list-signals", action="store_true", dest="list_signals", help="List available signals")
    parser.add_argument("--universe", default="quick", choices=["quick", "fno", "nse500"], help="Stock universe (default: quick)")
    parser.add_argument("--period", default="2y", help="Data period (default: 2y)")
    parser.add_argument("--holding-days", type=int, default=20, dest="holding_days", help="Max holding period in days")
    parser.add_argument("--sl-pct", type=float, default=7, dest="sl_pct", help="Stop loss %% (default: 7)")
    parser.add_argument("--target-pct", type=float, default=12, dest="target_pct", help="Target %% (default: 12)")
    parser.add_argument("--compare", nargs=2, metavar="SIGNAL", help="Compare two signals head-to-head")
    parser.add_argument("--verbose", action="store_true", help="Include per-trade details")
    args = parser.parse_args()

    if args.list_signals:
        print("\nAvailable signals:\n")
        for name, sig in SIGNALS.items():
            lesson = f"  ({sig['lesson']})" if sig["lesson"] else ""
            print(f"  {name:30s} {sig['description']}{lesson}")
        print()
        return

    if args.compare:
        signals_to_run = args.compare
    elif args.signal:
        signals_to_run = [args.signal]
    else:
        parser.error("Specify --signal or --compare or --list-signals")
        return

    for s in signals_to_run:
        if s not in SIGNALS:
            print(f"Unknown signal: {s}. Use --list-signals.", file=sys.stderr)
            sys.exit(1)

    # Download data
    symbols = get_universe(args.universe)
    print(f"\nBacktest: {', '.join(signals_to_run)} | Universe: {args.universe} ({len(symbols)} stocks) | Period: {args.period}", file=sys.stderr)
    print(f"Params: hold={args.holding_days}d, SL={args.sl_pct}%, Target={args.target_pct}%\n", file=sys.stderr)

    t0 = time.time()
    all_data = download_universe(symbols, args.period)
    print(f"Downloaded {len(all_data)} stocks with sufficient data in {time.time() - t0:.1f}s\n", file=sys.stderr)

    # Run backtests
    results = {}
    for sig_name in signals_to_run:
        sig_func = SIGNALS[sig_name]["func"]
        all_trades = []
        for sym, df in all_data.items():
            trades = run_backtest(sym, df, sig_func,
                                  holding_days=args.holding_days,
                                  sl_pct=args.sl_pct,
                                  target_pct=args.target_pct)
            all_trades.extend(trades)
            if trades:
                print(f"  {sig_name}: {sym} — {len(trades)} triggers", file=sys.stderr)

        summary = summarise(all_trades, sig_name)
        summary["universe"] = args.universe
        summary["period"] = args.period
        summary["parameters"] = {
            "holding_days": args.holding_days,
            "sl_pct": args.sl_pct,
            "target_pct": args.target_pct,
        }
        if SIGNALS[sig_name]["lesson"]:
            summary["lesson_reference"] = SIGNALS[sig_name]["lesson"]
        if args.verbose:
            summary["trades"] = all_trades
        results[sig_name] = summary

    # Comparison output
    output = {}
    if args.compare and len(results) == 2:
        names = list(results.keys())
        a, b = results[names[0]], results[names[1]]
        output["comparison"] = {
            "signal_a": names[0],
            "signal_b": names[1],
            "win_rate_a": a.get("win_rate", 0),
            "win_rate_b": b.get("win_rate", 0),
            "win_rate_delta": round(a.get("win_rate", 0) - b.get("win_rate", 0), 3),
            "avg_return_a": a.get("avg_return", 0),
            "avg_return_b": b.get("avg_return", 0),
            "profit_factor_a": a.get("profit_factor", 0),
            "profit_factor_b": b.get("profit_factor", 0),
            "verdict": f"{names[0]} is {'BETTER' if a.get('profit_factor', 0) > b.get('profit_factor', 0) else 'WORSE'} than {names[1]}",
        }
        output["results"] = results
    else:
        output = results

    print(json.dumps(output, indent=2, default=lambda x: float(x) if isinstance(x, (np.floating, np.integer)) else x))


if __name__ == "__main__":
    main()