walk-forward optimization - how i avoid overfitting my strategies

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
import itertools
from dataclasses import dataclass
import json

@dataclass
class WalkForwardConfig:
    """
    Configuration for walk-forward optimization
    """
    in_sample_days: int = 180  # 6 months training
    out_sample_days: int = 60  # 2 months testing
    step_days: int = 30  # Roll forward 1 month at a time
    min_trades: int = 20  # Minimum trades per period

    # Parameter ranges to optimize
    param_ranges: Dict[str, List] = None

    # Metrics for optimization
    primary_metric: str = 'sharpe_ratio'  # What to maximize
    min_sharpe: float = 1.0  # Minimum acceptable
    max_drawdown: float = 0.15  # Maximum acceptable (15%)

class Strategy:
    """
    Example mean reversion strategy
    """
    def __init__(self, lookback: int, entry_threshold: float, exit_threshold: float):
        self.lookback = lookback
        self.entry_threshold = entry_threshold
        self.exit_threshold = exit_threshold

    def generate_signals(self, prices: pd.Series) -> pd.Series:
        """
        Generate trading signals
        Returns: Series of 1 (long), -1 (short), 0 (flat)
        """
        # Calculate z-score
        rolling_mean = prices.rolling(window=self.lookback).mean()
        rolling_std = prices.rolling(window=self.lookback).std()
        zscore = (prices - rolling_mean) / rolling_std

        # Generate signals
        signals = pd.Series(0, index=prices.index)

        # Entry: price far from mean
        signals[zscore < -self.entry_threshold] = 1  # Long when oversold
        signals[zscore > self.entry_threshold] = -1  # Short when overbought

        # Exit: price returns to mean
        signals[abs(zscore) < self.exit_threshold] = 0

        # Forward fill to maintain positions
        signals = signals.replace(0, np.nan).ffill().fillna(0)

        return signals

class PerformanceMetrics:
    """
    Calculate strategy performance metrics
    """
    @staticmethod
    def calculate_returns(prices: pd.Series, signals: pd.Series) -> pd.Series:
        """
        Calculate strategy returns from prices and signals
        """
        # Daily returns
        daily_returns = prices.pct_change()

        # Strategy returns = signal × next day's return
        strategy_returns = signals.shift(1) * daily_returns

        return strategy_returns.dropna()

    @staticmethod
    def sharpe_ratio(returns: pd.Series, risk_free_rate: float = 0.0) -> float:
        """
        Annualized Sharpe ratio
        """
        if len(returns) == 0 or returns.std() == 0:
            return 0.0

        excess_returns = returns - risk_free_rate / 252
        sharpe = np.sqrt(252) * excess_returns.mean() / excess_returns.std()

        return sharpe

    @staticmethod
    def max_drawdown(returns: pd.Series) -> float:
        """
        Maximum drawdown from peak
        """
        if len(returns) == 0:
            return 0.0

        cumulative = (1 + returns).cumprod()
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max

        return abs(drawdown.min())

    @staticmethod
    def win_rate(returns: pd.Series) -> float:
        """
        Percentage of positive return days
        """
        if len(returns) == 0:
            return 0.0

        winning_days = (returns > 0).sum()
        total_days = len(returns[returns != 0])

        return winning_days / total_days if total_days > 0 else 0.0

    @staticmethod
    def profit_factor(returns: pd.Series) -> float:
        """
        Ratio of gross profit to gross loss
        """
        wins = returns[returns > 0].sum()
        losses = abs(returns[returns < 0].sum())

        return wins / losses if losses != 0 else 0.0

class WalkForwardOptimizer:
    """
    Walk-forward optimization framework
    """
    def __init__(self, config: WalkForwardConfig):
        self.config = config
        self.results = []

    def generate_parameter_combinations(self) -> List[Dict]:
        """
        Generate all parameter combinations to test
        """
        if self.config.param_ranges is None:
            raise ValueError("param_ranges not configured")

        # Get all possible combinations
        keys = self.config.param_ranges.keys()
        values = self.config.param_ranges.values()

        combinations = []
        for combo in itertools.product(*values):
            param_dict = dict(zip(keys, combo))
            combinations.append(param_dict)

        return combinations

    def split_data(self, data: pd.DataFrame, start_date: datetime) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """
        Split data into in-sample and out-of-sample periods
        """
        in_sample_end = start_date + timedelta(days=self.config.in_sample_days)
        out_sample_end = in_sample_end + timedelta(days=self.config.out_sample_days)

        in_sample = data[(data.index >= start_date) & (data.index < in_sample_end)]
        out_sample = data[(data.index >= in_sample_end) & (data.index < out_sample_end)]

        return in_sample, out_sample

    def optimize_period(self, in_sample: pd.DataFrame, out_sample: pd.DataFrame) -> Dict:
        """
        Optimize on in-sample, validate on out-of-sample
        """
        param_combinations = self.generate_parameter_combinations()

        best_in_sample = None
        best_params = None
        best_sharpe = -999

        # Test all parameter combinations on in-sample
        for params in param_combinations:
            strategy = Strategy(**params)
            signals = strategy.generate_signals(in_sample['close'])
            returns = PerformanceMetrics.calculate_returns(in_sample['close'], signals)

            if len(returns) < self.config.min_trades:
                continue

            sharpe = PerformanceMetrics.sharpe_ratio(returns)
            max_dd = PerformanceMetrics.max_drawdown(returns)

            # Check constraints
            if sharpe < self.config.min_sharpe:
                continue
            if max_dd > self.config.max_drawdown:
                continue

            # Track best
            if sharpe > best_sharpe:
                best_sharpe = sharpe
                best_params = params
                best_in_sample = {
                    'sharpe': sharpe,
                    'max_dd': max_dd,
                    'win_rate': PerformanceMetrics.win_rate(returns),
                    'profit_factor': PerformanceMetrics.profit_factor(returns),
                    'trades': len(returns[returns != 0])
                }

        if best_params is None:
            return None

        # Test best parameters on out-of-sample
        strategy = Strategy(**best_params)
        signals = strategy.generate_signals(out_sample['close'])
        returns = PerformanceMetrics.calculate_returns(out_sample['close'], signals)

        out_sample_metrics = {
            'sharpe': PerformanceMetrics.sharpe_ratio(returns),
            'max_dd': PerformanceMetrics.max_drawdown(returns),
            'win_rate': PerformanceMetrics.win_rate(returns),
            'profit_factor': PerformanceMetrics.profit_factor(returns),
            'trades': len(returns[returns != 0])
        }

        return {
            'params': best_params,
            'in_sample': best_in_sample,
            'out_sample': out_sample_metrics,
            'in_sample_start': in_sample.index[0],
            'in_sample_end': in_sample.index[-1],
            'out_sample_start': out_sample.index[0],
            'out_sample_end': out_sample.index[-1]
        }

    def run(self, data: pd.DataFrame) -> List[Dict]:
        """
        Run full walk-forward optimization
        """
        current_date = data.index[0]
        end_date = data.index[-1]

        results = []

        while current_date < end_date:
            # Split data
            in_sample, out_sample = self.split_data(data, current_date)

            if len(in_sample) == 0 or len(out_sample) == 0:
                break

            # Optimize this period
            result = self.optimize_period(in_sample, out_sample)

            if result is not None:
                results.append(result)

            # Roll forward
            current_date += timedelta(days=self.config.step_days)

        self.results = results
        return results

    def analyze_results(self) -> Dict:
        """
        Analyze walk-forward results
        """
        if len(self.results) == 0:
            return None

        # Extract out-of-sample metrics
        oos_sharpes = [r['out_sample']['sharpe'] for r in self.results]
        oos_max_dds = [r['out_sample']['max_dd'] for r in self.results]
        oos_win_rates = [r['out_sample']['win_rate'] for r in self.results]

        # Calculate degradation (in-sample vs out-sample)
        is_sharpes = [r['in_sample']['sharpe'] for r in self.results]
        degradation = [(is_s - oos_s) / is_s for is_s, oos_s in zip(is_sharpes, oos_sharpes)]

        analysis = {
            'num_periods': len(self.results),
            'avg_oos_sharpe': np.mean(oos_sharpes),
            'median_oos_sharpe': np.median(oos_sharpes),
            'avg_oos_max_dd': np.mean(oos_max_dds),
            'avg_oos_win_rate': np.mean(oos_win_rates),
            'avg_degradation': np.mean(degradation),
            'periods_profitable': sum(1 for s in oos_sharpes if s > 0),
            'consistency': sum(1 for s in oos_sharpes if s > self.config.min_sharpe) / len(oos_sharpes)
        }

        return analysis

    def export_results(self, filename: str):
        """
        Export results to JSON
        """
        # Convert datetime objects to strings
        exportable = []
        for result in self.results:
            r = result.copy()
            r['in_sample_start'] = str(r['in_sample_start'])
            r['in_sample_end'] = str(r['in_sample_end'])
            r['out_sample_start'] = str(r['out_sample_start'])
            r['out_sample_end'] = str(r['out_sample_end'])
            exportable.append(r)

        with open(filename, 'w') as f:
            json.dump({
                'config': {
                    'in_sample_days': self.config.in_sample_days,
                    'out_sample_days': self.config.out_sample_days,
                    'step_days': self.config.step_days,
                    'param_ranges': self.config.param_ranges
                },
                'results': exportable,
                'analysis': self.analyze_results()
            }, f, indent=2)

# Example usage
if __name__ == "__main__":
    # Load data (example with random data)
    dates = pd.date_range('2022-01-01', '2024-06-12', freq='D')
    prices = pd.DataFrame({
        'close': 100 * (1 + np.random.randn(len(dates)).cumsum() * 0.01)
    }, index=dates)

    # Configure walk-forward
    config = WalkForwardConfig(
        in_sample_days=180,
        out_sample_days=60,
        step_days=30,
        param_ranges={
            'lookback': [10, 20, 30, 40, 50],
            'entry_threshold': [1.5, 2.0, 2.5, 3.0],
            'exit_threshold': [0.25, 0.5, 0.75, 1.0]
        },
        primary_metric='sharpe_ratio',
        min_sharpe=1.0,
        max_drawdown=0.15
    )

    # Run optimization
    optimizer = WalkForwardOptimizer(config)
    results = optimizer.run(prices)

    # Analyze
    analysis = optimizer.analyze_results()

    print("Walk-Forward Results:")
    print(f"Periods tested: {analysis['num_periods']}")
    print(f"Avg OOS Sharpe: {analysis['avg_oos_sharpe']:.2f}")
    print(f"Median OOS Sharpe: {analysis['median_oos_sharpe']:.2f}")
    print(f"Avg OOS Max DD: {analysis['avg_oos_max_dd']:.1%}")
    print(f"Avg Degradation: {analysis['avg_degradation']:.1%}")
    print(f"Consistency: {analysis['consistency']:.1%}")

    # Export
    optimizer.export_results('walk_forward_results.json')