Суть: Walk-Forward оптимизация (WFO) — это метод тестирования торговых стратегий, при котором параметры стратегии оптимизируются на историческом (обучающем) периоде, а затем фиксируются и тестируются на следующем, невидимом (тестовом) периоде. Этот процесс повторяется, сдвигая окно обучения и тестирования вперед, что позволяет оценить робастность стратегии к изменениям рынка и предотвратить переобучение (overfitting).
В мире количественной торговли разработка и тестирование стратегий являются краеугольным камнем успеха. Однако одной из самых коварных ловушек является переобучение (overfitting) — когда стратегия показывает отличные результаты на исторических данных, но полностью проваливается на новых, невидимых данных. Walk-Forward оптимизация (WFO) — это мощный метод, разработанный для борьбы с этой проблемой, обеспечивая более реалистичную оценку производительности стратегии.
WFO имитирует реальный процесс торговли: вы оптимизируете стратегию на доступных исторических данных, а затем применяете найденные параметры к следующему, еще не виденному рыночному периоду. Этот цикл повторяется, постоянно адаптируя стратегию к меняющимся рыночным условиям и предоставляя более надежную оценку ее долгосрочной жизнеспособности.
В этой статье мы подробно рассмотрим, как реализовать Walk-Forward оптимизацию в Python с использованием библиотеки Backtrader, демонстрируя, как построить робастную систему тестирования, которая поможет вам избежать ловушек переобучения.
Содержание
Исходный код
Ниже представлен полный исходный код для реализации Walk-Forward оптимизации. Мы определим простую стратегию на основе скользящих средних (SMACrossover), а затем создадим функцию run_walk_forward_optimization, которая будет управлять окнами обучения и тестирования, выполнять оптимизацию параметров и собирать результаты.
import backtrader as bt
import datetime
import pandas as pd
import numpy as np
# 1. Определение простой торговой стратегии (SMA Crossover)
class SMACrossover(bt.Strategy):
params = (('fast_length', 10), ('slow_length', 30),)
def __init__(self):
self.dataclose = self.datas[0].close
self.fast_sma = bt.indicators.SMA(self.dataclose, period=self.p.fast_length)
self.slow_sma = bt.indicators.SMA(self.dataclose, period=self.p.slow_length)
self.crossover = bt.indicators.CrossOver(self.fast_sma, self.slow_sma)
self.order = None
def next(self):
if self.order:
return
if not self.position: # Если нет открытой позиции
if self.crossover > 0: # Быстрая SMA пересекает медленную SMA снизу вверх (покупка)
self.order = self.buy()
elif self.crossover < 0: # Быстрая SMA пересекает медленную SMA сверху вниз (продажа/закрытие)
self.order = self.close()
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy:
self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}')
elif order.issell:
self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}')
self.bar_executed = len(self)
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
self.order = None
def log(self, txt, dt=None):
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()} {txt}')
# 2. Функция для выполнения Walk-Forward оптимизации
def run_walk_forward_optimization(
strategy_class,
data_feed,
start_date,
end_date,
train_period_days,
walk_forward_period_days,
optimization_params_space,
cash=100000,
commission=0.001
):
results = []
current_train_start = start_date
current_train_end = start_date + datetime.timedelta(days=train_period_days)
while current_train_end <= end_date:
current_test_start = current_train_end
current_test_end = current_test_start + datetime.timedelta(days=walk_forward_period_days)
# Корректировка последнего тестового периода, чтобы не выходить за end_date
if current_test_end > end_date:
current_test_end = end_date
if current_test_start >= current_test_end: # Убедимся, что тестовый период валиден
break
print(f"\n--- Walk-Forward Шаг ---")
print(f"Период обучения: {current_train_start.isoformat()} по {current_train_end.isoformat()}")
print(f"Период тестирования: {current_test_start.isoformat()} по {current_test_end.isoformat()}")
# --- Оптимизация на периоде обучения ---
cerebro_opt = bt.Cerebro(optreturn=False) # optreturn=False для получения экземпляров стратегий
cerebro_opt.broker.setcash(cash)
cerebro_opt.broker.setcommission(commission=commission)
# Фильтрация данных для периода обучения
data_train = bt.feeds.PandasData(
dataname=data_feed.loc[current_train_start:current_train_end],
fromdate=current_train_start,
todate=current_train_end,
timeframe=bt.TimeFrame.Days
)
cerebro_opt.adddata(data_train)
# Добавление стратегии с пространством параметров для оптимизации
cerebro_opt.optstrategy(
strategy_class,
**optimization_params_space
)
# Запуск оптимизации
strat_runs = cerebro_opt.run()
# Поиск лучших параметров на основе конечной стоимости портфеля
best_params = None
best_value = -float('inf')
# strat_runs - это список списков экземпляров стратегий
# Каждый внутренний список соответствует одной комбинации параметров
for run_group in strat_runs:
for s in run_group:
final_value = s.broker.getvalue()
if final_value > best_value:
best_value = final_value
best_params = s.p._getkwargs() # Получаем параметры как словарь
if not best_params:
print("Не удалось найти успешный запуск оптимизации в этом периоде обучения.")
# Сдвигаем окна вперед, даже если оптимизация не удалась, чтобы избежать зацикливания
current_train_start += datetime.timedelta(days=walk_forward_period_days)
current_train_end += datetime.timedelta(days=walk_forward_period_days)
continue
print(f"Найдены лучшие параметры: {best_params} с конечной стоимостью: {best_value:.2f}")
# --- Бэктест на периоде тестирования с лучшими параметрами ---
cerebro_test = bt.Cerebro()
cerebro_test.broker.setcash(cash)
cerebro_test.broker.setcommission(commission=commission)
# Фильтрация данных для периода тестирования
data_test = bt.feeds.PandasData(
dataname=data_feed.loc[current_test_start:current_test_end],
fromdate=current_test_start,
todate=current_test_end,
timeframe=bt.TimeFrame.Days
)
cerebro_test.adddata(data_test)
# Добавление стратегии с лучшими параметрами
cerebro_test.addstrategy(strategy_class, **best_params)
# Добавление анализаторов для метрик производительности
cerebro_test.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro_test.addanalyzer(bt.analyzers.Returns, _name='returns')
cerebro_test.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro_test.addanalyzer(bt.analyzers.TradeAnalyzer, _name='tradeanalyzer')
# Запуск бэктеста
test_runs = cerebro_test.run()
test_strat = test_runs[0] # Получаем единственный экземпляр стратегии
# Сбор результатов
sharpe_ratio = test_strat.analyzers.sharpe.get_analysis()['sharperatio'] if test_strat.analyzers.sharpe.get_analysis() else float('nan')
total_return = test_strat.analyzers.returns.get_analysis()['rtot'] if test_strat.analyzers.returns.get_analysis() else float('nan')
max_drawdown = test_strat.analyzers.drawdown.get_analysis()['max']['drawdown'] if test_strat.analyzers.drawdown.get_analysis() else float('nan')
pnl = test_strat.broker.getvalue() - cash
print(f"PnL тестового периода: {pnl:.2f}, Sharpe Ratio: {sharpe_ratio:.2f}, Макс. просадка: {max_drawdown:.2f}%")
results.append({
'train_start': current_train_start,
'train_end': current_train_end,
'test_start': current_test_start,
'test_end': current_test_end,
'best_params': best_params,
'test_pnl': pnl,
'test_sharpe_ratio': sharpe_ratio,
'test_max_drawdown': max_drawdown,
'test_total_return': total_return
})
# Сдвиг окон вперед
current_train_start += datetime.timedelta(days=walk_forward_period_days)
current_train_end += datetime.timedelta(days=walk_forward_period_days)
return pd.DataFrame(results)
# 3. Основной блок выполнения
if __name__ == '__main__':
# Генерация фиктивных данных для примера
dates = pd.date_range(start='2010-01-01', end='2020-12-31', freq='D')
np.random.seed(42)
prices = 100 + np.cumsum(np.random.randn(len(dates))) # Случайное блуждание
data = pd.DataFrame({'Open': prices, 'High': prices + 1, 'Low': prices - 1, 'Close': prices, 'Volume': 1000}, index=dates)
data.index.name = 'datetime'
# Определение параметров WFO
start_date = datetime.datetime(2010, 1, 1)
end_date = datetime.datetime(2020, 12, 31)
train_period_days = 365 * 2 # 2 года для обучения
walk_forward_period_days = 365 # 1 год для тестирования (сдвиг окна)
# Определение пространства параметров для оптимизации (сетка поиска)
optimization_params_space = dict(
fast_length=range(5, 20, 5), # Например: 5, 10, 15
slow_length=range(20, 50, 10) # Например: 20, 30, 40
)
# Запуск WFO
wfo_results = run_walk_forward_optimization(
SMACrossover,
data,
start_date,
end_date,
train_period_days,
walk_forward_period_days,
optimization_params_space
)
print("\n--- Сводка Walk-Forward Оптимизации ---")
print(wfo_results)
# Расчет общей производительности по всем тестовым периодам
total_pnl = wfo_results['test_pnl'].sum()
print(f"\nОбщий PnL по всем тестовым периодам: {total_pnl:.2f}")
# Дополнительно: Визуализация результатов (можно реализовать, объединив PnL из каждого тестового периода)
# Для простоты, здесь мы ограничиваемся выводом сводной таблицы.
Разбор параметров
Рассмотрим ключевые параметры, используемые в функции run_walk_forward_optimization:
strategy_class: Класс торговой стратегииBacktrader, которую необходимо оптимизировать (например,SMACrossover).data_feed: Объектpandas.DataFrame, содержащий исторические данные (OHLCV), индексированные по дате/времени.start_date: Начальная дата всего исторического периода, для которого будет выполняться WFO (типdatetime.datetime).end_date: Конечная дата всего исторического периода WFO (типdatetime.datetime).train_period_days: Длительность периода обучения (in-sample) в днях. На этом периоде происходит оптимизация параметров стратегии.walk_forward_period_days: Длительность периода тестирования (out-of-sample) в днях. После оптимизации параметры фиксируются и тестируются на этом периоде. Также это определяет, насколько сильно сдвигаются окна обучения и тестирования на каждом шаге.optimization_params_space: Словарь, определяющий пространство параметров для оптимизации. Ключи — это имена параметров стратегии, а значения — итерируемые объекты (например,range,list,tuple), содержащие возможные значения для каждого параметра. Для более продвинутых методов оптимизации, таких как байесовская оптимизация, можно использовать библиотеки вроде Optuna, как описано в нашей статье: Оптимизация гиперпараметров торговой стратегии: Байесовская оптимизация с Optuna в Python.cash: Начальный капитал для каждого бэктеста.commission: Комиссия за сделку.
Как запустить
Для запуска Walk-Forward оптимизации выполните следующие шаги:
- Установите необходимые библиотеки: Если у вас еще нет
Backtrader,pandasиnumpy, установите их с помощью pip:pip install backtrader pandas numpy - Сохраните код: Скопируйте предоставленный выше код в файл Python (например,
wfo_backtrader.py). - Подготовьте данные: В примере используются сгенерированные фиктивные данные. В реальном сценарии вы должны заменить их своими историческими данными. Убедитесь, что ваш
DataFrameимеет колонкиOpen,High,Low,Close,Volumeи индексирован по дате/времени. - Настройте параметры WFO: Измените значения
start_date,end_date,train_period_days,walk_forward_period_daysиoptimization_params_spaceв блокеif __name__ == '__main__':в соответствии с вашими требованиями. - Запустите скрипт: Откройте терминал или командную строку, перейдите в директорию, где сохранен файл, и выполните команду:
python wfo_backtrader.py - Анализируйте результаты: Скрипт выведет в консоль информацию о каждом шаге WFO (периоды обучения/тестирования, найденные параметры, производительность на тестовом периоде), а затем сводную таблицу
pandas.DataFrameс общими результатами. Вы можете использовать метрики, такие как Sharpe Ratio, для оценки производительности, а также рассмотреть возможность реализации кастомных функций потерь, как описано в статье Реализация кастомной функции потерь Sharpe Ratio в PyTorch, для более специфичных целей оптимизации.
Walk-Forward оптимизация является мощным инструментом для создания более робастных и надежных торговых стратегий, помогая трейдерам и количественным аналитикам избежать ловушки переобучения и принимать более обоснованные решения на реальных рынках.
