mirror of
https://github.com/Arianhgh/fx-quant-research.git
synced 2026-08-01 21:17:44 +00:00
Arian Haghparast
281 lines
11 KiB
Python
281 lines
11 KiB
Python
"""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%}")
|