Backtesting Your Scanner — Proving It Works Before Risking Real Money

class BacktestEngine:
    """Simulate scanner trades over historical data."""

    def __init__(self, scanner, universe_bars, atr_stop=1.5, atr_tp1=2.0, atr_tp2=3.0):
        self.scanner = scanner
        self.bars = universe_bars  # dict of ticker -> DataFrame
        self.atr_stop = atr_stop
        self.atr_tp1 = atr_tp1
        self.atr_tp2 = atr_tp2
        self.trade_journal = []

    def run(self, start_date, end_date):
        """Run scanner for each trading day and simulate trades."""
        dates = pd.bdate_range(start_date, end_date)

        for date in dates:
            # Build universe snapshot for this date
            snapshot = self._build_snapshot(date)
            if snapshot is None: continue

            # Run scanner (regime + filter + score)
            top10 = self.scanner.daily_scan(snapshot, date)

            # Simulate each trade
            for _, setup in top10.iterrows():
                trade = self._simulate_trade(setup, date)
                if trade:
                    self.trade_journal.append(trade)

        return pd.DataFrame(self.trade_journal)

    def _simulate_trade(self, setup, signal_date):
        """Simulate a single trade with ATR-based stops/targets."""
        ticker = setup['ticker']
        bars = self.bars[ticker]

        # Entry: next day's open
        entry_idx = bars.index.get_loc(signal_date) + 1
        if entry_idx >= len(bars): return None
        entry_price = bars.iloc[entry_idx]['open']
        atr = setup['atr_14']

        # Levels
        stop = entry_price - self.atr_stop * atr
        tp1 = entry_price + self.atr_tp1 * atr
        tp2 = entry_price + self.atr_tp2 * atr

        # Walk forward day by day
        for i in range(entry_idx + 1, min(entry_idx + 30, len(bars))):
            bar = bars.iloc[i]
            if bar['low'] <= stop:
                return self._log_trade(setup, entry_price, stop,
                    'stopped', bars.index[i])
            if bar['high'] >= tp1:
                return self._log_trade(setup, entry_price, tp1,
                    'tp1_hit', bars.index[i])

        # Max holding period reached — exit at close
        exit_price = bars.iloc[min(entry_idx+29, len(bars)-1)]['close']
        return self._log_trade(setup, entry_price, exit_price,
            'timeout', bars.index[min(entry_idx+29, len(bars)-1)])

def compute_metrics(journal_df):
    """Compute key backtest metrics from trade journal."""
    wins = journal_df[journal_df['pnl'] > 0]
    losses = journal_df[journal_df['pnl'] <= 0]

    win_rate = len(wins) / len(journal_df) * 100
    profit_factor = wins['pnl'].sum() / abs(losses['pnl'].sum()) if len(losses) > 0 else float('inf')
    avg_rr = wins['pnl'].mean() / abs(losses['pnl'].mean()) if len(losses) > 0 else float('inf')

    # Sharpe (annualized)
    daily_returns = journal_df.groupby('exit_date')['pnl_pct'].sum()
    sharpe = daily_returns.mean() / daily_returns.std() * (252 ** 0.5)

    # Max drawdown
    cumulative = (1 + daily_returns / 100).cumprod()
    peak = cumulative.expanding().max()
    drawdown = (cumulative - peak) / peak * 100
    max_dd = drawdown.min()

    return {
        'trades': len(journal_df),
        'win_rate': round(win_rate, 1),
        'profit_factor': round(profit_factor, 2),
        'sharpe': round(sharpe, 2),
        'max_drawdown': round(max_dd, 1),
        'avg_rr': round(avg_rr, 2),
    }

def walk_forward(engine, full_data, train_months=24, test_months=6, step=6):
    """Walk-forward validation with sliding windows."""
    results = []
    start = full_data.index.min()
    end = full_data.index.max()

    window_start = start
    while True:
        train_end = window_start + pd.DateOffset(months=train_months)
        test_end = train_end + pd.DateOffset(months=test_months)
        if test_end > end: break

        # Train: optimize thresholds on training data
        train_data = full_data[window_start:train_end]
        engine.optimize(train_data)

        # Test: run scanner on unseen data
        test_data = full_data[train_end:test_end]
        journal = engine.run(test_data)
        metrics = compute_metrics(journal)
        metrics['window'] = f"{train_end:%Y-%m} to {test_end:%Y-%m}"
        results.append(metrics)

        window_start += pd.DateOffset(months=step)

    return pd.DataFrame(results)