Разработка бэктестинга торговой стратегии

Разработка бэктестинга торговой стратегии

Бэктестинг — это тестирование торговой стратегии на исторических данных для оценки её потенциальной эффективности. Хороший бэктест — это минимизация lookahead bias и максимально реалистичная симуляция реального исполнения: с комиссиями, slippage, частичными заполнениями и задержками.

Почему большинство бэктестов ненадёжны

Lookahead bias — самая распространённая ошибка. Стратегия использует данные из будущего при генерации сигналов настоящего.

# НЕПРАВИЛЬНО: используем current high для входа на current open
signal = df['high'].rolling(20).max() > df['close'] * 1.05  # текущие max и close

# ПРАВИЛЬНО: сигнал формируется на закрытой свече, вход на следующей
signal = df['high'].shift(1).rolling(20).max() > df['close'].shift(1) * 1.05
entry_price = df['open']  # вход по открытию следующей свечи

Survivorship bias: бэктест только на активах, которые ещё существуют. LUNA, FTX token — делистингованные активы не включаются в исторические данные многих провайдеров.

Overfitting: стратегия оптимизирована под конкретный исторический период. Работает отлично на train data, проваливается на out-of-sample.

Правильная структура бэктест-движка

from dataclasses import dataclass, field
from decimal import Decimal
from typing import Optional
import pandas as pd

@dataclass
class BacktestConfig:
    initial_capital: Decimal = Decimal('10000')
    commission_rate: Decimal = Decimal('0.001')  # 0.1%
    slippage_bps: int = 5  # 5 basis points
    position_size_percent: float = 95.0  # % капитала на позицию
    allow_short: bool = True

@dataclass
class Trade:
    entry_time: pd.Timestamp
    exit_time: Optional[pd.Timestamp]
    side: str
    symbol: str
    entry_price: Decimal
    exit_price: Optional[Decimal]
    quantity: Decimal
    commission: Decimal
    pnl: Optional[Decimal] = None

class BacktestEngine:
    def __init__(self, strategy, config: BacktestConfig):
        self.strategy = strategy
        self.config = config
        self.capital = config.initial_capital
        self.position: Optional[Trade] = None
        self.completed_trades: list[Trade] = []
        self.equity_curve: list[tuple] = []

    def apply_slippage(self, price: Decimal, side: str) -> Decimal:
        """Моделируем ухудшение цены исполнения"""
        slippage = price * Decimal(self.config.slippage_bps) / Decimal(10000)
        if side == 'buy':
            return price + slippage  # покупаем дороже
        else:
            return price - slippage  # продаём дешевле

    def run(self, df: pd.DataFrame) -> 'BacktestResult':
        warmup = 50  # свечей для прогрева индикаторов

        for i in range(warmup, len(df)):
            candle = df.iloc[i]
            history = df.iloc[:i]

            # Цена исполнения = open следующей свечи (реалистично)
            exec_price = Decimal(str(candle['open']))

            # Проверяем выход для открытой позиции
            if self.position:
                exit_signal = self.strategy.should_exit(self.position, history)
                if exit_signal:
                    self.close_position(exec_price, candle.name, exit_signal)

            # Проверяем сигнал входа
            if not self.position:
                signal = self.strategy.generate_signal(history)

                if signal in ('BUY', 'SELL') and (signal == 'BUY' or self.config.allow_short):
                    self.open_position(signal, exec_price, candle.name)

            # Записываем equity
            current_equity = self.calculate_current_equity(candle['close'])
            self.equity_curve.append((candle.name, float(current_equity)))

        # Закрываем открытую позицию по последней цене
        if self.position:
            self.close_position(Decimal(str(df.iloc[-1]['close'])), df.index[-1], 'end_of_data')

        return self.build_result()

    def open_position(self, signal: str, price: Decimal, timestamp):
        exec_price = self.apply_slippage(price, 'buy' if signal == 'BUY' else 'sell')
        quantity = self.capital * Decimal(str(self.config.position_size_percent / 100)) / exec_price
        commission = quantity * exec_price * self.config.commission_rate

        self.capital -= (quantity * exec_price + commission)

        self.position = Trade(
            entry_time=timestamp,
            exit_time=None,
            side=signal,
            symbol='BTC',
            entry_price=exec_price,
            exit_price=None,
            quantity=quantity,
            commission=commission
        )

    def close_position(self, price: Decimal, timestamp, reason: str):
        side = 'sell' if self.position.side == 'BUY' else 'buy'
        exec_price = self.apply_slippage(price, side)
        commission = self.position.quantity * exec_price * self.config.commission_rate

        if self.position.side == 'BUY':
            gross_pnl = (exec_price - self.position.entry_price) * self.position.quantity
        else:
            gross_pnl = (self.position.entry_price - exec_price) * self.position.quantity

        net_pnl = gross_pnl - commission - self.position.commission

        self.capital += self.position.quantity * exec_price - commission

        self.position.exit_time = timestamp
        self.position.exit_price = exec_price
        self.position.pnl = net_pnl

        self.completed_trades.append(self.position)
        self.position = None

Walk-Forward Analysis

def walk_forward_analysis(
    strategy_class,
    df: pd.DataFrame,
    train_size: int = 365,   # свечей (дней)
    test_size: int = 90,
    step_size: int = 30,
    param_grid: dict = None
) -> list[dict]:
    results = []
    n = len(df)

    for start in range(0, n - train_size - test_size, step_size):
        train_df = df.iloc[start : start + train_size]
        test_df = df.iloc[start + train_size : start + train_size + test_size]

        # Оптимизация на train данных
        if param_grid:
            best_params = optimize_params(strategy_class, train_df, param_grid)
        else:
            best_params = {}

        # Тест на out-of-sample данных
        strategy = strategy_class(**best_params)
        engine = BacktestEngine(strategy, BacktestConfig())
        result = engine.run(test_df)

        results.append({
            'period_start': test_df.index[0],
            'period_end': test_df.index[-1],
            'params': best_params,
            'roi': result.roi,
            'sharpe': result.sharpe_ratio,
            'max_drawdown': result.max_drawdown,
            'win_rate': result.win_rate,
        })

    return results

Анализ результатов

Ключевые метрики

def analyze_results(trades: list[Trade], equity_curve: list, initial_capital: float) -> dict:
    pnls = [float(t.pnl) for t in trades]
    wins = [p for p in pnls if p > 0]
    losses = [p for p in pnls if p <= 0]

    # Sharpe Ratio
    equity_values = [e[1] for e in equity_curve]
    daily_returns = pd.Series(equity_values).pct_change().dropna()
    sharpe = daily_returns.mean() / daily_returns.std() * (365 ** 0.5) if daily_returns.std() > 0 else 0

    # Max Drawdown
    peak = equity_values[0]
    max_dd = 0
    for val in equity_values:
        peak = max(peak, val)
        dd = (peak - val) / peak
        max_dd = max(max_dd, dd)

    return {
        'roi_percent': (equity_values[-1] / initial_capital - 1) * 100,
        'total_trades': len(trades),
        'win_rate': len(wins) / len(trades) * 100 if trades else 0,
        'profit_factor': sum(wins) / abs(sum(losses)) if losses else float('inf'),
        'sharpe_ratio': sharpe,
        'max_drawdown_percent': max_dd * 100,
        'avg_win': sum(wins) / len(wins) if wins else 0,
        'avg_loss': sum(losses) / len(losses) if losses else 0,
        'expectancy': (sum(pnls) / len(pnls)) if pnls else 0,  # средний P&L на сделку
    }

Интерпретация результатов

Profit Factor важнее Win Rate: стратегия с 35% win rate, но avg_win в 3x avg_loss — прибыльна. Стратегия с 65% win rate, но avg_win в 0.5x avg_loss — убыточна.

Никогда не запускайте стратегию live без прохождения walk-forward analysis. Хорошие результаты на одном периоде — может быть случайностью. Стабильность на множестве walk-forward окон — признак реального edge.