Разработка системы бэктестинга с поддержкой multi-timeframe

Разработка системы бэктестинга с поддержкой multi-timeframe

Multi-timeframe (MTF) бэктестинг — одна из наиболее технически сложных задач в алготрейдинге. Большинство эффективных стратегий используют несколько таймфреймов: старший для определения тренда, младший для точки входа. Реализовать это без look-ahead bias — нетривиальная инженерная задача.

Проблема look-ahead в MTF

Представим стратегию: сигнал на входе по 1h EMA, фильтр по 4h тренду. Кажется простым, но есть ловушка:

На момент закрытия 1h свечи в 14:00, свеча 4h за 12:00–16:00 ещё не закрылась. Если использовать 4h close этой свечи — это look-ahead bias: мы используем данные, которые ещё не известны в реальности.

Правило: на каждый момент времени T доступны только те данные с больших таймфреймов, которые уже закрылись до T.

Архитектура MTF бэктестера

from dataclasses import dataclass
from typing import Optional
import pandas as pd

@dataclass
class MTFContext:
    """Контекст с данными разных таймфреймов, корректно синхронизированных"""
    current_timestamp: int

    # Словарь: таймфрейм → DataFrame с доступными данными
    _bars: dict[str, pd.DataFrame]

    def get_bars(self, timeframe: str, n: int = 100) -> pd.DataFrame:
        """Возвращает последние N баров таймфрейма, доступных на текущий момент"""
        bars = self._bars.get(timeframe, pd.DataFrame())
        if bars.empty:
            return bars

        # Только закрытые бары: timestamp + duration < current_timestamp
        tf_duration_ms = self._timeframe_to_ms(timeframe)
        available = bars[bars.index + tf_duration_ms <= self.current_timestamp]

        return available.tail(n)

    def get_last_closed_bar(self, timeframe: str) -> Optional[pd.Series]:
        bars = self.get_bars(timeframe, n=1)
        return bars.iloc[-1] if not bars.empty else None

    @staticmethod
    def _timeframe_to_ms(timeframe: str) -> int:
        mapping = {
            '1m': 60_000, '5m': 300_000, '15m': 900_000,
            '1h': 3_600_000, '4h': 14_400_000, '1d': 86_400_000,
        }
        return mapping.get(timeframe, 3_600_000)

Синхронизация данных разных таймфреймов

class MTFDataSynchronizer:
    def __init__(self, timeframes: list[str], symbol: str):
        self.timeframes = timeframes
        self.symbol = symbol
        self.bars: dict[str, pd.DataFrame] = {}

    def load_all(self, source, start: str, end: str) -> None:
        for tf in self.timeframes:
            self.bars[tf] = source.fetch_ohlcv(
                symbol=self.symbol,
                timeframe=tf,
                start=start,
                end=end,
            )
            self.bars[tf].set_index('timestamp', inplace=True)

    def create_context(self, timestamp: int) -> MTFContext:
        """Создаём контекст для конкретного момента времени"""
        return MTFContext(
            current_timestamp=timestamp,
            _bars=self.bars,
        )

MTF стратегия: пример

class TrendFollowingMTF:
    """
    Стратегия: торгуем в направлении 4h тренда, вход по 1h сигналу
    """

    def on_bar_1h(self, ctx: MTFContext, bar_1h: pd.Series):
        # Получаем данные 4h (только закрытые свечи)
        bars_4h = ctx.get_bars('4h', n=50)
        if len(bars_4h) < 21:
            return None  # недостаточно данных

        # 4h тренд: EMA(21)
        ema_21_4h = bars_4h['close'].ewm(span=21).mean().iloc[-1]
        last_4h_close = bars_4h['close'].iloc[-1]
        trend_up = last_4h_close > ema_21_4h

        # 1h сигнал: EMA(9) кроссовер
        bars_1h = ctx.get_bars('1h', n=20)
        ema_9 = bars_1h['close'].ewm(span=9).mean()
        ema_21_1h = bars_1h['close'].ewm(span=21).mean()

        # Кроссовер вверх
        cross_up = ema_9.iloc[-1] > ema_21_1h.iloc[-1] and ema_9.iloc[-2] <= ema_21_1h.iloc[-2]
        # Кроссовер вниз
        cross_down = ema_9.iloc[-1] < ema_21_1h.iloc[-1] and ema_9.iloc[-2] >= ema_21_1h.iloc[-2]

        if trend_up and cross_up:
            return Signal.LONG
        elif cross_down:
            return Signal.CLOSE

        return None

MTF бэктест runner

class MTFBacktester:
    def run(
        self,
        strategy,
        synchronizer: MTFDataSynchronizer,
        base_timeframe: str,  # таймфрейм для основного цикла
        initial_cash: float = 100_000,
    ) -> BacktestResult:

        portfolio = Portfolio(initial_cash)
        primary_bars = synchronizer.bars[base_timeframe]

        for timestamp, bar in primary_bars.iterrows():
            # Создаём контекст с правильной синхронизацией
            ctx = synchronizer.create_context(timestamp)

            # Обрабатываем pending ордера
            self._process_orders(portfolio, bar)

            # Вызываем стратегию с MTF контекстом
            signal = strategy.on_bar_1h(ctx, bar)
            if signal:
                self._execute_signal(portfolio, signal, bar)

            # Snapshot equity
            portfolio.equity_curve.append((timestamp, portfolio.get_equity(bar['close'])))

        return BacktestResult(portfolio)

Верификация корректности

Тест на отсутствие look-ahead:

def test_mtf_no_lookahead(synchronizer: MTFDataSynchronizer):
    """Убеждаемся, что на момент T не доступны данные 4h свечи, закрывающейся после T"""
    # Момент: 14:30 (внутри 4h свечи 12:00-16:00)
    timestamp_14_30 = pd.Timestamp('2024-01-01 14:30:00').value // 10**6

    ctx = synchronizer.create_context(timestamp_14_30)
    bars_4h = ctx.get_bars('4h', n=5)

    # Последняя доступная 4h свеча должна быть 08:00-12:00, не 12:00-16:00
    last_bar_ts = bars_4h.index[-1]
    last_bar_close_ts = last_bar_ts + 4 * 3600 * 1000

    assert last_bar_close_ts <= timestamp_14_30, \
        f"Look-ahead bias detected! Bar closing at {last_bar_close_ts} is visible at {timestamp_14_30}"

Этот тест — обязательная часть тест-сьюта любого MTF бэктестера. Без него очень легко случайно допустить look-ahead и получить фантастически хорошие результаты на истории, которые не воспроизводятся в live trading.