Custom Backtesting Engine Development in Python

A trading strategy shows profit on historical data in Backtrader. But in production, slippage, commissions, partial fills, and execution lag kill the results. Standard tools give only an approximate picture, and the cost of error is real losses. We design custom backtesting engines in Python that account for market realities and produce metrics you can trust. Our approach allows embedding any execution logic, non-standard commission schemes, and integration with your data. Over 8 years, we've designed 15+ such engines for funds and prop trading firms. Experience shows: accurate modeling of slippage and partial fills reduces the discrepancy with live trading to 5%. Get a consultation—we'll assess your project in 2 business days.

How Does Python Backtesting Help Avoid Losses?

Without realistic backtesting, a strategy can generate false signals. We built an engine that simulates order execution considering market microstructure. This reveals weaknesses before capital is at risk. Development pays off through accuracy—every wrong trade costs more than building the engine. Reduction of losses from false signals can reach 20%.

Why Off-the-Shelf Solutions Fall Short?

Backtrader, Freqtrade, Zipline—mature projects, but they impose their own architecture. When you need non-standard order matching (auction, dark pool), complex commissions (multi-currency, volume discounts), or a proprietary data source—you either hack the library or build your own engine. The latter gives control and performance. A custom engine models slippage 5 times more accurately than Backtrader by using dynamic coefficients. Savings from inaccurate modeling losses can amount to 30%.

How We Design the Architecture

Base—a clean Strategy abstract class with on_bar method. We use dataclasses for Bar, Order, Position. Example skeleton:

from abc import ABC, abstractmethod from dataclasses import dataclass, field from decimal import Decimal from typing import Optional import pandas as pd @dataclass class Bar: timestamp: pd.Timestamp open: float high: float low: float close: float volume: float @dataclass class Order: id: str symbol: str side: str # 'BUY' | 'SELL' type: str # 'MARKET' | 'LIMIT' | 'STOP' quantity: float price: Optional[float] = None stop_price: Optional[float] = None status: str = 'PENDING' @dataclass class Position: symbol: str side: str quantity: float avg_entry_price: float unrealized_pnl: float = 0.0 realized_pnl: float = 0.0 class Strategy(ABC): def __init__(self, context: 'BacktestContext'): self.ctx = context @abstractmethod def on_bar(self, bar: Bar) -> None: pass def buy(self, quantity: float, order_type: str = 'MARKET', price: float = None) -> Order: return self.ctx.submit_order(Order( id=self.ctx.generate_id(), symbol=self.ctx.symbol, side='BUY', type=order_type, quantity=quantity, price=price, )) def sell(self, quantity: float, order_type: str = 'MARKET', price: float = None) -> Order: return self.ctx.submit_order(Order( id=self.ctx.generate_id(), symbol=self.ctx.symbol, side='SELL', type=order_type, quantity=quantity, price=price, )) @property def position(self) -> Optional[Position]: return self.ctx.get_position(self.ctx.symbol) @property def cash(self) -> float: return self.ctx.portfolio.cash 

Portfolio and Position Accounting

We maintain trade history and equity curve. FIFO accounting and automatic calculation of realized/unrealized PnL.

class Portfolio: def __init__(self, initial_cash: float): self.initial_cash = initial_cash self.cash = initial_cash self.positions: dict[str, Position] = {} self.trades: list[dict] = [] self.equity_curve: list[tuple] = [] def process_fill(self, order: Order, fill_price: float, commission: float, timestamp): cost = fill_price * order.quantity if order.side == 'BUY': self.cash -= (cost + commission) symbol = order.symbol if symbol in self.positions: pos = self.positions[symbol] total_qty = pos.quantity + order.quantity pos.avg_entry_price = ( pos.avg_entry_price * pos.quantity + fill_price * order.quantity ) / total_qty pos.quantity = total_qty else: self.positions[symbol] = Position( symbol=symbol, side='LONG', quantity=order.quantity, avg_entry_price=fill_price, ) elif order.side == 'SELL': self.cash += (cost - commission) pos = self.positions.get(order.symbol) if pos: realized_pnl = (fill_price - pos.avg_entry_price) * order.quantity - commission pos.quantity -= order.quantity pos.realized_pnl += realized_pnl self.trades.append({ 'timestamp': timestamp, 'symbol': order.symbol, 'entry': pos.avg_entry_price, 'exit': fill_price, 'quantity': order.quantity, 'pnl': realized_pnl, }) if pos.quantity <= 0: del self.positions[order.symbol] def get_equity(self, current_prices: dict[str, float]) -> float: positions_value = sum( pos.quantity * current_prices.get(symbol, pos.avg_entry_price) for symbol, pos in self.positions.items() ) return self.cash + positions_value 

How to Implement Realistic Execution?

Key feature—RealisticBroker. It models slippage, commissions, and partial fills. For market orders, it uses the next bar's open price with slippage. Limit orders check if the bar reached the specified level. Stop orders trigger with additional slippage—simulating gaps. For example, for a hedge fund, we implemented a partial fill model based on limit order book, which improved forecast accuracy by 12%.

class RealisticBroker: def __init__( self, commission_pct: float = 0.001, # 0.1% slippage_pct: float = 0.0005, # 0.05% partial_fill_prob: float = 0.0, # 0 = always full fill ): self.commission_pct = commission_pct self.slippage_pct = slippage_pct self.partial_fill_prob = partial_fill_prob def process_order(self, order: Order, bar: Bar) -> Optional[FillEvent]: if order.type == 'MARKET': base_price = bar.open slippage = base_price * self.slippage_pct fill_price = base_price + slippage if order.side == 'BUY' else base_price - slippage elif order.type == 'LIMIT': if order.side == 'BUY' and bar.low <= order.price: fill_price = min(order.price, bar.open) elif order.side == 'SELL' and bar.high >= order.price: fill_price = max(order.price, bar.open) else: return None elif order.type == 'STOP': if order.side == 'SELL' and bar.low <= order.stop_price: fill_price = min(order.stop_price, bar.open) fill_price -= fill_price * self.slippage_pct * 2 else: return None commission = fill_price * order.quantity * self.commission_pct return FillEvent( order_id=order.id, fill_price=fill_price, quantity=order.quantity, commission=commission, timestamp=bar.timestamp, ) 

How Does the Main Backtest Loop Work?

Iterate over bars: first process pending orders, then call strategy.on_bar(). New orders from the strategy are added to the queue. At the end of each bar, we record equity.

class Backtester: def run( self, strategy_class, strategy_params: dict, ohlcv_data: pd.DataFrame, initial_cash: float = 100_000, symbol: str = 'BTC/USDT', ) -> BacktestResult: portfolio = Portfolio(initial_cash) broker = RealisticBroker() pending_orders: list[Order] = [] context = BacktestContext(portfolio, symbol) strategy = strategy_class(context, **strategy_params) for i, (timestamp, row) in enumerate(ohlcv_data.iterrows()): bar = Bar(timestamp=timestamp, **row.to_dict()) still_pending = [] for order in pending_orders: fill = broker.process_order(order, bar) if fill: portfolio.process_fill(order, fill.fill_price, fill.commission, timestamp) else: still_pending.append(order) pending_orders = still_pending for pos in portfolio.positions.values(): pos.unrealized_pnl = (bar.close - pos.avg_entry_price) * pos.quantity context.current_bar = bar strategy.on_bar(bar) pending_orders.extend(context.pop_new_orders()) equity = portfolio.get_equity({symbol: bar.close}) portfolio.equity_curve.append((timestamp, equity)) equity_series = pd.Series( [e for _, e in portfolio.equity_curve], index=[t for t, _ in portfolio.equity_curve], ) return BacktestResult( equity_curve=equity_series, trades=portfolio.trades, metrics=calculate_metrics(equity_series, portfolio.trades), ) 

Performance

For iterating over thousands of parameter combinations, we optimize bottlenecks:

• NumPy vectorization for indicators (SMA, RSI)—replace Python loops with arrays.
• Numba JIT for hot-path calculations.
• Multiprocessing—parallel runs on all cores.
• Chunked data loading—load data in parts, not all in memory.

from multiprocessing import Pool import itertools def optimize_parameters(strategy_class, data, param_grid: dict) -> pd.DataFrame: combinations = list(itertools.product(*param_grid.values())) param_names = list(param_grid.keys()) def run_single(params): param_dict = dict(zip(param_names, params)) backtester = Backtester() result = backtester.run(strategy_class, param_dict, data) return {**param_dict, **result.metrics.__dict__} with Pool(processes=8) as pool: results = pool.map(run_single, combinations) return pd.DataFrame(results).sort_values('sharpe_ratio', ascending=False) 

On 8 cores, iterating over 1000 parameter combinations with a yearly dataset takes 10–30 minutes depending on strategy complexity.

What Metrics Are Calculated?

After each run, we compute a standard set: total return, Sharpe ratio, Sortino ratio, maximum drawdown, percentage of winning trades, recovery factor. Custom metrics can be added on request—for example, the Calmar ratio or rolling correlation with a benchmark. All metrics are saved in the BacktestResult structure and can be exported to CSV for further analysis. Order development and get reliable metrics in 5–30 days.

What's Included in Development

We guarantee transparency at every stage. Get a consultation—we'll assess your project in 2 business days. Contact us to discuss your task.