parameter tuning - avoiding overfitting with walk-forward validation

import pandas as pd
import numpy as np
from backtrader import Cerebro, Strategy
from datetime import datetime, timedelta

class WalkForwardOptimizer:
    def __init__(
        self,
        strategy_class,
        data,
        param_ranges,
        in_sample_days=365,
        out_sample_days=90,
        step_days=90
    ):
        self.strategy_class = strategy_class
        self.data = data
        self.param_ranges = param_ranges
        self.in_sample_days = in_sample_days
        self.out_sample_days = out_sample_days
        self.step_days = step_days
        self.results = []

    def generate_windows(self):
        """Create rolling in-sample and out-of-sample windows"""
        start_date = self.data.index[0]
        end_date = self.data.index[-1]
        windows = []

        current_start = start_date
        while current_start + timedelta(days=self.in_sample_days + self.out_sample_days) <= end_date:
            in_sample_end = current_start + timedelta(days=self.in_sample_days)
            out_sample_end = in_sample_end + timedelta(days=self.out_sample_days)

            windows.append({
                'in_sample': (current_start, in_sample_end),
                'out_sample': (in_sample_end, out_sample_end)
            })

            current_start += timedelta(days=self.step_days)

        return windows

    def optimize_parameters(self, data_slice):
        """Grid search optimization on in-sample data"""
        best_params = None
        best_sharpe = -np.inf

        # Grid search over parameter ranges
        param_combinations = self._generate_param_grid()

        for params in param_combinations:
            cerebro = Cerebro()
            cerebro.addstrategy(self.strategy_class, **params)
            cerebro.adddata(data_slice)
            cerebro.broker.setcash(100000)

            # Run backtest
            results = cerebro.run()
            sharpe = self._calculate_sharpe(cerebro)

            if sharpe > best_sharpe:
                best_sharpe = sharpe
                best_params = params

        return best_params, best_sharpe

    def validate_parameters(self, params, data_slice):
        """Test parameters on out-of-sample data"""
        cerebro = Cerebro()
        cerebro.addstrategy(self.strategy_class, **params)
        cerebro.adddata(data_slice)
        cerebro.broker.setcash(100000)

        results = cerebro.run()
        sharpe = self._calculate_sharpe(cerebro)
        total_return = (cerebro.broker.getvalue() - 100000) / 100000

        return {
            'sharpe': sharpe,
            'return': total_return,
            'final_value': cerebro.broker.getvalue()
        }

    def run_walk_forward(self):
        """Execute complete walk-forward validation"""
        windows = self.generate_windows()

        for i, window in enumerate(windows):
            print(f"Window {i+1}/{len(windows)}")

            # Get in-sample and out-of-sample data
            in_start, in_end = window['in_sample']
            out_start, out_end = window['out_sample']

            in_sample_data = self.data.loc[in_start:in_end]
            out_sample_data = self.data.loc[out_start:out_end]

            # Optimize on in-sample
            best_params, in_sample_sharpe = self.optimize_parameters(in_sample_data)

            # Validate on out-of-sample
            out_sample_results = self.validate_parameters(best_params, out_sample_data)

            # Store results
            self.results.append({
                'window': i + 1,
                'in_sample_period': (in_start, in_end),
                'out_sample_period': (out_start, out_end),
                'params': best_params,
                'in_sample_sharpe': in_sample_sharpe,
                'out_sample_sharpe': out_sample_results['sharpe'],
                'out_sample_return': out_sample_results['return']
            })

        return self.results

    def _generate_param_grid(self):
        """Generate all parameter combinations"""
        import itertools
        keys = self.param_ranges.keys()
        values = self.param_ranges.values()
        combinations = [dict(zip(keys, v)) for v in itertools.product(*values)]
        return combinations

    def _calculate_sharpe(self, cerebro):
        """Calculate Sharpe ratio from backtest results"""
        # Get returns from cerebro portfolio value
        portfolio_values = cerebro.broker.get_value_history()
        returns = pd.Series(portfolio_values).pct_change().dropna()

        if len(returns) == 0 or returns.std() == 0:
            return 0

        sharpe = (returns.mean() / returns.std()) * np.sqrt(252)
        return sharpe

# Example usage
if __name__ == "__main__":
    # Load market data
    data = pd.read_csv('es_futures_5min.csv', index_col='datetime', parse_dates=True)

    # Define parameter ranges to test
    param_ranges = {
        'lookback_period': [10, 20, 30, 50],
        'volatility_threshold': [0.5, 1.0, 1.5, 2.0],
        'stop_loss_pct': [1.0, 1.5, 2.0],
        'take_profit_pct': [2.0, 3.0, 4.0]
    }

    # Run walk-forward optimization
    optimizer = WalkForwardOptimizer(
        strategy_class=MyStrategy,
        data=data,
        param_ranges=param_ranges,
        in_sample_days=365,  # 1 year in-sample
        out_sample_days=90,  # 3 months out-of-sample
        step_days=90         # Roll forward 3 months
    )

    results = optimizer.run_walk_forward()

    # Analyze results
    results_df = pd.DataFrame(results)
    print(f"Average out-of-sample Sharpe: {results_df['out_sample_sharpe'].mean():.2f}")
    print(f"Average out-of-sample return: {results_df['out_sample_return'].mean():.2%}")