Розробка високопродуктивного двигуна бектестування на Python

Торгова стратегія показує прибуток на історичних даних у Backtrader. Але в продакшні slippage, комісії, partial fills та лаг виконання вбивають результати. Стандартні інструменти дають лише приблизну картину, а ціна помилки — реальні збитки. Ми проєктуємо кастомні двигуни бектестування на Python, які враховують ринкові реалії та видають метрики, яким можна вірити. Наш підхід дозволяє закласти будь-яку логіку виконання, нестандартні комісійні схеми та інтеграцію з вашими даними. За 8 років ми спроєктували 15+ таких двигунів для фондів і проп-трейдингових компаній. Досвід показує: точне моделювання slippage та partial fills знижує розбіжність з реальною торгівлею до 5%. Отримайте консультацію — оцінимо ваш проект за 2 робочих дні.

Як бектестування на Python допомагає уникнути збитків?

Без реалістичного бектестування стратегія може давати хибні сигнали. Ми побудували двигун, який симулює виконання ордерів з урахуванням ринкової мікроструктури. Це дозволяє виявити слабкі місця до того, як капітал опиниться під загрозою. Розробка окупається за рахунок точності — кожна неправильна угода обходиться дорожче, ніж вартість створення двигуна. Зниження збитків від хибних сигналів досягає 20%.

Чому готові рішення не підходять?

Backtrader, Freqtrade, Zipline — зрілі проекти, але вони нав'язують свою архітектуру. Коли потрібен нестандартний матчинг ордерів (аукціон, dark pool), складні комісії (мультивалютні, знижки за обсяг) або власне джерело даних — доводиться або хакнути бібліотеку, або писати свій двигун. Друге дає контроль та продуктивність. Кастомний двигун моделює slippage точніше Backtrader у 5 разів за рахунок динамічних коефіцієнтів. Економія на втратах від неточного моделювання може становити до 30%.

Як ми проєктуємо архітектуру

Основа — чиста абстракція Strategy з методом on_bar. Використовуємо датакласи для Bar, Order, Position. Приклад каркасу:

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 

Портфель та облік позицій

Ведемо історію угод та equity curve. Реалізовано FIFO-облік та автоматичний розрахунок реалізованого/нереалізованого 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 

Як реалізувати реалістичне виконання?

Ключова особливість — RealisticBroker. Він моделює slippage, комісії та часткове заповнення. Для маркет-ордерів використовується ціна відкриття наступного бару з прослизанням. Лімітні ордери перевіряють, чи досяг бар заданого рівня. Стоп-ордери спрацьовують з додатковим slippage — імітація гепів. Наприклад, для хедж-фонду ми реалізували модель часткового заповнення на основі книги лімітних ордерів, що підвищило точність прогнозів на 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 = завжди повне виконання ): 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, ) 

Як влаштований основний loop бектесту?

Запускаємо ітерацію по барах: спочатку обробляємо pending-ордери, потім викликаємо strategy.on_bar(). Нові ордери від стратегії додаються в чергу. В кінці кожного бару записуємо 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), ) 

Продуктивність

Для перебору тисяч комбінацій параметрів ми оптимізуємо вузькі місця:

• NumPy vectorization для індикаторів (SMA, RSI) — Python-цикли замінюємо на масиви.
• Numba JIT для hot-path обчислень.
• Multiprocessing — паралельний прогін на всіх ядрах.
• Chunked data loading — підвантажуємо дані частинами, не тримаємо все в пам'яті.

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) 

На 8 ядрах перебір 1000 комбінацій параметрів з річним датасетом займає 10–30 хвилин залежно від складності стратегії.

Які метрики розраховуються?

Після кожного прогону ми обчислюємо стандартний набір метрик: загальна дохідність, Sharpe ratio, Sortino ratio, максимальний спад, відсоток прибуткових угод, фактор відновлення. При необхідності додаємо користувацькі метрики — наприклад, коефіцієнт Калмара або ковзну кореляцію з бенчмарком. Всі метрики зберігаються в структурі BacktestResult і можуть бути експортовані в CSV для подальшого аналізу. Замовте розробку та отримайте достовірні метрики за 5–30 днів.

Що входить в розробку

Ми гарантуємо прозорість на кожному етапі. Отримайте консультацію — оцінимо ваш проект за 2 робочих дні. Звертайтеся, щоб обговорити ваше завдання.