Skip to main content
backtesting-frameworks 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.
Jump to install Skills Marketplace Discover and explore AI skills built by the community.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Show prompt details A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/Sheshiyer/skill-clusters --skill backtesting-frameworksThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Download Zip Downloading... Related occupations SOC
Based on SOC occupation classification
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. version 1.0.0 cluster databases-data
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
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 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" ]))
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():
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()
prices = {data.index.name or "default" : Decimal(str (bar["close" ]))}
equity = self .portfolio.get_equity(prices)
self .equity_curve.append((timestamp, float (equity)))
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) 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 ,
slippage: float = 0.0005
):
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
"""
signals = signal_func(prices).shift(1 ).fillna(0 )
returns = prices["close" ].pct_change()
position_changes = signals.diff().abs ()
trading_costs = position_changes * (self .commission + self .slippage)
strategy_returns = signals * returns - trading_costs
equity = (1 + strategy_returns).cumprod() * self .initial_capital
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
rolling_max = equity.cummax()
drawdown = (equity - rolling_max) / rolling_max
max_drawdown = drawdown.min ()
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 ())
}
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 )
Pattern 3: Walk-Forward Optimization 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):
best_params, best_metric = self ._grid_search(
train_data, strategy_func, param_grid, metric
)
optimal_params_history.append(best_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:.4 f} , 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
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 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):
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 def calculate_metrics (returns: pd.Series, rf_rate: float = 0.02 ) -> Dict [str , float ]:
"""Calculate comprehensive performance metrics."""
ann_factor = 252
total_return = (1 + returns).prod() - 1
annual_return = (1 + total_return) ** (ann_factor / len (returns)) - 1
annual_vol = returns.std() * np.sqrt(ann_factor)
sharpe = (annual_return - rf_rate) / annual_vol if annual_vol > 0 else 0
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
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
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
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 ())
}
Best Practices
Do's
Use point-in-time data - Avoid look-ahead biasInclude transaction costs - Realistic estimatesTest out-of-sample - Always reserve dataUse walk-forward - Not just train/testMonte Carlo analysis - Understand uncertainty
Don'ts
Don't overfit - Limit parametersDon't ignore survivorship - Include delistedDon't use adjusted data carelessly - Understand adjustmentsDon't optimize on full history - Reserve test setDon't ignore capacity - Market impact matters
Resources More from this repository Arcplume runs Grok through the Grok Build CLI's own OAuth-authenticated session (grok login) for image generation, with strict preflight validation, secret-safe handling, and headless CLI-driven execution -- no separate XAI_API_KEY billing. Video falls back to the billed xAI API. USE WHEN a user wants to generate an image via a locally logged-in Grok Build CLI session, e.g. 'generate an image with grok', 'use grok build', or 'use my logged-in grok session'.
Shared reference for the Selemene cluster: the two report surfaces (deterministic Rust reports vs. narrative witness-pipeline readings), the @selemene/bridge CLI contract, the output manifest format, and non-prescriptive witnessing tone. USE WHEN deciding which Selemene surface to invoke or when routing between birth/compatibility/transit reports and solo/dyadic readings.
Route Selemene Engine tasks to the right surface: deterministic reports (birth/compatibility/transit) via selemene-core and the @selemene/bridge CLI, or narrative witness readings via selemene-report. USE WHEN the user wants anything under the Selemene/Noesis umbrella but has not named the exact surface.
