Walk-Forward Optimization¶

Validate strategy robustness using walk-forward analysis.

---

📋 Notebook Information

• RustyBT Version: 0.1.2+
• Last Validated: 2025-11-07
• API Compatibility: Verified ✅
• Documentation: API Reference

---

In [1]:

from rustybt.analytics import setup_notebook

setup_notebook()

from rustybt.analytics import setup_notebook setup_notebook()

✅ Notebook environment configured successfully
   - Async/await support enabled
   - Pandas display options optimized
   - Progress bars configured

Walk-Forward Testing¶

Prevents overfitting by testing on unseen data.

In [2]:

# Example: Walk-forward optimization
import numpy as np
import polars as pl
from rustybt.optimization import (
    WalkForwardOptimizer,
    WindowConfig,
    ParameterSpace,
    DiscreteParameter,
    ObjectiveFunction,
)
from rustybt.optimization.search import GridSearchAlgorithm

# Define parameter space
param_space = ParameterSpace(
    parameters=[
        DiscreteParameter(name='fast_period', min_value=10, max_value=30, step=10),
        DiscreteParameter(name='slow_period', min_value=50, max_value=100, step=25),
    ]
)

# Configure walk-forward windows
# Train on 250 days, validate on 50 days, test on 50 days
# Advance by 50 days each iteration
config = WindowConfig(
    train_period=250,
    validation_period=50,
    test_period=50,
    step_size=50,
    window_type='rolling',  # or 'expanding' for growing training window
)

# Define backtest function that accepts parameters and data
def backtest_with_data(params, data):
    """
    Run backtest with given parameters on specific data window.
    
    Args:
        params: Dictionary of strategy parameters
        data: Polars DataFrame with market data for this window
        
    Returns:
        Dictionary with metrics (must include objective metric)
    """
    # Example: Run your strategy on the data window
    # results = run_algorithm(
    #     start=data['date'].min(),
    #     end=data['date'].max(),
    #     initialize=lambda context: MyStrategy().initialize(
    #         context,
    #         fast_period=params['fast_period'],
    #         slow_period=params['slow_period']
    #     ),
    #     handle_data=MyStrategy().handle_data,
    #     capital_base=100000.0,
    #     bundle='custom',  # Use data from the window
    # )
    # return {'sharpe_ratio': results['sharpe'].iloc[-1]}
    
    # Placeholder return for demonstration
    return {
        'sharpe_ratio': np.random.uniform(0.5, 2.0),
        'total_return': np.random.uniform(0.1, 0.5),
    }

# Create walk-forward optimizer
wf_optimizer = WalkForwardOptimizer(
    parameter_space=param_space,
    search_algorithm_factory=lambda: GridSearchAlgorithm(parameter_space=param_space),
    objective_function=ObjectiveFunction(metric="sharpe_ratio"),
    backtest_function=backtest_with_data,
    config=config,
    max_trials_per_window=12,  # Run 12 trials per training window
)

# Prepare historical data as Polars DataFrame
# In practice, this would be your actual market data
# Example structure:
# market_data = pl.DataFrame({
#     'date': [...],
#     'symbol': [...],
#     'open': [...],
#     'high': [...],
#     'low': [...],
#     'close': [...],
#     'volume': [...],
# })

# Run walk-forward optimization
# wf_result = wf_optimizer.run(market_data)

# Analyze results
# print(f"Number of windows tested: {len(wf_result.window_results)}")
# print(f"Average out-of-sample Sharpe: {wf_result.out_of_sample_sharpe:.2f}")
# print(f"Stability score: {wf_result.stability_score:.2f}")

# Access results per window
# for i, window_result in enumerate(wf_result.window_results):
#     print(f"\nWindow {i+1}:")
#     print(f"  Best in-sample params: {window_result.best_in_sample_params}")
#     print(f"  In-sample Sharpe: {window_result.in_sample_score:.2f}")
#     print(f"  Out-of-sample Sharpe: {window_result.out_of_sample_score:.2f}")

# Example: Walk-forward optimization import numpy as np import polars as pl from rustybt.optimization import ( WalkForwardOptimizer, WindowConfig, ParameterSpace, DiscreteParameter, ObjectiveFunction, ) from rustybt.optimization.search import GridSearchAlgorithm # Define parameter space param_space = ParameterSpace( parameters=[ DiscreteParameter(name='fast_period', min_value=10, max_value=30, step=10), DiscreteParameter(name='slow_period', min_value=50, max_value=100, step=25), ] ) # Configure walk-forward windows # Train on 250 days, validate on 50 days, test on 50 days # Advance by 50 days each iteration config = WindowConfig( train_period=250, validation_period=50, test_period=50, step_size=50, window_type='rolling', # or 'expanding' for growing training window ) # Define backtest function that accepts parameters and data def backtest_with_data(params, data): """ Run backtest with given parameters on specific data window. Args: params: Dictionary of strategy parameters data: Polars DataFrame with market data for this window Returns: Dictionary with metrics (must include objective metric) """ # Example: Run your strategy on the data window # results = run_algorithm( # start=data['date'].min(), # end=data['date'].max(), # initialize=lambda context: MyStrategy().initialize( # context, # fast_period=params['fast_period'], # slow_period=params['slow_period'] # ), # handle_data=MyStrategy().handle_data, # capital_base=100000.0, # bundle='custom', # Use data from the window # ) # return {'sharpe_ratio': results['sharpe'].iloc[-1]} # Placeholder return for demonstration return { 'sharpe_ratio': np.random.uniform(0.5, 2.0), 'total_return': np.random.uniform(0.1, 0.5), } # Create walk-forward optimizer wf_optimizer = WalkForwardOptimizer( parameter_space=param_space, search_algorithm_factory=lambda: GridSearchAlgorithm(parameter_space=param_space), objective_function=ObjectiveFunction(metric="sharpe_ratio"), backtest_function=backtest_with_data, config=config, max_trials_per_window=12, # Run 12 trials per training window ) # Prepare historical data as Polars DataFrame # In practice, this would be your actual market data # Example structure: # market_data = pl.DataFrame({ # 'date': [...], # 'symbol': [...], # 'open': [...], # 'high': [...], # 'low': [...], # 'close': [...], # 'volume': [...], # }) # Run walk-forward optimization # wf_result = wf_optimizer.run(market_data) # Analyze results # print(f"Number of windows tested: {len(wf_result.window_results)}") # print(f"Average out-of-sample Sharpe: {wf_result.out_of_sample_sharpe:.2f}") # print(f"Stability score: {wf_result.stability_score:.2f}") # Access results per window # for i, window_result in enumerate(wf_result.window_results): # print(f"\nWindow {i+1}:") # print(f" Best in-sample params: {window_result.best_in_sample_params}") # print(f" In-sample Sharpe: {window_result.in_sample_score:.2f}") # print(f" Out-of-sample Sharpe: {window_result.out_of_sample_score:.2f}")