Розробка бектестингу торгової стратегії

Розробка бэктестингу торговельної стратегії

Бэктестинг — це тестування торговельної стратегії на історичних даних для оцінки її потенціальної ефективності. Хороший бэктест — це мінімізація 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.