Trading Strategy Backtesting Development

Trading Strategy Backtesting Development

Backtesting is testing a trading strategy on historical data to assess its potential effectiveness. Good backtesting minimizes lookahead bias and simulates real execution as realistically as possible: with commissions, slippage, partial fills, and delays.

Why Most Backtests Are Unreliable

Lookahead bias — the most common mistake. The strategy uses future data when generating present signals.

# WRONG: use current high for entry on current open
signal = df['high'].rolling(20).max() > df['close'] * 1.05  # current max and close

# CORRECT: signal forms on closed candle, entry on next candle
signal = df['high'].shift(1).rolling(20).max() > df['close'].shift(1) * 1.05
entry_price = df['open']  # entry at next candle open

Survivorship bias: backtest only on assets that still exist. LUNA, FTX token — delisted assets are not included in many providers' historical data.

Overfitting: strategy optimized for a specific historical period. Works perfectly on train data, fails on out-of-sample.

Correct Backtest Engine Structure

from dataclasses import dataclass, field
from decimal import Decimal
from typing import Optional
import pandas as pd

@dataclass
class BacktestConfig:
    initial_capital: Decimal = Decimal('10000')
    commission_rate: Decimal = Decimal('0.001')  # 0.1%
    slippage_bps: int = 5  # 5 basis points
    position_size_percent: float = 95.0  # % of capital per position
    allow_short: bool = True

@dataclass
class Trade:
    entry_time: pd.Timestamp
    exit_time: Optional[pd.Timestamp]
    side: str
    symbol: str
    entry_price: Decimal
    exit_price: Optional[Decimal]
    quantity: Decimal
    commission: Decimal
    pnl: Optional[Decimal] = None

class BacktestEngine:
    def __init__(self, strategy, config: BacktestConfig):
        self.strategy = strategy
        self.config = config
        self.capital = config.initial_capital
        self.position: Optional[Trade] = None
        self.completed_trades: list[Trade] = []
        self.equity_curve: list[tuple] = []

    def apply_slippage(self, price: Decimal, side: str) -> Decimal:
        """Simulate execution price deterioration"""
        slippage = price * Decimal(self.config.slippage_bps) / Decimal(10000)
        if side == 'buy':
            return price + slippage  # buy more expensive
        else:
            return price - slippage  # sell cheaper

    def run(self, df: pd.DataFrame) -> 'BacktestResult':
        warmup = 50  # candles for indicator warmup

        for i in range(warmup, len(df)):
            candle = df.iloc[i]
            history = df.iloc[:i]

            # Execution price = next candle open (realistic)
            exec_price = Decimal(str(candle['open']))

            # Check exit for open position
            if self.position:
                exit_signal = self.strategy.should_exit(self.position, history)
                if exit_signal:
                    self.close_position(exec_price, candle.name, exit_signal)

            # Check entry signal
            if not self.position:
                signal = self.strategy.generate_signal(history)

                if signal in ('BUY', 'SELL') and (signal == 'BUY' or self.config.allow_short):
                    self.open_position(signal, exec_price, candle.name)

            # Record equity
            current_equity = self.calculate_current_equity(candle['close'])
            self.equity_curve.append((candle.name, float(current_equity)))

        # Close open position at last price
        if self.position:
            self.close_position(Decimal(str(df.iloc[-1]['close'])), df.index[-1], 'end_of_data')

        return self.build_result()

    def open_position(self, signal: str, price: Decimal, timestamp):
        exec_price = self.apply_slippage(price, 'buy' if signal == 'BUY' else 'sell')
        quantity = self.capital * Decimal(str(self.config.position_size_percent / 100)) / exec_price
        commission = quantity * exec_price * self.config.commission_rate

        self.capital -= (quantity * exec_price + commission)

        self.position = Trade(
            entry_time=timestamp,
            exit_time=None,
            side=signal,
            symbol='BTC',
            entry_price=exec_price,
            exit_price=None,
            quantity=quantity,
            commission=commission
        )

    def close_position(self, price: Decimal, timestamp, reason: str):
        side = 'sell' if self.position.side == 'BUY' else 'buy'
        exec_price = self.apply_slippage(price, side)
        commission = self.position.quantity * exec_price * self.config.commission_rate

        if self.position.side == 'BUY':
            gross_pnl = (exec_price - self.position.entry_price) * self.position.quantity
        else:
            gross_pnl = (self.position.entry_price - exec_price) * self.position.quantity

        net_pnl = gross_pnl - commission - self.position.commission

        self.capital += self.position.quantity * exec_price - commission

        self.position.exit_time = timestamp
        self.position.exit_price = exec_price
        self.position.pnl = net_pnl

        self.completed_trades.append(self.position)
        self.position = None

Walk-Forward Analysis

def walk_forward_analysis(
    strategy_class,
    df: pd.DataFrame,
    train_size: int = 365,   # candles (days)
    test_size: int = 90,
    step_size: int = 30,
    param_grid: dict = None
) -> list[dict]:
    results = []
    n = len(df)

    for start in range(0, n - train_size - test_size, step_size):
        train_df = df.iloc[start : start + train_size]
        test_df = df.iloc[start + train_size : start + train_size + test_size]

        # Optimization on train data
        if param_grid:
            best_params = optimize_params(strategy_class, train_df, param_grid)
        else:
            best_params = {}

        # Test on out-of-sample data
        strategy = strategy_class(**best_params)
        engine = BacktestEngine(strategy, BacktestConfig())
        result = engine.run(test_df)

        results.append({
            'period_start': test_df.index[0],
            'period_end': test_df.index[-1],
            'params': best_params,
            'roi': result.roi,
            'sharpe': result.sharpe_ratio,
            'max_drawdown': result.max_drawdown,
            'win_rate': result.win_rate,
        })

    return results

Results Analysis

Key Metrics

def analyze_results(trades: list[Trade], equity_curve: list, initial_capital: float) -> dict:
    pnls = [float(t.pnl) for t in trades]
    wins = [p for p in pnls if p > 0]
    losses = [p for p in pnls if p <= 0]

    # Sharpe Ratio
    equity_values = [e[1] for e in equity_curve]
    daily_returns = pd.Series(equity_values).pct_change().dropna()
    sharpe = daily_returns.mean() / daily_returns.std() * (365 ** 0.5) if daily_returns.std() > 0 else 0

    # Max Drawdown
    peak = equity_values[0]
    max_dd = 0
    for val in equity_values:
        peak = max(peak, val)
        dd = (peak - val) / peak
        max_dd = max(max_dd, dd)

    return {
        'roi_percent': (equity_values[-1] / initial_capital - 1) * 100,
        'total_trades': len(trades),
        'win_rate': len(wins) / len(trades) * 100 if trades else 0,
        'profit_factor': sum(wins) / abs(sum(losses)) if losses else float('inf'),
        'sharpe_ratio': sharpe,
        'max_drawdown_percent': max_dd * 100,
        'avg_win': sum(wins) / len(wins) if wins else 0,
        'avg_loss': sum(losses) / len(losses) if losses else 0,
        'expectancy': (sum(pnls) / len(pnls)) if pnls else 0,  # avg P&L per trade
    }

Results Interpretation

Profit Factor is more important than Win Rate: a strategy with 35% win rate but avg_win = 3x avg_loss — is profitable. Strategy with 65% win rate but avg_win = 0.5x avg_loss — is loss-making.

Never run a strategy live without passing walk-forward analysis. Good results on one period — may be random. Stability across multiple walk-forward windows — sign of real edge.