trading_bot_v4/cluster/v11_full_worker.py

#!/usr/bin/env python3
"""
V11 Full Sweep Worker - Processes parameter combinations in parallel

Runs on EPYC workers, processes assigned chunk of parameter space.
"""

import argparse
import itertools
import json
import multiprocessing as mp
import sys
import time
from pathlib import Path

# Add backtester to path
sys.path.insert(0, str(Path(__file__).parent.parent / 'backtester'))

from backtester.data_loader import load_csv
from backtester.simulator import run_backtest
from backtester.v11_moneyline_all_filters import MoneyLineV11Inputs

# Parameter grid (matches coordinator)
PARAM_GRID = {
    'flip_threshold': [0.25, 0.3, 0.35, 0.4, 0.45, 0.5],
    'adx_min': [0, 5, 10, 15, 20, 25],
    'long_pos_max': [90, 95, 100],
    'short_pos_min': [0, 5, 10],
    'vol_min': [0.0, 0.5, 1.0],
    'entry_buffer_atr': [0.0, 0.05, 0.10],
    'rsi_long_min': [20, 25, 30],
    'rsi_short_max': [70, 75, 80],
}


def test_single_config(args):
    """Test a single parameter configuration"""
    combo_idx, params_tuple, data = args
    
    # Unpack parameters
    keys = sorted(PARAM_GRID.keys())
    params = dict(zip(keys, params_tuple))
    
    # Create inputs
    inputs = MoneyLineV11Inputs(
        flip_threshold=params['flip_threshold'],
        adx_min=params['adx_min'],
        long_pos_max=params['long_pos_max'],
        short_pos_min=params['short_pos_min'],
        vol_min=params['vol_min'],
        entry_buffer_atr=params['entry_buffer_atr'],
        rsi_long_min=params['rsi_long_min'],
        rsi_short_max=params['rsi_short_max'],
    )
    
    try:
        # Run backtest
        results = run_backtest(data, inputs, 'v11')
        
        return {
            'params': params,
            'pnl': results['total_pnl'],
            'win_rate': results['win_rate'],
            'profit_factor': results['profit_factor'],
            'max_drawdown': results['max_drawdown'],
            'total_trades': results['total_trades'],
        }
    except Exception as e:
        print(f"Error testing combo {combo_idx}: {e}")
        return None


def main():
    parser = argparse.ArgumentParser(description='V11 Full Sweep Worker')
    parser.add_argument('--chunk-id', required=True, help='Chunk ID')
    parser.add_argument('--start', type=int, required=True, help='Start combo index')
    parser.add_argument('--end', type=int, required=True, help='End combo index')
    parser.add_argument('--workers', type=int, default=24, help='Parallel workers')
    args = parser.parse_args()
    
    print(f"V11 Full Worker: {args.chunk_id}")
    print(f"Range: {args.start} - {args.end}")
    print(f"Workers: {args.workers}")
    print()
    
    # Load data
    print("Loading market data...")
    data = load_csv(Path('data/solusdt_5m.csv'), 'SOLUSDT', '5m')
    print(f"Loaded {len(data)} candles")
    print()
    
    # Generate all combinations
    keys = sorted(PARAM_GRID.keys())
    values = [PARAM_GRID[k] for k in keys]
    all_combos = list(itertools.product(*values))
    
    # Get this chunk's combos
    chunk_combos = all_combos[args.start:args.end]
    print(f"Testing {len(chunk_combos)} combinations...")
    print()
    
    # Prepare args for parallel processing
    test_args = [
        (args.start + i, combo, data) 
        for i, combo in enumerate(chunk_combos)
    ]
    
    # Run tests in parallel
    start_time = time.time()
    results = []
    
    with mp.Pool(processes=args.workers) as pool:
        for i, result in enumerate(pool.imap_unordered(test_single_config, test_args)):
            if result:
                results.append(result)
            
            # Progress update every 10 combos
            if (i + 1) % 10 == 0:
                elapsed = time.time() - start_time
                rate = (i + 1) / elapsed
                remaining = len(chunk_combos) - (i + 1)
                eta = remaining / rate if rate > 0 else 0
                
                print(f"Progress: {i+1}/{len(chunk_combos)} "
                      f"({rate:.1f} combos/s, ETA: {eta/60:.1f}m)")
    
    # Sort by P&L descending
    results.sort(key=lambda x: x['pnl'], reverse=True)
    
    # Save results to CSV
    output_path = Path(f"v11_results/{args.chunk_id}_results.csv")
    output_path.parent.mkdir(exist_ok=True)
    
    with open(output_path, 'w') as f:
        # Header
        param_cols = sorted(PARAM_GRID.keys())
        metric_cols = ['pnl', 'win_rate', 'profit_factor', 'max_drawdown', 'total_trades']
        header = ','.join(param_cols + metric_cols)
        f.write(header + '\n')
        
        # Data rows
        for result in results:
            param_values = [str(result['params'][k]) for k in param_cols]
            metric_values = [
                f"{result['pnl']:.1f}",
                f"{result['win_rate']:.1f}",
                f"{result['profit_factor']:.3f}",
                f"{result['max_drawdown']:.1f}",
                str(result['total_trades'])
            ]
            row = ','.join(param_values + metric_values)
            f.write(row + '\n')
    
    duration = time.time() - start_time
    print()
    print("=" * 60)
    print(f"Chunk {args.chunk_id} COMPLETE")
    print(f"Duration: {duration/60:.1f} minutes")
    print(f"Rate: {len(chunk_combos)/duration:.2f} combos/second")
    print(f"Results saved: {output_path}")
    
    if results:
        best = results[0]
        print()
        print("Best configuration in chunk:")
        print(f"  P&L: ${best['pnl']:.2f}")
        print(f"  Win Rate: {best['win_rate']:.1f}%")
        print(f"  Profit Factor: {best['profit_factor']:.3f}")
        print(f"  Parameters: {best['params']}")
    
    print("=" * 60)


if __name__ == '__main__':
    main()