zhutoutoutousan
360 lines
14 KiB
Python
360 lines
14 KiB
Python
"""
|
|
Parameter Optimization for ONNX Strategy
|
|
|
|
This script optimizes strategy parameters (prediction_threshold, min_confidence,
|
|
stop_loss_pips, take_profit_pips, lot_size) using grid search or random search.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
import MetaTrader5 as mt5
|
|
import pandas as pd
|
|
import numpy as np
|
|
from itertools import product
|
|
import json
|
|
|
|
# Add paths
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
backtest_dir = os.path.join(os.path.dirname(current_dir), 'backtesting', 'MT5')
|
|
sys.path.insert(0, backtest_dir)
|
|
|
|
from backtest_engine import BacktestEngine
|
|
from onnx_backtest_strategy import ONNXBacktestStrategy
|
|
from performance_analyzer import PerformanceAnalyzer
|
|
|
|
|
|
class ONNXParameterOptimizer:
|
|
"""Optimize ONNX strategy parameters."""
|
|
|
|
def __init__(self, symbol: str, timeframe: int, model_path: str, scaler_path: str,
|
|
start_date: datetime, end_date: datetime, initial_balance: float = 10000.0):
|
|
"""
|
|
Initialize optimizer.
|
|
|
|
Args:
|
|
symbol: Trading symbol
|
|
timeframe: MT5 timeframe
|
|
model_path: Path to ONNX model
|
|
scaler_path: Path to scaler
|
|
start_date: Backtest start date
|
|
end_date: Backtest end date
|
|
initial_balance: Starting balance
|
|
"""
|
|
self.symbol = symbol
|
|
self.timeframe = timeframe
|
|
self.model_path = model_path
|
|
self.scaler_path = scaler_path
|
|
self.start_date = start_date
|
|
self.end_date = end_date
|
|
self.initial_balance = initial_balance
|
|
|
|
def grid_search(self, param_grid: dict, metric: str = 'sharpe_ratio') -> pd.DataFrame:
|
|
"""
|
|
Perform grid search optimization.
|
|
|
|
Args:
|
|
param_grid: Dictionary of parameter ranges
|
|
Example: {
|
|
'prediction_threshold': [0.0001, 0.0002, 0.0005],
|
|
'min_confidence': [0.2, 0.3, 0.4],
|
|
'stop_loss_pips': [30, 50, 70],
|
|
'take_profit_pips': [60, 100, 150],
|
|
'lot_size': [0.1, 0.2]
|
|
}
|
|
metric: Metric to optimize ('sharpe_ratio', 'total_return', 'max_drawdown', 'profit_factor')
|
|
|
|
Returns:
|
|
DataFrame with results sorted by metric
|
|
"""
|
|
print("="*60)
|
|
print("Grid Search Parameter Optimization")
|
|
print("="*60)
|
|
|
|
# Generate all parameter combinations
|
|
param_names = list(param_grid.keys())
|
|
param_values = list(param_grid.values())
|
|
combinations = list(product(*param_values))
|
|
|
|
total_combinations = len(combinations)
|
|
print(f"\nTotal parameter combinations: {total_combinations}")
|
|
print(f"Optimizing for: {metric}\n")
|
|
|
|
results = []
|
|
|
|
for i, combo in enumerate(combinations, 1):
|
|
params = dict(zip(param_names, combo))
|
|
|
|
print(f"[{i}/{total_combinations}] Testing: {params}")
|
|
|
|
try:
|
|
# Create strategy with these parameters
|
|
strategy = ONNXBacktestStrategy(
|
|
symbol=self.symbol,
|
|
timeframe=self.timeframe,
|
|
model_path=self.model_path,
|
|
scaler_path=self.scaler_path,
|
|
initial_balance=self.initial_balance,
|
|
**params
|
|
)
|
|
|
|
# Run backtest
|
|
engine = BacktestEngine(strategy, self.start_date, self.end_date)
|
|
backtest_results = engine.run()
|
|
|
|
# Calculate metrics
|
|
analyzer = PerformanceAnalyzer(backtest_results)
|
|
metrics = analyzer.metrics
|
|
|
|
# Store results
|
|
result = params.copy()
|
|
# Map metric names to match what we're looking for
|
|
result['total_return'] = metrics.get('total_return_pct', 0) / 100.0
|
|
result['max_drawdown'] = metrics.get('max_drawdown_pct', 0) / 100.0
|
|
result['sharpe_ratio'] = metrics.get('sharpe_ratio', 0.0) if 'sharpe_ratio' in metrics else 0.0
|
|
result['profit_factor'] = metrics.get('profit_factor', 0.0)
|
|
result['win_rate'] = metrics.get('win_rate_pct', 0) / 100.0
|
|
result['total_trades'] = metrics.get('total_trades', 0)
|
|
result['final_balance'] = metrics.get('final_balance', self.initial_balance)
|
|
results.append(result)
|
|
|
|
metric_value = result.get(metric, 0)
|
|
print(f" -> {metric}: {metric_value:.4f} | Trades: {result['total_trades']}")
|
|
|
|
except Exception as e:
|
|
print(f" X Error: {e}")
|
|
continue
|
|
|
|
# Convert to DataFrame
|
|
df_results = pd.DataFrame(results)
|
|
|
|
if len(df_results) == 0:
|
|
raise ValueError("No successful backtests!")
|
|
|
|
# Sort by metric (descending for most metrics, ascending for max_drawdown)
|
|
if metric == 'max_drawdown':
|
|
df_results = df_results.sort_values(metric, ascending=True)
|
|
else:
|
|
df_results = df_results.sort_values(metric, ascending=False)
|
|
|
|
return df_results
|
|
|
|
def random_search(self, param_ranges: dict, n_iter: int = 50,
|
|
metric: str = 'sharpe_ratio') -> pd.DataFrame:
|
|
"""
|
|
Perform random search optimization.
|
|
|
|
Args:
|
|
param_ranges: Dictionary of parameter ranges
|
|
Example: {
|
|
'prediction_threshold': (0.0001, 0.001),
|
|
'min_confidence': (0.1, 0.5),
|
|
'stop_loss_pips': (20, 100),
|
|
'take_profit_pips': (40, 200),
|
|
'lot_size': (0.1, 0.5)
|
|
}
|
|
n_iter: Number of random combinations to test
|
|
metric: Metric to optimize
|
|
|
|
Returns:
|
|
DataFrame with results sorted by metric
|
|
"""
|
|
print("="*60)
|
|
print("Random Search Parameter Optimization")
|
|
print("="*60)
|
|
print(f"\nTesting {n_iter} random parameter combinations")
|
|
print(f"Optimizing for: {metric}\n")
|
|
|
|
results = []
|
|
np.random.seed(42) # For reproducibility
|
|
|
|
for i in range(1, n_iter + 1):
|
|
# Generate random parameters
|
|
params = {}
|
|
for param_name, (min_val, max_val) in param_ranges.items():
|
|
if isinstance(min_val, int) and isinstance(max_val, int):
|
|
params[param_name] = np.random.randint(min_val, max_val + 1)
|
|
else:
|
|
params[param_name] = np.random.uniform(min_val, max_val)
|
|
|
|
print(f"[{i}/{n_iter}] Testing: {params}")
|
|
|
|
try:
|
|
# Create strategy
|
|
strategy = ONNXBacktestStrategy(
|
|
symbol=self.symbol,
|
|
timeframe=self.timeframe,
|
|
model_path=self.model_path,
|
|
scaler_path=self.scaler_path,
|
|
initial_balance=self.initial_balance,
|
|
**params
|
|
)
|
|
|
|
# Run backtest
|
|
engine = BacktestEngine(strategy, self.start_date, self.end_date)
|
|
backtest_results = engine.run()
|
|
|
|
# Calculate metrics
|
|
analyzer = PerformanceAnalyzer(backtest_results)
|
|
metrics = analyzer.metrics
|
|
|
|
# Store results
|
|
result = params.copy()
|
|
# Map metric names to match what we're looking for
|
|
result['total_return'] = metrics.get('total_return_pct', 0) / 100.0
|
|
result['max_drawdown'] = metrics.get('max_drawdown_pct', 0) / 100.0
|
|
result['sharpe_ratio'] = metrics.get('sharpe_ratio', 0.0) if 'sharpe_ratio' in metrics else 0.0
|
|
result['profit_factor'] = metrics.get('profit_factor', 0.0)
|
|
result['win_rate'] = metrics.get('win_rate_pct', 0) / 100.0
|
|
result['total_trades'] = metrics.get('total_trades', 0)
|
|
result['final_balance'] = metrics.get('final_balance', self.initial_balance)
|
|
results.append(result)
|
|
|
|
metric_value = result.get(metric, 0)
|
|
print(f" -> {metric}: {metric_value:.4f} | Trades: {result['total_trades']}")
|
|
|
|
except Exception as e:
|
|
print(f" X Error: {e}")
|
|
continue
|
|
|
|
# Convert to DataFrame
|
|
df_results = pd.DataFrame(results)
|
|
|
|
if len(df_results) == 0:
|
|
raise ValueError("No successful backtests!")
|
|
|
|
# Sort by metric
|
|
if metric == 'max_drawdown':
|
|
df_results = df_results.sort_values(metric, ascending=True)
|
|
else:
|
|
df_results = df_results.sort_values(metric, ascending=False)
|
|
|
|
return df_results
|
|
|
|
def save_results(self, df_results: pd.DataFrame, output_file: str = 'optimization_results.csv'):
|
|
"""Save optimization results to CSV."""
|
|
df_results.to_csv(output_file, index=False)
|
|
print(f"\nResults saved to: {output_file}")
|
|
|
|
# Also save top 10 as JSON
|
|
top_10 = df_results.head(10).to_dict('records')
|
|
json_file = output_file.replace('.csv', '_top10.json')
|
|
with open(json_file, 'w') as f:
|
|
json.dump(top_10, f, indent=2, default=str)
|
|
print(f"Top 10 results saved to: {json_file}")
|
|
|
|
|
|
def main():
|
|
"""Main optimization function."""
|
|
# Configuration
|
|
symbol = 'XAUUSD'
|
|
timeframe = mt5.TIMEFRAME_H1
|
|
model_path = 'models/XAUUSD_H1_model.onnx'
|
|
scaler_path = 'models/XAUUSD_H1_scaler.pkl'
|
|
initial_balance = 10000.0
|
|
|
|
# Backtest date range (use last 6 months for optimization)
|
|
end_date = datetime.now()
|
|
start_date = end_date - timedelta(days=180)
|
|
|
|
# Check if model exists
|
|
if not os.path.exists(model_path):
|
|
print(f"ERROR: Model not found: {model_path}")
|
|
print("Please train the model first using train_onnx_model.py")
|
|
return
|
|
|
|
# Initialize MT5
|
|
if not mt5.initialize():
|
|
print("ERROR: Failed to initialize MT5")
|
|
return
|
|
|
|
try:
|
|
# Create optimizer
|
|
optimizer = ONNXParameterOptimizer(
|
|
symbol=symbol,
|
|
timeframe=timeframe,
|
|
model_path=model_path,
|
|
scaler_path=scaler_path,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
initial_balance=initial_balance
|
|
)
|
|
|
|
# Choose optimization method (use command line args or defaults)
|
|
import sys
|
|
choice = "2" # Default to random search
|
|
n_iter = 30 # Default iterations
|
|
|
|
if len(sys.argv) > 1:
|
|
choice = sys.argv[1]
|
|
if len(sys.argv) > 2:
|
|
n_iter = int(sys.argv[2])
|
|
|
|
print("\nOptimization Configuration:")
|
|
print(f"Method: {'Grid Search' if choice == '1' else 'Random Search'}")
|
|
if choice != "1":
|
|
print(f"Iterations: {n_iter}")
|
|
print()
|
|
|
|
if choice == "1":
|
|
# Grid search parameters
|
|
param_grid = {
|
|
'prediction_threshold': [0.0001, 0.0002, 0.0005, 0.001],
|
|
'min_confidence': [0.1, 0.2, 0.3, 0.4],
|
|
'stop_loss_pips': [30, 50, 70, 100],
|
|
'take_profit_pips': [60, 100, 150, 200],
|
|
'lot_size': [0.1, 0.2]
|
|
}
|
|
|
|
results = optimizer.grid_search(param_grid, metric='sharpe_ratio')
|
|
|
|
else:
|
|
# Random search parameters
|
|
param_ranges = {
|
|
'prediction_threshold': (0.00005, 0.002), # Lower threshold to get more trades
|
|
'min_confidence': (0.05, 0.5), # Lower confidence requirement
|
|
'stop_loss_pips': (20, 150),
|
|
'take_profit_pips': (40, 300),
|
|
'lot_size': (0.05, 0.3)
|
|
}
|
|
|
|
results = optimizer.random_search(param_ranges, n_iter=n_iter, metric='sharpe_ratio')
|
|
|
|
# Display top results
|
|
print("\n" + "="*60)
|
|
print("Top 10 Results")
|
|
print("="*60)
|
|
print(results.head(10).to_string(index=False))
|
|
|
|
# Save results
|
|
optimizer.save_results(results, 'onnx_optimization_results.csv')
|
|
|
|
# Display best parameters
|
|
best = results.iloc[0]
|
|
print("\n" + "="*60)
|
|
print("Best Parameters")
|
|
print("="*60)
|
|
print(f"Prediction Threshold: {best['prediction_threshold']:.6f}")
|
|
print(f"Min Confidence: {best['min_confidence']:.2f}")
|
|
print(f"Stop Loss (pips): {best['stop_loss_pips']}")
|
|
print(f"Take Profit (pips): {best['take_profit_pips']}")
|
|
print(f"Lot Size: {best['lot_size']:.2f}")
|
|
print(f"\nPerformance Metrics:")
|
|
print(f" Sharpe Ratio: {best.get('sharpe_ratio', 'N/A'):.4f}")
|
|
print(f" Total Return: {best.get('total_return', 'N/A'):.2%}")
|
|
print(f" Max Drawdown: {best.get('max_drawdown', 'N/A'):.2%}")
|
|
print(f" Profit Factor: {best.get('profit_factor', 'N/A'):.2f}")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\nOptimization interrupted by user")
|
|
except Exception as e:
|
|
print(f"\nERROR: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
mt5.shutdown()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|