#!/usr/bin/env python3 """ Comprehensive parameter sweep to maximize v9 profitability. Tests all critical parameters that could boost profit. """ import sys from pathlib import Path from datetime import datetime import itertools import multiprocessing as mp from typing import List, Tuple, Optional # Add parent directories to path sys.path.insert(0, str(Path(__file__).parent.parent)) sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from backtester.data_loader import load_csv from backtester.simulator import simulate_money_line, TradeConfig from backtester.indicators.money_line import MoneyLineInputs def test_config(args): """Test a single parameter configuration.""" config_id, params, data_slice = args try: # Unpack parameters flip_thresh, ma_gap, adx_min, long_pos, short_pos, cooldown, \ pos_size, tp1_mult, tp2_mult, sl_mult, tp1_close, trail_mult, \ vol_min, max_bars = params # Create inputs inputs = MoneyLineInputs( flip_threshold_percent=flip_thresh, ma_gap_threshold=ma_gap, momentum_min_adx=adx_min, momentum_long_max_pos=long_pos, momentum_short_min_pos=short_pos, cooldown_bars=cooldown, momentum_spacing=3, momentum_cooldown=2 ) # Create config config = TradeConfig( position_size=pos_size, atr_multiplier_tp1=tp1_mult, atr_multiplier_tp2=tp2_mult, atr_multiplier_sl=sl_mult, take_profit_1_size_percent=tp1_close, trailing_atr_multiplier=trail_mult, max_bars_per_trade=max_bars ) # Quality filter (optional based on vol_min) if vol_min > 0: quality_filter = lambda s: s.adx >= adx_min and s.volume_ratio >= vol_min else: quality_filter = None # Run simulation results = simulate_money_line( data_slice.data, data_slice.symbol, inputs, config, quality_filter=quality_filter ) # Calculate per-$1000 profitability pnl_per_1k = (results.total_pnl / pos_size) * 1000.0 return ( config_id, len(results.trades), results.win_rate * 100, results.total_pnl, pnl_per_1k, params ) except Exception as e: return (config_id, 0, 0.0, 0.0, 0.0, params) def main(): print("=" * 80) print("COMPREHENSIVE V9 PARAMETER SWEEP") print("=" * 80) print(f"Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print() # Load data print("Loading data...") data_path = Path(__file__).parent.parent / 'data' / 'solusdt_5m_aug_nov.csv' data_slice = load_csv(data_path, 'SOL-PERP', '5m') print(f"Loaded {len(data_slice.data)} candles") print() # Define parameter ranges to test print("Setting up parameter grid...") # CRITICAL PARAMETERS (most impact on profit) flip_thresholds = [0.4, 0.5, 0.6, 0.7, 0.8] # 5 values ma_gaps = [0.2, 0.35, 0.5] # 3 values adx_mins = [15, 18, 21, 24] # 4 values long_pos_maxs = [75, 80, 85, 90] # 4 values short_pos_mins = [10, 15, 20, 25] # 4 values cooldowns = [0, 1, 2, 3] # 4 values # TRADE CONFIG PARAMETERS position_sizes = [210] # Keep bot's size for consistency tp1_multipliers = [1.5, 2.0, 2.5] # 3 values tp2_multipliers = [3.0, 4.0, 5.0] # 3 values sl_multipliers = [2.5, 3.0, 3.5] # 3 values tp1_close_percents = [50, 60, 70, 80] # 4 values trailing_multipliers = [1.5, 2.0, 2.5, 3.0] # 4 values # QUALITY FILTERS vol_mins = [0, 0.7, 0.9] # 3 values (0 = no filter) max_bars_list = [None, 144, 288] # 3 values # Calculate total combinations total_combos = ( len(flip_thresholds) * len(ma_gaps) * len(adx_mins) * len(long_pos_maxs) * len(short_pos_mins) * len(cooldowns) * len(position_sizes) * len(tp1_multipliers) * len(tp2_multipliers) * len(sl_multipliers) * len(tp1_close_percents) * len(trailing_multipliers) * len(vol_mins) * len(max_bars_list) ) print(f"Total combinations: {total_combos:,}") print() print("Parameter ranges:") print(f" Flip Threshold: {flip_thresholds}") print(f" MA Gap: {ma_gaps}") print(f" ADX Min: {adx_mins}") print(f" Long Pos Max: {long_pos_maxs}") print(f" Short Pos Min: {short_pos_mins}") print(f" Cooldown: {cooldowns}") print(f" TP1 Multiplier: {tp1_multipliers}") print(f" TP2 Multiplier: {tp2_multipliers}") print(f" SL Multiplier: {sl_multipliers}") print(f" TP1 Close %: {tp1_close_percents}") print(f" Trailing Multiplier: {trailing_multipliers}") print(f" Vol Min: {vol_mins}") print(f" Max Bars: {max_bars_list}") print() # Generate all parameter combinations print("Generating parameter combinations...") param_combos = list(itertools.product( flip_thresholds, ma_gaps, adx_mins, long_pos_maxs, short_pos_mins, cooldowns, position_sizes, tp1_multipliers, tp2_multipliers, sl_multipliers, tp1_close_percents, trailing_multipliers, vol_mins, max_bars_list )) print(f"Generated {len(param_combos):,} combinations") print() # Prepare arguments for multiprocessing args_list = [(i, params, data_slice) for i, params in enumerate(param_combos)] # Run multiprocessing sweep num_workers = mp.cpu_count() print(f"Starting sweep with {num_workers} workers...") print(f"Progress logged to: sweep_progress.log") print() results = [] with mp.Pool(processes=num_workers) as pool: completed = 0 for result in pool.imap_unordered(test_config, args_list, chunksize=10): completed += 1 results.append(result) # Log progress every 100 configs if completed % 100 == 0: pct = (completed / len(param_combos)) * 100 print(f"Progress: {completed:,}/{len(param_combos):,} ({pct:.1f}%)") # Show best so far if results: best = max(results, key=lambda x: x[4]) print(f" Best so far: ${best[4]:.2f}/1k (trades={best[1]}, WR={best[2]:.1f}%)") print() print() print("=" * 80) print("SWEEP COMPLETE!") print("=" * 80) print() # Sort by profitability per $1000 results.sort(key=lambda x: x[4], reverse=True) # Show top 20 results print("TOP 20 CONFIGURATIONS:") print("-" * 80) print(f"{'Rank':<6} {'Trades':<8} {'WR%':<8} {'P&L':<12} {'$/1k':<12} Parameters") print("-" * 80) for i, (config_id, trades, wr, pnl, pnl_per_1k, params) in enumerate(results[:20], 1): flip_t, ma_g, adx, long_p, short_p, cool, pos_sz, tp1_m, tp2_m, sl_m, tp1_c, trail_m, vol_m, max_b = params param_str = ( f"flip={flip_t:.1f}, ma={ma_g:.2f}, adx={adx:.0f}, " f"pos={long_p:.0f}/{short_p:.0f}, cool={cool}, " f"tp={tp1_m:.1f}/{tp2_m:.1f}, sl={sl_m:.1f}, " f"tp1%={tp1_c:.0f}, trail={trail_m:.1f}, " f"vol={vol_m:.1f}, bars={max_b}" ) print(f"{i:<6} {trades:<8} {wr:<8.2f} ${pnl:<11.2f} ${pnl_per_1k:<11.2f} {param_str}") print() print("=" * 80) print(f"Finished: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("=" * 80) # Save results to CSV output_file = Path(__file__).parent.parent / 'sweep_comprehensive.csv' with open(output_file, 'w') as f: f.write("rank,trades,win_rate,total_pnl,pnl_per_1k,") f.write("flip_threshold,ma_gap,adx_min,long_pos_max,short_pos_min,cooldown,") f.write("position_size,tp1_mult,tp2_mult,sl_mult,tp1_close_pct,trailing_mult,vol_min,max_bars\n") for i, (config_id, trades, wr, pnl, pnl_per_1k, params) in enumerate(results, 1): f.write(f"{i},{trades},{wr:.2f},{pnl:.2f},{pnl_per_1k:.2f},") f.write(",".join(str(p) for p in params)) f.write("\n") print(f"Full results saved to: {output_file}") if __name__ == '__main__': main()