backtester/engine.py

"""

backtester/engine.py — Vectorised backtester with zero lookahead bias.

Signal on bar[t], entry at open of bar[t+1].

Session-aware VWAP, corporate-action-adjusted prices only.

PDT simulation with settlement-date logic.



Supports regime-aware multi-strategy via backtester.strategies module.

"""
from __future__ import annotations

import argparse
import datetime
import logging
import sys
from pathlib import Path

import numpy as np
import pandas as pd

# Add parent to path for imports when run as module
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import config
from signals.technical import (
    compute_rsi,
    compute_macd,
    compute_bollinger_bands,
    compute_vwap,
    compute_atr,
    compute_ema,
)
from backtester.report import generate_report
from backtester.strategies import generate_signals, STRATEGY_PARAMS

logger = logging.getLogger("trading_system.backtester")

# ── Cost Model (REALISTIC — matches actual Alpaca fills) ─────────────────────
SPREAD_COST_PCT = 0.05   # per side (conservative for mega-caps via Alpaca)
COMMISSION = 0.0          # Alpaca zero commission
SLIPPAGE_PCT = 0.03       # per fill — accounts for limit order fill quality


def backtest(

    symbol: str,

    df: pd.DataFrame,

    daily_df: pd.DataFrame | None = None,

    hourly_df: pd.DataFrame | None = None,

    account_size: float = 100000.0,

    risk_per_trade_pct: float = 1.0,

    strategy: str = "momentum",

    allow_overnight: bool = False,

    pdt_enabled: bool = False,

    dry_run: bool = True,

    entry_threshold: float = 0.35,

    cooldown_bars: int = 12,

) -> dict:
    """Run a full backtest on historical 5m bars with multi-timeframe confirmation.



    Returns: Dict of results + trade list + daily PnL

    """
    if len(df) < 100:
        logger.error("Insufficient data for backtest: %d bars (need 100+)", len(df))
        return {"error": "insufficient_data"}

    # ══════════════════════════════════════════════════════════════════════
    #   SIGNAL GENERATION (delegated to strategies module)
    # ══════════════════════════════════════════════════════════════════════
    signals, atr, strat_params = generate_signals(strategy, df, daily_df, hourly_df)
    close = df["close"].astype(float)
    high = df["high"].astype(float)
    low = df["low"].astype(float)
    opn = df["open"].astype(float)

    stop_atr_mult = strat_params["stop_atr_mult"]
    tp_rr_ratio = strat_params["tp_rr_ratio"]
    conf_lo, conf_hi = strat_params["conf_range"]

    # Swing strategy overrides: allow overnight holding
    if strat_params.get("allow_overnight"):
        allow_overnight = True

    # ══════════════════════════════════════════════════════════════════════
    #   TRADE SIMULATION  (trailing stop, breakeven management)
    # ══════════════════════════════════════════════════════════════════════
    equity = account_size
    position = None  # single position per symbol (quality over quantity)
    trades = []
    daily_pnl: dict[str, float] = {}
    pdt_blocked = 0
    gap_losses = 0.0
    last_entry_bar = -cooldown_bars

    for i in range(1, len(df) - 1):
        entry_sig = int(signals["entry"].iloc[i])
        current_atr = float(atr.iloc[i]) if not np.isnan(atr.iloc[i]) else 0

        date_str = str(df.index[i].date()) if hasattr(df.index[i], 'date') else str(df.index[i])[:10]
        if date_str not in daily_pnl:
            daily_pnl[date_str] = 0.0

        # ── Manage open position ──
        if position is not None:
            current_price = float(close.iloc[i])
            entry_price = position["entry_price"]
            qty = position["qty"]
            side = position["side"]
            initial_risk = position["initial_risk"]

            # ── Adaptive trailing stop management ──
            # At 1R profit: move stop to breakeven
            # At 2R profit: trail by 1.5x initial risk (not current ATR — stable reference)
            trail_distance = 1.5 * initial_risk  # use initial stop distance as baseline
            if side == "buy":
                unrealized_r = (current_price - entry_price) / initial_risk if initial_risk > 0 else 0
                if unrealized_r >= 2.0 and not position.get("trailing", False):
                    new_stop = current_price - trail_distance
                    position["stop"] = max(position["stop"], new_stop)
                    position["trailing"] = True
                elif unrealized_r >= 1.0 and not position.get("be_moved", False):
                    position["stop"] = max(position["stop"], entry_price + 0.001)
                    position["be_moved"] = True
                elif position.get("trailing", False):
                    trail_stop = current_price - trail_distance
                    position["stop"] = max(position["stop"], trail_stop)
            elif side == "sell":
                unrealized_r = (entry_price - current_price) / initial_risk if initial_risk > 0 else 0
                if unrealized_r >= 2.0 and not position.get("trailing", False):
                    new_stop = current_price + trail_distance
                    position["stop"] = min(position["stop"], new_stop)
                    position["trailing"] = True
                elif unrealized_r >= 1.0 and not position.get("be_moved", False):
                    position["stop"] = min(position["stop"], entry_price - 0.001)
                    position["be_moved"] = True
                elif position.get("trailing", False):
                    trail_stop = current_price + trail_distance
                    position["stop"] = min(position["stop"], trail_stop)

            # Stop loss check
            stop_hit = False
            exit_price = 0.0
            if side == "buy" and float(low.iloc[i]) <= position["stop"]:
                if float(opn.iloc[i]) < position["stop"]:
                    exit_price = float(opn.iloc[i])
                    gap_losses += abs(position["stop"] - exit_price) * qty
                else:
                    exit_price = position["stop"]
                stop_hit = True
            elif side == "sell" and float(high.iloc[i]) >= position["stop"]:
                if float(opn.iloc[i]) > position["stop"]:
                    exit_price = float(opn.iloc[i])
                    gap_losses += abs(exit_price - position["stop"]) * qty
                else:
                    exit_price = position["stop"]
                stop_hit = True

            # Take profit check (2.5× risk)
            tp_hit = False
            if not stop_hit:
                if side == "buy" and float(high.iloc[i]) >= position["tp"]:
                    exit_price = position["tp"]
                    tp_hit = True
                elif side == "sell" and float(low.iloc[i]) <= position["tp"]:
                    exit_price = position["tp"]
                    tp_hit = True

            # Forced EOD close
            eod_close = False
            if not stop_hit and not tp_hit and not allow_overnight:
                if hasattr(df.index[i], 'hour'):
                    et = df.index[i].tz_convert("US/Eastern") if df.index[i].tzinfo else df.index[i]
                    if et.hour >= 15 and et.minute >= 45:
                        exit_price = current_price
                        eod_close = True

            # Time stop for swing: max 5 trading days
            time_stop = False
            if not stop_hit and not tp_hit and not eod_close and allow_overnight:
                entry_time = position.get("entry_time")
                if entry_time is not None:
                    hold_mins = (df.index[i] - entry_time).total_seconds() / 60
                    # 5 trading days = ~5 * 6.5h * 60min = ~1950 5m bars
                    if hold_mins > 5 * 390:  # 390 min per trading day
                        exit_price = current_price
                        time_stop = True

            if stop_hit or tp_hit or eod_close or time_stop:
                slippage = exit_price * SLIPPAGE_PCT / 100
                if side == "buy":
                    exit_price -= slippage
                else:
                    exit_price += slippage

                spread_cost = (entry_price + exit_price) * SPREAD_COST_PCT / 100 * qty
                if side == "buy":
                    pnl = (exit_price - entry_price) * qty - spread_cost
                else:
                    pnl = (entry_price - exit_price) * qty - spread_cost

                equity += pnl
                daily_pnl[date_str] = daily_pnl.get(date_str, 0) + pnl

                reason = ("stop_loss" if stop_hit else "take_profit" if tp_hit
                          else "time_stop" if time_stop else "end_of_day")
                trades.append({
                    "symbol": symbol, "side": side,
                    "entry_price": entry_price, "exit_price": exit_price,
                    "qty": qty, "pnl": round(pnl, 2),
                    "entry_time": str(position["entry_time"]),
                    "exit_time": str(df.index[i]),
                    "exit_reason": reason,
                    "duration_min": (df.index[i] - position["entry_time"]).total_seconds() / 60,
                    "confidence": position.get("confidence", 0),
                    "conf_multiplier": position.get("conf_multiplier", 1.0),
                })
                position = None

                # Don't enter same bar we exited
                continue

        # ── Check for entry ──
        if position is None and current_atr > 0 and (i - last_entry_bar) >= cooldown_bars:
            if entry_sig == 1 or entry_sig == -1:
                side = "buy" if entry_sig == 1 else "sell"

                entry_price = float(opn.iloc[i + 1])
                entry_slippage = entry_price * SLIPPAGE_PCT / 100
                if side == "buy":
                    entry_price += entry_slippage
                else:
                    entry_price -= entry_slippage

                # Stop: uses strategy-specific ATR multiplier
                stop_distance = stop_atr_mult * current_atr
                # Confidence-based position sizing: scale 0.5x to 1.5x
                conf = float(signals["confidence"].iloc[i])
                conf_multiplier = 0.5 + (min(conf, conf_hi) - conf_lo) / (conf_hi - conf_lo)
                conf_multiplier = max(0.5, min(1.5, conf_multiplier))
                dollar_risk = equity * (risk_per_trade_pct / 100) * conf_multiplier
                qty = dollar_risk / stop_distance

                if qty >= 0.01 and equity > 0:
                    if side == "buy":
                        stop = entry_price - stop_distance
                        tp = entry_price + tp_rr_ratio * stop_distance
                    else:
                        stop = entry_price + stop_distance
                        tp = entry_price - tp_rr_ratio * stop_distance

                    position = {
                        "side": side,
                        "entry_price": entry_price,
                        "qty": qty,
                        "stop": stop,
                        "tp": tp,
                        "initial_risk": stop_distance,
                        "be_moved": False,
                        "entry_time": df.index[i + 1],
                        "confidence": conf,
                        "conf_multiplier": conf_multiplier,
                    }
                    last_entry_bar = i

    # ── Close remaining position at last bar ──
    if position is not None:
        exit_price = float(close.iloc[-1])
        entry_price = position["entry_price"]
        qty = position["qty"]
        side = position["side"]
        spread_cost = (entry_price + exit_price) * SPREAD_COST_PCT / 100 * qty
        if side == "buy":
            pnl = (exit_price - entry_price) * qty - spread_cost
        else:
            pnl = (entry_price - exit_price) * qty - spread_cost
        equity += pnl
        date_str = str(df.index[-1].date()) if hasattr(df.index[-1], 'date') else str(df.index[-1])[:10]
        daily_pnl[date_str] = daily_pnl.get(date_str, 0) + pnl
        trades.append({
            "symbol": symbol, "side": side,
            "entry_price": entry_price, "exit_price": exit_price,
            "qty": qty, "pnl": round(pnl, 2),
            "entry_time": str(position["entry_time"]),
            "exit_time": str(df.index[-1]),
            "exit_reason": "end_of_backtest",
            "duration_min": 0,
        })

    results = {
        "symbol": symbol,
        "strategy": strategy,
        "start_date": str(df.index[0].date()),
        "end_date": str(df.index[-1].date()),
        "account_size": account_size,
        "final_equity": round(equity, 2),
        "total_trades": len(trades),
        "pdt_blocked_count": pdt_blocked,
        "gap_losses_usd": round(gap_losses, 2),
        "trades": trades,
        "daily_pnl": daily_pnl,
    }

    return results


def main():
    parser = argparse.ArgumentParser(description="Trading System Backtester")
    parser.add_argument("--symbols", type=str, default=None,
                        help="Comma-separated symbols. Default: all from config.UNIVERSE")
    parser.add_argument("--start", type=str, default=None,
                        help="Start date (YYYY-MM-DD). Default: use all available data")
    parser.add_argument("--end", type=str, default=None,
                        help="End date (YYYY-MM-DD). Default: use all available data")
    parser.add_argument("--strategy", type=str, default="momentum",
                        choices=["momentum", "mean_reversion", "swing"])
    parser.add_argument("--account-size", type=float, default=100000)
    parser.add_argument("--risk-pct", type=float, default=1.0)
    parser.add_argument("--threshold", type=float, default=0.35,
                        help="Signal entry threshold (lower = more trades, default 0.35)")
    parser.add_argument("--cooldown", type=int, default=6,
                        help="Min bars between entries per symbol (default 6 = 30min)")
    parser.add_argument("--dry-run", action="store_true", default=True)
    args = parser.parse_args()

    from monitoring.logger import setup_logging
    setup_logging("INFO")

    from data.downloader import download_historical_range
    from data.storage import get_all_bars

    # Determine symbols
    symbols = [s.strip().upper() for s in args.symbols.split(",")] if args.symbols else config.UNIVERSE
    logger.info("Backtesting %d symbols: %s", len(symbols), symbols)

    all_trades = []
    all_daily_pnl: dict[str, float] = {}
    total_equity = args.account_size
    per_symbol_equity = args.account_size / len(symbols)  # equal allocation
    symbol_results = []
    failed_downloads = []
    
    # Parse start and end dates for historical download
    # Default to a reasonable range if not provided to avoid downloading forever
    end_dt = pd.Timestamp(args.end, tz="UTC").to_pydatetime() if args.end else datetime.datetime.now(datetime.timezone.utc)
    start_dt = pd.Timestamp(args.start, tz="UTC").to_pydatetime() if args.start else (end_dt - datetime.timedelta(days=500))

    for sym in symbols:
        logger.info("--- Processing %s ---", sym)

        # Download historical data for backtest range
        try:
            download_historical_range(sym, "5Min", start_dt, end_dt)
            download_historical_range(sym, "1Hour", start_dt, end_dt)
            download_historical_range(sym, "1Day", start_dt, end_dt)
        except Exception as e:
            logger.warning("Download failed for %s: %s, skipping", sym, e)
            failed_downloads.append(sym)
            continue

        df_5m = get_all_bars(sym, "5Min")
        df_1h = get_all_bars(sym, "1Hour")
        df_1d = get_all_bars(sym, "1Day")

        if df_5m.empty or len(df_5m) < 100:
            logger.warning("Insufficient data for %s (%d bars), skipping", sym, len(df_5m))
            continue

        # Filter date range
        if args.start:
            start = pd.Timestamp(args.start, tz="UTC")
            df_5m = df_5m[df_5m.index >= start]
        if args.end:
            end = pd.Timestamp(args.end, tz="UTC")
            df_5m = df_5m[df_5m.index <= end]

        if len(df_5m) < 100:
            logger.warning("Insufficient data for %s after filter (%d bars)", sym, len(df_5m))
            continue

        result = backtest(
            symbol=sym,
            df=df_5m,
            daily_df=df_1d,
            hourly_df=df_1h if not df_1h.empty else None,
            account_size=per_symbol_equity,
            risk_per_trade_pct=args.risk_pct,
            strategy=args.strategy,
            entry_threshold=args.threshold,
            cooldown_bars=args.cooldown,
        )

        if "error" in result:
            logger.warning("Backtest error for %s: %s", sym, result["error"])
            continue

        symbol_results.append(result)
        all_trades.extend(result.get("trades", []))
        for date, pnl in result.get("daily_pnl", {}).items():
            all_daily_pnl[date] = all_daily_pnl.get(date, 0) + pnl

        sym_trades = result.get("total_trades", 0)
        sym_pnl = result.get("final_equity", per_symbol_equity) - per_symbol_equity
        logger.info("%s: %d trades, P&L: $%.2f", sym, sym_trades, sym_pnl)

    if not all_trades:
        if failed_downloads:
            logger.error("Data downloading failed for: %s. Fix the API/data errors.", ", ".join(failed_downloads))
            sys.exit(1)
        else:
            logger.error("No trades generated across any symbol. Try lowering --threshold.")
            sys.exit(1)

    # Compute combined results
    combined_final = sum(r.get("final_equity", 0) for r in symbol_results)
    # Any leftover from skipped symbols stays at par
    skipped = len(symbols) - len(symbol_results)
    combined_final += skipped * per_symbol_equity

    # Sort all trades by entry time for the merged CSV
    all_trades.sort(key=lambda t: t.get("entry_time", ""))

    # Get overall date range
    all_starts = [r["start_date"] for r in symbol_results]
    all_ends = [r["end_date"] for r in symbol_results]

    combined = {
        "symbol": "+".join(symbols) if len(symbols) <= 3 else "PORTFOLIO",
        "strategy": args.strategy,
        "start_date": min(all_starts),
        "end_date": max(all_ends),
        "account_size": args.account_size,
        "final_equity": round(combined_final, 2),
        "total_trades": len(all_trades),
        "pdt_blocked_count": sum(r.get("pdt_blocked_count", 0) for r in symbol_results),
        "gap_losses_usd": round(sum(r.get("gap_losses_usd", 0) for r in symbol_results), 2),
        "trades": all_trades,
        "daily_pnl": all_daily_pnl,
    }

    # Per-symbol summary table
    days = len(all_daily_pnl) or 1
    print()
    print("=" * 70)
    print("  PER-SYMBOL SUMMARY")
    print("=" * 70)
    print(f"  {'Symbol':<8} {'Trades':>7} {'Trades/Day':>10} {'P&L':>12} {'Win Rate':>9} {'Final Eq':>12}")
    print("  " + "-" * 64)
    for r in symbol_results:
        sym_trades_list = r.get("trades", [])
        wins = sum(1 for t in sym_trades_list if t["pnl"] > 0)
        wr = (wins / len(sym_trades_list) * 100) if sym_trades_list else 0
        sym_pnl = r["final_equity"] - per_symbol_equity
        per_day = r["total_trades"] / days
        print(f"  {r['symbol']:<8} {r['total_trades']:>7} {per_day:>10.1f} "
              f"{'$' + f'{sym_pnl:,.2f}':>12} {wr:>8.1f}% ${r['final_equity']:>11,.2f}")
    print("  " + "-" * 64)
    total_pnl = combined_final - args.account_size
    total_per_day = len(all_trades) / days
    total_wins = sum(1 for t in all_trades if t["pnl"] > 0)
    total_wr = (total_wins / len(all_trades) * 100) if all_trades else 0
    print(f"  {'TOTAL':<8} {len(all_trades):>7} {total_per_day:>10.1f} "
          f"{'$' + f'{total_pnl:,.2f}':>12} {total_wr:>8.1f}% ${combined_final:>11,.2f}")
    print()

    report = generate_report(combined)
    print(report)

    # ── ML Training: generate enriched training data from backtest trades ──
    try:
        from ml.trade_predictor import get_predictor
        ml_pred = get_predictor()

        # Enrich trades with features computed from daily bars at entry time
        enriched_records = []
        for trade in all_trades:
            sym = trade["symbol"]
            conf = trade.get("confidence", 0)
            conf_mult = trade.get("conf_multiplier", 1.0)
            pnl = trade.get("pnl", 0)

            # Get daily bars for feature computation
            try:
                df_1d = get_all_bars(sym, "1Day")
            except Exception:
                df_1d = None

            # Parse entry time to find matching daily bar
            entry_time_str = trade.get("entry_time", "")
            try:
                entry_ts = pd.Timestamp(entry_time_str)
                entry_date = entry_ts.date() if hasattr(entry_ts, 'date') else None
            except Exception:
                entry_date = None

            # Compute features from daily bars as of entry date
            signal_str = min(conf / 6.5, 1.0) if conf else 0.5
            rel_strength = 0.5
            vol_adj_mom = 0.5
            vol_score = 0.5
            trend_cons = 0.5
            atr_pct_val = 0.02
            rsi_daily = 0.5
            daily_trend_val = 1.0 if trade.get("side") == "buy" else -1.0

            if df_1d is not None and len(df_1d) >= 50 and entry_date is not None:
                try:
                    # Find bars up to entry date (no lookahead)
                    if hasattr(df_1d.index, 'date'):
                        mask = df_1d.index.date <= entry_date
                    else:
                        mask = pd.to_datetime(df_1d.index).date <= entry_date
                    d = df_1d[mask]
                    if len(d) >= 21:
                        d_close = d["close"].astype(float)
                        d_high = d["high"].astype(float)
                        d_low = d["low"].astype(float)

                        # ATR%
                        d_atr = compute_atr(d, 14)
                        last_atr = float(d_atr.iloc[-1]) if not d_atr.dropna().empty else 0
                        last_close = float(d_close.iloc[-1])
                        if last_close > 0 and last_atr > 0:
                            atr_pct_val = last_atr / last_close

                        # Relative strength (vs simple 20-day return as proxy)
                        stock_ret = (last_close / float(d_close.iloc[-21]) - 1.0)
                        rel_strength = max(0, min((stock_ret + 0.1) / 0.2, 1.0))

                        # Vol-adjusted momentum
                        if len(d) >= 6:
                            ret_5d = (last_close - float(d_close.iloc[-6])) / float(d_close.iloc[-6])
                            if atr_pct_val > 0:
                                vol_adj_mom = min(abs(ret_5d) / atr_pct_val, 2.0) / 2.0

                        # Volume score
                        if "volume" in d.columns:
                            recent_v = float(d["volume"].iloc[-1])
                            avg_v = float(d["volume"].iloc[-21:-1].mean())
                            if avg_v > 0:
                                vol_score = max(0, min((recent_v / avg_v - 1.0), 1.0))

                        # Trend consistency
                        lookback = min(10, len(d) - 1)
                        if lookback >= 4:
                            h = d_high.iloc[-lookback:].values
                            l = d_low.iloc[-lookback:].values
                            hh = sum(1 for j in range(1, len(h)) if h[j] > h[j-1])
                            hl = sum(1 for j in range(1, len(l)) if l[j] > l[j-1])
                            trend_cons = (hh + hl) / (2 * (lookback - 1))

                        # RSI
                        from signals.technical import compute_rsi as _rsi
                        d_rsi = _rsi(d_close, 14)
                        if not d_rsi.dropna().empty:
                            rsi_daily = float(d_rsi.iloc[-1]) / 100.0

                        # Daily trend
                        from signals.technical import compute_ema as _ema
                        e20 = float(_ema(d_close, 20).iloc[-1])
                        e50 = float(_ema(d_close, 50).iloc[-1])
                        if last_close > e20 > e50:
                            daily_trend_val = 1.0
                        elif last_close > e50:
                            daily_trend_val = 0.5
                        elif last_close < e20 < e50:
                            daily_trend_val = -1.0
                        else:
                            daily_trend_val = 0.0
                except Exception:
                    pass

            record = {
                "signal_strength": signal_str,
                "relative_strength": rel_strength,
                "vol_adj_momentum": vol_adj_mom,
                "volume_score": vol_score,
                "trend_consistency": trend_cons,
                "rank_score": signal_str * 0.6 + rel_strength * 0.4,
                "regime_multiplier": 1.0,
                "spy_trend_score": 0.5,
                "daily_trend": daily_trend_val,
                "daily_macd": 0.5 if daily_trend_val > 0 else 0.0,
                "hourly_pullback": 0.5,
                "momentum_accel": 0.5,
                "overextension": 0.2,
                "atr_pct": atr_pct_val,
                "rsi_daily": rsi_daily,
                "pnl": pnl,
                "exit_reason": trade.get("exit_reason", "unknown"),
                "duration_min": trade.get("duration_min", 0),
                "conf_multiplier": conf_mult,
                "win": 1 if pnl > 0 else 0,
                "symbol": sym,
            }
            enriched_records.append(record)

        if enriched_records:
            success = ml_pred.train_from_enriched_backtest(enriched_records)
            if success:
                logger.info("ML model trained from %d backtest trades", len(enriched_records))
                print(f"\n  ML MODEL TRAINED on {len(enriched_records)} trades")
                print(f"  Model saved to: {ml_pred.MODEL_DIR}")
            else:
                logger.info("ML training skipped (insufficient data)")
    except Exception as e:
        logger.warning("ML backtest training failed (non-fatal): %s", e)

    logger.info("Backtest complete: %d symbols, %d trades, %.1f trades/day",
                len(symbol_results), len(all_trades), total_per_day)


if __name__ == "__main__":
    main()