Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V132.5 - GEM-Architect: Full Diagnostics)
+# 🧪 backtest_engine.py (V134.0 - GEM-Architect: Feature Parity Edition)
 # ============================================================
 
 import asyncio
@@ -10,15 +10,13 @@ import time
 import logging
 import itertools 
 import os
-import glob
 import gc 
 import sys
 import traceback
-from numpy.lib.stride_tricks import sliding_window_view 
 from datetime import datetime, timezone
 from typing import Dict, Any, List
+from numpy.lib.stride_tricks import sliding_window_view 
 
-# ✅ استيراد المحركات
 try:
     from ml_engine.processor import MLProcessor, SystemLimits 
     from ml_engine.data_manager import DataManager
@@ -26,303 +24,22 @@ try:
     from r2 import R2Service
     import ccxt.async_support as ccxt 
     import xgboost as xgb 
+    import lightgbm as lgb 
 except ImportError:
-    print("❌ [Import Error] Critical ML modules missing.")
     pass
 
 logging.getLogger('ml_engine').setLevel(logging.WARNING)
 CACHE_DIR = "backtest_real_scores"
 
-# ============================================================
-# 🛡️ GLOBAL SANITIZATION & HELPERS
-# ============================================================
-
-def sanitize_features(df):
-    """
-    Cleans DataFrame from Infinity and NaNs to prevent crashes.
-    """
-    if df is None or df.empty: return df
-    return df.replace([np.inf, -np.inf], np.nan).fillna(0.0)
-
-def _zv(x):
-    with np.errstate(divide='ignore', invalid='ignore'):
-        x = np.asarray(x, dtype="float32")
-        m = np.nanmean(x, axis=0)
-        s = np.nanstd(x, axis=0) + 1e-9
-        return np.nan_to_num((x - m) / s, nan=0.0)
-
-def _z_score_rolling(x, w=500): 
-    r = x.rolling(w).mean()
-    s = x.rolling(w).std().replace(0, np.nan)
-    return ((x - r) / s).fillna(0)
-
-# ============================================================
-# 🔧 1. FEATURE ENGINEERING
-# ============================================================
-
-def _transform_window_for_pattern(df_window):
-    try:
-        c = df_window['close'].values.astype('float32')
-        o = df_window['open'].values.astype('float32')
-        h = df_window['high'].values.astype('float32')
-        l = df_window['low'].values.astype('float32')
-        v = df_window['volume'].values.astype('float32')
-
-        base = np.stack([o, h, l, c, v], axis=1)
-        base_z = _zv(base)
-
-        lr = np.zeros_like(c); lr[1:] = np.diff(np.log1p(c))
-        rng = (h - l) / (c + 1e-9)
-        extra = np.stack([lr, rng], axis=1)
-        extra_z = _zv(extra)
-
-        def _ema(arr, n):
-            return pd.Series(arr).ewm(span=n, adjust=False).mean().values
-
-        ema9 = _ema(c, 9); ema21 = _ema(c, 21); ema50 = _ema(c, 50); ema200 = _ema(c, 200)
-        slope21 = np.gradient(ema21); slope50 = np.gradient(ema50)
-        
-        delta = np.diff(c, prepend=c[0])
-        up, down = delta.copy(), delta.copy()
-        up[up < 0] = 0; down[down > 0] = 0
-        roll_up = pd.Series(up).ewm(alpha=1/14, adjust=False).mean().values
-        roll_down = pd.Series(down).abs().ewm(alpha=1/14, adjust=False).mean().values
-        rs = roll_up / (roll_down + 1e-9)
-        rsi = 100.0 - (100.0 / (1.0 + rs))
-
-        indicators = np.stack([ema9, ema21, ema50, ema200, slope21, slope50, rsi], axis=1)
-        padding = np.zeros((200, 5), dtype='float32') 
-        indicators_full = np.concatenate([indicators, padding], axis=1)
-        indicators_z = _zv(indicators_full)
-
-        X_seq = np.concatenate([base_z, extra_z, indicators_z], axis=1)
-        X_flat = X_seq.flatten()
-        X_stat = np.array([0.5, 0.0, 0.5], dtype="float32")
-
-        return np.concatenate([X_flat, X_stat])
-    except: return None
-
-def calculate_sniper_features_exact(df):
-    d = df.copy()
-    c = d['close']; h = d['high']; l = d['low']; v = d['volume']; o = d['open']
-    
-    def _z_roll(x, w=200):
-        r = x.rolling(w).mean()
-        s = x.rolling(w).std().replace(0, np.nan)
-        return ((x - r) / s).fillna(0)
-
-    d['return_1m'] = c.pct_change(1).fillna(0)
-    d['return_3m'] = c.pct_change(3).fillna(0)
-    d['return_5m'] = c.pct_change(5).fillna(0)
-    d['return_15m'] = c.pct_change(15).fillna(0)
-
-    d['rsi_14'] = ta.rsi(c, length=14).fillna(50)
-    ema_9 = ta.ema(c, length=9).fillna(c)
-    ema_21 = ta.ema(c, length=21).fillna(c)
-    d['ema_9_slope'] = ((ema_9 - ema_9.shift(1)) / ema_9.shift(1).replace(0, np.nan)).fillna(0)
-    d['ema_21_dist'] = ((c - ema_21) / ema_21.replace(0, np.nan)).fillna(0)
-    
-    atr_raw = ta.atr(h, l, c, length=100).fillna(0)
-    d['atr'] = _z_roll(atr_raw, 500) 
-    d['vol_zscore_50'] = _z_roll(v, 50)
-    rng = (h - l).replace(0, 1e-9)
-    d['candle_range'] = _z_roll(rng, 500)
-    d['close_pos_in_range'] = ((c - l) / rng).fillna(0.5)
-
-    d['dollar_vol'] = c * v
-    amihud_raw = (d['return_1m'].abs() / d['dollar_vol'].replace(0, np.nan)).fillna(0)
-    d['amihud'] = _z_roll(amihud_raw, 500)
-    
-    dp = c.diff()
-    roll_cov = dp.rolling(64).cov(dp.shift(1)).fillna(0)
-    roll_spread_raw = (2 * np.sqrt(np.maximum(0, -roll_cov)))
-    d['roll_spread'] = _z_roll(roll_spread_raw, 500)
-    
-    sign = np.sign(c.diff()).fillna(0)
-    d['signed_vol'] = sign * v
-    ofi_raw = d['signed_vol'].rolling(30).sum().fillna(0)
-    d['ofi'] = _z_roll(ofi_raw, 500)
-    
-    buy_vol = (sign > 0) * v
-    sell_vol = (sign < 0) * v
-    imb = (buy_vol.rolling(60).sum() - sell_vol.rolling(60).sum()).abs()
-    tot = v.rolling(60).sum().replace(0, np.nan)
-    d['vpin'] = (imb / tot).fillna(0)
-    
-    rv_gk_raw = ((np.log(h / l)**2) / 2) - ((2 * np.log(2) - 1) * (np.log(c / o)**2))
-    d['rv_gk'] = _z_roll(rv_gk_raw.fillna(0), 500)
-    
-    vwap_win = 20
-    vwap = (d['dollar_vol'].rolling(vwap_win).sum() / v.rolling(vwap_win).sum().replace(0, np.nan)).fillna(c)
-    d['vwap_dev'] = _z_roll((c - vwap), 500)
-    
-    d['L_score'] = (
-        d['vol_zscore_50'] + (-d['amihud']) + (-d['roll_spread']) + 
-        (-d['rv_gk'].abs()) + (-d['vwap_dev'].abs()) + d['ofi']
-    ).fillna(0)
-    
-    return sanitize_features(d)
-
-def calculate_titan_features_real(df):
-    df = df.copy()
-    df['RSI'] = ta.rsi(df['close'], 14)
-    macd = ta.macd(df['close'])
-    if macd is not None: 
-        df['MACD'] = macd.iloc[:, 0]; df['MACD_h'] = macd.iloc[:, 1]
-    else: df['MACD'] = 0.0; df['MACD_h'] = 0.0
-
-    df['CCI'] = ta.cci(df['high'], df['low'], df['close'], 20)
-    adx = ta.adx(df['high'], df['low'], df['close'], 14)
-    if adx is not None: df['ADX'] = adx.iloc[:, 0]
-    else: df['ADX'] = 0.0
-    
-    for p in [9, 21, 50, 200]:
-        ema = ta.ema(df['close'], p)
-        if ema is not None:
-            df[f'EMA_{p}_dist'] = (df['close'] / ema.replace(0, np.nan)) - 1
-        else: df[f'EMA_{p}_dist'] = 0.0
-
-    bb = ta.bbands(df['close'], 20, 2.0)
-    if bb is not None:
-        df['BB_w'] = (bb.iloc[:, 2] - bb.iloc[:, 0]) / bb.iloc[:, 1].replace(0, np.nan)
-        df['BB_p'] = (df['close'] - bb.iloc[:, 0]) / (bb.iloc[:, 2] - bb.iloc[:, 0]).replace(0, np.nan)
-    else: df['BB_w'] = 0.0; df['BB_p'] = 0.0
-
-    df['MFI'] = ta.mfi(df['high'], df['low'], df['close'], df['volume'], 14)
-    vwap = ta.vwap(df['high'], df['low'], df['close'], df['volume'])
-    if vwap is not None: df['VWAP_dist'] = (df['close'] / vwap.replace(0, np.nan)) - 1
-    else: df['VWAP_dist'] = 0.0
-
-    return sanitize_features(df)
-
-# ============================================================
-# 🔧 LEGACY GUARD (V2 & V3)
-# ============================================================
-def calculate_legacy_v2_vectorized(df_1m, df_5m, df_15m):
-    try:
-        def calc_basic(df, suffix):
-            c = df['close']; h = df['high']; l = df['low']
-            res = pd.DataFrame(index=df.index)
-            res[f'log_ret_{suffix}'] = np.log(c / c.shift(1).replace(0, np.nan)).fillna(0)
-            res[f'rsi_{suffix}'] = (ta.rsi(c, 14) / 100.0).fillna(0.5)
-            roll_max = h.rolling(50).max(); roll_min = l.rolling(50).min()
-            diff = (roll_max - roll_min).replace(0, 1e-9)
-            res[f'fib_pos_{suffix}'] = ((c - roll_min) / diff).fillna(0.5)
-            if suffix == '1m':
-                res[f'volatility_{suffix}'] = (ta.atr(h, l, c, 14) / c.replace(0, np.nan)).fillna(0)
-            else:
-                ema = ta.ema(c, 20)
-                if ema is not None:
-                    res[f'trend_slope_{suffix}'] = ((ema - ema.shift(5)) / ema.shift(5).replace(0, np.nan)).fillna(0)
-                else: res[f'trend_slope_{suffix}'] = 0.0
-                if suffix == '15m':
-                    fib618 = roll_max - (diff * 0.382)
-                    res[f'dist_fib618_{suffix}'] = ((c - fib618) / c.replace(0, np.nan)).fillna(0)
-            return res
-
-        f1 = calc_basic(df_1m, '1m')
-        f5 = calc_basic(df_5m, '5m').reindex(df_1m.index, method='ffill')
-        f15 = calc_basic(df_15m, '15m').reindex(df_1m.index, method='ffill')
-
-        FEATS_1M = ['log_ret_1m', 'rsi_1m', 'fib_pos_1m', 'volatility_1m']
-        FEATS_5M = ['log_ret_5m', 'rsi_5m', 'fib_pos_5m', 'trend_slope_5m']
-        FEATS_15M = ['log_ret_15m', 'rsi_15m', 'dist_fib618_15m', 'trend_slope_15m']
-
-        parts = [f1[FEATS_1M], f5[FEATS_5M], f15[FEATS_15M]]
-        lags = [1, 2, 3, 5, 10, 20]
-        for lag in lags: parts.append(f1[FEATS_1M].shift(lag).fillna(0))
-            
-        X_df = pd.concat(parts, axis=1)
-        return sanitize_features(X_df).values 
-    except Exception as e:
-        return None
-
-def calculate_legacy_v3_vectorized(df_1m, df_5m, df_15m):
-    try:
-        def calc_v3_base(df, prefix=""):
-            d = df.copy()
-            targets = ['rsi', 'rsi_slope', 'macd_h', 'macd_h_slope', 'adx', 'dmp', 'dmn', 
-                       'trend_net_force', 'ema_20', 'ema_50', 'ema_200', 'dist_ema20', 
-                       'dist_ema50', 'dist_ema200', 'slope_ema50', 'atr', 'atr_rel', 
-                       'obv', 'obv_slope', 'cmf', 'log_ret', 'mc_skew', 'mc_kurt', 
-                       'mc_prob_gain', 'mc_var_95', 'mc_shock']
-            for t in targets: d[t] = 0.0
-
-            c = d['close']; h = d['high']; l = d['low']; v = d['volume']
-            try:
-                d['log_ret'] = np.log(c / c.shift(1).replace(0, np.nan)).fillna(0)
-                d['rsi'] = ta.rsi(c, 14).fillna(50)
-                d['rsi_slope'] = (d['rsi'] - d['rsi'].shift(3).fillna(50)) / 3
-                macd = ta.macd(c)
-                if macd is not None:
-                    d['macd_h'] = macd.iloc[:, 1].fillna(0)
-                    d['macd_h_slope'] = (d['macd_h'] - d['macd_h'].shift(3).fillna(0)) / 3
-                adx = ta.adx(h, l, c, 14)
-                if adx is not None:
-                    d['adx'] = adx.iloc[:, 0].fillna(0); d['dmp'] = adx.iloc[:, 1].fillna(0); d['dmn'] = adx.iloc[:, 2].fillna(0)
-                    d['trend_net_force'] = (d['dmp'] - d['dmn']) * (d['adx'] / 100.0)
-                d['ema_20'] = ta.ema(c, 20).fillna(c); d['ema_50'] = ta.ema(c, 50).fillna(c); d['ema_200'] = ta.ema(c, 200).fillna(c)
-                d['dist_ema20'] = (c - d['ema_20']) / d['ema_20'].replace(0, np.nan)
-                d['dist_ema50'] = (c - d['ema_50']) / d['ema_50'].replace(0, np.nan)
-                d['dist_ema200'] = (c - d['ema_200']) / d['ema_200'].replace(0, np.nan)
-                d['slope_ema50'] = (d['ema_50'] - d['ema_50'].shift(5).fillna(0)) / d['ema_50'].shift(5).replace(0, np.nan)
-                d['atr'] = ta.atr(h, l, c, 14).fillna(0); d['atr_rel'] = d['atr'] / c.replace(0, np.nan)
-                d['obv'] = ta.obv(c, v).fillna(0); d['obv_slope'] = d['obv'] - d['obv'].shift(5).fillna(0)
-                d['cmf'] = ta.cmf(h, l, c, v, 20).fillna(0)
-                
-                win = 30; roll = d['log_ret'].rolling(win)
-                d['mc_skew'] = roll.skew().fillna(0); d['mc_kurt'] = roll.kurt().fillna(0)
-                d['mc_prob_gain'] = (d['log_ret'] > 0).rolling(win).mean().fillna(0.5)
-                d['mc_var_95'] = roll.quantile(0.05).fillna(-0.02)
-                d['mc_shock'] = ((d['log_ret'] - roll.mean()) / (roll.std().replace(0, np.nan))).fillna(0)
-            except: pass
-
-            if prefix:
-                d.columns = [f"{col}_{prefix}" if col not in ['timestamp'] else col for col in d.columns]
-                return sanitize_features(d)
-            return sanitize_features(d)
-
-        df1 = calc_v3_base(df_1m); df5 = calc_v3_base(df_5m, "5m").reindex(df_1m.index, method='ffill')
-        df15 = calc_v3_base(df_15m, "15m").reindex(df_1m.index, method='ffill')
-
-        final_df = pd.DataFrame(index=df_1m.index)
-        for i, col_name in enumerate(["6", "7", "8", "9", "10", "11"], 1):
-            final_df[col_name] = df1['log_ret'].shift(i)
-
-        cols_1m = ['rsi', 'rsi_slope', 'macd_h', 'macd_h_slope', 'adx', 'dmp', 'dmn', 'trend_net_force',
-                   'ema_20', 'ema_50', 'ema_200', 'dist_ema20', 'dist_ema50', 'dist_ema200', 'slope_ema50',
-                   'atr', 'atr_rel', 'obv', 'obv_slope', 'cmf', 'log_ret', 'mc_skew', 'mc_kurt', 
-                   'mc_prob_gain', 'mc_var_95', 'mc_shock']
-        for c in cols_1m: final_df[c] = df1[c]
-
-        cols_5m = {'rsi_5m': 'rsi_5m', 'rsi_slope_5m': 'rsi_slope_5m', 'macd_h_5m': 'macd_h_5m',
-                   'mc_prob_gain_5m': 'mc_prob_gain_5m', 'mc_shock_5m': 'mc_shock_5m'}
-        for k, v in cols_5m.items(): final_df[k] = df5[v]
-
-        cols_15m = {'rsi_15m': 'rsi_15m', 'macd_h_15m': 'macd_h_15m', 'trend_net_force_15m': 'trend_net_force_15m',
-                    'mc_prob_gain_15m': 'mc_prob_gain_15m', 'dist_ema200_15m': 'dist_ema200_15m'}
-        for k, v in cols_15m.items(): final_df[k] = df15[v]
-
-        expected = ["6", "7", "8", "9", "10", "11", "rsi", "rsi_slope", "macd_h", "macd_h_slope", "adx", "dmp", "dmn", 
-                    "trend_net_force", "ema_20", "ema_50", "ema_200", "dist_ema20", "dist_ema50", "dist_ema200", 
-                    "slope_ema50", "atr", "atr_rel", "obv", "obv_slope", "cmf", "log_ret", "mc_skew", "mc_kurt", 
-                    "mc_prob_gain", "mc_var_95", "mc_shock", "rsi_5m", "rsi_slope_5m", "macd_h_5m", "mc_prob_gain_5m", 
-                    "mc_shock_5m", "rsi_15m", "macd_h_15m", "trend_net_force_15m", "mc_prob_gain_15m", "dist_ema200_15m"]
-        
-        return sanitize_features(final_df.reindex(columns=expected, fill_value=0.0))
-    except: return None
-
-# ============================================================
-# 🧪 THE BACKTESTER
-# ============================================================
 class HeavyDutyBacktester:
     def __init__(self, data_manager, processor):
         self.dm = data_manager
         self.proc = processor
-        self.GRID_DENSITY = 10
+        self.GRID_DENSITY = 5 
         self.INITIAL_CAPITAL = 10.0
         self.TRADING_FEES = 0.001 
         self.MAX_SLOTS = 4         
+        
         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', 
@@ -334,379 +51,606 @@ class HeavyDutyBacktester:
             'STRK/USDT', 'BLUR/USDT', 'ALT/USDT', 'JUP/USDT', 'PENDLE/USDT', 'ETHFI/USDT', 
             'MEME/USDT', 'ATOM/USDT'
         ]
-        self.force_start_date = None; self.force_end_date = None   
-        if os.path.exists(CACHE_DIR):
-            for f in glob.glob(os.path.join(CACHE_DIR, "*")): os.remove(f)
-        else: os.makedirs(CACHE_DIR)
-        self._check_engines()
-
-    def _check_engines(self):
-        status = []
-        if self.proc.titan and self.proc.titan.model: status.append("Titan")
-        if self.proc.pattern_engine and self.proc.pattern_engine.models: status.append("Patterns")
-        if self.proc.oracle: status.append("Oracle")
-        if self.proc.sniper: status.append("Sniper")
-        if self.proc.guardian_hydra: status.append("Hydra")
-        if self.proc.guardian_legacy: status.append("Legacy")
-        print(f"   ✅ Engines Ready: {', '.join(status)}")
+        
+        self.force_start_date = None 
+        self.force_end_date = None   
+        
+        if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
+        print(f"🧪 [Backtest V134.0] Feature Parity Mode (Exact Live System Logic).")
 
     def set_date_range(self, start_str, end_str):
-        self.force_start_date = start_str; self.force_end_date = end_str
-
-    def _smart_predict(self, model, X, model_name="Generic"):
-        try:
-            if hasattr(model, "predict_proba"):
-                raw = model.predict_proba(X)
-                if raw.ndim == 2: return raw[:, -1] 
-                return raw 
-            return model.predict(X)
-        except Exception as e:
-            return np.zeros(len(X) if hasattr(X, '__len__') else 0)
-
-    def _extract_probs(self, raw_preds):
-        if isinstance(raw_preds, list): raw_preds = np.array(raw_preds)
-        if raw_preds.ndim == 1: return raw_preds
-        elif raw_preds.ndim == 2:
-            if raw_preds.shape[1] >= 2: return raw_preds[:, -1]
-            return raw_preds.flatten()
-        return raw_preds.flatten()
+        self.force_start_date = start_str
+        self.force_end_date = end_str
 
+    # ==============================================================
+    # ⚡ FAST DATA DOWNLOADER
+    # ==============================================================
     async def _fetch_all_data_fast(self, sym, start_ms, end_ms):
         print(f"   ⚡ [Network] Downloading {sym}...", flush=True)
-        limit = 1000; duration = limit * 60 * 1000
-        tasks = []; curr = start_ms
-        while curr < end_ms: tasks.append(curr); curr += duration
-        all_c = []; sem = asyncio.Semaphore(20)
-        async def _fetch(ts):
+        limit = 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) 
+
+        async def _fetch_batch(timestamp):
             async with sem:
-                try: return await self.dm.exchange.fetch_ohlcv(sym, '1m', since=ts, limit=limit)
-                except: await asyncio.sleep(0.5); return []
-        chunk = 50
-        for i in range(0, len(tasks), chunk):
-            res = await asyncio.gather(*[_fetch(t) for t in tasks[i:i+chunk]])
-            for r in res: all_c.extend(r)
-        seen = set(); unique = []
-        for c in all_c:
-            if c[0] not in seen and c[0] >= start_ms and c[0] <= end_ms:
-                unique.append(c); seen.add(c[0])
-        unique.sort(key=lambda x: x[0])
-        print(f"     ✅ Downloaded {len(unique)} candles.", flush=True)
-        return unique
+                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 = 25
+        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]
+            results = await asyncio.gather(*futures)
+            for res in results:
+                if res: all_candles.extend(res)
+
+        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)]
+        df.sort_values('timestamp', inplace=True)
+        print(f"     ✅ Downloaded {len(df)} candles.", flush=True)
+        return df.values.tolist()
+
+    # ==============================================================
+    # 🏎️ HELPER: Rolling Z-Score (For Sniper)
+    # ==============================================================
+    def _z_roll(self, x, w=500):
+        r = x.rolling(w).mean()
+        s = x.rolling(w).std().replace(0, np.nan)
+        return ((x - r) / s).fillna(0)
 
+    # ==============================================================
+    # 🏎️ VECTORIZED INDICATORS (EXACT MATCH TO LIVE SYSTEM)
+    # ==============================================================
+    def _calculate_indicators_vectorized(self, df, timeframe='1m'):
+        # 1. Clean Types
+        cols = ['close', 'high', 'low', 'volume', 'open']
+        for c in cols: df[c] = df[c].astype(np.float64) # Use float64 for precision match
+        
+        # ---------------------------------------------------------
+        # 🧠 PART 1: TITAN FEATURES (Exact Replica of TitanEngine)
+        # ---------------------------------------------------------
+        # RSI
+        df['RSI'] = ta.rsi(df['close'], length=14).fillna(50)
+        
+        # MACD
+        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.0; df['MACD_h'] = 0.0
+
+        # CCI
+        df['CCI'] = ta.cci(df['high'], df['low'], df['close'], length=20).fillna(0)
+        
+        # ADX
+        adx = ta.adx(df['high'], df['low'], df['close'], length=14)
+        if adx is not None: df['ADX'] = adx.iloc[:, 0].fillna(0)
+        else: df['ADX'] = 0.0
+
+        # EMAs & Distances
+        for p in [9, 21, 50, 200]:
+            ema = ta.ema(df['close'], length=p)
+            df[f'EMA_{p}_dist'] = ((df['close'] / ema) - 1).fillna(0)
+            df[f'ema{p}'] = ema # Keep raw for others
+
+        # Bollinger Bands (Width & %B)
+        bb = ta.bbands(df['close'], length=20, std=2.0)
+        if bb is not None:
+            # Width = (Upper - Lower) / Middle
+            df['BB_w'] = ((bb.iloc[:, 2] - bb.iloc[:, 0]) / bb.iloc[:, 1]).fillna(0)
+            # %B = (Price - Lower) / (Upper - Lower)
+            df['BB_p'] = ((df['close'] - bb.iloc[:, 0]) / (bb.iloc[:, 2] - bb.iloc[:, 0])).fillna(0)
+            
+            # Helper for Hydra
+            df['bb_width'] = df['BB_w'] # Alias
+        
+        # MFI
+        df['MFI'] = ta.mfi(df['high'], df['low'], df['close'], df['volume'], length=14).fillna(50)
+        
+        # VWAP
+        vwap = ta.vwap(df['high'], df['low'], df['close'], df['volume'])
+        if vwap is not None:
+            df['VWAP_dist'] = ((df['close'] / vwap) - 1).fillna(0)
+            df['vwap'] = vwap
+        else: 
+            df['VWAP_dist'] = 0.0
+            df['vwap'] = df['close']
+
+        # ATR (for others)
+        df['atr'] = ta.atr(df['high'], df['low'], df['close'], length=14).fillna(0)
+        df['atr_pct'] = df['atr'] / df['close']
+
+        # ---------------------------------------------------------
+        # 🎯 PART 2: SNIPER FEATURES (1m Only)
+        # ---------------------------------------------------------
+        if timeframe == '1m':
+            df['return_1m'] = df['close'].pct_change().fillna(0)
+            df['return_3m'] = df['close'].pct_change(3).fillna(0)
+            df['return_5m'] = df['close'].pct_change(5).fillna(0)
+            df['return_15m'] = df['close'].pct_change(15).fillna(0)
+            
+            df['rsi_14'] = df['RSI'] # Alias
+            
+            # Sniper specific derivations
+            df['ema_9_slope'] = ((df['ema9'] - df['ema9'].shift(1)) / df['ema9'].shift(1)).fillna(0)
+            df['ema_21_dist'] = df['EMA_21_dist'] # Reuse
+            
+            # Z-Scores for Sniper
+            atr_100 = ta.atr(df['high'], df['low'], df['close'], length=100).fillna(0)
+            df['atr_z'] = self._z_roll(atr_100) # Mapped later
+            
+            df['vol_zscore_50'] = self._z_roll(df['volume'], 50)
+            
+            rng = (df['high'] - df['low']).replace(0, 1e-9)
+            df['candle_range'] = self._z_roll(rng, 500)
+            df['close_pos_in_range'] = ((df['close'] - df['low']) / rng).fillna(0.5)
+            
+            # Liquidity Proxies
+            df['dollar_vol'] = df['close'] * df['volume']
+            amihud_raw = (df['return_1m'].abs() / df['dollar_vol'].replace(0, np.nan)).fillna(0)
+            df['amihud'] = self._z_roll(amihud_raw)
+            
+            dp = df['close'].diff()
+            roll_cov = dp.rolling(64).cov(dp.shift(1))
+            roll_spread_raw = (2 * np.sqrt(np.maximum(0, -roll_cov)))
+            df['roll_spread'] = self._z_roll(roll_spread_raw)
+            
+            sign = np.sign(df['close'].diff()).fillna(0)
+            signed_vol = sign * df['volume']
+            ofi_raw = signed_vol.rolling(30).sum()
+            df['ofi'] = self._z_roll(ofi_raw)
+            
+            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_win = 20
+            v_short = (df['dollar_vol'].rolling(vwap_win).sum() / df['volume'].rolling(vwap_win).sum().replace(0, np.nan)).fillna(df['close'])
+            df['vwap_dev'] = self._z_roll(df['close'] - v_short)
+            
+            rv_gk = ((np.log(df['high'] / df['low'])**2) / 2) - ((2 * np.log(2) - 1) * (np.log(df['close'] / df['open'])**2))
+            df['rv_gk'] = self._z_roll(rv_gk)
+            
+            # L_Score approximation
+            df['L_score'] = (df['vol_zscore_50'] - df['amihud'] - df['roll_spread'] - df['rv_gk'].abs() - df['vwap_dev'].abs() + df['ofi']).fillna(0)
+
+        # ---------------------------------------------------------
+        # 🧠 PART 3: ORACLE / HYDRA / LEGACY EXTRAS
+        # ---------------------------------------------------------
+        df['slope'] = ta.slope(df['close'], length=7).fillna(0)
+        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)).fillna(0)
+        
+        df['rel_vol'] = df['volume'] / (df['volume'].rolling(50).mean() + 1e-9)
+        
+        df['log_ret'] = np.log(df['close'] / df['close'].shift(1)).fillna(0)
+        roll_max = df['high'].rolling(50).max()
+        roll_min = df['low'].rolling(50).min()
+        diff = (roll_max - roll_min).replace(0, 1e-9)
+        df['fib_pos'] = ((df['close'] - roll_min) / diff).fillna(0.5)
+        df['trend_slope'] = ((df['ema20'] - df['ema20'].shift(5)) / df['ema20'].shift(5)).fillna(0)
+        df['volatility'] = (df['atr'] / df['close']).fillna(0)
+        
+        fib618 = roll_max - (diff * 0.382)
+        df['dist_fib618'] = ((df['close'] - fib618) / df['close']).fillna(0)
+        
+        # Legacy Lags
+        if timeframe == '1m':
+            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) / 100.0).fillna(0.5)
+                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)
+    # ==============================================================
     async def _process_data_in_memory(self, sym, candles, start_ms, end_ms):
         safe_sym = sym.replace('/', '_')
-        scores_file = f"{CACHE_DIR}/{safe_sym}_{start_ms}_{end_ms}_scores.pkl"
+        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} (ALL REAL MODELS)...", flush=True)
+        print(f"   ⚙️ [CPU] Analyzing {sym} (Global Inference)...", flush=True)
         t0 = time.time()
 
         df_1m = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-        for c in ['open', 'high', 'low', 'close', 'volume']: df_1m[c] = df_1m[c].astype(float)
         df_1m['datetime'] = pd.to_datetime(df_1m['timestamp'], unit='ms')
         df_1m.set_index('datetime', inplace=True)
         df_1m = df_1m.sort_index()
 
-        df_5m = df_1m.resample('5T').agg({'open':'first', 'high':'max', 'low':'min', 'close':'last', 'volume':'sum'}).dropna()
-        df_15m = df_1m.resample('15T').agg({'open':'first', 'high':'max', 'low':'min', 'close':'last', 'volume':'sum'}).dropna()
-
-        # 1. Sniper
-        df_sniper_feats = calculate_sniper_features_exact(df_1m)
-        rel_vol = df_1m['volume'] / (df_1m['volume'].rolling(50).mean() + 1e-9)
-        l1_score = (rel_vol * 10) + ((ta.atr(df_1m['high'], df_1m['low'], df_1m['close'], 14)/df_1m['close']) * 1000)
+        frames = {}
+        agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
         
-        valid_mask = (l1_score >= 5.0) & (np.arange(len(df_1m)) > 500)
-        df_candidates = df_1m[valid_mask].copy()
-        if df_candidates.empty: return
-
-        print(f"     🎯 Candidates: {len(df_candidates)}. Running Deep Inference...", flush=True)
+        # 1. Calc 1m (Base)
+        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. 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()
+            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}
+
+        # 3. Global Index Maps
+        arr_ts_1m = fast_1m['timestamp']
+        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)
+        
+        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
+        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 []
+        legacy_v2 = getattr(self.proc.guardian_legacy, 'model_v2', 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', [])
+        
+        titan_model = getattr(self.proc.titan, 'model', None)
+        titan_cols = getattr(self.proc.titan, 'feature_names', [])
 
-        # 2. Patterns
-        res_patterns = np.full(len(df_candidates), 0.5)
-        pattern_models = getattr(self.proc.pattern_engine, 'models', {})
-        if pattern_models and '15m' in pattern_models:
-            try:
-                df_15m_res = df_1m.resample('15T').agg({'open':'first', 'high':'max', 'low':'min', 'close':'last', 'volume':'sum'}).dropna()
-                pat_scores_15m = np.full(len(df_15m_res), 0.5)
-                pat_inputs = []; valid_15m_idxs = []
-                for i in range(200, len(df_15m_res)):
-                    window = df_15m_res.iloc[i-200:i]
-                    vec = _transform_window_for_pattern(window)
-                    if vec is not None:
-                        pat_inputs.append(vec); valid_15m_idxs.append(i)
-                if pat_inputs:
-                    X_pat = np.array(pat_inputs)
-                    pat_preds = self._smart_predict(pattern_models['15m'], xgb.DMatrix(X_pat), "Pattern")
-                    pat_scores_15m[valid_15m_idxs] = pat_preds
-                ts_15m = df_15m_res.index.astype(np.int64) // 10**6
-                map_idxs = np.searchsorted(ts_15m, df_candidates['timestamp'].values) - 1
-                res_patterns = pat_scores_15m[np.clip(map_idxs, 0, len(pat_scores_15m)-1)]
-            except Exception as e: print(f"Patterns Error: {e}")
-
-        # 3. Titan
-        res_titan = np.full(len(df_candidates), 0.5)
-        if self.proc.titan and self.proc.titan.model:
+        # ======================================================================
+        # 🔥 GLOBAL INFERENCE (Batch)
+        # ======================================================================
+        
+        # A. TITAN (Map 5m -> 1m)
+        global_titan_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
+        if titan_model and titan_cols:
+            print("     🚀 Running Global Titan...", flush=True)
             try:
-                df_5m_feat = calculate_titan_features_real(df_5m).add_prefix('5m_')
-                ts_5m = df_5m.index.astype(np.int64) // 10**6
-                map_idxs = np.clip(np.searchsorted(ts_5m, df_candidates['timestamp'].values) - 1, 0, len(df_5m_feat)-1)
-                feats = self.proc.titan.feature_names
-                X_titan_df = sanitize_features(df_5m_feat.iloc[map_idxs].reindex(columns=feats, fill_value=0))
-                res_titan = self.proc.titan.model.predict(xgb.DMatrix(X_titan_df.values, feature_names=feats))
+                # Titan needs 5m features aligned to 1m
+                # Build feature matrix from numpy_htf['5m'] using map_5m
+                t_vecs = []
+                for col in titan_cols:
+                    # Titan features usually have no prefix in the pickle list, 
+                    # but in htf dict we have raw names. 
+                    # Need to verify if titan_cols expects "RSI" or "5m_RSI"??
+                    # Usually Titan is trained on ONE timeframe (5m).
+                    # So we just pull the raw column from numpy_htf['5m'].
+                    
+                    # Fix: Clean name (e.g. if trained as 'RSI', grab 'RSI')
+                    if col in numpy_htf['5m']:
+                        t_vecs.append(numpy_htf['5m'][col][map_5m])
+                    else:
+                        t_vecs.append(np.zeros(len(arr_ts_1m)))
+                
+                X_TITAN = np.column_stack(t_vecs)
+                preds_t = titan_model.predict(xgb.DMatrix(X_TITAN))
+                global_titan_scores = preds_t
             except Exception as e: print(f"Titan Error: {e}")
 
-        # 4. Sniper
-        res_sniper = np.full(len(df_candidates), 0.5)
-        sniper_instance = self.proc.sniper
-        if sniper_instance and getattr(sniper_instance, 'models', []):
+        # B. SNIPER (1m Direct)
+        global_sniper_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
+        if sniper_models:
+            print("     🚀 Running Global Sniper...", flush=True)
             try:
-                required_features = getattr(sniper_instance, 'feature_names', [])
-                if not required_features and hasattr(sniper_instance.models[0], 'feature_name'):
-                     required_features = sniper_instance.models[0].feature_name()
-
-                source_cols = df_sniper_feats.columns
-                def normalize_name(s): return s.lower().replace('_', '').replace('-', '').replace(' ', '')
-                col_map_lower = {col.lower(): col for col in source_cols}
-                col_map_fuzzy = {normalize_name(col): col for col in source_cols}
+                s_vecs = []
+                for col in sniper_cols:
+                    if col in fast_1m: s_vecs.append(fast_1m[col])
+                    # Fix mapping for 'atr' -> 'atr_z' if needed
+                    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_final = pd.DataFrame(index=df_candidates.index)
-                if required_features:
-                    for req_feat in required_features:
-                        req_norm = normalize_name(req_feat)
-                        if req_feat in source_cols: X_final[req_feat] = df_sniper_feats.loc[df_candidates.index, req_feat]
-                        elif req_feat.lower() in col_map_lower: X_final[req_feat] = df_sniper_feats.loc[df_candidates.index, col_map_lower[req_feat.lower()]]
-                        elif req_norm in col_map_fuzzy: X_final[req_feat] = df_sniper_feats.loc[df_candidates.index, col_map_fuzzy[req_norm]]
-                        else: X_final[req_feat] = 0.0
-                else: X_final = df_sniper_feats.loc[df_candidates.index]
-
-                preds = []
-                for m in sniper_instance.models:
-                    X_in = X_final.astype(np.float32)
-                    raw_p = self._extract_probs(self._smart_predict(m, X_in, "Sniper"))
-                    preds.append(raw_p)
-                res_sniper = np.mean(preds, axis=0)
-            except Exception as e: 
-                print(f"❌ Sniper Inference Error: {e}")
-                traceback.print_exc()
-
-        # 5. Oracle
-        res_oracle = np.full(len(df_candidates), 0.5)
-        oracle_model = getattr(self.proc.oracle, 'model_direction', None)
-        if oracle_model:
+                X_SNIPER = np.column_stack(s_vecs)
+                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}")
+
+        # C. ORACLE (HTF Mix)
+        global_oracle_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
+        if oracle_dir:
+            print("     🚀 Running Global Oracle...", flush=True)
             try:
-                oracle_feats = getattr(self.proc.oracle, 'feature_cols', [])
-                X_orc_df = pd.DataFrame(0.0, index=range(len(df_candidates)), columns=oracle_feats)
-                if 'sim_titan_score' in X_orc_df: X_orc_df['sim_titan_score'] = res_titan
-                if 'sim_pattern_score' in X_orc_df: X_orc_df['sim_pattern_score'] = res_patterns
-                if 'sim_mc_score' in X_orc_df: X_orc_df['sim_mc_score'] = 0.5
-                res_oracle = self._extract_probs(self._smart_predict(oracle_model, X_orc_df.values, "Oracle"))
+                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)
+                global_oracle_scores = preds_o if isinstance(preds_o, np.ndarray) and len(preds_o.shape)==1 else preds_o[:, 0]
+                # Adjust if binary (assuming 0=Long, 1=Short or vice versa, check training)
+                # Usually we want Confidence > 0.6. Assuming output is Long Prob.
             except Exception as e: print(f"Oracle Error: {e}")
 
-        # 6. Hydra
-        res_hydra_risk = np.zeros(len(df_candidates))
-        hydra_models = getattr(self.proc.guardian_hydra, 'models', {})
-        if hydra_models and 'crash' in hydra_models:
+        # D. LEGACY V2 (Global)
+        global_v2_scores = np.zeros(len(arr_ts_1m), dtype=np.float32)
+        if legacy_v2:
             try:
-                df_sniper_base = df_1m.copy() 
-                df_sniper_base['rsi_14'] = ta.rsi(df_sniper_base['close'], 14).fillna(50)
-                df_sniper_base['atr'] = ta.atr(df_sniper_base['high'], df_sniper_base['low'], df_sniper_base['close'], 14).fillna(0)
-                df_sniper_base['rel_vol'] = rel_vol 
-
-                global_hydra_feats = np.column_stack([
-                    df_sniper_base['rsi_14'], df_sniper_base['rsi_14'], df_sniper_base['rsi_14'],
-                    (df_sniper_base['close']-df_sniper_base['close'].rolling(20).mean())/df_sniper_base['close'],
-                    df_sniper_base['rel_vol'], df_sniper_base['atr'], df_sniper_base['close']
-                ]).astype(np.float32)
+                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]
                 
-                global_hydra_feats = np.nan_to_num(global_hydra_feats, nan=0.0, posinf=0.0, neginf=0.0) 
-                window_view = sliding_window_view(global_hydra_feats, 240, axis=0).transpose(0, 2, 1)
-                c_idxs = np.searchsorted(df_1m.index, df_candidates.index)
-                valid_s = c_idxs + 1
-                valid_mask_h = valid_s < (len(global_hydra_feats) - 240)
-                final_s = valid_s[valid_mask_h]; res_idxs = np.where(valid_mask_h)[0]
+                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}']])
                 
-                for i in range(0, len(final_s), 5000):
-                    b_idxs = final_s[i:i+5000]; r_idxs = res_idxs[i:i+5000]
-                    static = window_view[b_idxs]
-                    B = len(b_idxs)
-                    entry = df_1m['close'].values[b_idxs-1].reshape(B, 1)
-                    s_c = static[:, 6, :]; s_atr = static[:, 5, :]
-                    dist = np.maximum(1.5*s_atr, entry*0.015) + 1e-9
-                    pnl = (s_c - entry)/dist
-                    max_pnl = (np.maximum.accumulate(s_c, axis=1) - entry)/dist
-                    atr_p = s_atr/(s_c+1e-9)
-                    zeros = np.zeros((B, 240)); ones = np.ones((B, 240)); t = np.tile(np.arange(1, 241), (B, 1))
-                    X = np.stack([
-                        static[:,0], static[:,1], static[:,2], static[:,3], static[:,4],
-                        zeros, atr_p, pnl, max_pnl, zeros, zeros, t, zeros, ones*0.6, ones*0.7, ones*3
-                    ], axis=2).reshape(-1, 16)
-                    X = np.nan_to_num(X, nan=0.0, posinf=0.0, neginf=0.0)
-                    preds = hydra_models['crash'].predict_proba(X)[:, 1].reshape(B, 240)
-                    res_hydra_risk[r_idxs] = np.max(preds, axis=1)
+                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
             except: pass
 
-        # 7. Legacy
-        res_legacy_v2 = np.zeros(len(df_candidates))
-        res_legacy_v3 = np.zeros(len(df_candidates))
-        if self.proc.guardian_legacy:
+        # E. HYDRA STATIC (Global)
+        global_hydra_static = None
+        if hydra_models:
             try:
-                X_v2_full = calculate_legacy_v2_vectorized(df_1m, df_5m, df_15m)
-                v3_df_full = calculate_legacy_v3_vectorized(df_1m, df_5m, df_15m)
-                all_indices = np.arange(len(df_1m))
-                cand_indices = all_indices[valid_mask]
-                max_len = len(df_1m)
-                cand_indices = cand_indices[cand_indices < max_len]
-                if len(cand_indices) > 0:
-                    if self.proc.guardian_legacy.model_v2 and X_v2_full is not None:
-                        subset_v2 = X_v2_full[cand_indices]
-                        preds_v2 = self.proc.guardian_legacy.model_v2.predict(xgb.DMatrix(subset_v2))
-                        if len(preds_v2.shape) > 1: res_legacy_v2[:len(cand_indices)] = preds_v2[:, 2] 
-                        else: res_legacy_v2[:len(cand_indices)] = preds_v2
-                    if self.proc.guardian_legacy.model_v3 and v3_df_full is not None:
-                        subset_v3_df = v3_df_full.iloc[cand_indices]
-                        preds_v3 = self.proc.guardian_legacy.model_v3.predict(xgb.DMatrix(subset_v3_df))
-                        res_legacy_v3[:len(cand_indices)] = preds_v3
-            except Exception as e: print(f"Legacy Error: {e}")
-
-        # 8. Assembly
-        print(f"     📊 [Stats] Titan:{res_titan.mean():.2f} | Patterns:{res_patterns.mean():.2f} | Sniper:{res_sniper.mean():.2f} | Oracle:{res_oracle.mean():.2f}")
-        ai_df = pd.DataFrame({
-            'timestamp': df_candidates['timestamp'],
-            'symbol': sym,
-            'close': df_candidates['close'],
-            'real_titan': res_titan,
-            'oracle_conf': res_oracle,
-            'sniper_score': res_sniper,
-            'l1_score': l1_score[valid_mask],
-            'risk_hydra_crash': res_hydra_risk,
-            'risk_legacy_v2': res_legacy_v2,    
-            'risk_legacy_v3': res_legacy_v3     
-        })
+                # [rsi1, rsi5, rsi15, bb, vol, atr, close]
+                global_hydra_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']
+                ])
+            except: pass
+
+        # --- 5. Filtering Candidates ---
+        # Using Oracle and Sniper to filter BEFORE loop
+        # This saves simulating trades that would never be entered
+        
+        # Valid: (Titan > 0.5) & (Oracle > 0.5) & (Sniper > 0.3) & (RSI < 70)
+        # This reduces loop count drastically
+        is_candidate = (
+            (numpy_htf['1h']['RSI'][map_1h] <= 70) & 
+            (global_titan_scores > 0.4) & 
+            (global_oracle_scores > 0.4)
+        )
+        
+        candidate_indices = np.where(is_candidate)[0]
+        
+        # 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]
+        max_idx = len(arr_ts_1m) - 245
+        candidate_indices = candidate_indices[candidate_indices < max_idx]
+
+        print(f"     🎯 Candidates: {len(candidate_indices)}. Simulating Trades...", flush=True)
+
+        ai_results = []
+        time_vec = np.arange(1, 241)
+
+        # --- 6. SIMULATION LOOP (Lite) ---
+        for idx_entry in candidate_indices:
+            
+            entry_price = fast_1m['close'][idx_entry]
+            entry_ts = int(arr_ts_1m[idx_entry])
+            
+            s_titan = global_titan_scores[idx_entry]
+            s_oracle = global_oracle_scores[idx_entry]
+            s_sniper = global_sniper_scores[idx_entry]
+            
+            idx_exit = idx_entry + 240
+            
+            # Legacy Max Risk
+            max_v2 = np.max(global_v2_scores[idx_entry:idx_exit])
+            v2_time = 0
+            if max_v2 > 0.8:
+                rel = np.argmax(global_v2_scores[idx_entry:idx_exit])
+                v2_time = int(arr_ts_1m[idx_entry + rel])
+
+            # Hydra Dynamic Risk
+            max_hydra = 0.0; hydra_time = 0
+            if hydra_models and global_hydra_static is not None:
+                sl_st = global_hydra_static[idx_entry:idx_exit]
+                sl_close = sl_st[:, 6]
+                sl_atr = sl_st[:, 5]
+                
+                dist = np.maximum(1.5 * sl_atr, entry_price * 0.015)
+                pnl = sl_close - entry_price
+                norm_pnl = pnl / dist
+                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
+                
+                zeros = np.zeros(240)
+                oracle_arr = np.full(240, s_oracle)
+                l2_arr = np.full(240, 0.7)
+                tgt_arr = np.full(240, 3.0)
+                
+                # [rsi1, rsi5, rsi15, bb, vol, dist_ema, atr_p, norm, max, dists, time, entry, oracle, l2, target]
+                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,
+                    oracle_arr, l2_arr, tgt_arr
+                ])
+                
+                try:
+                    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])
+                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',
+                'l1_score': 50.0
+            })
+
         dt = time.time() - t0
-        if not ai_df.empty:
-            ai_df.to_pickle(scores_file)
-            print(f"   ✅ [{sym}] Completed {len(ai_df)} signals in {dt:.2f} seconds.", flush=True)
+        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)
+        else:
+            print(f"   ⚠️ [{sym}] No signals.", flush=True)
+            
+        del frames, fast_1m, numpy_htf, global_v2_scores, global_oracle_scores, global_sniper_scores, global_titan_scores
         gc.collect()
 
+    # ==============================================================
+    # PHASE 1 & 2 (Standard Optimization)
+    # ==============================================================
     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)
+        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)
+            dt_end = datetime.strptime(self.force_end_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
+            start_time_ms = int(dt_start.timestamp() * 1000)
+            end_time_ms = int(dt_end.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)
+        else: return 
+        
+        for sym in self.TARGET_COINS:
+            try:
+                candles = await self._fetch_all_data_fast(sym, start_time_ms, end_time_ms)
+                if candles: await self._process_data_in_memory(sym, candles, start_time_ms, end_time_ms)
+            except Exception as e: print(f"   ❌ SKIP {sym}: {e}", flush=True)
+            gc.collect()
 
     @staticmethod
     def _worker_optimize(combinations_batch, scores_files, initial_capital, fees_pct, max_slots):
-        print(f"     ⏳ [System] Loading {len(scores_files)} datasets...", flush=True)
-        data = []
-        for f in scores_files:
-            try: data.append(pd.read_pickle(f))
+        results = []
+        all_data = []
+        for fp in scores_files:
+            try: 
+                df = pd.read_pickle(fp)
+                if not df.empty: all_data.append(df)
             except: pass
-        if not data: return []
-        df = pd.concat(data).sort_values('timestamp')
+        if not all_data: return []
+        global_df = pd.concat(all_data)
+        global_df.sort_values('timestamp', inplace=True)
+        grouped_by_time = global_df.groupby('timestamp')
         
-        ts = df['timestamp'].values; close = df['close'].values.astype(float)
-        sym = df['symbol'].values; sym_map = {s:i for i,s in enumerate(np.unique(sym))}
-        sym_id = np.array([sym_map[s] for s in sym])
-        
-        oracle = df['oracle_conf'].values; sniper = df['sniper_score'].values
-        hydra = df['risk_hydra_crash'].values; titan = df['real_titan'].values
-        l1 = df['l1_score'].values
-        legacy_v2 = df['risk_legacy_v2'].values; legacy_v3 = df['risk_legacy_v3'].values 
-        
-        N = len(ts)
-        print(f"     🚀 [System] Testing {len(combinations_batch)} configs on {N} candles...", flush=True)
-        
-        res = []
-        for cfg in combinations_batch:
-            pos = {}; log = [] 
-            bal = initial_capital; alloc = 0.0
-            mask = (l1 >= cfg['l1_thresh']) & (oracle >= cfg['oracle_thresh']) & (sniper >= cfg['sniper_thresh'])
-            
-            for i in range(N):
-                s = sym_id[i]; p = close[i]
-                if s in pos:
-                    entry = pos[s][0]; h_r = pos[s][1]; titan_entry = pos[s][3]
-                    crash_hydra = (h_r > cfg['hydra_thresh']) 
-                    panic_legacy = (legacy_v2[i] > cfg['legacy_thresh']) or (legacy_v3[i] > cfg['legacy_thresh'])
-                    pnl = (p - entry)/entry
-                    
-                    if crash_hydra or panic_legacy or pnl > 0.04 or pnl < -0.02:
-                        realized = pnl - fees_pct*2
-                        bal += pos[s][2] * (1 + realized)
-                        alloc -= pos[s][2]
-                        is_consensus = (titan_entry > 0.55)
-                        log.append({'pnl': realized, 'consensus': is_consensus})
-                        del pos[s]
-                
-                if len(pos) < max_slots and mask[i]:
-                    if s not in pos and bal >= 5.0:
-                        size = min(10.0, bal * 0.98)
-                        pos[s] = (p, hydra[i], size, titan[i])
-                        bal -= size; alloc += size
-            
-            final_bal = bal + alloc
-            profit = final_bal - initial_capital
+        for config in combinations_batch:
+            wallet = { "balance": initial_capital, "allocated": 0.0, "positions": {}, "trades_history": [] }
+            w_titan = config['w_titan']; oracle_thresh = config.get('oracle_thresh', 0.6)
+            sniper_thresh = config.get('sniper_thresh', 0.4); hydra_thresh = config['hydra_thresh']
+            peak_balance = initial_capital; max_drawdown = 0.0
+
+            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]
+                        pos = wallet["positions"][sym]
+                        h_risk = pos.get('risk_hydra_crash', 0)
+                        h_time = pos.get('time_hydra_crash', 0)
+                        is_crash = (h_risk > hydra_thresh) and (h_time > 0) and (ts >= h_time)
+                        pnl = (curr - pos['entry']) / pos['entry']
+                        if is_crash or pnl > 0.04 or pnl < -0.02:
+                            wallet['balance'] += pos['size'] * (1 + pnl - (fees_pct*2))
+                            wallet['allocated'] -= pos['size']
+                            del wallet['positions'][sym]
+                            wallet['trades_history'].append({'pnl': pnl})
+
+                total_eq = wallet['balance'] + wallet['allocated']
+                if total_eq > peak_balance: peak_balance = total_eq
+                dd = (peak_balance - total_eq) / peak_balance
+                if dd > max_drawdown: max_drawdown = dd
+
+                if len(wallet['positions']) < max_slots:
+                    for _, row in group.iterrows():
+                        if row['symbol'] in wallet['positions']: continue
+                        if row['oracle_conf'] < oracle_thresh: continue
+                        if row['sniper_score'] < sniper_thresh: continue
+                        if row['real_titan'] < w_titan: continue # Titan Check
+                        
+                        size = 10.0
+                        if wallet['balance'] >= size:
+                            wallet['positions'][row['symbol']] = {
+                                'entry': row['close'], 'size': size,
+                                'risk_hydra_crash': row['risk_hydra_crash'],
+                                'time_hydra_crash': row['time_hydra_crash']
+                            }
+                            wallet['balance'] -= size
+                            wallet['allocated'] += size
             
-            # --- Detailed Stats Calculation ---
-            tot = len(log)
-            winning_trades = [x for x in log if x['pnl'] > 0]
-            losing_trades = [x for x in log if x['pnl'] <= 0]
+            final_bal = wallet['balance'] + wallet['allocated']
+            net_profit = final_bal - initial_capital
+            trades = wallet['trades_history']
+            total_t = len(trades)
+            win_count = len([t for t in trades if t['pnl'] > 0])
+            loss_count = len([t for t in trades if t['pnl'] <= 0])
+            win_rate = (win_count / total_t * 100) if total_t > 0 else 0
+            max_win = max([t['pnl'] for t in trades]) if trades else 0
+            max_loss = min([t['pnl'] for t in trades]) if trades else 0
             
-            win_count = len(winning_trades)
-            loss_count = len(losing_trades)
-            win_rate = (win_count/tot*100) if tot else 0
-            
-            avg_win = np.mean([x['pnl'] for x in winning_trades]) if winning_trades else 0
-            avg_loss = np.mean([x['pnl'] for x in losing_trades]) if losing_trades else 0
-            
-            gross_profit = sum([x['pnl'] for x in winning_trades])
-            gross_loss = abs(sum([x['pnl'] for x in losing_trades]))
-            profit_factor = (gross_profit / gross_loss) if gross_loss > 0 else 99.9
-
-            # Streaks
-            max_win_s = 0; max_loss_s = 0; curr_w = 0; curr_l = 0
-            for t in log:
+            max_win_streak = 0; max_loss_streak = 0; curr_w = 0; curr_l = 0
+            for t in trades:
                 if t['pnl'] > 0:
                     curr_w += 1; curr_l = 0
-                    if curr_w > max_win_s: max_win_s = curr_w
+                    if curr_w > max_win_streak: max_win_streak = curr_w
                 else:
                     curr_l += 1; curr_w = 0
-                    if curr_l > max_loss_s: max_loss_s = curr_l
-
-            cons_trades = [x for x in log if x['consensus']]
-            n_cons = len(cons_trades)
-            agree_rate = (n_cons/tot*100) if tot else 0
-            cons_win_rate = (sum(1 for x in cons_trades if x['pnl']>0)/n_cons*100) if n_cons else 0
-            cons_avg_pnl = (sum(x['pnl'] for x in cons_trades)/n_cons*100) if n_cons else 0
-
-            res.append({
-                'config': cfg, 'final_balance': final_bal, 'net_profit': profit, 
-                'total_trades': tot, 'win_rate': win_rate, 'max_drawdown': 0,
-                'win_count': win_count, 'loss_count': loss_count,
-                'avg_win': avg_win, 'avg_loss': avg_loss,
-                'max_win_streak': max_win_s, 'max_loss_streak': max_loss_s,
-                'profit_factor': profit_factor,
-                'consensus_agreement_rate': agree_rate,
-                'high_consensus_win_rate': cons_win_rate,
-                'high_consensus_avg_pnl': cons_avg_pnl
+                    if curr_l > max_loss_streak: max_loss_streak = curr_l
+
+            results.append({
+                'config': config, 'final_balance': final_bal, 'net_profit': net_profit,
+                'total_trades': total_t, 'win_count': win_count, 'loss_count': loss_count,
+                'win_rate': win_rate, 'max_single_win': max_win, 'max_single_loss': max_loss,
+                'max_drawdown': max_drawdown * 100
             })
-        return res
+            
+        return results
 
     async def run_optimization(self, target_regime="RANGE"):
         await self.generate_truth_data()
-        oracle_r = np.linspace(0.3, 0.7, 3); sniper_r = np.linspace(0.2, 0.6, 3)
-        hydra_r = [0.8, 0.9]; l1_r = [5.0, 10.0]
+        oracle_r = np.linspace(0.4, 0.7, 3); sniper_r = np.linspace(0.4, 0.7, 3)
+        hydra_r = [0.85, 0.95]
         
         combos = []
-        for o, s, h, l1 in itertools.product(oracle_r, sniper_r, hydra_r, l1_r):
+        for o, s, h in itertools.product(oracle_r, sniper_r, hydra_r):
             combos.append({
-                'w_titan': 0.4, 'w_struct': 0.3, 'thresh': l1, 'l1_thresh': l1, 
+                'w_titan': 0.5, 'w_struct': 0.3, 'thresh': 0.5, 'l1_thresh': 50.0,
                 'oracle_thresh': o, 'sniper_thresh': s, 'hydra_thresh': h, 'legacy_thresh': 0.95
             })
             
@@ -717,18 +661,6 @@ class HeavyDutyBacktester:
         results_list.sort(key=lambda x: x['net_profit'], reverse=True)
         best = results_list[0]
 
-        # --- AUTO-DIAGNOSIS LOGIC ---
-        diag = []
-        if best['total_trades'] > 2000 and best['net_profit'] < 10:
-            diag.append("⚠️ Overtrading: Too many trades for low profit.")
-        if best['win_rate'] > 55 and best['net_profit'] < 0:
-            diag.append("⚠️ Fee Burn: High win rate but fees are eating profits.")
-        if abs(best['avg_loss']) > best['avg_win']:
-            diag.append("⚠️ Risk/Reward Inversion: Avg Loss > Avg Win.")
-        if best['max_loss_streak'] > 10:
-            diag.append("⚠️ Consecutive Loss Risk: Strategy prone to streak failures.")
-        if not diag: diag.append("✅ System Healthy.")
-
         print("\n" + "="*60)
         print(f"🏆 CHAMPION REPORT [{target_regime}]:")
         print(f"   💰 Final Balance:   ${best['final_balance']:,.2f}")
@@ -736,22 +668,13 @@ class HeavyDutyBacktester:
         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['consensus_agreement_rate']:.1f}%")
-        print(f"   🌟 High-Consensus Win Rate:  {best['high_consensus_win_rate']:.1f}%")
         print("-" * 60)
-        print(f"   🩺 DIAGNOSIS: {' '.join(diag)}")
         print(f"   ⚙️ Oracle={best['config']['oracle_thresh']:.2f} | Sniper={best['config']['sniper_thresh']:.2f} | Hydra={best['config']['hydra_thresh']:.2f}")
         print("="*60)
         return best['config'], best
 
 async def run_strategic_optimization_task():
-    print("\n🧪 [STRATEGIC BACKTEST] Full Spectrum Mode...")
+    print("\n🧪 [STRATEGIC BACKTEST] Feature Parity Mode...")
     r2 = R2Service(); dm = DataManager(None, None, r2); proc = MLProcessor(dm)
     try:
         await dm.initialize(); await proc.initialize()