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