Разработка платформы бэктестинга торговых стратегий

Разработка платформы бэктестинга торговых стратегий

Платформа бэктестинга — это инфраструктура для проверки торговых стратегий на исторических данных. Хорошая платформа не просто воспроизводит стратегию на прошлых данных, а делает это максимально реалистично: с учётом комиссий, проскальзывания, ликвидности, и защитой от data leakage.

Архитектура платформы

Data Layer — хранилище исторических данных: OHLCV свечи, tick data, order book snapshots. ClickHouse или Arctic (Python) для эффективного доступа.

Backtest Engine — ядро симуляции. Event-driven или vectorized архитектура.

Strategy Runtime — среда исполнения пользовательских стратегий с изоляцией.

Results & Reporting — расчёт метрик, визуализация P&L, сравнение стратегий.

Optimization Engine — перебор параметров (grid search, genetic algorithm, Bayesian).

Event-driven vs Vectorized

Vectorized — вся логика применяется к pandas DataFrame сразу. Быстро (NumPy операции), но сложно моделировать реальное поведение ордеров:

# Vectorized подход
import pandas as pd
import numpy as np

def backtest_ma_crossover(df: pd.DataFrame, fast: int, slow: int) -> pd.Series:
    fast_ma = df['close'].rolling(fast).mean()
    slow_ma = df['close'].rolling(slow).mean()

    # Сигналы
    signal = np.where(fast_ma > slow_ma, 1, -1)
    signal = pd.Series(signal, index=df.index)

    # Returns
    returns = df['close'].pct_change()
    strategy_returns = signal.shift(1) * returns  # shift(1) = нет look-ahead

    return strategy_returns.cumsum()

Event-driven — более реалистичная симуляция. Каждое рыночное событие (свеча, тик) обрабатывается последовательно. Можно моделировать partial fills, order slippage, margin calls:

class EventDrivenBacktester:
    def run(self, strategy: Strategy, data_feed: DataFeed) -> BacktestResult:
        portfolio = Portfolio(initial_cash=100_000)
        broker = SimulatedBroker(portfolio, slippage=0.001, commission=0.0005)

        for event in data_feed:
            if isinstance(event, MarketEvent):
                strategy.on_market_data(event)

            elif isinstance(event, SignalEvent):
                order = strategy.generate_order(event)
                broker.submit_order(order)

            elif isinstance(event, FillEvent):
                portfolio.update(event)
                strategy.on_fill(event)

        return BacktestResult(portfolio.equity_curve, portfolio.trades)

Симуляция исполнения ордеров

Реалистичная симуляция — ключевое отличие хорошего бэктестера от плохого:

class SimulatedBroker:
    def __init__(self, slippage_pct: float = 0.001, commission_pct: float = 0.0005):
        self.slippage = slippage_pct
        self.commission = commission_pct
        self.pending_orders: list[Order] = []

    def simulate_fill(self, order: Order, bar: OHLCV) -> FillEvent:
        if order.type == "MARKET":
            # Market order исполняется по следующему open + slippage
            fill_price = bar.open * (1 + self.slippage if order.side == "BUY" else 1 - self.slippage)

        elif order.type == "LIMIT":
            # Limit order исполняется если цена достигла уровня
            if order.side == "BUY" and bar.low <= order.price:
                fill_price = min(order.price, bar.open)  # консервативный fill
            elif order.side == "SELL" and bar.high >= order.price:
                fill_price = max(order.price, bar.open)
            else:
                return None  # не исполнен

        commission = fill_price * order.quantity * self.commission
        return FillEvent(
            order_id=order.id,
            fill_price=fill_price,
            quantity=order.quantity,
            commission=commission,
            timestamp=bar.timestamp,
        )

Метрики бэктеста

def calculate_metrics(equity_curve: pd.Series, trades: list[Trade]) -> BacktestMetrics:
    returns = equity_curve.pct_change().dropna()
    annual_factor = 252  # торговых дней

    # Базовые метрики
    total_return = (equity_curve.iloc[-1] / equity_curve.iloc[0]) - 1
    annual_return = (1 + total_return) ** (annual_factor / len(returns)) - 1

    # Risk-adjusted
    sharpe = returns.mean() / returns.std() * np.sqrt(annual_factor) if returns.std() > 0 else 0
    sortino = returns.mean() / returns[returns < 0].std() * np.sqrt(annual_factor)

    # Drawdown
    rolling_max = equity_curve.cummax()
    drawdown = (equity_curve - rolling_max) / rolling_max
    max_drawdown = drawdown.min()
    calmar = annual_return / abs(max_drawdown) if max_drawdown != 0 else 0

    # Trade-level метрики
    winning_trades = [t for t in trades if t.pnl > 0]
    losing_trades = [t for t in trades if t.pnl < 0]
    win_rate = len(winning_trades) / len(trades) if trades else 0

    gross_profit = sum(t.pnl for t in winning_trades)
    gross_loss = abs(sum(t.pnl for t in losing_trades))
    profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf')

    return BacktestMetrics(
        total_return=total_return,
        annual_return=annual_return,
        sharpe_ratio=sharpe,
        sortino_ratio=sortino,
        max_drawdown=max_drawdown,
        calmar_ratio=calmar,
        win_rate=win_rate,
        profit_factor=profit_factor,
        total_trades=len(trades),
        avg_trade_pnl=sum(t.pnl for t in trades) / len(trades) if trades else 0,
    )

Защита от data leakage

Самая распространённая ошибка в бэктестинге — data leakage: использование данных будущего при принятии решений в прошлом.

Look-ahead bias — стратегия использует данные за текущий период (например, close цену) для принятия решения, которое должно быть сделано до закрытия бара.

# Неправильно: использование close той же свечи
signal = df['close'].rolling(20).mean()  # signal[-1] включает текущий close
entry_price = df['close']  # вход по тому же close

# Правильно: вход на следующий bar
signal = df['close'].rolling(20).mean().shift(1)  # shift(1) = прошлый период
entry_price = df['open']  # вход по следующему open

Survivorship bias — если тестируете на активных в данный момент символах, исключая деликтованные, результаты завышены. Используйте исторические списки индексов.

Parameter optimization leakage — если параметры оптимизировались на всех данных, а тест проводится на тех же данных — это не тест, это подгонка. Всегда выделяйте out-of-sample период.

Walk-forward Validation

def walk_forward_backtest(
    strategy_class,
    data: pd.DataFrame,
    train_period: int,  # дней
    test_period: int,   # дней
    optimization_func,
) -> list[BacktestResult]:
    results = []
    start_idx = 0

    while start_idx + train_period + test_period <= len(data):
        train_data = data.iloc[start_idx:start_idx + train_period]
        test_data = data.iloc[start_idx + train_period:start_idx + train_period + test_period]

        # Оптимизируем параметры на train данных
        best_params = optimization_func(strategy_class, train_data)

        # Тестируем на out-of-sample
        strategy = strategy_class(**best_params)
        result = run_backtest(strategy, test_data)
        results.append(result)

        start_idx += test_period

    return results

Оптимизация параметров

Grid search, genetic algorithm, Bayesian optimization — разные подходы к нахождению оптимальных параметров. Ключевой принцип: оптимизация на train, оценка на test, никогда не наоборот.

Для платформы с пользовательскими стратегиями — очередь задач (Celery, RQ) для распределённого выполнения бэктестов на нескольких воркерах. Один сложный бэктест (1 год данных, тысячи комбинаций параметров) может занимать часы — асинхронное выполнение с уведомлением по завершению обязательно.