"""Walk-forward optimization harness.""" import numpy as np import pandas as pd from pathlib import Path from tradingbot.models.model_manager import GoldModelManager from tradingbot.signals.generator import GoldSignalGenerator class WalkForwardOptimizer: """ Implements walk-forward optimization for model training and validation """ def __init__( self, train_window=120, # Initial training days (approx. 34560 5-min bars) step_size=20, # Days to move forward in each step (approx. 5760 bars) feature_selection_interval=3, # How often to perform feature selection test_window=10, # Days to test on after each training (approx. 2880 bars) n_jobs=-1, model_path='models', random_state=42 ): """ Initialize the walk-forward optimizer Parameters: ----------- train_window : int Number of days for initial training window step_size : int Number of days to move forward in each step feature_selection_interval : int How many steps between feature selection (to save time) test_window : int Number of days to test on after each training n_jobs : int Number of parallel jobs for training model_path : str Directory to save models """ self.train_window = train_window self.step_size = step_size self.feature_selection_interval = feature_selection_interval self.test_window = test_window self.n_jobs = n_jobs self.model_path = Path(model_path) self.model_path.mkdir(exist_ok=True) self.random_state = random_state # Initialize metrics storage self.metrics = [] self.signals = [] def optimize(self, data, min_train_size=8000): """ Perform walk-forward optimization with improved error handling Parameters: ----------- data : pd.DataFrame DataFrame with features and target min_train_size : int Minimum number of samples required for training Returns: -------- results : dict Walk-forward optimization results """ # Convert time windows from days to number of 5-minute bars # Assuming 288 5-minute bars per day (24 hours) bars_per_day = 288 train_bars = self.train_window * bars_per_day step_bars = self.step_size * bars_per_day test_bars = self.test_window * bars_per_day # Ensure data has a datetime index if not isinstance(data.index, pd.DatetimeIndex): raise ValueError("Data must have a DatetimeIndex") # Calculate number of steps total_bars = len(data) n_steps = max(1, (total_bars - train_bars) // step_bars) print(f"Starting walk-forward optimization with {n_steps} steps") # Initialize results storage all_signals = pd.DataFrame() all_metrics = [] # Track whether to do feature selection in this step do_feature_selection = True selected_features = None # For each step for step in range(n_steps): print(f"\n{'-'*50}") print(f"Step {step+1}/{n_steps}") print(f"{'-'*50}") try: # Calculate indices for this step train_start = step * step_bars train_end = train_start + train_bars test_start = train_end test_end = min(test_start + test_bars, total_bars) # Get data for this step train_data = data.iloc[train_start:train_end] test_data = data.iloc[test_start:test_end] print(f"Train period: {train_data.index[0]} to {train_data.index[-1]}") print(f"Test period: {test_data.index[0]} to {test_data.index[-1]}") # Skip if not enough training data if len(train_data) < min_train_size: print(f"Skipping step {step+1} - not enough training data ({len(train_data)} < {min_train_size})") continue # Create model manager for this step model_dir = self.model_path / f'step_{step+1}' model_dir.mkdir(exist_ok=True, parents=True) model_manager = GoldModelManager( n_splits=3, feature_selection_method='boruta' if do_feature_selection else 'importance', max_features=25, ensemble_models=3, volatility_based_models=True, n_jobs=self.n_jobs, model_path=model_dir, random_state=self.random_state + step # Vary random state by step ) # Extract y_train for class distribution check before training X_train, y_train = model_manager.prepare_data(train_data) unique_classes = np.unique(y_train) class_counts = {cls: np.sum(y_train == cls) for cls in unique_classes} print("Class distribution in training data:") expected_classes = set(range(-2, 3)) # -2, -1, 0, 1, 2 missing_classes = expected_classes - set(unique_classes) for cls in sorted(expected_classes): count = class_counts.get(cls, 0) percentage = (count / len(y_train)) * 100 if len(y_train) > 0 else 0 status = "PRESENT" if cls in unique_classes else "MISSING" print(f" Class {cls}: {count} samples ({percentage:.2f}%) - {status}") if missing_classes: print(f"Warning: Missing classes in training data: {missing_classes}") print("Continuing anyway with appropriate class weights...") # Set selected features if not doing feature selection if not do_feature_selection and selected_features is not None: model_manager.selected_features = selected_features # Train models model_manager.fit(train_data) # Store selected features for future steps if do_feature_selection or selected_features is None: selected_features = model_manager.selected_features # Update feature selection flag for next step do_feature_selection = ((step + 1) % self.feature_selection_interval == 0) # Skip signal generation if model training failed if model_manager.meta_model is None: print("Warning: Meta-model training failed, skipping signal generation") continue # Create signal generator signal_generator = GoldSignalGenerator( confidence_threshold=0.7, risk_reward_min=1.5, model_manager=model_manager ) # Generate signals on test data signals = signal_generator.generate_signals(test_data) # Check if signals DataFrame is empty if signals.empty or 'signal' not in signals.columns: print("Warning: No signals generated, skipping analysis") continue # Check if any signals were generated signal_count = (signals['signal'] != 0).sum() if 'signal' in signals.columns else 0 if signal_count == 0: print("Warning: No active signals found in test period") else: print(f"Generated {signal_count} active signals") # Analyze signals metrics = signal_generator.analyze_signals(signals, test_data) # Store results signals['step'] = step + 1 all_signals = pd.concat([all_signals, signals]) metrics['step'] = step + 1 metrics['train_start'] = train_data.index[0] metrics['train_end'] = train_data.index[-1] metrics['test_start'] = test_data.index[0] metrics['test_end'] = test_data.index[-1] all_metrics.append(metrics) except Exception as e: import traceback print(f"Error in step {step+1}: {str(e)}") print("Detailed traceback:") print(traceback.format_exc()) continue # Compile final results results = { 'signals': all_signals, 'metrics': all_metrics } # Print overall performance self._print_overall_performance(all_metrics) return results def _print_overall_performance(self, metrics): """Print overall performance statistics""" if not metrics: print("No metrics available for performance analysis") return print("\n=== Overall Walk-Forward Performance ===") # Calculate average metrics total_signals = sum(m['total_signals'] for m in metrics) avg_win_rate = np.mean([m['overall_win_rate'] for m in metrics if not np.isnan(m['overall_win_rate'])]) avg_return = np.mean([m['overall_avg_return'] for m in metrics if not np.isnan(m['overall_avg_return'])]) print(f"Total Steps: {len(metrics)}") print(f"Total Signals: {total_signals}") print(f"Average Win Rate: {avg_win_rate:.2%}") print(f"Average Return: {avg_return:.2%}") # Calculate performance by regime if available regime_performance = {} for m in metrics: if m['regime_stats'] is not None: for regime, stats in m['regime_stats'].items(): if regime not in regime_performance: regime_performance[regime] = { 'count': 0, 'win_rate': [], 'avg_return': [] } regime_performance[regime]['count'] += stats['count'] # Long performance if not np.isnan(stats['long_win_rate']): regime_performance[regime]['win_rate'].append(stats['long_win_rate']) if not np.isnan(stats['long_avg_return']): regime_performance[regime]['avg_return'].append(stats['long_avg_return']) # Short performance if not np.isnan(stats['short_win_rate']): regime_performance[regime]['win_rate'].append(stats['short_win_rate']) if not np.isnan(stats['short_avg_return']): regime_performance[regime]['avg_return'].append(stats['short_avg_return']) if regime_performance: print("\nPerformance by Volatility Regime:") for regime, stats in regime_performance.items(): avg_win_rate = np.mean(stats['win_rate']) if stats['win_rate'] else np.nan avg_return = np.mean(stats['avg_return']) if stats['avg_return'] else np.nan print(f"Regime {regime}: {stats['count']} signals, Win Rate: {avg_win_rate:.2%}, Avg Return: {avg_return:.2%}")