run_backtest.py

#!/usr/bin/env python3
"""
NQ Futures Bot — Backtest + Walk-Forward Validation

Usage
─────
  venv/bin/python run_backtest.py                   # 5-min bars, 60 days
  venv/bin/python run_backtest.py --interval 1m     # 1-min bars, 7 days
  venv/bin/python run_backtest.py --days 45 --wf    # 45-day window + walk-forward
  venv/bin/python run_backtest.py --sweep           # parameter sweep
  venv/bin/python run_backtest.py --csv bars.csv    # your own CSV data

Output
──────
  Terminal report with summary, trade log, equity curve
  equity_curve.png  (requires matplotlib)
"""
import argparse
import sys
from dotenv import load_dotenv
load_dotenv()

from data.fetcher import get_nq_bars, load_csv
from backtest.simulator import SimParams, simulate
from backtest.walk_forward import run_walk_forward
from backtest.report import (
    print_summary, print_trade_log, print_exit_breakdown,
    print_direction_breakdown, print_equity_curve,
    print_walk_forward_report, save_equity_curve_png,
)

W = 80


def _sep(char="─"):
    print(char * W)


def _sweep(df) -> None:
    """Quick parameter sweep over stop/target/volume combinations."""
    from itertools import product

    STOPS   = [12, 16, 20, 24]
    TARGETS = [24, 32, 40, 48]
    VOLS    = [1.3, 1.5, 2.0]

    print(f"\n{'='*W}")
    print(f"  PARAMETER SWEEP  ({len(STOPS)*len(TARGETS)*len(VOLS)} combinations)")
    print(f"{'='*W}")
    print(f"  {'Stop':>5}  {'Target':>7}  {'Vol':>4}  {'N':>4}  {'WR%':>5}  "
          f"{'PF':>5}  {'P&L':>9}  {'AvgTk':>7}  {'MaxDD':>9}")
    _sep()

    results = []
    for stop, target, vol in product(STOPS, TARGETS, VOLS):
        if target <= stop:
            continue
        p      = SimParams(stop_ticks=stop, target_ticks=target, vol_multiplier=vol)
        trades = simulate(df.copy(), p)
        if not trades:
            continue
        wins  = [t for t in trades if t.pnl > 0]
        losses= [t for t in trades if t.pnl <= 0]
        gw    = sum(t.pnl for t in wins)
        gl    = abs(sum(t.pnl for t in losses))
        pf    = round(gw / gl, 2) if gl > 0 else 0.0
        pnl   = sum(t.pnl for t in trades)
        wr    = len(wins) / len(trades) * 100
        avgtk = sum(t.ticks for t in trades) / len(trades)
        running = peak = maxdd = 0.0
        for t in trades:
            running += t.pnl
            peak = max(peak, running)
            maxdd = min(maxdd, running - peak)
        results.append((pf, stop, target, vol, len(trades), wr, pnl, avgtk, maxdd))

    results.sort(reverse=True)
    for rank, (pf, stop, target, vol, n, wr, pnl, avgtk, maxdd) in enumerate(results):
        marker = " ◄ BEST" if rank == 0 else ""
        print(f"  {stop:>5}  {target:>7}  {vol:>4.1f}  {n:>4}  {wr:>4.1f}%  "
              f"{pf:>5.2f}  ${pnl:>+8.2f}  {avgtk:>+6.1f}tk  "
              f"${maxdd:>8.2f}{marker}")
    _sep()


def main():
    parser = argparse.ArgumentParser(description="NQ Futures Backtest")
    parser.add_argument("--days",     type=int,   default=60,  help="Calendar days lookback")
    parser.add_argument("--interval", type=str,   default="5m",help="Bar interval: 1m / 5m / 15m")
    parser.add_argument("--csv",      type=str,   default="",  help="Path to CSV file (overrides yfinance)")
    parser.add_argument("--sweep",    action="store_true",      help="Run parameter sweep")
    parser.add_argument("--wf",       action="store_true",      help="Run walk-forward validation")
    parser.add_argument("--is-days",  type=int,   default=45,  help="Walk-forward in-sample days")
    parser.add_argument("--oos-days", type=int,   default=15,  help="Walk-forward out-of-sample days")
    parser.add_argument("--stop",     type=int,   default=16,  help="Stop ticks")
    parser.add_argument("--target",   type=int,   default=32,  help="Target ticks")
    parser.add_argument("--vol",      type=float, default=1.5, help="Volume multiplier")
    parser.add_argument("--no-vwap",  action="store_true",     help="Disable VWAP filter")
    parser.add_argument("--no-vol",   action="store_true",     help="Disable volume filter")
    parser.add_argument("--log",      action="store_true",     help="Print full trade log")
    args = parser.parse_args()

    print(f"\n{'='*W}")
    print(f"  NQ Futures Bot — ORB + VWAP + Volume Strategy Backtest")
    print(f"  Target: 25-35 ticks/day net  |  2:1 R:R  |  Max 2 trades/day")
    print(f"{'='*W}\n")

    # ── Load data ─────────────────────────────────────────────
    if args.csv:
        print(f"Loading CSV: {args.csv}")
        df = load_csv(args.csv)
    else:
        print(f"Fetching NQ=F via yfinance ({args.interval}, {args.days} days)…")
        df = get_nq_bars(days=args.days, interval=args.interval)

    trading_days = df.index.normalize().nunique()
    print(f"  Bars: {len(df):,}  |  Trading days: {trading_days}")
    print(f"  Range: {df.index[0].date()} → {df.index[-1].date()}")
    print(f"  NQ price range: {df['low'].min():.0f} – {df['high'].max():.0f}\n")

    if len(df) < 20:
        print("ERROR: Not enough data. Try --interval 5m or --days 60.")
        sys.exit(1)

    # ── Parameter sweep ────────────────────────────────────────
    if args.sweep:
        _sweep(df)
        return

    # ── Walk-forward ───────────────────────────────────────────
    if args.wf:
        if trading_days < args.is_days + args.oos_days:
            print(f"ERROR: Need at least {args.is_days + args.oos_days} trading days for walk-forward.")
            sys.exit(1)
        print(f"Running walk-forward (IS={args.is_days}d / OOS={args.oos_days}d)…")
        windows = run_walk_forward(df, is_days=args.is_days, oos_days=args.oos_days)
        if not windows:
            print("No complete walk-forward windows — need more data.")
            sys.exit(1)
        print_walk_forward_report(windows)
        all_oos = [t for w in windows for t in w.oos_trades]
        save_equity_curve_png(all_oos, "equity_curve_wf.png")
        return

    # ── Single full backtest ───────────────────────────────────
    params = SimParams(
        stop_ticks     = args.stop,
        target_ticks   = args.target,
        vol_multiplier = args.vol,
        require_vwap   = not args.no_vwap,
        require_volume = not args.no_vol,
    )

    print(f"Running backtest:")
    print(f"  Stop={params.stop_ticks}tk ({params.stop_ticks*0.25:.2f}pts)  "
          f"Target={params.target_ticks}tk ({params.target_ticks*0.25:.2f}pts)  "
          f"R:R={params.target_ticks/params.stop_ticks:.1f}:1")
    print(f"  Trail: activate@{params.trail_activate_ticks}tk / trail@{params.trail_stop_ticks}tk")
    print(f"  Vol filter: {params.vol_multiplier}×avg  |  VWAP filter: {params.require_vwap}")
    print(f"  Commission: ${params.commission_per_rt}/RT  |  "
          f"Slippage: {params.slippage_ticks}tk/side\n")

    trades = simulate(df, params)

    if not trades:
        print("No trades generated — try relaxing filters (--no-vwap, --no-vol, lower --vol)")
        return

    print_summary(trades, label="BACKTEST RESULTS — Full Period")

    # ── Filter breakdown (how each layer adds value) ────────────
    from backtest.simulator import simulate as _sim
    b_nofilter = _sim(df.copy(), SimParams(require_vwap=False, require_volume=False,
                                           stop_ticks=args.stop, target_ticks=args.target))
    b_vwap     = _sim(df.copy(), SimParams(require_vwap=True,  require_volume=False,
                                           stop_ticks=args.stop, target_ticks=args.target))

    def _pf(ts):
        if not ts: return 0.0
        gw = sum(t.pnl for t in ts if t.pnl > 0)
        gl = abs(sum(t.pnl for t in ts if t.pnl <= 0))
        return round(gw/gl, 2) if gl else 0.0

    print(f"\n  Filter waterfall (how each layer improves quality):")
    print(f"  {'Filter':<30}  {'N':>4}  {'WR':>5}  {'PF':>5}  {'P&L':>9}")
    _sep()
    for label_f, ts in [
        ("ORB only (no filters)", b_nofilter),
        ("+ VWAP alignment",      b_vwap),
        ("+ Volume surge",        trades),
    ]:
        if not ts:
            print(f"  {label_f:<30}  {'—':>4}  {'—':>5}  {'—':>5}  {'—':>9}")
            continue
        wins = [t for t in ts if t.pnl > 0]
        wr   = round(len(wins)/len(ts)*100, 1)
        pnl  = round(sum(t.pnl for t in ts), 2)
        print(f"  {label_f:<30}  {len(ts):>4}  {wr:>4.1f}%  {_pf(ts):>5.2f}  ${pnl:>+8.2f}")
    _sep()

    print_exit_breakdown(trades)
    print_direction_breakdown(trades)
    print_equity_curve(trades)

    if args.log:
        print_trade_log(trades)

    save_equity_curve_png(trades)

    # ── Daily tick summary ────────────────────────────────────
    from collections import defaultdict
    by_day: dict = defaultdict(list)
    for t in trades:
        by_day[t.date].append(t)

    daily_ticks = sorted([
        (d, sum(t.ticks for t in ts), sum(t.pnl for t in ts), len(ts))
        for d, ts in by_day.items()
    ])
    if daily_ticks:
        all_ticks = [tk for _, tk, _, _ in daily_ticks]
        print(f"\n  Daily tick performance (net, 1 contract):")
        print(f"    Avg/day : {sum(all_ticks)/len(all_ticks):+.1f} ticks")
        print(f"    Median  : {sorted(all_ticks)[len(all_ticks)//2]:+.1f} ticks")
        print(f"    Best day: {max(all_ticks):+.1f} ticks")
        print(f"    Worst   : {min(all_ticks):+.1f} ticks")
        pos_days = sum(1 for tk in all_ticks if tk > 0)
        print(f"    Positive days: {pos_days}/{len(all_ticks)} "
              f"({pos_days/len(all_ticks)*100:.0f}%)")
        print()


if __name__ == "__main__":
    main()