Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V114.0 - GEM-Architect: Global Inference Velocity)
+# 🧪 backtest_engine.py (V113.0 - GEM-Architect: Global Pre-Inference)
 # ============================================================
 
 import asyncio
@@ -55,7 +55,7 @@ class HeavyDutyBacktester:
         self.force_end_date = None   
         
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V114.0] Global Inference Architecture (Max Speed).")
+        print(f"🧪 [Backtest V113.0] Pre-Inference Velocity Mode (Target: 60s).")
 
     def set_date_range(self, start_str, end_str):
         self.force_start_date = start_str
@@ -67,25 +67,24 @@ class HeavyDutyBacktester:
     async def _fetch_all_data_fast(self, sym, start_ms, end_ms):
         print(f"   ⚡ [Network] Downloading {sym}...", flush=True)
         limit = 1000
-        # Reduce delay slightly for speed
+        duration_per_batch = limit * 60 * 1000
         tasks = []
         current = start_ms
-        duration_per_batch = limit * 60 * 1000
         while current < end_ms:
             tasks.append(current)
             current += duration_per_batch
         all_candles = []
-        sem = asyncio.Semaphore(15) # Increased concurrency
+        sem = asyncio.Semaphore(10) 
 
         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)
+                    except: await asyncio.sleep(1)
                 return []
 
-        chunk_size = 25
+        chunk_size = 20
         for i in range(0, len(tasks), chunk_size):
             chunk_tasks = tasks[i:i + chunk_size]
             futures = [_fetch_batch(ts) for ts in chunk_tasks]
@@ -94,30 +93,31 @@ class HeavyDutyBacktester:
                 if res: all_candles.extend(res)
 
         if not all_candles: return None
-        # Fast Dedupe
-        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)]
-        df.sort_values('timestamp', inplace=True)
-        
-        print(f"     ✅ Downloaded {len(df)} candles.", flush=True)
-        return df.values.tolist()
+        filtered = [c for c in all_candles if c[0] >= start_ms and c[0] <= end_ms]
+        seen = set(); unique_candles = []
+        for c in filtered:
+            if c[0] not in seen:
+                unique_candles.append(c)
+                seen.add(c[0])
+        unique_candles.sort(key=lambda x: x[0])
+        print(f"     ✅ Downloaded {len(unique_candles)} candles.", flush=True)
+        return unique_candles
 
     # ==============================================================
-    # 🏎️ VECTORIZED INDICATORS (ALL LAYERS)
+    # 🏎️ VECTORIZED INDICATORS
     # ==============================================================
     def _calculate_indicators_vectorized(self, df, timeframe='1m'):
-        # Optimize types
-        cols = ['close', 'high', 'low', 'volume', 'open']
-        for c in cols: df[c] = df[c].astype(np.float32)
+        df['close'] = df['close'].astype(float)
+        df['high'] = df['high'].astype(float)
+        df['low'] = df['low'].astype(float)
+        df['volume'] = df['volume'].astype(float)
+        df['open'] = df['open'].astype(float)
         
-        # Standard
         df['rsi'] = ta.rsi(df['close'], length=14)
         df['ema20'] = ta.ema(df['close'], length=20)
         df['ema50'] = ta.ema(df['close'], length=50)
         df['atr'] = ta.atr(df['high'], df['low'], df['close'], length=14)
         
-        # Hydra
         if timeframe == '1m':
             sma20 = df['close'].rolling(20).mean()
             std20 = df['close'].rolling(20).std()
@@ -125,43 +125,37 @@ class HeavyDutyBacktester:
             df['vol_ma50'] = df['volume'].rolling(50).mean()
             df['rel_vol'] = df['volume'] / (df['vol_ma50'] + 1e-9)
 
-        # Oracle
         df['slope'] = ta.slope(df['close'], length=7)
         vol_mean = df['volume'].rolling(20).mean()
         vol_std = df['volume'].rolling(20).std()
         df['vol_z'] = (df['volume'] - vol_mean) / (vol_std + 1e-9)
         df['atr_pct'] = df['atr'] / df['close']
 
-        # Sniper (1m)
         if timeframe == '1m':
             df['ret'] = df['close'].pct_change()
             df['dollar_vol'] = df['close'] * df['volume']
             df['amihud'] = (df['ret'].abs() / df['dollar_vol'].replace(0, np.nan)).fillna(0)
-            
             dp = df['close'].diff()
             roll_cov = dp.rolling(64).cov(dp.shift(1))
             df['roll_spread'] = (2 * np.sqrt(np.maximum(0, -roll_cov))).fillna(0)
-            
             sign = np.sign(df['close'].diff()).fillna(0)
             df['signed_vol'] = sign * df['volume']
             df['ofi'] = df['signed_vol'].rolling(30).sum().fillna(0)
-            
             buy_vol = (sign > 0) * df['volume']
             sell_vol = (sign < 0) * df['volume']
             imb = (buy_vol.rolling(60).sum() - sell_vol.rolling(60).sum()).abs()
             tot = df['volume'].rolling(60).sum()
             df['vpin'] = (imb / tot.replace(0, np.nan)).fillna(0)
-            
             vwap = (df['close'] * df['volume']).rolling(20).sum() / df['volume'].rolling(20).sum()
             df['vwap_dev'] = (df['close'] - vwap).fillna(0)
-            
             df['rv_gk'] = (np.log(df['high'] / df['low'])**2) / 2 - (2 * np.log(2) - 1) * (np.log(df['close'] / df['open'])**2)
-            
+            df['return_1m'] = df['ret']
+            df['return_5m'] = df['close'].pct_change(5)
+            df['return_15m'] = df['close'].pct_change(15)
             r = df['volume'].rolling(500).mean()
             s = df['volume'].rolling(500).std()
             df['vol_zscore_50'] = ((df['volume'] - r) / s).fillna(0)
 
-        # Legacy
         df['log_ret'] = np.log(df['close'] / df['close'].shift(1))
         roll_max = df['high'].rolling(50).max()
         roll_min = df['low'].rolling(50).min()
@@ -175,19 +169,19 @@ class HeavyDutyBacktester:
         df['ema200'] = ta.ema(df['close'], length=200)
         df['dist_ema200'] = (df['close'] - df['ema200']) / df['close']
         
-        # Pre-calc Lags for Legacy V2
+        # Lags for V2
         if timeframe == '1m':
             for lag in [1, 2, 3, 5, 10, 20]:
-                df[f'log_ret_lag_{lag}'] = df['log_ret'].shift(lag)
-                df[f'rsi_lag_{lag}'] = df['rsi'].shift(lag) / 100.0
-                df[f'fib_pos_lag_{lag}'] = df['fib_pos'].shift(lag)
-                df[f'volatility_lag_{lag}'] = df['volatility'].shift(lag)
+                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
 
     # ==============================================================
-    # 🧠 CPU PROCESSING (GLOBAL INFERENCE)
+    # 🧠 CPU PROCESSING (PRE-INFERENCE OPTIMIZED)
     # ==============================================================
     async def _process_data_in_memory(self, sym, candles, start_ms, end_ms):
         safe_sym = sym.replace('/', '_')
@@ -198,7 +192,7 @@ class HeavyDutyBacktester:
              print(f"   📂 [{sym}] Data Exists -> Skipping.")
              return
 
-        print(f"   ⚙️ [CPU] Analyzing {sym} (Global Inference)...", flush=True)
+        print(f"   ⚙️ [CPU] Analyzing {sym} (Global Pre-Inference)...", flush=True)
         t0 = time.time()
 
         df_1m = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
@@ -209,11 +203,12 @@ class HeavyDutyBacktester:
         frames = {}
         agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
         
-        # 1. Vectorized Calculation (1m)
+        # 1. Calc 1m
         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}
         
-        # 2. Vectorized Calculation (HTF)
+        # 2. Calc HTF
         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()
@@ -222,243 +217,250 @@ class HeavyDutyBacktester:
             frames[tf_str] = resampled
             numpy_htf[tf_str] = {col: resampled[col].values for col in resampled.columns}
 
-        # 3. GLOBAL MAPPING (Pre-map indices for the whole array)
-        fast_1m = {col: frames['1m'][col].values for col in frames['1m'].columns}
-        arr_ts_1m = fast_1m['timestamp']
-        
-        # Using searchsorted once for the entire array is extremely fast
-        map_5m = np.searchsorted(numpy_htf['5m']['timestamp'], arr_ts_1m)
-        map_15m = np.searchsorted(numpy_htf['15m']['timestamp'], arr_ts_1m)
-        map_1h = np.searchsorted(numpy_htf['1h']['timestamp'], arr_ts_1m)
-        map_4h = np.searchsorted(numpy_htf['4h']['timestamp'], arr_ts_1m)
+        # 3. Global Index Maps
+        map_1m_to_1h = np.searchsorted(numpy_htf['1h']['timestamp'], fast_1m['timestamp'])
+        map_1m_to_5m = np.searchsorted(numpy_htf['5m']['timestamp'], fast_1m['timestamp'])
+        map_1m_to_15m = np.searchsorted(numpy_htf['15m']['timestamp'], fast_1m['timestamp'])
         
         # Clip
-        map_5m = np.clip(map_5m, 0, len(numpy_htf['5m']['timestamp']) - 1)
-        map_15m = np.clip(map_15m, 0, len(numpy_htf['15m']['timestamp']) - 1)
-        map_1h = np.clip(map_1h, 0, len(numpy_htf['1h']['timestamp']) - 1)
-        map_4h = np.clip(map_4h, 0, len(numpy_htf['4h']['timestamp']) - 1)
-
-        # 4. LOAD MODELS
-        legacy_v2 = getattr(self.proc.guardian_legacy, 'model_v2', None)
-        oracle_model = 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', [])
+        max_idx_1h = len(numpy_htf['1h']['timestamp']) - 1
+        max_idx_5m = len(numpy_htf['5m']['timestamp']) - 1
+        max_idx_15m = len(numpy_htf['15m']['timestamp']) - 1
         
+        map_1m_to_1h = np.clip(map_1m_to_1h, 0, max_idx_1h)
+        map_1m_to_5m = np.clip(map_1m_to_5m, 0, max_idx_5m)
+        map_1m_to_15m = np.clip(map_1m_to_15m, 0, max_idx_15m)
+
+        # 4. Load Models
         hydra_models = getattr(self.proc.guardian_hydra, 'models', {}) if self.proc.guardian_hydra else {}
         hydra_cols = getattr(self.proc.guardian_hydra, 'feature_cols', []) if self.proc.guardian_hydra else []
-
-        # ======================================================================
-        # 🔥 GLOBAL INFERENCE: Run Models on ALL data at once (No Loop) 🔥
-        # ======================================================================
+        legacy_v2 = getattr(self.proc.guardian_legacy, 'model_v2', None)
+        
+        # 5. 🔥 PRE-CALCULATE LEGACY V2 (GLOBAL) 🔥
+        # V2 depends only on structure, not entry price. We can predict for ALL rows at once.
+        global_v2_probs = np.zeros(len(fast_1m['close']))
         
-        # A. Global Legacy V2
-        global_v2_scores = np.zeros(len(arr_ts_1m), dtype=np.float32)
         if legacy_v2:
-            print("     🚀 Running Global Legacy V2...", flush=True)
+            print(f"     🚀 Pre-calculating Legacy V2 for entire history...", flush=True)
             try:
-                # Construct Matrix
+                # 1m Feats
                 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]
+                # HTF Feats Mapped to 1m
+                l5_log = numpy_htf['5m']['log_ret'][map_1m_to_5m]
+                l5_rsi = numpy_htf['5m']['rsi'][map_1m_to_5m] / 100.0
+                l5_fib = numpy_htf['5m']['fib_pos'][map_1m_to_5m]
+                l5_trd = numpy_htf['5m']['trend_slope'][map_1m_to_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]
+                l15_log = numpy_htf['15m']['log_ret'][map_1m_to_15m]
+                l15_rsi = numpy_htf['15m']['rsi'][map_1m_to_15m] / 100.0
+                l15_fib618 = numpy_htf['15m']['dist_fib618'][map_1m_to_15m]
+                l15_trd = numpy_htf['15m']['trend_slope'][map_1m_to_15m]
                 
-                lags = []
+                # Lags
+                lag_cols = []
                 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}']])
+                    lag_cols.append(fast_1m[f'log_ret_lag_{lag}'])
+                    lag_cols.append(fast_1m[f'rsi_lag_{lag}'])
+                    lag_cols.append(fast_1m[f'fib_pos_lag_{lag}'])
+                    lag_cols.append(fast_1m[f'volatility_lag_{lag}'])
+                
+                # Huge Matrix
+                X_GLOBAL_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,
+                    *lag_cols
+                ])
                 
-                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])
+                # Predict All in One Go
+                dm_glob = xgb.DMatrix(X_GLOBAL_V2)
+                preds_glob = legacy_v2.predict(dm_glob)
+                global_v2_probs = preds_glob[:, 2] if len(preds_glob.shape) > 1 else preds_glob
                 
-                # Predict All
-                preds = legacy_v2.predict(xgb.DMatrix(X_V2))
-                global_v2_scores = preds[:, 2] if len(preds.shape) > 1 else preds
             except Exception as e: print(f"V2 Error: {e}")
 
-        # B. Global Oracle
-        global_oracle_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
-        if oracle_model and oracle_cols:
-            print("     🚀 Running Global Oracle...", flush=True)
-            try:
-                # Build Oracle Matrix
-                # This is tricky because mapping is by column name. 
-                # We assume consistent ordering or build list.
-                vecs = []
-                for col in oracle_cols:
-                    if col.startswith('1h_'): vecs.append(numpy_htf['1h'].get(col[3:], np.zeros(len(arr_ts_1m)))[map_1h])
-                    elif col.startswith('15m_'): vecs.append(numpy_htf['15m'].get(col[4:], np.zeros(len(arr_ts_1m)))[map_15m])
-                    elif col.startswith('4h_'): vecs.append(numpy_htf['4h'].get(col[3:], np.zeros(len(arr_ts_1m)))[map_4h])
-                    elif col == 'sim_titan_score': vecs.append(np.full(len(arr_ts_1m), 0.6))
-                    else: vecs.append(np.full(len(arr_ts_1m), 0.5))
-                
-                X_ORACLE = np.column_stack(vecs)
-                preds_o = oracle_model.predict(X_ORACLE)
-                # Ensure output format
-                global_oracle_scores = preds_o if isinstance(preds_o, np.ndarray) and len(preds_o.shape)==1 else preds_o[:, 0]
-                # Adjust short prob if needed logic: here assuming long confidence
-            except Exception as e: print(f"Oracle Error: {e}")
-
-        # C. Global Sniper
-        global_sniper_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
-        if sniper_models:
-            print("     🚀 Running Global Sniper...", flush=True)
+        # 6. 🔥 PRE-ASSEMBLE HYDRA STATIC (GLOBAL) 🔥
+        # Hydra needs PnL (dynamic), but 90% features are static.
+        global_hydra_static = None
+        if hydra_models:
+            print(f"     🚀 Pre-assembling Hydra features...", flush=True)
             try:
-                vecs_s = []
-                for col in sniper_cols:
-                    if col in fast_1m: vecs_s.append(fast_1m[col])
-                    elif col == 'L_score': vecs_s.append(fast_1m.get('vol_zscore_50', np.zeros(len(arr_ts_1m))))
-                    else: vecs_s.append(np.zeros(len(arr_ts_1m)))
+                # Map columns that don't depend on PnL
+                h_rsi_1m = fast_1m['rsi']
+                h_rsi_5m = numpy_htf['5m']['rsi'][map_1m_to_5m]
+                h_rsi_15m = numpy_htf['15m']['rsi'][map_1m_to_15m]
+                h_bb = fast_1m['bb_width']
+                h_vol = fast_1m['rel_vol']
+                h_atr = fast_1m['atr']
+                h_close = fast_1m['close']
                 
-                X_SNIPER = np.column_stack(vecs_s)
-                # Average of folds
-                preds_list = [m.predict(X_SNIPER) for m in sniper_models]
-                global_sniper_scores = np.mean(preds_list, axis=0)
-            except Exception as e: print(f"Sniper Error: {e}")
-
-        # --- 5. L1 FILTER (Fast) ---
-        # Map 1h RSI to 1m resolution for filtering
-        rsi_1h_mapped = numpy_htf['1h']['rsi'][map_1h]
-        # Valid: RSI < 70
-        is_candidate = (rsi_1h_mapped <= 70) 
-        
-        # We need to find "Start of signal" indices.
-        # Simple logic: Every index is potentially a signal start if it passes L1.
-        # To avoid spamming signals every minute, we can debounce or just take them all (HeavyDutyBacktester usually takes all)
-        
-        # Limit processing to valid candidates
-        candidate_indices = np.where(is_candidate)[0]
-        
-        # Start date filter
-        start_ts_val = frames['1m'].index[0] + pd.Timedelta(minutes=500)
-        start_idx_offset = np.searchsorted(arr_ts_1m, int(start_ts_val.timestamp()*1000))
-        candidate_indices = candidate_indices[candidate_indices >= start_idx_offset]
+                # We store these separate to combine inside loop efficiently
+                # [rsi1, rsi5, rsi15, bb, vol, atr, close]
+                global_hydra_static = np.column_stack([h_rsi_1m, h_rsi_5m, h_rsi_15m, h_bb, h_vol, h_atr, h_close])
+            except: pass
+
+        # 7. Candidate Filtering
+        df_1h = frames['1h'].reindex(frames['5m'].index, method='ffill')
+        df_5m = frames['5m'].copy()
+        is_valid = (df_1h['rsi'] <= 70) 
+        valid_indices = df_5m[is_valid].index
+        start_dt = df_1m.index[0] + pd.Timedelta(minutes=500)
+        final_valid_indices = [t for t in valid_indices if t >= start_dt]
         
-        # Limit end index to allow for 4h simulation
-        max_idx = len(arr_ts_1m) - 245
-        candidate_indices = candidate_indices[candidate_indices < max_idx]
+        total_signals = len(final_valid_indices)
+        print(f"     🎯 Candidates: {total_signals}. Running Models...", flush=True)
 
-        print(f"     🎯 Candidates: {len(candidate_indices)}. Simulating Trades...", flush=True)
+        oracle_dir_model = 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', [])
 
-        # --- 6. SIMULATION LOOP (Lightweight) ---
-        # Now the loop only needs to:
-        # 1. Lookup Oracle/Sniper/Titan scores (Instant)
-        # 2. Calculate Hydra (Dynamic PnL) - Simplified vectorization per trade
-        # 3. Lookup Legacy (Instant)
-        
         ai_results = []
         
-        # Pre-allocate for Hydra batching inside loop
-        # We can optimize Hydra further: The static features are already global.
-        # Only PnL changes per trade entry.
-        
-        if hydra_models:
-            # Prepare Global Static Matrix for Hydra
-            # [rsi1, rsi5, rsi15, bb, vol, atr, close]
-            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']
-            ])
-        
-        # Iterate candidates
-        for idx_entry in candidate_indices:
-            
-            entry_price = fast_1m['close'][idx_entry]
-            entry_ts = int(arr_ts_1m[idx_entry])
+        # Pre-allocate Hydra time vector (0 to 240)
+        time_vec = np.arange(1, 241)
+
+        # --- MAIN LOOP (Optimized Lookups) ---
+        for i, current_time in enumerate(final_valid_indices):
+            ts_val = int(current_time.timestamp() * 1000)
+            idx_1m = np.searchsorted(fast_1m['timestamp'], ts_val)
             
-            # Scores Lookup
-            s_oracle = global_oracle_scores[idx_entry]
-            s_sniper = global_sniper_scores[idx_entry]
-            s_titan = 0.6 # Placeholder
+            if idx_1m < 500 or idx_1m >= len(fast_1m['close']) - 245: continue
             
-            # Future Window (4h)
-            idx_exit = idx_entry + 240
+            idx_1h = map_1m_to_1h[idx_1m]
+            idx_15m = map_1m_to_15m[idx_1m]
+            idx_4h = np.searchsorted(numpy_htf['4h']['timestamp'], ts_val)
+            if idx_4h >= len(numpy_htf['4h']['close']): idx_4h = len(numpy_htf['4h']['close']) - 1
+
+            # === Oracle (Single Call) ===
+            oracle_conf = 0.5
+            if oracle_dir_model:
+                o_vec = []
+                for col in oracle_cols:
+                    val = 0.0
+                    if col.startswith('1h_'): val = numpy_htf['1h'].get(col[3:], [0])[idx_1h]
+                    elif col.startswith('15m_'): val = numpy_htf['15m'].get(col[4:], [0])[idx_15m]
+                    elif col.startswith('4h_'): val = numpy_htf['4h'].get(col[3:], [0])[idx_4h]
+                    elif col == 'sim_titan_score': val = 0.6
+                    elif col == 'sim_mc_score': val = 0.5
+                    elif col == 'sim_pattern_score': val = 0.5
+                    o_vec.append(val)
+                try:
+                    o_pred = oracle_dir_model.predict(np.array(o_vec).reshape(1, -1))[0]
+                    oracle_conf = float(o_pred[0]) if isinstance(o_pred, (list, np.ndarray)) else float(o_pred)
+                    if oracle_conf < 0.5: oracle_conf = 1 - oracle_conf 
+                except: pass
+
+            # === Sniper (Single Call) ===
+            sniper_score = 0.5
+            if sniper_models:
+                s_vec = []
+                for col in sniper_cols:
+                    if col in fast_1m: s_vec.append(fast_1m[col][idx_1m])
+                    elif col == 'L_score': s_vec.append(fast_1m.get('vol_zscore_50', [0])[idx_1m])
+                    else: s_vec.append(0.0)
+                try:
+                    s_preds = [m.predict(np.array(s_vec).reshape(1, -1))[0] for m in sniper_models]
+                    sniper_score = np.mean(s_preds)
+                except: pass
+
+            # === RISK SIMULATION (ULTRA FAST) ===
+            start_idx = idx_1m + 1
+            end_idx = start_idx + 240
             
-            # Legacy Risk (Max in window)
-            # Instant slice max
-            max_v2 = np.max(global_v2_scores[idx_entry:idx_exit])
-            # Time of panic?
-            panic_rel_idx = np.argmax(global_v2_scores[idx_entry:idx_exit])
-            v2_time = 0
-            if max_v2 > 0.8: v2_time = int(arr_ts_1m[idx_entry + panic_rel_idx])
-
-            # Hydra Risk (Dynamic)
-            max_hydra = 0.0; hydra_time = 0
-            if hydra_models:
-                # Slice Static
-                sl_st = h_static[idx_entry:idx_exit] # (240, 7)
-                sl_close = sl_st[:, 6]
-                sl_atr = sl_st[:, 5]
+            # 1. LEGACY V2 (Instant Lookup)
+            max_legacy_v2 = 0.0; legacy_panic_time = 0
+            if legacy_v2:
+                # Just slice the pre-calculated array!
+                probs_slice = global_v2_probs[start_idx:end_idx]
+                max_legacy_v2 = np.max(probs_slice)
+                panic_indices = np.where(probs_slice > 0.8)[0]
+                if len(panic_indices) > 0:
+                    legacy_panic_time = int(fast_1m['timestamp'][start_idx + panic_indices[0]])
+
+            # 2. HYDRA (Semi-Vectorized)
+            max_hydra_crash = 0.0; hydra_crash_time = 0
+            if hydra_models and global_hydra_static is not None:
+                # Slice Static Feats
+                sl_static = global_hydra_static[start_idx:end_idx] # [rsi1, rsi5, rsi15, bb, vol, atr, close]
+                
+                entry_price = fast_1m['close'][idx_1m]
+                sl_close = sl_static[:, 6]
+                sl_atr = sl_static[:, 5]
+                
+                # Calc Dynamic Feats
+                sl_dist = 1.5 * sl_atr
+                sl_dist = np.where(sl_dist > 0, sl_dist, entry_price * 0.015)
                 
-                # Dynamic Calcs (Vectorized for 240 steps)
-                dist = np.maximum(1.5 * sl_atr, entry_price * 0.015)
-                pnl = sl_close - entry_price
-                norm_pnl = pnl / dist
-                # Cumulative max for Giveback
-                cum_max = np.maximum.accumulate(sl_close)
-                max_pnl_r = (np.maximum(cum_max, entry_price) - entry_price) / dist
-                atr_pct = sl_atr / sl_close
-                time_vec = np.arange(1, 241)
+                sl_pnl = sl_close - entry_price
+                sl_norm_pnl = sl_pnl / sl_dist
                 
-                # Stack Hydra Matrix (240, 15)
-                # Map columns manually
-                # rsi1(0), rsi5(1), rsi15(2), bb(3), vol(4), ema(0), atr_pct, norm, max, dists(0), time, entry(0), oracle, l2(0.7), target(3)
+                sl_cum_max = np.maximum.accumulate(sl_close)
+                sl_cum_max = np.maximum(sl_cum_max, entry_price)
+                sl_max_pnl_r = (sl_cum_max - entry_price) / sl_dist
                 
-                # Constants
+                sl_atr_pct = sl_atr / sl_close
+                
+                # Map to Hydra Cols Order (Hardcoded for max speed)
+                # Cols: rsi_1m, rsi_5m, rsi_15m, bb_width, rel_vol, dist_ema20_1h, atr_pct, norm_pnl_r, max_pnl_r, dists..., time, entry, oracle, l2, target
+                
+                # Re-assemble only what is needed
+                # (Static 0-4) + (Zeros) + (Dynamic) + (Constants)
+                
+                # Create arrays for constants
                 zeros = np.zeros(240)
-                const_oracle = np.full(240, s_oracle)
-                const_l2 = np.full(240, 0.7)
-                const_tgt = np.full(240, 3.0)
+                oracle_arr = np.full(240, oracle_conf)
+                l2_arr = np.full(240, 0.7)
+                target_arr = np.full(240, 3.0)
                 
-                X_H = np.column_stack([
-                    sl_st[:,0], sl_st[:,1], sl_st[:,2], sl_st[:,3], sl_st[:,4],
-                    zeros, atr_pct, norm_pnl, max_pnl_r,
-                    zeros, zeros, time_vec, zeros,
-                    const_oracle, const_l2, const_tgt
+                X_hydra = np.column_stack([
+                    sl_static[:, 0], sl_static[:, 1], sl_static[:, 2], # RSIs
+                    sl_static[:, 3], sl_static[:, 4], # BB, Vol
+                    zeros, # dist_ema
+                    sl_atr_pct, sl_norm_pnl, sl_max_pnl_r,
+                    zeros, zeros, # dists
+                    time_vec, # time
+                    zeros, oracle_arr, l2_arr, target_arr
                 ])
                 
                 try:
-                    # Fast Prediction (240 rows is tiny for XGB)
-                    probs = hydra_models['crash'].predict_proba(X_H)[:, 1]
-                    max_hydra = np.max(probs)
-                    if max_hydra > 0.6:
-                        t_idx = np.argmax(probs)
-                        hydra_time = int(arr_ts_1m[idx_entry + t_idx])
+                    probs_crash = hydra_models['crash'].predict_proba(X_hydra)[:, 1]
+                    max_hydra_crash = np.max(probs_crash)
+                    crash_indices = np.where(probs_crash > 0.6)[0]
+                    if len(crash_indices) > 0:
+                        hydra_crash_time = int(fast_1m['timestamp'][start_idx + crash_indices[0]])
                 except: pass
 
             ai_results.append({
-                'timestamp': entry_ts, 'symbol': sym, 'close': entry_price,
-                'real_titan': s_titan, 
-                'oracle_conf': s_oracle, 
-                'sniper_score': s_sniper,
-                'risk_hydra_crash': max_hydra,
-                'time_hydra_crash': hydra_time,
-                'risk_legacy_v2': max_v2,
-                'time_legacy_panic': v2_time,
-                'signal_type': 'BREAKOUT', # Simplified
+                'timestamp': ts_val, 'symbol': sym, 'close': entry_price,
+                'real_titan': 0.6, 
+                'oracle_conf': oracle_conf, 
+                'sniper_score': sniper_score,
+                'risk_hydra_crash': max_hydra_crash,
+                'time_hydra_crash': hydra_crash_time,
+                'risk_legacy_v2': max_legacy_v2,
+                'time_legacy_panic': legacy_panic_time,
+                'signal_type': 'BREAKOUT',
                 'l1_score': 50.0
             })
-
+        
         dt = time.time() - t0
         if ai_results:
             pd.DataFrame(ai_results).to_pickle(scores_file)
-            print(f"   ✅ [{sym}] Completed in {dt:.2f} seconds. ({len(ai_results)} signals)", flush=True)
+            print(f"   ✅ [{sym}] Completed {len(ai_results)} signals in {dt:.2f} seconds.", flush=True)
         else:
-            print(f"   ⚠️ [{sym}] No signals. Time: {dt:.2f}s", flush=True)
+            print(f"   ⚠️ [{sym}] No valid signals. Time: {dt:.2f}s", flush=True)
             
-        del frames, fast_1m, numpy_htf, global_v2_scores, global_oracle_scores, global_sniper_scores
+        del frames, fast_1m, numpy_htf, global_v2_probs, global_hydra_static
         gc.collect()
 
+    # ==============================================================
+    # PHASE 1 & 2 (Unchanged - Standard Optimization Logic)
+    # ==============================================================
     async def generate_truth_data(self):
         if self.force_start_date and self.force_end_date:
             dt_start = datetime.strptime(self.force_start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
@@ -498,7 +500,6 @@ class HeavyDutyBacktester:
             for ts, group in grouped_by_time:
                 active = list(wallet["positions"].keys())
                 current_prices = {row['symbol']: row['close'] for _, row in group.iterrows()}
-                
                 for sym in active:
                     if sym in current_prices:
                         curr = current_prices[sym]
@@ -523,6 +524,7 @@ class HeavyDutyBacktester:
                         if row['symbol'] in wallet['positions']: continue
                         if row['oracle_conf'] < oracle_thresh: continue
                         if row['sniper_score'] < sniper_thresh: continue
+                        
                         size = 10.0
                         if wallet['balance'] >= size:
                             wallet['positions'][row['symbol']] = {
@@ -563,11 +565,9 @@ class HeavyDutyBacktester:
 
     async def run_optimization(self, target_regime="RANGE"):
         await self.generate_truth_data()
-        
         oracle_range = [0.5, 0.6, 0.7]
         sniper_range = [0.4, 0.5, 0.6]
         hydra_range = [0.75, 0.85, 0.95]
-        
         combinations = []
         for o, s, h in itertools.product(oracle_range, sniper_range, hydra_range):
             combinations.append({
@@ -630,7 +630,7 @@ async def run_strategic_optimization_task():
             best_cfg, best_stats = await optimizer.run_optimization(target_regime=target)
             if best_cfg:
                 hub.submit_challenger(target, best_cfg, best_stats)
-        
+                
         await hub._save_state_to_r2()
         print("✅ [System] ALL Strategic DNA Updated & Saved.")