| """
|
| Parameter optimization for the trading strategy.
|
|
|
| Tests different combinations of parameters to find the most profitable settings.
|
| Uses grid search with the real Binance data.
|
| """
|
|
|
| import pandas as pd
|
| import numpy as np
|
| import itertools
|
| from datetime import datetime
|
| import json
|
| import os
|
|
|
| from core.data_loader import load_data
|
| from core.features import calculate_all_features
|
| from core.signal_model import MACrossoverSignal
|
| from core.backtester import Backtester
|
| import config
|
|
|
|
|
| def optimize_parameters(
|
| data,
|
| ma_fast_range=[10, 15, 20, 25, 30],
|
| ma_slow_range=[30, 40, 50, 60, 80],
|
| stop_range=[0.005, 0.008, 0.010, 0.012, 0.015],
|
| tp_range=[0.010, 0.015, 0.020, 0.025, 0.030],
|
| leverage_range=[1.0, 2.0, 3.0],
|
| risk_range=[0.005, 0.01, 0.015],
|
| volatility_filter_range=[1.5, 2.0, 2.5, 3.0],
|
| max_combinations=100
|
| ):
|
| """
|
| Run grid search optimization on parameters.
|
|
|
| Parameters
|
| ----------
|
| data : pd.DataFrame
|
| Historical OHLCV data
|
| *_range : list
|
| Range of values to test for each parameter
|
| max_combinations : int
|
| Maximum combinations to test (for time management)
|
|
|
| Returns
|
| -------
|
| pd.DataFrame
|
| Results sorted by best performance
|
| """
|
| print("\n" + "="*70)
|
| print("PARAMETER OPTIMIZATION - GRID SEARCH")
|
| print("="*70)
|
|
|
|
|
| param_combinations = list(itertools.product(
|
| ma_fast_range,
|
| ma_slow_range,
|
| stop_range,
|
| tp_range,
|
| leverage_range,
|
| risk_range,
|
| volatility_filter_range
|
| ))
|
|
|
|
|
| valid_combinations = [
|
| (maf, mas, stop, tp, lev, risk, vf)
|
| for maf, mas, stop, tp, lev, risk, vf in param_combinations
|
| if maf < mas and tp > stop
|
| ]
|
|
|
| print(f"\nTotal valid combinations: {len(valid_combinations):,}")
|
|
|
|
|
| if len(valid_combinations) > max_combinations:
|
| print(f"Limiting to {max_combinations} random samples for time management...")
|
| np.random.seed(42)
|
| indices = np.random.choice(len(valid_combinations), max_combinations, replace=False)
|
| valid_combinations = [valid_combinations[i] for i in indices]
|
|
|
| print(f"Testing {len(valid_combinations)} parameter combinations...\n")
|
|
|
| results = []
|
|
|
| for idx, (ma_fast, ma_slow, s_stop, s_tp, leverage, rho, vol_filter) in enumerate(valid_combinations):
|
| try:
|
|
|
| features = calculate_all_features(
|
| data,
|
| ma_fast=ma_fast,
|
| ma_slow=ma_slow,
|
| atr_period=14,
|
| trend_window=60
|
| )
|
|
|
|
|
| signal_model = MACrossoverSignal(volatility_threshold=vol_filter)
|
| signals = signal_model.generate_signals(features)
|
|
|
|
|
| backtester = Backtester(
|
| data=data,
|
| features=features,
|
| signals=signals,
|
| signal_model=signal_model,
|
| e_base=config.E_BASE,
|
| leverage=leverage,
|
| risk_fraction=rho,
|
| stop_distance=s_stop,
|
| tp_distance=s_tp,
|
| d_max=config.D_MAX,
|
| commission=config.COMMISSION,
|
| slippage=config.SLIPPAGE,
|
| verbose=False
|
| )
|
|
|
| result = backtester.run()
|
| metrics = result['metrics']
|
|
|
|
|
| results.append({
|
| 'ma_fast': ma_fast,
|
| 'ma_slow': ma_slow,
|
| 'stop': s_stop,
|
| 'tp': s_tp,
|
| 'leverage': leverage,
|
| 'risk': rho,
|
| 'vol_filter': vol_filter,
|
| 'total_return': metrics['total_return_pct'],
|
| 'sharpe': metrics['sharpe_ratio'],
|
| 'max_dd': metrics['max_drawdown_pct'],
|
| 'win_rate': metrics['win_rate'],
|
| 'profit_factor': metrics['profit_factor'],
|
| 'num_trades': metrics['num_trades'],
|
| 'final_equity': metrics['total_equity']
|
| })
|
|
|
|
|
| if (idx + 1) % 10 == 0:
|
| print(f" Tested {idx + 1}/{len(valid_combinations)} combinations...")
|
|
|
| except Exception as e:
|
|
|
| continue
|
|
|
|
|
| results_df = pd.DataFrame(results)
|
|
|
|
|
| results_df = results_df.sort_values('total_return', ascending=False)
|
|
|
| print(f"\n✓ Optimization complete! Tested {len(results_df)} combinations")
|
|
|
| return results_df
|
|
|
|
|
| def print_top_results(results_df, top_n=10):
|
| """Print top performing parameter sets."""
|
| print("\n" + "="*70)
|
| print(f"TOP {top_n} PARAMETER COMBINATIONS")
|
| print("="*70 + "\n")
|
|
|
| for rank, (idx, row) in enumerate(results_df.head(top_n).iterrows(), 1):
|
| print(f"#{rank} - Return: {row['total_return']:+.2f}% | Sharpe: {row['sharpe']:.2f} | Win Rate: {row['win_rate']:.1f}%")
|
| print(f" MA: {int(row['ma_fast'])}/{int(row['ma_slow'])} | "
|
| f"Stop/TP: {row['stop']*100:.2f}%/{row['tp']*100:.2f}% | "
|
| f"Lev: {row['leverage']:.1f}x | Risk: {row['risk']*100:.1f}%")
|
| print(f" Trades: {int(row['num_trades'])} | "
|
| f"Profit Factor: {row['profit_factor']:.2f} | "
|
| f"Max DD: {row['max_dd']:.1f}%")
|
| print()
|
|
|
|
|
| def save_optimization_results(results_df, output_path):
|
| """Save results to CSV."""
|
| results_df.to_csv(output_path, index=False)
|
| print(f"✓ Saved full results to: {output_path}")
|
|
|
|
|
| def get_best_parameters(results_df):
|
| """Get the best parameter set."""
|
| best = results_df.iloc[0]
|
|
|
| return {
|
| 'MA_FAST': int(best['ma_fast']),
|
| 'MA_SLOW': int(best['ma_slow']),
|
| 'S_STOP': best['stop'],
|
| 'S_TP': best['tp'],
|
| 'L_DEFAULT': best['leverage'],
|
| 'RHO': best['risk'],
|
| 'VOLATILITY_FILTER': best['vol_filter']
|
| }
|
|
|
|
|
| def main():
|
| """Main optimization execution."""
|
| print("\n🔍 STRATEGY PARAMETER OPTIMIZATION")
|
| print("="*70)
|
|
|
|
|
| print("\n📥 Loading Binance data...")
|
| data = load_data(config.BTC_DATA_FILE)
|
| print(f" Loaded {len(data):,} bars")
|
|
|
|
|
| print("\n🎯 Defining parameter search space...")
|
| print(" MA Fast: [15, 20, 25, 30, 40]")
|
| print(" MA Slow: [40, 50, 60, 80, 100]")
|
| print(" Stop-Loss: [0.5%, 0.8%, 1.0%, 1.2%, 1.5%]")
|
| print(" Take-Profit: [1.0%, 1.5%, 2.0%, 2.5%, 3.0%]")
|
| print(" Leverage: [1x, 2x, 3x]")
|
| print(" Risk per trade: [0.5%, 1.0%, 1.5%]")
|
| print(" Volatility filter: [1.5, 2.0, 2.5, 3.0]")
|
|
|
|
|
| results_df = optimize_parameters(
|
| data,
|
| ma_fast_range=[15, 20, 25, 30, 40],
|
| ma_slow_range=[40, 50, 60, 80, 100],
|
| stop_range=[0.005, 0.008, 0.010, 0.012, 0.015],
|
| tp_range=[0.010, 0.015, 0.020, 0.025, 0.030],
|
| leverage_range=[1.0, 2.0, 3.0],
|
| risk_range=[0.005, 0.01, 0.015],
|
| volatility_filter_range=[1.5, 2.0, 2.5, 3.0],
|
| max_combinations=100
|
| )
|
|
|
|
|
| print_top_results(results_df, top_n=10)
|
|
|
|
|
| output_path = os.path.join(config.REPORTS_DIR, 'optimization_results.csv')
|
| save_optimization_results(results_df, output_path)
|
|
|
|
|
| best_params = get_best_parameters(results_df)
|
|
|
| print("\n" + "="*70)
|
| print("📌 RECOMMENDED PARAMETERS (Best Performance)")
|
| print("="*70)
|
| for param, value in best_params.items():
|
| print(f" {param} = {value}")
|
|
|
|
|
| best_params_path = os.path.join(config.REPORTS_DIR, 'best_parameters.json')
|
| with open(best_params_path, 'w') as f:
|
| json.dump(best_params, f, indent=2)
|
| print(f"\n✓ Saved best parameters to: {best_params_path}")
|
|
|
| print("\n💡 Next steps:")
|
| print(" 1. Update config.py with the recommended parameters")
|
| print(" 2. Run: python run_backtest.py (to verify)")
|
| print(" 3. If profitable, proceed to live paper trading!")
|
| print("="*70 + "\n")
|
|
|
|
|
| if __name__ == "__main__":
|
| main()
|
|
|
