\r \r

Backtesting Frameworks\r

\r Build robust, production-grade backtesting systems that avoid common pitfalls and produce reliable strategy performance estimates.\r \r

When to Use This Skill\r

\r

• Developing trading strategy backtests\r
• Building backtesting infrastructure\r
• Validating strategy performance\r
• Avoiding common backtesting biases\r
• Implementing walk-forward analysis\r
• Comparing strategy alternatives\r \r

Core Concepts\r

\r

1. Backtesting Biases\r

\r | Bias | Description | Mitigation |\r | ---------------- | ------------------------- | ----------------------- |\r | Look-ahead | Using future information | Point-in-time data |\r | Survivorship | Only testing on survivors | Use delisted securities |\r | Overfitting | Curve-fitting to history | Out-of-sample testing |\r | Selection | Cherry-picking strategies | Pre-registration |\r | Transaction | Ignoring trading costs | Realistic cost models |\r \r

2. Proper Backtest Structure\r

\r

Historical Data\r
      │\r
      ▼\r
┌─────────────────────────────────────────┐\r
│              Training Set               │\r
│  (Strategy Development & Optimization)  │\r
└─────────────────────────────────────────┘\r
      │\r
      ▼\r
┌─────────────────────────────────────────┐\r
│             Validation Set              │\r
│  (Parameter Selection, No Peeking)      │\r
└─────────────────────────────────────────┘\r
      │\r
      ▼\r
┌─────────────────────────────────────────┐\r
│               Test Set                  │\r
│  (Final Performance Evaluation)         │\r
└─────────────────────────────────────────┘\r
```\r
\r
### 3. Walk-Forward Analysis\r
\r
```\r
Window 1: [Train──────][Test]\r
Window 2:     [Train──────][Test]\r
Window 3:         [Train──────][Test]\r
Window 4:             [Train──────][Test]\r
                                     ─────▶ Time\r
```\r
\r
## Implementation Patterns\r
\r
### Pattern 1: Event-Driven Backtester\r
\r
```python\r
from abc import ABC, abstractmethod\r
from dataclasses import dataclass, field\r
from datetime import datetime\r
from decimal import Decimal\r
from enum import Enum\r
from typing import Dict, List, Optional\r
import pandas as pd\r
import numpy as np\r
\r
class OrderSide(Enum):\r
    BUY = "buy"\r
    SELL = "sell"\r
\r
class OrderType(Enum):\r
    MARKET = "market"\r
    LIMIT = "limit"\r
    STOP = "stop"\r
\r
@dataclass\r
class Order:\r
    symbol: str\r
    side: OrderSide\r
    quantity: Decimal\r
    order_type: OrderType\r
    limit_price: Optional[Decimal] = None\r
    stop_price: Optional[Decimal] = None\r
    timestamp: Optional[datetime] = None\r
\r
@dataclass\r
class Fill:\r
    order: Order\r
    fill_price: Decimal\r
    fill_quantity: Decimal\r
    commission: Decimal\r
    slippage: Decimal\r
    timestamp: datetime\r
\r
@dataclass\r
class Position:\r
    symbol: str\r
    quantity: Decimal = Decimal("0")\r
    avg_cost: Decimal = Decimal("0")\r
    realized_pnl: Decimal = Decimal("0")\r
\r
    def update(self, fill: Fill) -> None:\r
        if fill.order.side == OrderSide.BUY:\r
            new_quantity = self.quantity + fill.fill_quantity\r
            if new_quantity != 0:\r
                self.avg_cost = (\r
                    (self.quantity * self.avg_cost + fill.fill_quantity * fill.fill_price)\r
                    / new_quantity\r
                )\r
            self.quantity = new_quantity\r
        else:\r
            self.realized_pnl += fill.fill_quantity * (fill.fill_price - self.avg_cost)\r
            self.quantity -= fill.fill_quantity\r
\r
@dataclass\r
class Portfolio:\r
    cash: Decimal\r
    positions: Dict[str, Position] = field(default_factory=dict)\r
\r
    def get_position(self, symbol: str) -> Position:\r
        if symbol not in self.positions:\r
            self.positions[symbol] = Position(symbol=symbol)\r
        return self.positions[symbol]\r
\r
    def process_fill(self, fill: Fill) -> None:\r
        position = self.get_position(fill.order.symbol)\r
        position.update(fill)\r
\r
        if fill.order.side == OrderSide.BUY:\r
            self.cash -= fill.fill_price * fill.fill_quantity + fill.commission\r
        else:\r
            self.cash += fill.fill_price * fill.fill_quantity - fill.commission\r
\r
    def get_equity(self, prices: Dict[str, Decimal]) -> Decimal:\r
        equity = self.cash\r
        for symbol, position in self.positions.items():\r
            if position.quantity != 0 and symbol in prices:\r
                equity += position.quantity * prices[symbol]\r
        return equity\r
\r
class Strategy(ABC):\r
    @abstractmethod\r
    def on_bar(self, timestamp: datetime, data: pd.DataFrame) -> List[Order]:\r
        pass\r
\r
    @abstractmethod\r
    def on_fill(self, fill: Fill) -> None:\r
        pass\r
\r
class ExecutionModel(ABC):\r
    @abstractmethod\r
    def execute(self, order: Order, bar: pd.Series) -> Optional[Fill]:\r
        pass\r
\r
class SimpleExecutionModel(ExecutionModel):\r
    def __init__(self, slippage_bps: float = 10, commission_per_share: float = 0.01):\r
        self.slippage_bps = slippage_bps\r
        self.commission_per_share = commission_per_share\r
\r
    def execute(self, order: Order, bar: pd.Series) -> Optional[Fill]:\r
        if order.order_type == OrderType.MARKET:\r
            base_price = Decimal(str(bar["open"]))\r
\r
            # Apply slippage\r
            slippage_mult = 1 + (self.slippage_bps / 10000)\r
            if order.side == OrderSide.BUY:\r
                fill_price = base_price * Decimal(str(slippage_mult))\r
            else:\r
                fill_price = base_price / Decimal(str(slippage_mult))\r
\r
            commission = order.quantity * Decimal(str(self.commission_per_share))\r
            slippage = abs(fill_price - base_price) * order.quantity\r
\r
            return Fill(\r
                order=order,\r
                fill_price=fill_price,\r
                fill_quantity=order.quantity,\r
                commission=commission,\r
                slippage=slippage,\r
                timestamp=bar.name\r
            )\r
        return None\r
\r
class Backtester:\r
    def __init__(\r
        self,\r
        strategy: Strategy,\r
        execution_model: ExecutionModel,\r
        initial_capital: Decimal = Decimal("100000")\r
    ):\r
        self.strategy = strategy\r
        self.execution_model = execution_model\r
        self.portfolio = Portfolio(cash=initial_capital)\r
        self.equity_curve: List[tuple] = []\r
        self.trades: List[Fill] = []\r
\r
    def run(self, data: pd.DataFrame) -> pd.DataFrame:\r
        """Run backtest on OHLCV data with DatetimeIndex."""\r
        pending_orders: List[Order] = []\r
\r
        for timestamp, bar in data.iterrows():\r
            # Execute pending orders at today's prices\r
            for order in pending_orders:\r
                fill = self.execution_model.execute(order, bar)\r
                if fill:\r
                    self.portfolio.process_fill(fill)\r
                    self.strategy.on_fill(fill)\r
                    self.trades.append(fill)\r
\r
            pending_orders.clear()\r
\r
            # Get current prices for equity calculation\r
            prices = {data.index.name or "default": Decimal(str(bar["close"]))}\r
            equity = self.portfolio.get_equity(prices)\r
            self.equity_curve.append((timestamp, float(equity)))\r
\r
            # Generate new orders for next bar\r
            new_orders = self.strategy.on_bar(timestamp, data.loc[:timestamp])\r
            pending_orders.extend(new_orders)\r
\r
        return self._create_results()\r
\r
    def _create_results(self) -> pd.DataFrame:\r
        equity_df = pd.DataFrame(self.equity_curve, columns=["timestamp", "equity"])\r
        equity_df.set_index("timestamp", inplace=True)\r
        equity_df["returns"] = equity_df["equity"].pct_change()\r
        return equity_df\r
```\r
\r
### Pattern 2: Vectorized Backtester (Fast)\r
\r
```python\r
import pandas as pd\r
import numpy as np\r
from typing import Callable, Dict, Any\r
\r
class VectorizedBacktester:\r
    """Fast vectorized backtester for simple strategies."""\r
\r
    def __init__(\r
        self,\r
        initial_capital: float = 100000,\r
        commission: float = 0.001,  # 0.1%\r
        slippage: float = 0.0005   # 0.05%\r
    ):\r
        self.initial_capital = initial_capital\r
        self.commission = commission\r
        self.slippage = slippage\r
\r
    def run(\r
        self,\r
        prices: pd.DataFrame,\r
        signal_func: Callable[[pd.DataFrame], pd.Series]\r
    ) -> Dict[str, Any]:\r
        """\r
        Run backtest with signal function.\r
\r
        Args:\r
            prices: DataFrame with 'close' column\r
            signal_func: Function that returns position signals (-1, 0, 1)\r
\r
        Returns:\r
            Dictionary with results\r
        """\r
        # Generate signals (shifted to avoid look-ahead)\r
        signals = signal_func(prices).shift(1).fillna(0)\r
\r
        # Calculate returns\r
        returns = prices["close"].pct_change()\r
\r
        # Calculate strategy returns with costs\r
        position_changes = signals.diff().abs()\r
        trading_costs = position_changes * (self.commission + self.slippage)\r
\r
        strategy_returns = signals * returns - trading_costs\r
\r
        # Build equity curve\r
        equity = (1 + strategy_returns).cumprod() * self.initial_capital\r
\r
        # Calculate metrics\r
        results = {\r
            "equity": equity,\r
            "returns": strategy_returns,\r
            "signals": signals,\r
            "metrics": self._calculate_metrics(strategy_returns, equity)\r
        }\r
\r
        return results\r
\r
    def _calculate_metrics(\r
        self,\r
        returns: pd.Series,\r
        equity: pd.Series\r
    ) -> Dict[str, float]:\r
        """Calculate performance metrics."""\r
        total_return = (equity.iloc[-1] / self.initial_capital) - 1\r
        annual_return = (1 + total_return) ** (252 / len(returns)) - 1\r
        annual_vol = returns.std() * np.sqrt(252)\r
        sharpe = annual_return / annual_vol if annual_vol > 0 else 0\r
\r
        # Drawdown\r
        rolling_max = equity.cummax()\r
        drawdown = (equity - rolling_max) / rolling_max\r
        max_drawdown = drawdown.min()\r
\r
        # Win rate\r
        winning_days = (returns > 0).sum()\r
        total_days = (returns != 0).sum()\r
        win_rate = winning_days / total_days if total_days > 0 else 0\r
\r
        return {\r
            "total_return": total_return,\r
            "annual_return": annual_return,\r
            "annual_volatility": annual_vol,\r
            "sharpe_ratio": sharpe,\r
            "max_drawdown": max_drawdown,\r
            "win_rate": win_rate,\r
            "num_trades": int((returns != 0).sum())\r
        }\r
\r
# Example usage\r
def momentum_signal(prices: pd.DataFrame, lookback: int = 20) -> pd.Series:\r
    """Simple momentum strategy: long when price > SMA, else flat."""\r
    sma = prices["close"].rolling(lookback).mean()\r
    return (prices["close"] > sma).astype(int)\r
\r
# Run backtest\r
# backtester = VectorizedBacktester()\r
# results = backtester.run(price_data, lambda p: momentum_signal(p, 50))\r
```\r
\r
### Pattern 3: Walk-Forward Optimization\r
\r
```python\r
from typing import Callable, Dict, List, Tuple, Any\r
import pandas as pd\r
import numpy as np\r
from itertools import product\r
\r
class WalkForwardOptimizer:\r
    """Walk-forward analysis with anchored or rolling windows."""\r
\r
    def __init__(\r
        self,\r
        train_period: int,\r
        test_period: int,\r
        anchored: bool = False,\r
        n_splits: int = None\r
    ):\r
        """\r
        Args:\r
            train_period: Number of bars in training window\r
            test_period: Number of bars in test window\r
            anchored: If True, training always starts from beginning\r
            n_splits: Number of train/test splits (auto-calculated if None)\r
        """\r
        self.train_period = train_period\r
        self.test_period = test_period\r
        self.anchored = anchored\r
        self.n_splits = n_splits\r
\r
    def generate_splits(\r
        self,\r
        data: pd.DataFrame\r
    ) -> List[Tuple[pd.DataFrame, pd.DataFrame]]:\r
        """Generate train/test splits."""\r
        splits = []\r
        n = len(data)\r
\r
        if self.n_splits:\r
            step = (n - self.train_period) // self.n_splits\r
        else:\r
            step = self.test_period\r
\r
        start = 0\r
        while start + self.train_period + self.test_period \x3C= n:\r
            if self.anchored:\r
                train_start = 0\r
            else:\r
                train_start = start\r
\r
            train_end = start + self.train_period\r
            test_end = min(train_end + self.test_period, n)\r
\r
            train_data = data.iloc[train_start:train_end]\r
            test_data = data.iloc[train_end:test_end]\r
\r
            splits.append((train_data, test_data))\r
            start += step\r
\r
        return splits\r
\r
    def optimize(\r
        self,\r
        data: pd.DataFrame,\r
        strategy_func: Callable,\r
        param_grid: Dict[str, List],\r
        metric: str = "sharpe_ratio"\r
    ) -> Dict[str, Any]:\r
        """\r
        Run walk-forward optimization.\r
\r
        Args:\r
            data: Full dataset\r
            strategy_func: Function(data, **params) -> results dict\r
            param_grid: Parameter combinations to test\r
            metric: Metric to optimize\r
\r
        Returns:\r
            Combined results from all test periods\r
        """\r
        splits = self.generate_splits(data)\r
        all_results = []\r
        optimal_params_history = []\r
\r
        for i, (train_data, test_data) in enumerate(splits):\r
            # Optimize on training data\r
            best_params, best_metric = self._grid_search(\r
                train_data, strategy_func, param_grid, metric\r
            )\r
            optimal_params_history.append(best_params)\r
\r
            # Test with optimal params\r
            test_results = strategy_func(test_data, **best_params)\r
            test_results["split"] = i\r
            test_results["params"] = best_params\r
            all_results.append(test_results)\r
\r
            print(f"Split {i+1}/{len(splits)}: "\r
                  f"Best {metric}={best_metric:.4f}, params={best_params}")\r
\r
        return {\r
            "split_results": all_results,\r
            "param_history": optimal_params_history,\r
            "combined_equity": self._combine_equity_curves(all_results)\r
        }\r
\r
    def _grid_search(\r
        self,\r
        data: pd.DataFrame,\r
        strategy_func: Callable,\r
        param_grid: Dict[str, List],\r
        metric: str\r
    ) -> Tuple[Dict, float]:\r
        """Grid search for best parameters."""\r
        best_params = None\r
        best_metric = -np.inf\r
\r
        # Generate all parameter combinations\r
        param_names = list(param_grid.keys())\r
        param_values = list(param_grid.values())\r
\r
        for values in product(*param_values):\r
            params = dict(zip(param_names, values))\r
            results = strategy_func(data, **params)\r
\r
            if results["metrics"][metric] > best_metric:\r
                best_metric = results["metrics"][metric]\r
                best_params = params\r
\r
        return best_params, best_metric\r
\r
    def _combine_equity_curves(\r
        self,\r
        results: List[Dict]\r
    ) -> pd.Series:\r
        """Combine equity curves from all test periods."""\r
        combined = pd.concat([r["equity"] for r in results])\r
        return combined\r
```\r
\r
### Pattern 4: Monte Carlo Analysis\r
\r
```python\r
import numpy as np\r
import pandas as pd\r
from typing import Dict, List\r
\r
class MonteCarloAnalyzer:\r
    """Monte Carlo simulation for strategy robustness."""\r
\r
    def __init__(self, n_simulations: int = 1000, confidence: float = 0.95):\r
        self.n_simulations = n_simulations\r
        self.confidence = confidence\r
\r
    def bootstrap_returns(\r
        self,\r
        returns: pd.Series,\r
        n_periods: int = None\r
    ) -> np.ndarray:\r
        """\r
        Bootstrap simulation by resampling returns.\r
\r
        Args:\r
            returns: Historical returns series\r
            n_periods: Length of each simulation (default: same as input)\r
\r
        Returns:\r
            Array of shape (n_simulations, n_periods)\r
        """\r
        if n_periods is None:\r
            n_periods = len(returns)\r
\r
        simulations = np.zeros((self.n_simulations, n_periods))\r
\r
        for i in range(self.n_simulations):\r
            # Resample with replacement\r
            simulated_returns = np.random.choice(\r
                returns.values,\r
                size=n_periods,\r
                replace=True\r
            )\r
            simulations[i] = simulated_returns\r
\r
        return simulations\r
\r
    def analyze_drawdowns(\r
        self,\r
        returns: pd.Series\r
    ) -> Dict[str, float]:\r
        """Analyze drawdown distribution via simulation."""\r
        simulations = self.bootstrap_returns(returns)\r
\r
        max_drawdowns = []\r
        for sim_returns in simulations:\r
            equity = (1 + sim_returns).cumprod()\r
            rolling_max = np.maximum.accumulate(equity)\r
            drawdowns = (equity - rolling_max) / rolling_max\r
            max_drawdowns.append(drawdowns.min())\r
\r
        max_drawdowns = np.array(max_drawdowns)\r
\r
        return {\r
            "expected_max_dd": np.mean(max_drawdowns),\r
            "median_max_dd": np.median(max_drawdowns),\r
            f"worst_{int(self.confidence*100)}pct": np.percentile(\r
                max_drawdowns, (1 - self.confidence) * 100\r
            ),\r
            "worst_case": max_drawdowns.min()\r
        }\r
\r
    def probability_of_loss(\r
        self,\r
        returns: pd.Series,\r
        holding_periods: List[int] = [21, 63, 126, 252]\r
    ) -> Dict[int, float]:\r
        """Calculate probability of loss over various holding periods."""\r
        results = {}\r
\r
        for period in holding_periods:\r
            if period > len(returns):\r
                continue\r
\r
            simulations = self.bootstrap_returns(returns, period)\r
            total_returns = (1 + simulations).prod(axis=1) - 1\r
            prob_loss = (total_returns \x3C 0).mean()\r
            results[period] = prob_loss\r
\r
        return results\r
\r
    def confidence_interval(\r
        self,\r
        returns: pd.Series,\r
        periods: int = 252\r
    ) -> Dict[str, float]:\r
        """Calculate confidence interval for future returns."""\r
        simulations = self.bootstrap_returns(returns, periods)\r
        total_returns = (1 + simulations).prod(axis=1) - 1\r
\r
        lower = (1 - self.confidence) / 2\r
        upper = 1 - lower\r
\r
        return {\r
            "expected": total_returns.mean(),\r
            "lower_bound": np.percentile(total_returns, lower * 100),\r
            "upper_bound": np.percentile(total_returns, upper * 100),\r
            "std": total_returns.std()\r
        }\r
```\r
\r
## Performance Metrics\r
\r
```python\r
def calculate_metrics(returns: pd.Series, rf_rate: float = 0.02) -> Dict[str, float]:\r
    """Calculate comprehensive performance metrics."""\r
    # Annualization factor (assuming daily returns)\r
    ann_factor = 252\r
\r
    # Basic metrics\r
    total_return = (1 + returns).prod() - 1\r
    annual_return = (1 + total_return) ** (ann_factor / len(returns)) - 1\r
    annual_vol = returns.std() * np.sqrt(ann_factor)\r
\r
    # Risk-adjusted returns\r
    sharpe = (annual_return - rf_rate) / annual_vol if annual_vol > 0 else 0\r
\r
    # Sortino (downside deviation)\r
    downside_returns = returns[returns \x3C 0]\r
    downside_vol = downside_returns.std() * np.sqrt(ann_factor)\r
    sortino = (annual_return - rf_rate) / downside_vol if downside_vol > 0 else 0\r
\r
    # Calmar ratio\r
    equity = (1 + returns).cumprod()\r
    rolling_max = equity.cummax()\r
    drawdowns = (equity - rolling_max) / rolling_max\r
    max_drawdown = drawdowns.min()\r
    calmar = annual_return / abs(max_drawdown) if max_drawdown != 0 else 0\r
\r
    # Win rate and profit factor\r
    wins = returns[returns > 0]\r
    losses = returns[returns \x3C 0]\r
    win_rate = len(wins) / len(returns[returns != 0]) if len(returns[returns != 0]) > 0 else 0\r
    profit_factor = wins.sum() / abs(losses.sum()) if losses.sum() != 0 else np.inf\r
\r
    return {\r
        "total_return": total_return,\r
        "annual_return": annual_return,\r
        "annual_volatility": annual_vol,\r
        "sharpe_ratio": sharpe,\r
        "sortino_ratio": sortino,\r
        "calmar_ratio": calmar,\r
        "max_drawdown": max_drawdown,\r
        "win_rate": win_rate,\r
        "profit_factor": profit_factor,\r
        "num_trades": int((returns != 0).sum())\r
    }\r
```\r
\r
## Best Practices\r
\r
### Do's\r
\r
- **Use point-in-time data** - Avoid look-ahead bias\r
- **Include transaction costs** - Realistic estimates\r
- **Test out-of-sample** - Always reserve data\r
- **Use walk-forward** - Not just train/test\r
- **Monte Carlo analysis** - Understand uncertainty\r
\r
### Don'ts\r
\r
- **Don't overfit** - Limit parameters\r
- **Don't ignore survivorship** - Include delisted\r
- **Don't use adjusted data carelessly** - Understand adjustments\r
- **Don't optimize on full history** - Reserve test set\r
- **Don't ignore capacity** - Market impact matters\r