Upload backtest_engine (19).py

backtest_engine (19).py

@@ -0,0 +1,618 @@
+# ============================================================
+# 🧪 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())