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