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.