Розробка системи бектестування з підтримкою multi-timeframe

Ви запустили стратегію на історії: неймовірна дохідність, мінімальна просадка. В live торгівлі — збиток. Знайома ситуація? З імовірністю 70% проблема в look-ahead bias при multi-timeframe тестуванні. Коли стратегія використовує дані старшого таймфрейму, які на момент прийняття рішення ще не були закриті, результати стають нереалістичними. Наша команда спеціалізується на створенні коректних MTF-бектестерів, що виключають цю помилку.

Multi-timeframe бектестування — одна з найбільш технічно складних задач в алготрейдингу. Більшість ефективних криптостратегій використовують кілька таймфреймів: старший для тренду, молодший для точки входу. Без правильної синхронізації ви гарантовано отримаєте look-ahead bias. Цей дефект робить історичні тести нереалістичними. За нашими даними, близько 70% 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.

Як уникнути look-ahead bias в multi-timeframe бектестуванні?

Наша архітектура базується на контексті, який гарантує коректну синхронізацію. Ключові класи:

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, ) 

Чому важлива синхронізація таймфреймів?

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

Додатково: наша система підтримує всі популярні часові інтервали:

Порівняння з готовими рішеннями

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

• Аналіз стратегії та виділення необхідних таймфреймів
• Проєктування архітектури синхронізації з гарантією відсутності look-ahead
• Реалізація ядра бектестера з класами MTFContext та MTFDataSynchronizer
• Написання стратегії з прикладом
• Набір юніт-тестів для верифікації коректності
• Документація по інтеграції
• Навчання команди

Повний список перевірок на look-ahead включає:

• Тест на доступність 4h свічки всередині її інтервалу
• Тест на коректність індексації часових міток
• Тест на граничні випадки (початок/кінець торгової сесії)
• Тест з кількома таймфреймами (3 і більше)

Процес розробки

1. Аналітика — розбираємо стратегію, виділяємо всі таймфрейми та залежності
2. Проєктування — створюємо архітектуру синхронізації, визначаємо структуру контексту
3. Реалізація — пишемо код на Python з використанням pandas та numpy
4. Тестування — перевіряємо на історичних даних з обов'язковим тестом на look-ahead
5. Деплой — розгортаємо в інфраструктуру, проводимо інтеграцію

Приклад 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('2023-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 бектестера. Без нього можна випадково отримати фантастично хороші результати на історії, які не відтворюються в live trading. Ми гарантуємо повну відсутність look-ahead bias в розробленій системі — це підтверджується формальними тестами.

Замовте розробку системи з гарантією відсутності look-ahead bias. Наш досвід — понад 15 проєктів з автоматизації криптотрейдингу. Кожна система проходить сувору верифікацію на історичних даних, що дозволяє уникнути втрат на реальному ринку. Отримайте консультацію прямо зараз — ми допоможемо оцінити вашу стратегію та запропонуємо оптимальне рішення.