Delete backtest_engine.py

backtest_engine.py

@@ -1,618 +0,0 @@
-# ============================================================
-# 🧪 backtest_engine.py (V159.0 - GEM-Architect: Hyper-Speed Jump Logic)
-# ============================================================
-
-import asyncio
-import pandas as pd
-import numpy as np
-import time
-import logging
-import itertools 
-import os
-import glob
-import gc 
-import sys
-import traceback
-from datetime import datetime, timezone
-from typing import Dict, Any, List
-
-# محاولة استيراد المكتبات
-try:
-    import pandas_ta as ta
-except ImportError:
-    ta = None
-
-try:
-    from ml_engine.processor import MLProcessor 
-    from ml_engine.data_manager import DataManager
-    from learning_hub.adaptive_hub import AdaptiveHub
-    from r2 import R2Service
-    import xgboost as xgb 
-except ImportError:
-    pass
-
-logging.getLogger('ml_engine').setLevel(logging.WARNING)
-CACHE_DIR = "backtest_real_scores"
-
-# ============================================================
-# ⚡ VECTORIZED HELPERS
-# ============================================================
-def _z_roll_np(arr, w=500):
-    if len(arr) < w: return np.zeros_like(arr)
-    mean = pd.Series(arr).rolling(w).mean().fillna(0).values
-    std = pd.Series(arr).rolling(w).std().fillna(1).values
-    return np.nan_to_num((arr - mean) / (std + 1e-9))
-
-def _revive_score_distribution(scores):
-    scores = np.array(scores, dtype=np.float32).flatten()
-    s_min, s_max = np.min(scores), np.max(scores)
-    if (s_max - s_min) < 1e-6: return scores
-    if s_max < 0.8 or s_min > 0.2:
-        return (scores - s_min) / (s_max - s_min)
-    return scores
-
-# ============================================================
-# 🧪 THE BACKTESTER CLASS
-# ============================================================
-class HeavyDutyBacktester:
-    def __init__(self, data_manager, processor):
-        self.dm = data_manager
-        self.proc = processor
-        
-        # 🎛️ الكثافة (Density): عدد الخطوات في النطاق
-        self.GRID_DENSITY = 3 # 3 is enough for quick checks, 5 for deep dive
-        
-        self.INITIAL_CAPITAL = 10.0
-        self.TRADING_FEES = 0.001 
-        self.MAX_SLOTS = 4
-        
-        # 🎛️ CONTROL PANEL - DYNAMIC RANGES
-        self.GRID_RANGES = {
-            'TITAN':  np.linspace(0.10, 0.50, self.GRID_DENSITY),
-            'ORACLE': np.linspace(0.40, 0.80, self.GRID_DENSITY),
-            'SNIPER': np.linspace(0.30, 0.70, self.GRID_DENSITY),
-            'PATTERN': np.linspace(0.10, 0.50, self.GRID_DENSITY),
-            'L1_SCORE': [10.0],
-            # Guardians
-            'HYDRA_CRASH':    np.linspace(0.60, 0.85, self.GRID_DENSITY), 
-            'HYDRA_GIVEBACK': np.linspace(0.60, 0.85, self.GRID_DENSITY),       
-            'LEGACY_V2':      np.linspace(0.85, 0.98, self.GRID_DENSITY),            
-        }
-        
-        self.TARGET_COINS = [
-    'SOL/USDT', 'XRP/USDT', 'DOGE/USDT', 'ADA/USDT', 'AVAX/USDT', 'LINK/USDT',
-    'TON/USDT', 'INJ/USDT', 'APT/USDT', 'OP/USDT', 'ARB/USDT', 'SUI/USDT',
-    'SEI/USDT', 'MINA/USDT', 'MATIC/USDT', 'NEAR/USDT', 'RUNE/USDT', 'API3/USDT',
-    'FLOKI/USDT', 'BABYDOGE/USDT', 'SHIB/USDT', 'TRX/USDT', 'DOT/USDT', 'UNI/USDT',
-    'ONDO/USDT', 'SNX/USDT', 'HBAR/USDT', 'XLM/USDT', 'AGIX/USDT', 'IMX/USDT',
-    'LRC/USDT', 'KCS/USDT', 'ICP/USDT', 'SAND/USDT', 'AXS/USDT', 'APE/USDT',
-    'GMT/USDT', 'CHZ/USDT', 'CFX/USDT', 'LDO/USDT', 'FET/USDT', 'RPL/USDT',
-    'MNT/USDT', 'RAY/USDT', 'CAKE/USDT', 'SRM/USDT', 'PENDLE/USDT', 'ATOM/USDT'
-]
-        self.force_start_date = None 
-        self.force_end_date = None   
-        
-        if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V159.0] Hyper-Speed Jump Engine (CPU Optimized).")
-
-    def set_date_range(self, start_str, end_str):
-        self.force_start_date = start_str
-        self.force_end_date = end_str
-
-    async def _fetch_all_data_fast(self, sym, start_ms, end_ms):
-        print(f"   ⚡ [Network] Downloading {sym}...", flush=True)
-        limit = 1000
-        tasks = []
-        curr = start_ms
-        while curr < end_ms:
-            tasks.append(curr)
-            curr += limit * 60 * 1000
-        
-        all_candles = []
-        sem = asyncio.Semaphore(20) 
-
-        async def _fetch_batch(timestamp):
-            async with sem:
-                for _ in range(3): 
-                    try: return await self.dm.exchange.fetch_ohlcv(sym, '1m', since=timestamp, limit=limit)
-                    except: await asyncio.sleep(0.5)
-                return []
-
-        chunk_size = 50
-        for i in range(0, len(tasks), chunk_size):
-            res = await asyncio.gather(*[_fetch_batch(t) for t in tasks[i:i+chunk_size]])
-            for r in res: 
-                if r: all_candles.extend(r)
-
-        if not all_candles: return None
-        df = pd.DataFrame(all_candles, columns=['timestamp', 'o', 'h', 'l', 'c', 'v'])
-        df.drop_duplicates('timestamp', inplace=True)
-        df = df[(df['timestamp'] >= start_ms) & (df['timestamp'] <= end_ms)].sort_values('timestamp')
-        print(f"     ✅ Downloaded {len(df)} candles.", flush=True)
-        return df.values.tolist()
-
-    # ----------------------------------------------------------------------
-    # 🏎️ VECTORIZED INDICATORS
-    # ----------------------------------------------------------------------
-    def _calculate_indicators_vectorized(self, df, timeframe='1m'):
-        if df.empty: return df
-        cols = ['close', 'high', 'low', 'volume', 'open']
-        for c in cols: df[c] = df[c].astype(np.float64)
-        
-        # EMAs
-        df['ema9'] = df['close'].ewm(span=9, adjust=False).mean()
-        df['ema20'] = df['close'].ewm(span=20, adjust=False).mean()
-        df['ema21'] = df['close'].ewm(span=21, adjust=False).mean()
-        df['ema50'] = df['close'].ewm(span=50, adjust=False).mean()
-        df['ema200'] = df['close'].ewm(span=200, adjust=False).mean()
-        
-        if ta:
-            df['RSI'] = ta.rsi(df['close'], length=14).fillna(50)
-            df['ATR'] = ta.atr(df['high'], df['low'], df['close'], length=14).fillna(0)
-            bb = ta.bbands(df['close'], length=20, std=2.0)
-            df['bb_width'] = bb.iloc[:, 3].fillna(0) if bb is not None else 0.0
-            macd = ta.macd(df['close'])
-            if macd is not None:
-                df['MACD'] = macd.iloc[:, 0].fillna(0)
-                df['MACD_h'] = macd.iloc[:, 1].fillna(0)
-            else: df['MACD'] = 0; df['MACD_h'] = 0
-            df['ADX'] = ta.adx(df['high'], df['low'], df['close'], length=14).iloc[:, 0].fillna(0)
-            df['CCI'] = ta.cci(df['high'], df['low'], df['close'], length=20).fillna(0)
-            df['MFI'] = ta.mfi(df['high'], df['low'], df['close'], df['volume'], length=14).fillna(50)
-            df['slope'] = ta.slope(df['close'], length=7).fillna(0)
-            vwap = ta.vwap(df['high'], df['low'], df['close'], df['volume'])
-            df['vwap'] = vwap.fillna(df['close']) if vwap is not None else df['close']
-
-        c = df['close'].values
-        df['EMA_9_dist'] = (c / df['ema9'].values) - 1
-        df['EMA_21_dist'] = (c / df['ema21'].values) - 1
-        df['EMA_50_dist'] = (c / df['ema50'].values) - 1
-        df['EMA_200_dist'] = (c / df['ema200'].values) - 1
-        df['VWAP_dist'] = (c / df['vwap'].values) - 1
-        df['ATR_pct'] = df['ATR'] / (c + 1e-9)
-        
-        if timeframe == '1d': df['Trend_Strong'] = np.where(df['ADX'] > 25, 1.0, 0.0)
-        
-        df['vol_z'] = _z_roll_np(df['volume'].values, 20)
-        df['rel_vol'] = df['volume'] / (df['volume'].rolling(50).mean() + 1e-9)
-        df['log_ret'] = np.concatenate([[0], np.diff(np.log(c + 1e-9))])
-        
-        roll_min = df['low'].rolling(50).min(); roll_max = df['high'].rolling(50).max()
-        df['fib_pos'] = (c - roll_min) / (roll_max - roll_min + 1e-9)
-        df['volatility'] = df['ATR_pct']
-        
-        e20 = df['ema20'].values
-        e20_s = np.roll(e20, 5); e20_s[:5] = e20[0]
-        df['trend_slope'] = (e20 - e20_s) / (e20_s + 1e-9)
-        
-        fib618 = roll_max - ((roll_max - roll_min) * 0.382)
-        df['dist_fib618'] = (c - fib618) / (c + 1e-9)
-        df['dist_ema50'] = df['EMA_50_dist']
-        df['dist_ema200'] = df['EMA_200_dist']
-
-        if timeframe == '1m':
-            df['return_1m'] = df['log_ret']
-            df['rsi_14'] = df['RSI']
-            e9 = df['ema9'].values; e9_s = np.roll(e9, 1); e9_s[0] = e9[0]
-            df['ema_9_slope'] = (e9 - e9_s) / (e9_s + 1e-9)
-            df['ema_21_dist'] = df['EMA_21_dist']
-            
-            df['atr_z'] = _z_roll_np(df['ATR'].values, 100)
-            df['vol_zscore_50'] = _z_roll_np(df['volume'].values, 50)
-            rng = df['high'].values - df['low'].values
-            df['candle_range'] = _z_roll_np(rng, 500)
-            df['close_pos_in_range'] = (c - df['low'].values) / (rng + 1e-9)
-            
-            dollar_vol = c * df['volume'].values
-            amihud = np.abs(df['log_ret']) / (dollar_vol + 1e-9)
-            df['amihud'] = _z_roll_np(amihud, 500)
-            
-            sign = np.sign(np.diff(c, prepend=c[0]))
-            signed_vol = sign * df['volume'].values
-            ofi = pd.Series(signed_vol).rolling(30).sum().fillna(0).values
-            df['ofi'] = _z_roll_np(ofi, 500)
-            df['vwap_dev'] = _z_roll_np(c - df['vwap'].values, 500)
-            
-            for lag in [1, 2, 3, 5, 10, 20]:
-                df[f'log_ret_lag_{lag}'] = df['log_ret'].shift(lag).fillna(0)
-                df[f'rsi_lag_{lag}'] = df['RSI'].shift(lag).fillna(50)/100.0
-                df[f'fib_pos_lag_{lag}'] = df['fib_pos'].shift(lag).fillna(0.5)
-                df[f'volatility_lag_{lag}'] = df['volatility'].shift(lag).fillna(0)
-
-        df.fillna(0, inplace=True)
-        return df
-
-    async def _process_data_in_memory(self, sym, candles, start_ms, end_ms):
-        safe_sym = sym.replace('/', '_')
-        period_suffix = f"{start_ms}_{end_ms}"
-        scores_file = f"{CACHE_DIR}/{safe_sym}_{period_suffix}_scores.pkl"
-        if os.path.exists(scores_file):
-             print(f"   📂 [{sym}] Data Exists -> Skipping.")
-             return
-
-        print(f"   ⚙️ [CPU] Analyzing {sym}...", flush=True)
-        t0 = time.time()
-
-        df_1m = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-        df_1m['datetime'] = pd.to_datetime(df_1m['timestamp'], unit='ms')
-        df_1m.set_index('datetime', inplace=True)
-        df_1m = df_1m.sort_index()
-
-        frames = {}
-        frames['1m'] = self._calculate_indicators_vectorized(df_1m.copy(), timeframe='1m')
-        frames['1m']['timestamp'] = frames['1m'].index.floor('1min').astype(np.int64) // 10**6
-        fast_1m = {col: frames['1m'][col].values for col in frames['1m'].columns}
-        
-        agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
-        numpy_htf = {} 
-        for tf_str, tf_code in [('5m', '5T'), ('15m', '15T'), ('1h', '1h'), ('4h', '4h'), ('1d', '1D')]:
-            resampled = df_1m.resample(tf_code).agg(agg_dict).dropna()
-            if resampled.empty:
-                numpy_htf[tf_str] = {} 
-                continue
-            resampled = self._calculate_indicators_vectorized(resampled, timeframe=tf_str)
-            resampled['timestamp'] = resampled.index.astype(np.int64) // 10**6
-            frames[tf_str] = resampled
-            numpy_htf[tf_str] = {col: resampled[col].values for col in resampled.columns}
-
-        arr_ts_1m = fast_1m['timestamp']
-        def get_map(tf):
-            if tf not in numpy_htf or 'timestamp' not in numpy_htf[tf]: return np.zeros(len(arr_ts_1m), dtype=int)
-            return np.clip(np.searchsorted(numpy_htf[tf]['timestamp'], arr_ts_1m), 0, len(numpy_htf[tf]['timestamp']) - 1)
-        
-        map_5m = get_map('5m'); map_15m = get_map('15m'); map_1h = get_map('1h'); map_4h = get_map('4h')
-
-        titan_model = getattr(self.proc.titan, 'model', None)
-        oracle_dir = getattr(self.proc.oracle, 'model_direction', None)
-        oracle_cols = getattr(self.proc.oracle, 'feature_cols', [])
-        sniper_models = getattr(self.proc.sniper, 'models', [])
-        sniper_cols = getattr(self.proc.sniper, 'feature_names', [])
-        hydra_models = getattr(self.proc.guardian_hydra, 'models', {}) if self.proc.guardian_hydra else {}
-        legacy_v2 = getattr(self.proc.guardian_legacy, 'model_v2', None)
-        
-        # --- BATCH PREDICTIONS ---
-        global_titan_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
-        if titan_model:
-            titan_cols = [
-                '5m_open', '5m_high', '5m_low', '5m_close', '5m_volume', '5m_RSI', '5m_MACD', '5m_MACD_h', 
-                '5m_CCI', '5m_ADX', '5m_EMA_9_dist', '5m_EMA_21_dist', '5m_EMA_50_dist', '5m_EMA_200_dist', 
-                '5m_BB_w', '5m_BB_p', '5m_MFI', '5m_VWAP_dist', '15m_timestamp', '15m_RSI', '15m_MACD', 
-                '15m_MACD_h', '15m_CCI', '15m_ADX', '15m_EMA_9_dist', '15m_EMA_21_dist', '15m_EMA_50_dist', 
-                '15m_EMA_200_dist', '15m_BB_w', '15m_BB_p', '15m_MFI', '15m_VWAP_dist', '1h_timestamp', 
-                '1h_RSI', '1h_MACD_h', '1h_EMA_50_dist', '1h_EMA_200_dist', '1h_ATR_pct', '4h_timestamp', 
-                '4h_RSI', '4h_MACD_h', '4h_EMA_50_dist', '4h_EMA_200_dist', '4h_ATR_pct', '1d_timestamp', 
-                '1d_RSI', '1d_EMA_200_dist', '1d_Trend_Strong'
-            ]
-            try:
-                t_vecs = []
-                for col in titan_cols:
-                    parts = col.split('_', 1); tf = parts[0]; feat = parts[1]
-                    target_arr = numpy_htf.get(tf, {})
-                    target_map = locals().get(f"map_{tf}", np.zeros(len(arr_ts_1m), dtype=int))
-                    if feat in target_arr: t_vecs.append(target_arr[feat][target_map])
-                    elif feat == 'timestamp' and 'timestamp' in target_arr: t_vecs.append(target_arr['timestamp'][target_map])
-                    elif feat in ['open','high','low','close','volume'] and feat in target_arr: t_vecs.append(target_arr[feat][target_map])
-                    else: t_vecs.append(np.zeros(len(arr_ts_1m)))
-                X_TITAN = np.column_stack(t_vecs)
-                global_titan_scores = _revive_score_distribution(titan_model.predict(xgb.DMatrix(X_TITAN, feature_names=titan_cols)))
-            except: pass
-
-        global_oracle_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
-        if oracle_dir:
-            try:
-                o_vecs = []
-                for col in oracle_cols:
-                    if col.startswith('1h_'): o_vecs.append(numpy_htf['1h'].get(col[3:], np.zeros(len(arr_ts_1m)))[map_1h])
-                    elif col.startswith('15m_'): o_vecs.append(numpy_htf['15m'].get(col[4:], np.zeros(len(arr_ts_1m)))[map_15m])
-                    elif col.startswith('4h_'): o_vecs.append(numpy_htf['4h'].get(col[3:], np.zeros(len(arr_ts_1m)))[map_4h])
-                    elif col == 'sim_titan_score': o_vecs.append(global_titan_scores)
-                    elif col == 'sim_mc_score': o_vecs.append(np.full(len(arr_ts_1m), 0.5))
-                    elif col == 'sim_pattern_score': o_vecs.append(np.full(len(arr_ts_1m), 0.5))
-                    else: o_vecs.append(np.zeros(len(arr_ts_1m)))
-                X_ORACLE = np.column_stack(o_vecs)
-                preds_o = oracle_dir.predict(X_ORACLE)
-                preds_o = preds_o if isinstance(preds_o, np.ndarray) and len(preds_o.shape)==1 else preds_o[:, 0]
-                global_oracle_scores = _revive_score_distribution(preds_o)
-            except: pass
-
-        global_sniper_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
-        if sniper_models:
-            try:
-                s_vecs = []
-                for col in sniper_cols:
-                    if col in fast_1m: s_vecs.append(fast_1m[col])
-                    elif col == 'atr' and 'atr_z' in fast_1m: s_vecs.append(fast_1m['atr_z'])
-                    else: s_vecs.append(np.zeros(len(arr_ts_1m)))
-                X_SNIPER = np.column_stack(s_vecs)
-                preds = [m.predict(X_SNIPER) for m in sniper_models]
-                global_sniper_scores = _revive_score_distribution(np.mean(preds, axis=0))
-            except: pass
-
-        global_v2_scores = np.zeros(len(arr_ts_1m), dtype=np.float32)
-        if legacy_v2:
-            try:
-                l_log = fast_1m['log_ret']; l_rsi = fast_1m['RSI'] / 100.0; l_fib = fast_1m['fib_pos']; l_vol = fast_1m['volatility']
-                l5_log = numpy_htf['5m']['log_ret'][map_5m]; l5_rsi = numpy_htf['5m']['RSI'][map_5m] / 100.0; l5_fib = numpy_htf['5m']['fib_pos'][map_5m]; l5_trd = numpy_htf['5m']['trend_slope'][map_5m]
-                l15_log = numpy_htf['15m']['log_ret'][map_15m]; l15_rsi = numpy_htf['15m']['RSI'][map_15m] / 100.0; l15_fib618 = numpy_htf['15m']['dist_fib618'][map_15m]; l15_trd = numpy_htf['15m']['trend_slope'][map_15m]
-                lags = []
-                for lag in [1, 2, 3, 5, 10, 20]:
-                    lags.extend([fast_1m[f'log_ret_lag_{lag}'], fast_1m[f'rsi_lag_{lag}'], fast_1m[f'fib_pos_lag_{lag}'], fast_1m[f'volatility_lag_{lag}']])
-                X_V2 = np.column_stack([l_log, l_rsi, l_fib, l_vol, l5_log, l5_rsi, l5_fib, l5_trd, l15_log, l15_rsi, l15_fib618, l15_trd, *lags])
-                preds = legacy_v2.predict(xgb.DMatrix(X_V2))
-                global_v2_scores = preds[:, 2] if len(preds.shape) > 1 else preds
-                global_v2_scores = global_v2_scores.flatten() 
-            except: pass
-
-        global_hydra_crash = np.zeros(len(arr_ts_1m), dtype=np.float32)
-        global_hydra_give = np.zeros(len(arr_ts_1m), dtype=np.float32)
-        if hydra_models:
-            try:
-                zeros = np.zeros(len(arr_ts_1m))
-                h_static = np.column_stack([
-                    fast_1m['RSI'], numpy_htf['5m']['RSI'][map_5m], numpy_htf['15m']['RSI'][map_15m],
-                    fast_1m['bb_width'], fast_1m['rel_vol'], fast_1m['atr'], fast_1m['close']
-                ])
-                X_H = np.column_stack([
-                    h_static[:,0], h_static[:,1], h_static[:,2], h_static[:,3], h_static[:,4],
-                    zeros, fast_1m['ATR_pct'], zeros, zeros, zeros, zeros, zeros, zeros,
-                    global_oracle_scores, np.full(len(arr_ts_1m), 0.7), np.full(len(arr_ts_1m), 3.0)
-                ])
-                
-                probs_c = hydra_models['crash'].predict_proba(X_H)[:, 1]
-                global_hydra_crash = probs_c.astype(np.float32)
-                
-                probs_g = hydra_models['giveback'].predict_proba(X_H)[:, 1]
-                global_hydra_give = probs_g.astype(np.float32)
-            except: pass
-
-        # Filter
-        rsi_1h = numpy_htf['1h'].get('RSI', np.zeros(len(arr_ts_1m)))[map_1h]
-        # Keep candles where at least minimal promise exists (reduces size)
-        is_candidate_mask = (rsi_1h <= 70) & (global_titan_scores > 0.3) & (global_oracle_scores > 0.3)
-        candidate_indices = np.where(is_candidate_mask)[0]
-        end_limit = len(arr_ts_1m) - 60
-        candidate_indices = candidate_indices[candidate_indices < end_limit]
-        candidate_indices = candidate_indices[candidate_indices >= 500]
-
-        print(f"     🌪️ Final List: {len(candidate_indices)} candidates ready for testing.", flush=True)
-
-        ai_results = pd.DataFrame({
-            'timestamp': arr_ts_1m[candidate_indices],
-            'symbol': sym,
-            'close': fast_1m['close'][candidate_indices],
-            'real_titan': global_titan_scores[candidate_indices],
-            'oracle_conf': global_oracle_scores[candidate_indices],
-            'sniper_score': global_sniper_scores[candidate_indices],
-            'pattern_score': np.full(len(candidate_indices), 0.5), 
-            'risk_hydra_crash': global_hydra_crash[candidate_indices],
-            'risk_hydra_giveback': global_hydra_give[candidate_indices],
-            'risk_legacy_v2': global_v2_scores[candidate_indices],
-            'time_hydra_crash': np.zeros(len(candidate_indices), dtype=int),
-            'l1_score': 50.0
-        })
-
-        dt = time.time() - t0
-        if not ai_results.empty:
-            ai_results.to_pickle(scores_file)
-            print(f"   ✅ [{sym}] Completed in {dt:.2f} seconds. ({len(ai_results)} signals)", flush=True)
-        gc.collect()
-
-    async def generate_truth_data(self):
-        if self.force_start_date:
-            dt_s = datetime.strptime(self.force_start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
-            dt_e = datetime.strptime(self.force_end_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
-            ms_s = int(dt_s.timestamp()*1000); ms_e = int(dt_e.timestamp()*1000)
-            print(f"\n🚜 [Phase 1] Processing Era: {self.force_start_date} -> {self.force_end_date}")
-            for sym in self.TARGET_COINS:
-                c = await self._fetch_all_data_fast(sym, ms_s, ms_e)
-                if c: await self._process_data_in_memory(sym, c, ms_s, ms_e)
-
-    @staticmethod
-    def _worker_optimize(combinations_batch, scores_files, initial_capital, fees_pct, max_slots):
-        """🚀 HYPER-SPEED JUMP LOGIC (NO LOOPING OVER IDLE CANDLES)"""
-        print(f"     ⏳ [System] Loading {len(scores_files)} datasets...", flush=True)
-        data = []
-        for f in scores_files:
-            try: data.append(pd.read_pickle(f))
-            except: pass
-        if not data: return []
-        df = pd.concat(data).sort_values('timestamp').reset_index(drop=True)
-        
-        # Pre-load arrays for max speed
-        ts = df['timestamp'].values
-        close = df['close'].values.astype(float)
-        sym = df['symbol'].values
-        u_syms = np.unique(sym); sym_map = {s: i for i, s in enumerate(u_syms)}; sym_id = np.array([sym_map[s] for s in sym])
-        
-        oracle = df['oracle_conf'].values
-        sniper = df['sniper_score'].values
-        titan = df['real_titan'].values
-        pattern = df['pattern_score'].values
-        l1 = df['l1_score'].values
-        hydra = df['risk_hydra_crash'].values
-        hydra_give = df['risk_hydra_giveback'].values
-        legacy = df['risk_legacy_v2'].values
-        
-        N = len(ts)
-        print(f"     🚀 [System] Testing {len(combinations_batch)} configs on {N} candidates...", flush=True)
-        
-        res = []
-        for cfg in combinations_batch:
-            # 1. Vectorized Entry Mask (The Speed Secret)
-            # Instead of checking every candle, we calculate ALL valid entries at once
-            entry_mask = (l1 >= cfg['L1_SCORE']) & \
-                         (oracle >= cfg['ORACLE']) & \
-                         (sniper >= cfg['SNIPER']) & \
-                         (titan >= cfg['TITAN']) & \
-                         (pattern >= cfg.get('PATTERN', 0.10))
-            
-            # Get only the indices where entry is possible
-            valid_entry_indices = np.where(entry_mask)[0]
-            
-            # Extract thresholds locally to avoid dictionary lookups in inner loop
-            h_crash_thresh = cfg['HYDRA_CRASH']
-            h_give_thresh = cfg['HYDRA_GIVEBACK']
-            leg_thresh = cfg['LEGACY_V2']
-            
-            # Simulation State
-            pos = {} # sym_id -> (entry_price, size)
-            bal = float(initial_capital)
-            alloc = 0.0
-            log = []
-            
-            # Iterate ONLY on relevant indices (Jump!)
-            # But we must respect time. So we iterate valid indices, 
-            # and check exits for OPEN positions at that time step? 
-            # Problem: If we jump, we miss exits between entries.
-            # Fix: We must iterate all rows for exits, but we can skip logic if no pos.
-            # OR: Since df is filtered candidates only, gaps exist.
-            # We assume candidates are frequent enough or we only check exits on candidate candles.
-            # *Refinement*: The dataframe `df` only contains ~30k candidates out of 100k candles.
-            # Exiting only on candidate candles is an approximation, but acceptable for optimization speed.
-            
-            for i in range(N):
-                s = sym_id[i]; p = float(close[i])
-                
-                # A. Check Exits (If holding this symbol)
-                if s in pos:
-                    entry_p, size_val = pos[s]
-                    pnl = (p - entry_p) / entry_p
-                    
-                    # Guardian Logic (Local vars)
-                    is_guard = (hydra[i] > h_crash_thresh) or \
-                               (hydra_give[i] > h_give_thresh) or \
-                               (legacy[i] > leg_thresh)
-                    
-                    # VETO (Price Confirmation)
-                    confirmed = is_guard and (pnl < -0.0015)
-                    
-                    if confirmed or (pnl > 0.04) or (pnl < -0.02):
-                        realized = pnl - (fees_pct * 2)
-                        bal += size_val * (1.0 + realized)
-                        alloc -= size_val
-                        del pos[s]
-                        log.append({'pnl': realized})
-                        continue # Can't buy same candle we sold
-
-                # B. Check Entries (Only if mask is True)
-                if entry_mask[i] and len(pos) < max_slots:
-                    if s not in pos and bal >= 5.0:
-                        size = min(10.0, bal * 0.98)
-                        pos[s] = (p, size)
-                        bal -= size; alloc += size
-
-            # Calc Stats
-            final_bal = bal + alloc
-            profit = final_bal - initial_capital
-            tot = len(log)
-            winning = [x for x in log if x['pnl'] > 0]
-            losing = [x for x in log if x['pnl'] <= 0]
-            
-            win_rate = (len(winning)/tot*100) if tot > 0 else 0.0
-            avg_win = np.mean([x['pnl'] for x in winning]) if winning else 0.0
-            avg_loss = np.mean([x['pnl'] for x in losing]) if losing else 0.0
-            gross_p = sum([x['pnl'] for x in winning])
-            gross_l = abs(sum([x['pnl'] for x in losing]))
-            profit_factor = (gross_p / gross_l) if gross_l > 0 else 99.9
-            
-            # Simple streaks
-            max_win_s = 0; max_loss_s = 0; curr_w = 0; curr_l = 0
-            for t in log:
-                if t['pnl'] > 0: curr_w +=1; curr_l = 0; max_win_s = max(max_win_s, curr_w)
-                else: curr_l +=1; curr_w = 0; max_loss_s = max(max_loss_s, curr_l)
-
-            res.append({
-                'config': cfg, 'final_balance': final_bal, 'net_profit': profit, 
-                'total_trades': tot, 'win_rate': win_rate, 'profit_factor': profit_factor,
-                'win_count': len(winning), 'loss_count': len(losing),
-                'avg_win': avg_win, 'avg_loss': avg_loss,
-                'max_win_streak': max_win_s, 'max_loss_streak': max_loss_s,
-                'consensus_agreement_rate': 0.0, 'high_consensus_win_rate': 0.0
-            })
-        return res
-
-    async def run_optimization(self, target_regime="RANGE"):
-        await self.generate_truth_data()
-        
-        keys = list(self.GRID_RANGES.keys())
-        values = list(self.GRID_RANGES.values())
-        combos = [dict(zip(keys, c)) for c in itertools.product(*values)]
-            
-        files = glob.glob(os.path.join(CACHE_DIR, "*.pkl"))
-        results_list = self._worker_optimize(combos, files, self.INITIAL_CAPITAL, self.TRADING_FEES, self.MAX_SLOTS)
-        if not results_list: return None, {'net_profit': 0.0, 'win_rate': 0.0}
-
-        results_list.sort(key=lambda x: x['net_profit'], reverse=True)
-        best = results_list[0]
-        
-        mapped_config = {
-            'w_titan': best['config']['TITAN'],
-            'w_struct': best['config']['PATTERN'],
-            'thresh': best['config']['L1_SCORE'],
-            'oracle_thresh': best['config']['ORACLE'],
-            'sniper_thresh': best['config']['SNIPER'],
-            'hydra_thresh': best['config']['HYDRA_CRASH'],
-            'legacy_thresh': best['config']['LEGACY_V2']
-        }
-
-        # Diagnosis
-        diag = []
-        if best['total_trades'] > 2000 and best['net_profit'] < 10: diag.append("⚠️ Overtrading")
-        if best['win_rate'] > 55 and best['net_profit'] < 0: diag.append("⚠️ Fee Burn")
-        if abs(best['avg_loss']) > best['avg_win'] and best['win_count'] > 0: diag.append("⚠️ Risk/Reward Inversion")
-        if best['max_loss_streak'] > 10: diag.append("⚠️ Consecutive Loss Risk")
-        if not diag: diag.append("✅ System Healthy")
-
-        print("\n" + "="*60)
-        print(f"🏆 CHAMPION REPORT [{target_regime}]:")
-        print(f"   💰 Final Balance:   ${best['final_balance']:,.2f}")
-        print(f"   🚀 Net PnL:         ${best['net_profit']:,.2f}")
-        print("-" * 60)
-        print(f"   📊 Total Trades:    {best['total_trades']}")
-        print(f"   📈 Win Rate:        {best['win_rate']:.1f}%")
-        print(f"   ✅ Winning Trades:  {best['win_count']} (Avg: {best['avg_win']*100:.2f}%)")
-        print(f"   ❌ Losing Trades:   {best['loss_count']} (Avg: {best['avg_loss']*100:.2f}%)")
-        print(f"   🌊 Max Streaks:     Win {best['max_win_streak']} | Loss {best['max_loss_streak']}")
-        print(f"   ⚖️ Profit Factor:   {best['profit_factor']:.2f}")
-        print("-" * 60)
-        print(f"   🧠 CONSENSUS ANALYTICS:")
-        print(f"   🤝 Model Agreement Rate:     {best.get('consensus_agreement_rate', 0.0):.1f}%")
-        print(f"   🌟 High-Consensus Win Rate:  {best.get('high_consensus_win_rate', 0.0):.1f}%")
-        print("-" * 60)
-        print(f"   🩺 DIAGNOSIS: {' '.join(diag)}")
-        
-        p_str = ""
-        for k, v in mapped_config.items():
-            if isinstance(v, float): p_str += f"{k}={v:.2f} | "
-            else: p_str += f"{k}={v} | "
-        print(f"   ⚙️ Config: {p_str}")
-        print("="*60)
-        
-        return mapped_config, best
-
-async def run_strategic_optimization_task():
-    print("\n🧪 [STRATEGIC BACKTEST] Hyper-Speed Jump Engine...")
-    r2 = R2Service(); dm = DataManager(None, None, r2); proc = MLProcessor(dm)
-    try:
-        await dm.initialize(); await proc.initialize()
-        if proc.guardian_hydra: proc.guardian_hydra.set_silent_mode(True)
-        hub = AdaptiveHub(r2); await hub.initialize()
-        opt = HeavyDutyBacktester(dm, proc)
-        scenarios = [
-                {"regime": "DEAD", "start": "2023-06-01", "end": "2023-08-01"},
-                {"regime": "RANGE", "start": "2024-07-01", "end": "2024-09-30"},
-                {"regime": "BULL", "start": "2024-01-01", "end": "2024-03-30"},
-                {"regime": "BEAR", "start": "2023-08-01", "end": "2023-09-15"},
-                  ]
-        for s in scenarios:
-            opt.set_date_range(s["start"], s["end"])
-            best_cfg, best_stats = await opt.run_optimization(s["regime"])
-            if best_cfg: hub.submit_challenger(s["regime"], best_cfg, best_stats)
-        await hub._save_state_to_r2()
-        print("✅ [System] DNA Updated.")
-    finally:
-        print("🔌 [System] Closing connections...")
-        await dm.close()
-
-if __name__ == "__main__":
-    asyncio.run(run_strategic_optimization_task())