Разработка системы walk-forward анализа стратегии

Разработка системы walk-forward анализа стратегии

Walk-forward анализ — наиболее строгий метод оценки торговой стратегии. Он симулирует реальный процесс трейдинга: периодическая переоптимизация на свежих данных, немедленное применение к следующему периоду, накопление результатов. Это значительно честнее одноразового бэктеста на всём периоде.

Концепция Walk-Forward

|--- Train Window 1 ---| Test 1 |
          |--- Train Window 2 ---| Test 2 |
                    |--- Train Window 3 ---| Test 3 |
                              |--- Train Window 4 ---| Test 4 |

Данные делятся на окна: train (оптимизация параметров) + test (оценка). Окно сдвигается вперёд по времени. Итоговый результат = конкатенация всех test-периодов.

Anchored walk-forward — train-период растёт, начало фиксировано. Rolling walk-forward — train-период фиксированной длины, сдвигается вместе с окном. Предпочтительнее: модель не устаревает.

Реализация

import pandas as pd
import numpy as np
from dataclasses import dataclass

@dataclass
class WalkForwardResult:
    period_results: list[dict]
    combined_equity: pd.Series
    combined_metrics: dict
    parameter_evolution: pd.DataFrame  # как менялись параметры

class WalkForwardAnalyzer:
    def __init__(
        self,
        optimizer,        # оптимизатор (Bayesian, GridSearch, etc.)
        backtester,       # движок бэктестинга
        train_months: int = 12,
        test_months: int = 3,
        anchored: bool = False,  # rolling (False) или anchored (True)
    ):
        self.optimizer = optimizer
        self.backtester = backtester
        self.train_months = train_months
        self.test_months = test_months
        self.anchored = anchored

    def run(self, data: pd.DataFrame, param_space: dict) -> WalkForwardResult:
        period_results = []
        param_history = []
        test_equities = []

        # Генерируем окна
        windows = self._generate_windows(data)
        print(f"Walk-forward windows: {len(windows)}")

        for i, (train_data, test_data) in enumerate(windows):
            print(f"\n=== Window {i+1}/{len(windows)} ===")
            print(f"Train: {train_data.index[0].date()} → {train_data.index[-1].date()}")
            print(f"Test:  {test_data.index[0].date()} → {test_data.index[-1].date()}")

            # Оптимизируем параметры на train данных
            best_params, _ = self.optimizer.run(
                param_space=param_space,
                data=train_data,
            )

            # Тестируем найденные параметры на test данных
            test_result = self.backtester.run(
                params=best_params,
                data=test_data,
            )

            param_history.append({
                'window': i,
                'test_start': test_data.index[0],
                **best_params,
            })

            period_results.append({
                'window': i,
                'test_start': test_data.index[0],
                'test_end': test_data.index[-1],
                'sharpe': test_result.metrics.sharpe_ratio,
                'return_pct': test_result.metrics.total_return_pct,
                'max_drawdown': test_result.metrics.max_drawdown_pct,
                'win_rate': test_result.metrics.win_rate,
                'n_trades': test_result.metrics.total_trades,
                'params': best_params,
            })

            test_equities.append(test_result.equity_curve)

        # Конкатенируем equity из всех test-периодов
        combined_equity = self._combine_equities(test_equities)
        combined_metrics = self._compute_combined_metrics(period_results, combined_equity)

        return WalkForwardResult(
            period_results=period_results,
            combined_equity=combined_equity,
            combined_metrics=combined_metrics,
            parameter_evolution=pd.DataFrame(param_history),
        )

    def _generate_windows(self, data: pd.DataFrame) -> list[tuple]:
        windows = []
        train_days = self.train_months * 21  # торговых дней
        test_days = self.test_months * 21

        if self.anchored:
            # Train-период растёт от начала
            start = 0
            while start + train_days + test_days <= len(data):
                train = data.iloc[0:start + train_days]
                test = data.iloc[start + train_days:start + train_days + test_days]
                windows.append((train, test))
                start += test_days
        else:
            # Rolling window
            start = 0
            while start + train_days + test_days <= len(data):
                train = data.iloc[start:start + train_days]
                test = data.iloc[start + train_days:start + train_days + test_days]
                windows.append((train, test))
                start += test_days

        return windows

    def _combine_equities(self, test_equities: list[pd.Series]) -> pd.Series:
        """Нормализуем и склеиваем equity из разных периодов"""
        combined = []
        multiplier = 1.0

        for equity in test_equities:
            normalized = equity / equity.iloc[0] * multiplier
            combined.append(normalized)
            multiplier = normalized.iloc[-1]

        return pd.concat(combined)

    def _compute_combined_metrics(self, period_results: list, equity: pd.Series) -> dict:
        returns = equity.pct_change().dropna()
        wf_efficiency = np.mean([r['sharpe'] for r in period_results])

        return {
            'wf_efficiency': wf_efficiency,  # средний Sharpe по периодам
            'combined_sharpe': returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0,
            'combined_total_return': (equity.iloc[-1] / equity.iloc[0] - 1) * 100,
            'combined_max_drawdown': ((equity - equity.cummax()) / equity.cummax()).min() * 100,
            'pct_profitable_windows': sum(1 for r in period_results if r['return_pct'] > 0) / len(period_results) * 100,
            'consistency': np.std([r['sharpe'] for r in period_results]),  # разброс по периодам
        }

Walk-Forward Efficiency (WFE)

def calculate_wfe(in_sample_results: list[dict], out_of_sample_results: list[dict]) -> float:
    """
    Walk-Forward Efficiency = средний OOS Sharpe / средний IS Sharpe.
    Значение > 0.3–0.5 считается хорошим.
    Низкое WFE указывает на overfitting на IS данных.
    """
    avg_is_sharpe = np.mean([r['sharpe'] for r in in_sample_results])
    avg_oos_sharpe = np.mean([r['sharpe'] for r in out_of_sample_results])

    if avg_is_sharpe <= 0:
        return 0.0

    return avg_oos_sharpe / avg_is_sharpe

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

Хороший walk-forward результат:

• Большинство тест-периодов прибыльны (> 60%)
• WFE > 0.4
• Параметры относительно стабильны (не скачут кардинально между периодами)
• Equity curve из тест-периодов растёт без катастрофических просадок

Плохой результат:

• Нестабильные параметры (fast_period меняется с 7 до 25 в разных периодах)
• WFE < 0.2 (сильный overfitting в IS)
• Чередование очень хороших и очень плохих периодов

Нестабильность параметров — главный красный флаг. Если оптимальный fast_period меняется с 5 до 30 от периода к периоду, стратегия подстраивается под шум, а не под реальную закономерность.