""" GPU/CPU Discrepancy Debugging Tool This script runs a single backtest on a fixed set of parameters to help diagnose discrepancies between the CPU and GPU execution paths. """ import os import sys # --- WSL GPU Fix --- # This MUST be at the top of the script, before any other imports, # especially before numba is imported (indirectly via strategy_module). if os.path.exists("/usr/lib/wsl/lib"): wsl_lib_path = "/usr/lib/wsl/lib" # 1. Set the LD_LIBRARY_PATH current_ld_path = os.environ.get("LD_LIBRARY_PATH", "") if wsl_lib_path not in current_ld_path: print(f" >> [WSL CONFIG] Injecting {wsl_lib_path} into LD_LIBRARY_PATH for GPU access.") os.environ["LD_LIBRARY_PATH"] = f"{current_ld_path}:{wsl_lib_path}" if current_ld_path else wsl_lib_path # 2. Set the Numba CUDA driver path numba_driver_path = f"{wsl_lib_path}/libcuda.so.1" if os.environ.get("NUMBA_CUDA_DRIVER") != numba_driver_path: print(f" >> [WSL CONFIG] Setting NUMBA_CUDA_DRIVER to {numba_driver_path}") os.environ["NUMBA_CUDA_DRIVER"] = numba_driver_path import json import pandas as pd import numpy as np # Add project root to path to allow imports from 'strategies' sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) import strategies.strategy_activation_scores as strategy_module # --- Configuration --- # Prefer the TradingView export as it contains pre-calculated indicators DATA_FILE = "results/TV_Export.csv" RESULTS_DIR = "results" # --- Fixed Parameters for Debugging --- # These parameters are taken from a previous sweep run to have a realistic test case. FIXED_PARAMS = { "i_long_entry_activation_threshold": 126.0, "i_long_exit_activation_threshold": 80.6, "i_long_exit_activation_confirmation_threshold": 32.475, "i_w_stoch": 6.85, "i_w_macd_pred": 36.5, "i_w_osc": -7.25, "i_w_macd_bullish": 0.553, "i_w_m3_momentum": 0.692, "i_w_m2_tiny": 23.5, "i_w_rsid_osc": 2.79, "i_w_stoch_div_osc": -3.35, "i_w_vwap_div_osc": 12.0, "i_w_stoch_peaking": -7.7, "i_w_stoch_bottoming": 162.5, "i_w_m3_div_osc": -3.71, "i_w_bearish_engulfing": 36.0, "i_use_long_entry_confirmation": 87.5, "i_m3_momentum_period": -152, "i_w_m2_div_osc": 25.0, "i_w_m2_div_osc_nooffset": 0.0, # Added missing parameter "i_cs_body_quality_ratio": 0.97, "i_cs_confidence_scaling_factor": 12.0 } def get_cpu_timeseries(df, params): """Runs the CPU backtest and extracts detailed time series.""" print("--- Running CPU Backtest ---") df_res = strategy_module.generate_signals(df.copy(), **params) # --- Re-calculate signals and equity curve to get the series --- score_series = df_res['activation_score'] # Replicate signal logic from generate_signals for comparison long_entry_threshold = params.get('i_long_entry_activation_threshold', 114.0) long_exit_threshold = params.get('i_long_exit_activation_threshold', 92.5) long_exit_conf_threshold = params.get('i_long_exit_activation_confirmation_threshold', 32.51) use_long_entry_conf = params.get('i_use_long_entry_confirmation', -54.7) use_long_exit_conf = params.get('i_use_long_exit_confirmation', 58.0) stoch_is_peaking = df_res['stoch_is_peaking'].astype(bool) if use_long_entry_conf > 0.0: entry_signal = (score_series.shift(1) < long_entry_threshold) & (score_series > score_series.shift(1)) else: entry_signal = (score_series < long_entry_threshold) & (score_series.shift(1) >= long_entry_threshold) if use_long_exit_conf > 0.0: exit_cond1 = score_series.shift(1) > long_exit_threshold exit_cond2 = score_series < long_exit_threshold exit_cond3 = score_series < long_exit_conf_threshold exit_signal = exit_cond1 & exit_cond2 & exit_cond3 else: exit_cond1 = score_series.shift(1) > long_exit_threshold exit_cond2 = score_series < long_exit_threshold exit_cond3 = stoch_is_peaking exit_signal = exit_cond1 & exit_cond2 & exit_cond3 # Re-calculate equity curve close = df_res['close'].values in_trade = df_res['in_trade'].shift(1).fillna(0).values asset_ret = np.zeros_like(close) asset_ret[1:] = np.divide( (close[1:] - close[:-1]), close[:-1], out=np.zeros_like(close[1:]), where=close[:-1] != 0 ) strat_ret = asset_ret * in_trade equity_curve = np.cumprod(1 + strat_ret) cpu_data = pd.DataFrame({ 'score': df_res['activation_score'], 'equity': equity_curve, 'daily_ret': strat_ret, 'in_position': df_res['in_trade'], 'entry_signal': entry_signal.fillna(False), 'exit_signal': exit_signal.fillna(False), 'exit_cond1': exit_cond1.fillna(False), 'exit_cond2': exit_cond2.fillna(False), 'exit_cond3': exit_cond3.fillna(False), 'threshold_val': long_exit_conf_threshold, }) print("CPU Metrics:", strategy_module.calculate_metrics(df_res)) return cpu_data def get_gpu_timeseries(df, params): """Runs the GPU debug kernel and returns detailed time series.""" print("\n--- Running GPU Debug Backtest ---") if not strategy_module.HAS_GPU: print("GPU not available (numba.cuda not found). Skipping.") return None # 1. Prepare data for GPU gpu_data_prepared = strategy_module.prepare_gpu_data(df.copy()) # 2. Extract parameters into numpy arrays in the correct order weight_keys = [ 'i_w_stoch', 'i_w_macd_pred', 'i_w_osc', 'i_w_macd_bullish', 'i_w_m3_momentum', 'i_w_m2_tiny', 'i_w_rsid_osc', 'i_w_stoch_div_osc', 'i_w_vwap_div_osc', 'i_w_stoch_peaking', 'i_w_stoch_bottoming', 'i_w_m3_div_osc', 'i_w_m2_div_osc', 'i_w_m2_div_osc_nooffset', 'i_w_bearish_engulfing' ] weights = np.array([params.get(k, 0.0) for k in weight_keys], dtype=np.float64) thresholds = np.array([ params['i_long_entry_activation_threshold'], params['i_long_exit_activation_threshold'], params['i_long_exit_activation_confirmation_threshold'] ], dtype=np.float64) configs = np.array([ params['i_use_long_entry_confirmation'], params.get('i_use_long_exit_confirmation', 58.0) ], dtype=np.float64) # 3. Execute the debug kernel gpu_debug_output = strategy_module.execute_gpu_debug_batch( gpu_data_prepared, weights, thresholds, configs ) # 4. Format output into a DataFrame gpu_data = pd.DataFrame({ 'score': gpu_debug_output['scores'], 'equity': gpu_debug_output['equity'], 'daily_ret': gpu_debug_output['daily_returns'], 'in_position': gpu_debug_output['in_position'], }) # 5. Print final metrics from GPU run for verification final_metrics = gpu_debug_output['final_metrics'] metrics_dict = { "Total P&L %": final_metrics[0], "Max Drawdown %": final_metrics[1], "Sharpe Ratio": final_metrics[2], "Sortino Ratio": final_metrics[3], "Total Trades": final_metrics[4] } print("GPU Metrics:", metrics_dict) return gpu_data def compare_results(df_cpu, df_gpu): """Compares the CPU and GPU DataFrames and reports the first divergence.""" print("\n--- Comparing CPU and GPU Time Series ---") if df_gpu is None: print("GPU results not available. Cannot compare.") return # Check for equal length if len(df_cpu) != len(df_gpu): print(f"ERROR: DataFrames have different lengths! CPU: {len(df_cpu)}, GPU: {len(df_gpu)}") return # Compare column by column for col in df_cpu.columns: print(f"Comparing column: {col}...") # Skip columns that don't exist in the GPU output yet if col not in df_gpu.columns: print(f" -- SKIPPED (column not in GPU output)") continue cpu_series = df_cpu[col] gpu_series = df_gpu[col] diverged = False # Ensure consistent types before comparison if cpu_series.dtype == bool: # Use array_equal for boolean comparison if not np.array_equal(cpu_series.astype(bool), gpu_series.astype(bool)): diverged = True else: # Assume float/numeric # Safely cast to float64 for isclose comparison cpu_series_float = cpu_series.astype(np.float64) gpu_series_float = gpu_series.astype(np.float64) if not np.all(np.isclose(cpu_series_float, gpu_series_float, atol=1e-9, rtol=1e-9, equal_nan=True)): diverged = True if diverged: print(f"\n!!!!!!!!!! DIVERGENCE DETECTED !!!!!!!!!!!") print(f"Column '{col}' diverged.") # Set pandas display options for full output pd.set_option('display.float_format', '{:.15f}'.format) pd.set_option('display.max_columns', None) # Find the first index where they diverge if cpu_series.dtype == bool: diff_mask = cpu_series.astype(bool) != gpu_series.astype(bool) else: diff_mask = ~np.isclose(cpu_series.astype(np.float64), gpu_series.astype(np.float64), atol=1e-9, rtol=1e-9, equal_nan=True) first_diff_idx = np.argmax(diff_mask) print(f"First divergence at index: {first_diff_idx}") comparison_df = pd.DataFrame({ 'CPU': cpu_series, 'GPU': gpu_series }) print("\n--- Values at point of divergence (and surroundings) ---") start_index = max(0, first_diff_idx - 2) print(comparison_df.iloc[start_index : first_diff_idx+3]) print("\n--- Full Context at point of divergence ---") full_context_df = pd.concat([df_cpu.add_prefix('cpu_'), df_gpu.add_prefix('gpu_')], axis=1) print(full_context_df.iloc[start_index : first_diff_idx+3]) print("\nAborting after first divergence.") return print("\nSUCCESS: All time series are identical between CPU and GPU.") def main(): """ Main function to load data and run the debug process. """ print("--- GPU/CPU Discrepancy Debugger ---") print(f"Loading data from: {DATA_FILE}") if not os.path.exists(DATA_FILE): print(f"ERROR: Data file not found at {DATA_FILE}. Please ensure the TradingView export is available.") sys.exit(1) # Load Data df_data = pd.read_csv(DATA_FILE) if 'time' in df_data.columns: df_data['time'] = pd.to_datetime(df_data['time']) # Apply the same date range filter as the main script mask = (df_data['time'] >= '2018-01-01') & (df_data['time'] <= '2024-07-01') df_data = df_data.loc[mask].copy().reset_index(drop=True) # Step 3: Get CPU time series cpu_timeseries = get_cpu_timeseries(df_data, FIXED_PARAMS) # Step 2 (execution): Get GPU time series gpu_timeseries = get_gpu_timeseries(df_data, FIXED_PARAMS) # Step 4: Compare results compare_results(cpu_timeseries, gpu_timeseries) if __name__ == "__main__": main()