Update ml_engine/processor.py

ml_engine/processor.py

@@ -1,393 +1,724 @@
 # ============================================================
-# 🧠 ml_engine/processor.py (V36.0 - GEM-Architect: The Cybernetic Processor)
+# 🧪 backtest_engine.py (V145.0 - GEM-Architect: Patterns Integrated)
 # ============================================================
 
 import asyncio
-import traceback
+import pandas as pd
+import numpy as np
+import pandas_ta as ta
+import time
 import logging
+import itertools 
 import os
+import glob
+import gc 
 import sys
-import numpy as np
-from typing import Dict, Any, List, Optional
-
-# --- استيراد المحركات (كما هي) ---
-try: from .titan_engine import TitanEngine
-except ImportError: TitanEngine = None
-try: from .patterns import ChartPatternAnalyzer
-except ImportError: ChartPatternAnalyzer = None
-try: from .monte_carlo import MonteCarloEngine
-except ImportError: MonteCarloEngine = None
-try: from .oracle_engine import OracleEngine
-except ImportError: OracleEngine = None
-try: from .sniper_engine import SniperEngine
-except ImportError: SniperEngine = None
-try: from .hybrid_guardian import HybridDeepSteward
-except ImportError: HybridDeepSteward = None
-try: from .guardian_hydra import GuardianHydra
-except ImportError: GuardianHydra = None
+import traceback
+from numpy.lib.stride_tricks import sliding_window_view 
+from datetime import datetime, timezone
+from typing import Dict, Any, List
+from scipy.special import expit
 
-# ============================================================
-# 📂 مسارات النماذج
-# ============================================================
-BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-MODELS_L2_DIR = os.path.join(BASE_DIR, "ml_models", "layer2")
-MODELS_PATTERN_DIR = os.path.join(BASE_DIR, "ml_models", "xgboost_pattern2")
-MODELS_UNIFIED_DIR = os.path.join(BASE_DIR, "ml_models", "Unified_Models_V1")
-MODELS_SNIPER_DIR = os.path.join(BASE_DIR, "ml_models", "guard_v2") 
-MODELS_HYDRA_DIR = os.path.join(BASE_DIR, "ml_models", "guard_v1")
-MODEL_V2_PATH = os.path.join(BASE_DIR, "ml_models", "DeepSteward_V2_Production.json")
-MODEL_V3_PATH = os.path.join(BASE_DIR, "ml_models", "DeepSteward_V3_Production.json")
-MODEL_V3_FEAT = os.path.join(BASE_DIR, "ml_models", "DeepSteward_V3_Features.json")
+try:
+    from ml_engine.processor import MLProcessor, SystemLimits 
+    from ml_engine.data_manager import DataManager
+    from learning_hub.adaptive_hub import StrategyDNA, AdaptiveHub
+    from r2 import R2Service
+    import ccxt.async_support as ccxt 
+    import xgboost as xgb 
+    import lightgbm as lgb 
+except ImportError:
+    pass
+
+logging.getLogger('ml_engine').setLevel(logging.WARNING)
+CACHE_DIR = "backtest_real_scores"
 
 # ============================================================
-# 🎛️ SYSTEM LIMITS & THRESHOLDS (Dynamic DNA Container)
+# 🛡️ GLOBAL HELPERS
 # ============================================================
-class SystemLimits:
-    """
-    GEM-Architect: The Dynamic Constitution.
-    يتم تحديث هذه القيم آلياً بواسطة AdaptiveHub بناءً على حالة السوق (Bull/Bear/etc).
-    """
-    
-    # --- Layer 1 (Data Manager Control) ---
-    L1_MIN_AFFINITY_SCORE = 15.0  # سيتم الكتابة عليها من الـ DNA
-    
-    # --- Layer 2 Weights (Dynamic) ---
-    # هذه الأوزان تتغير حسب أداء النماذج (Tactical Loop) وحالة السوق
-    L2_WEIGHT_TITAN    = 0.40
-    L2_WEIGHT_PATTERNS = 0.30
-    L2_WEIGHT_MC       = 0.10
-    # (Hydra/Sniper قد يساهمون في L2 أو L4 حسب التصميم)
-
-    # إعدادات الأنماط (تتغير حسب الاستراتيجية)
-    PATTERN_TF_WEIGHTS = {'15m': 0.40, '1h': 0.30, '5m': 0.20, '4h': 0.10, '1d': 0.00}
-    PATTERN_THRESH_BULLISH = 0.60
-    PATTERN_THRESH_BEARISH = 0.40
-
-    # --- Layer 3 (Oracle) ---
-    L3_CONFIDENCE_THRESHOLD = 0.65
-    L3_WHALE_IMPACT_MAX = 0.10
-    L3_NEWS_IMPACT_MAX  = 0.05
-    L3_MC_ADVANCED_MAX  = 0.10
-
-    # --- Layer 4 (Sniper & Execution) ---
-    L4_ENTRY_THRESHOLD  = 0.40
-    # أوزان داخلية لـ Sniper (ML vs OrderBook)
-    L4_WEIGHT_ML        = 0.60
-    L4_WEIGHT_OB        = 0.40
-    # نسبة الجدار المسموح بها (تتغير جذرياً بين Bull و Bear)
-    L4_OB_WALL_RATIO    = 0.40
-
-    # --- Layer 0: Hydra & Guardian Thresholds ---
-    HYDRA_CRASH_THRESH      = 0.60
-    HYDRA_GIVEBACK_THRESH   = 0.70
-    HYDRA_STAGNATION_THRESH = 0.50
-
-    # Legacy Guard Thresholds
-    LEGACY_V2_PANIC_THRESH = 0.95
-    LEGACY_V3_HARD_THRESH  = 0.95
-    LEGACY_V3_SOFT_THRESH  = 0.85
-    LEGACY_V3_ULTRA_THRESH = 0.98
-
-    @classmethod
-    def to_dict(cls) -> Dict[str, Any]:
-        return {k: v for k, v in cls.__dict__.items() if not k.startswith('__') and not callable(v)}
-
-    @classmethod
-    def update_from_dict(cls, config: Dict[str, Any]):
-        """تحديث القيم من AdaptiveHub"""
-        if not config: return
-        for k, v in config.items():
-            if hasattr(cls, k): 
-                setattr(cls, k, v)
-        # print(f"🔄 [SystemLimits] Updated. TitanW={cls.L2_WEIGHT_TITAN:.2f}, WallRatio={cls.L4_OB_WALL_RATIO}")
+def sanitize_features(df):
+    if df is None or df.empty: return df
+    return df.replace([np.inf, -np.inf], np.nan).ffill().bfill().fillna(0.0)
+
+def _z_roll(x, w=500):
+    if not isinstance(x, pd.Series): x = pd.Series(x)
+    r = x.rolling(w).mean()
+    s = x.rolling(w).std().replace(0, np.nan)
+    return ((x - r) / s).fillna(0)
+
+def _revive_score_distribution(scores):
+    scores = np.array(scores, dtype=np.float32).flatten()
+    if len(scores) < 10: return scores
+    s_min, s_max = np.min(scores), np.max(scores)
+    std = np.std(scores)
+    if std < 0.001: return scores 
+    if s_max < 0.8 or s_min > 0.2:
+        return (scores - s_min) / (s_max - s_min + 1e-9)
+    return scores
+
+def safe_ta(ind_output, index, fill_method='smart'):
+    if ind_output is None: 
+        return pd.Series(0.0, index=index, dtype='float64')
+    if not isinstance(ind_output, pd.Series):
+        s = pd.Series(ind_output, index=index)
+    else:
+        s = ind_output
+    s = s.bfill().ffill()
+    return s.fillna(0.0).astype('float64')
+
+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 _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)
+        X_seq = np.concatenate([base_z, extra_z, _zv(indicators)], 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
 
 # ============================================================
-# 🧠 MLProcessor Class
+# 🧪 THE BACKTESTER CLASS
 # ============================================================
-class MLProcessor:
-    def __init__(self, data_manager=None):
-        self.data_manager = data_manager
-        self.initialized = False
-        
-        self.titan = TitanEngine(model_dir=MODELS_L2_DIR) if TitanEngine else None
-        self.pattern_engine = ChartPatternAnalyzer(models_dir=MODELS_PATTERN_DIR) if ChartPatternAnalyzer else None
-        self.mc_analyzer = MonteCarloEngine() if MonteCarloEngine else None
-        self.oracle = OracleEngine(model_dir=MODELS_UNIFIED_DIR) if OracleEngine else None
-        self.sniper = SniperEngine(models_dir=MODELS_SNIPER_DIR) if SniperEngine else None
+class HeavyDutyBacktester:
+    def __init__(self, data_manager, processor):
+        self.dm = data_manager
+        self.proc = processor
+        self.GRID_DENSITY = 5 
+        self.INITIAL_CAPITAL = 10.0
+        self.TRADING_FEES = 0.001 
+        self.MAX_SLOTS = 4
         
-        self.guardian_hydra = None
-        if GuardianHydra:
-            self.guardian_hydra = GuardianHydra(model_dir=MODELS_HYDRA_DIR)
-
-        self.guardian_legacy = None
-        if HybridDeepSteward:
-            self.guardian_legacy = HybridDeepSteward(
-                v2_model_path=MODEL_V2_PATH,
-                v3_model_path=MODEL_V3_PATH,
-                v3_features_map_path=MODEL_V3_FEAT
-            )
-
-        print(f"🧠 [MLProcessor V36.0] Cybernetic Control Active.")
-
-    async def initialize(self):
-        if self.initialized: return
-        print("⚙️ [Processor] Initializing Neural Grid...")
-        try:
-            tasks = []
-            if self.titan: tasks.append(self.titan.initialize())
-            
-            if self.pattern_engine: 
-                # التكوين الأولي (سيتم تحديثه لاحقاً ديناميكياً)
-                self.pattern_engine.configure_thresholds(
-                    weights=SystemLimits.PATTERN_TF_WEIGHTS, 
-                    bull_thresh=SystemLimits.PATTERN_THRESH_BULLISH, 
-                    bear_thresh=SystemLimits.PATTERN_THRESH_BEARISH
-                )
-                tasks.append(self.pattern_engine.initialize())
-            
-            if self.oracle: 
-                if hasattr(self.oracle, 'set_threshold'): 
-                    self.oracle.set_threshold(SystemLimits.L3_CONFIDENCE_THRESHOLD)
-                tasks.append(self.oracle.initialize())
+        # ==============================================================================
+        # 🎛️ CONTROL PANEL: FULL GRANULARITY (Including Patterns)
+        # ==============================================================================
+        self.GRID_RANGES = {
+            # --- Entry Models ---
+            'TITAN':  [0.10, 0.50],                     
+            'ORACLE': np.linspace(0.4, 0.7, 3),   
+            'SNIPER': np.linspace(0.1, 0.7, 3),   
+            'PATTERN': [0.10, 0.50], # ✅ ADDED: Pattern Threshold
+            'L1_SCORE': [10.0],
             
-            if self.sniper: 
-                # التكوين الأولي
-                if hasattr(self.sniper, 'configure_settings'): 
-                    self.sniper.configure_settings(
-                        threshold=SystemLimits.L4_ENTRY_THRESHOLD, 
-                        wall_ratio=SystemLimits.L4_OB_WALL_RATIO,
-                        w_ml=SystemLimits.L4_WEIGHT_ML,
-                        w_ob=SystemLimits.L4_WEIGHT_OB
-                    )
-                tasks.append(self.sniper.initialize())
+            # --- Hydra Heads (Exit) ---
+            'HYDRA_CRASH':    [0.50, 0.80], 
+            'HYDRA_GIVEBACK': [0.20, 0.80],       
             
-            if tasks: await asyncio.gather(*tasks)
+            # --- Legacy Heads (Exit) ---
+            'LEGACY_V2': [0.50, 0.95],            
+            'LEGACY_V3': [0.50, 0.95]             
+        }
+        # ==============================================================================
+        
+        self.TARGET_COINS = [
+            'SOL/USDT', 'XRP/USDT', 'DOGE/USDT', 'ADA/USDT', 'AVAX/USDT', 'LINK/USDT', 
+            'TON/USDT', 'INJ/USDT', 'APT/USDT', 'OP/USDT', 'ARB/USDT', 'SUI/USDT', 
+            'SEI/USDT', 'TIA/USDT', 'MATIC/USDT', 'NEAR/USDT', 'RUNE/USDT', 'PYTH/USDT', 
+            'WIF/USDT', 'PEPE/USDT', 'SHIB/USDT', 'TRX/USDT', 'DOT/USDT', 'UNI/USDT', 
+            'ONDO/USDT', 'ENA/USDT', 'HBAR/USDT', 'XLM/USDT', 'TAO/USDT', 'ZK/USDT', 
+            'ZRO/USDT', 'KCS/USDT', 'ICP/USDT', 'SAND/USDT', 'AXS/USDT', 'APE/USDT', 
+            'GMT/USDT', 'CHZ/USDT', 'CFX/USDT', 'LDO/USDT', 'FET/USDT', 'JTO/USDT', 
+            '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 not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
+        print(f"🧪 [Backtest V145.0] Full Stack (Titan + Patterns + Oracle + Sniper + Hydra + Legacy).")
 
-            if self.guardian_hydra: 
-                self.guardian_hydra.initialize()
-                print("   🛡️ [Guard 1] Hydra X-Ray: Active")
-            
-            if self.guardian_legacy:
-                if asyncio.iscoroutinefunction(self.guardian_legacy.initialize):
-                    await self.guardian_legacy.initialize()
-                else:
-                    self.guardian_legacy.initialize()
-                
-                # تطبيق العتبات المبدئية
-                self.guardian_legacy.configure_thresholds(
-                    v2_panic=SystemLimits.LEGACY_V2_PANIC_THRESH,
-                    v3_hard=SystemLimits.LEGACY_V3_HARD_THRESH,
-                    v3_soft=SystemLimits.LEGACY_V3_SOFT_THRESH,
-                    v3_ultra=SystemLimits.LEGACY_V3_ULTRA_THRESH
-                )
-                print(f"   🛡️ [Guard 2] Legacy Steward: Active")
-
-            self.initialized = True
-            print("✅ [Processor] All Systems Operational.")
-            
-        except Exception as e:
-            print(f"❌ [Processor FATAL] Init failed: {e}")
-            traceback.print_exc()
-
-    async def process_compound_signal(self, raw_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
-        """
-        L2 Processing:
-        هنا يتم دمج الدرجات بناءً على الأوزان الديناميكية الحالية (SystemLimits).
-        """
-        if not self.initialized: await self.initialize()
+    def set_date_range(self, start_str, end_str):
+        self.force_start_date = start_str
+        self.force_end_date = end_str
+
+    async def _fetch_all_data_fast(self, sym, start_ms, end_ms):
+        print(f"   ⚡ [Network] Downloading {sym}...", flush=True)
+        limit = 1000
+        tasks = []
+        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(20) 
+
+        async def _fetch_batch(timestamp):
+            async with sem:
+                for _ in range(3): 
+                    try:
+                        return await self.dm.exchange.fetch_ohlcv(sym, '1m', since=timestamp, limit=limit)
+                    except: await asyncio.sleep(0.5)
+                return []
+
+        chunk_size = 50
+        for i in range(0, len(tasks), chunk_size):
+            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()
+
+    # ==============================================================
+    # 🏎️ VECTORIZED INDICATORS
+    # ==============================================================
+    def _calculate_indicators_vectorized(self, df, timeframe='1m'):
+        cols = ['close', 'high', 'low', 'volume', 'open']
+        for c in cols: df[c] = df[c].astype(np.float64)
+        idx = df.index
         
-        symbol = raw_data.get('symbol')
-        ohlcv_data = raw_data.get('ohlcv')
-        current_price = raw_data.get('current_price', 0.0)
+        df['RSI'] = safe_ta(ta.rsi(df['close'], length=14), idx, 50)
+        macd = ta.macd(df['close'])
+        if macd is not None:
+            df['MACD'] = safe_ta(macd.iloc[:, 0], idx, 0)
+            df['MACD_h'] = safe_ta(macd.iloc[:, 1], idx, 0)
+        else: df['MACD'] = 0.0; df['MACD_h'] = 0.0
+        df['CCI'] = safe_ta(ta.cci(df['high'], df['low'], df['close'], length=20), idx, 0)
+        adx = ta.adx(df['high'], df['low'], df['close'], length=14)
+        if adx is not None: df['ADX'] = safe_ta(adx.iloc[:, 0], idx, 0)
+        else: df['ADX'] = 0.0
+        if timeframe == '1d': df['Trend_Strong'] = np.where(df['ADX'] > 25, 1.0, 0.0)
+        for p in [9, 21, 50, 200]:
+            ema = safe_ta(ta.ema(df['close'], length=p), idx, 0)
+            df[f'EMA_{p}_dist'] = ((df['close'] / ema.replace(0, np.nan)) - 1).fillna(0)
+            df[f'ema{p}'] = ema 
+        df['ema20'] = safe_ta(ta.ema(df['close'], length=20), idx, df['close'])
+        bb = ta.bbands(df['close'], length=20, std=2.0)
+        if bb is not None:
+            w = ((bb.iloc[:, 2] - bb.iloc[:, 0]) / bb.iloc[:, 1].replace(0, np.nan)).fillna(0)
+            p = ((df['close'] - bb.iloc[:, 0]) / (bb.iloc[:, 2] - bb.iloc[:, 0]).replace(0, np.nan)).fillna(0)
+            df['BB_w'] = w; df['BB_p'] = p; df['bb_width'] = w
+        else: df['BB_w'] = 0; df['BB_p'] = 0; df['bb_width'] = 0
+        df['MFI'] = safe_ta(ta.mfi(df['high'], df['low'], df['close'], df['volume'], length=14), idx, 50)
+        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).fillna(0)
+            df['vwap'] = vwap
+        else: df['VWAP_dist'] = 0.0; df['vwap'] = df['close']
+        df['atr'] = safe_ta(ta.atr(df['high'], df['low'], df['close'], length=14), idx, 0)
+        df['atr_pct'] = (df['atr'] / df['close'].replace(0, np.nan)).fillna(0)
+        df['ATR_pct'] = df['atr_pct'] 
+
+        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']
+            e9 = df['ema9'].replace(0, np.nan)
+            df['ema_9_slope'] = ((df['ema9'] - df['ema9'].shift(1)) / e9.shift(1)).fillna(0)
+            df['ema_21_dist'] = df['EMA_21_dist']
+            atr_100 = safe_ta(ta.atr(df['high'], df['low'], df['close'], length=100), idx, 0)
+            df['atr_z'] = _z_roll(atr_100)
+            df['vol_zscore_50'] = _z_roll(df['volume'], 50)
+            rng = (df['high'] - df['low']).replace(0, 1e-9)
+            df['candle_range'] = _z_roll(rng, 500)
+            df['close_pos_in_range'] = ((df['close'] - df['low']) / rng).fillna(0.5)
+            df['dollar_vol'] = df['close'] * df['volume']
+            amihud_raw = (df['return_1m'].abs() / df['dollar_vol'].replace(0, np.nan)).fillna(0)
+            df['amihud'] = _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))).fillna(0)
+            df['roll_spread'] = _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().fillna(0)
+            df['ofi'] = _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().replace(0, np.nan)
+            df['vpin'] = (imb / tot).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'] = _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'] = _z_roll(rv_gk)
+            df['L_score'] = (df['vol_zscore_50'] - df['amihud'] - df['roll_spread'] - df['rv_gk'].abs() - df['vwap_dev'].abs() + df['ofi']).fillna(0)
+
+        df['slope'] = safe_ta(ta.slope(df['close'], length=7), idx, 0)
+        vol_mean = df['volume'].rolling(20).mean()
+        vol_std = df['volume'].rolling(20).std().replace(0, np.nan)
+        df['vol_z'] = ((df['volume'] - vol_mean) / vol_std).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).replace(0, np.nan)).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)
+        e20_s = df['ema20'].shift(5).replace(0, np.nan)
+        df['trend_slope'] = ((df['ema20'] - df['ema20'].shift(5)) / e20_s).fillna(0)
+        df['volatility'] = (df['atr'] / df['close'].replace(0, np.nan)).fillna(0)
+        fib618 = roll_max - (diff * 0.382)
+        df['dist_fib618'] = ((df['close'] - fib618) / df['close'].replace(0, np.nan)).fillna(0)
+        df['dist_ema50'] = ((df['close'] - df['ema50']) / df['ema50'].replace(0, np.nan)).fillna(0)
+        e200 = safe_ta(ta.ema(df['close'], length=200), idx, df['close'])
+        df['ema200'] = e200
+        df['dist_ema200'] = ((df['close'] - e200) / e200.replace(0, np.nan)).fillna(0)
+        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
+    # ==============================================================
+    async def _process_data_in_memory(self, sym, candles, start_ms, end_ms):
+        safe_sym = sym.replace('/', '_')
+        period_suffix = f"{start_ms}_{end_ms}"
+        scores_file = f"{CACHE_DIR}/{safe_sym}_{period_suffix}_scores.pkl"
+        if os.path.exists(scores_file):
+             print(f"   📂 [{sym}] Data Exists -> Skipping.")
+             return
+
+        print(f"   ⚙️ [CPU] Analyzing {sym}...", flush=True)
+        t0 = time.time()
+
+        df_1m = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+        df_1m['datetime'] = pd.to_datetime(df_1m['timestamp'], unit='ms')
+        df_1m.set_index('datetime', inplace=True)
+        df_1m = df_1m.sort_index()
+
+        frames = {}
+        frames['1m'] = self._calculate_indicators_vectorized(df_1m.copy(), timeframe='1m')
+        frames['1m']['timestamp'] = frames['1m'].index.floor('1min').astype(np.int64) // 10**6
+        fast_1m = {col: frames['1m'][col].values for col in frames['1m'].columns}
         
-        if not symbol or not ohlcv_data: return None
+        agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
+        numpy_htf = {} 
+        for tf_str, tf_code in [('5m', '5T'), ('15m', '15T'), ('1h', '1h'), ('4h', '4h'), ('1d', '1D')]:
+            resampled = df_1m.resample(tf_code).agg(agg_dict).dropna()
+            resampled = self._calculate_indicators_vectorized(resampled, timeframe=tf_str)
+            resampled['timestamp'] = resampled.index.astype(np.int64) // 10**6
+            frames[tf_str] = resampled
+            numpy_htf[tf_str] = {col: resampled[col].values for col in resampled.columns}
+
+        arr_ts_1m = fast_1m['timestamp']
+        map_5m = np.clip(np.searchsorted(numpy_htf['5m']['timestamp'], arr_ts_1m), 0, len(numpy_htf['5m']['timestamp']) - 1)
+        map_15m = np.clip(np.searchsorted(numpy_htf['15m']['timestamp'], arr_ts_1m), 0, len(numpy_htf['15m']['timestamp']) - 1)
+        map_1h = np.clip(np.searchsorted(numpy_htf['1h']['timestamp'], arr_ts_1m), 0, len(numpy_htf['1h']['timestamp']) - 1)
+        map_4h = np.clip(np.searchsorted(numpy_htf['4h']['timestamp'], arr_ts_1m), 0, len(numpy_htf['4h']['timestamp']) - 1)
+        map_1d = np.clip(np.searchsorted(numpy_htf['1d']['timestamp'], arr_ts_1m), 0, len(numpy_htf['1d']['timestamp']) - 1) if '1d' in numpy_htf else np.zeros(len(arr_ts_1m), dtype=int)
+
+        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)
+        legacy_v3 = getattr(self.proc.guardian_legacy, 'model_v3', None)
+        v3_feat_names = getattr(self.proc.guardian_legacy, 'v3_feature_names', [])
+        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)
+        pattern_models = getattr(self.proc.pattern_engine, 'models', {})
         
-        try:
-            # 1. Titan Engine
-            score_titan = 0.5
-            titan_res = {}
-            if self.titan:
-                titan_res = await asyncio.to_thread(self.titan.predict, ohlcv_data)
-                score_titan = titan_res.get('score', 0.5)
-
-            # 2. Pattern Engine (Inject Dynamic Config First)
-            score_patterns = 0.5
-            pattern_res = {}
-            pattern_name = "Neutral"
-            if self.pattern_engine:
-                # تحديث التكوين قبل التحليل لضمان استخدام أحدث أوزان الاستراتيجية
-                self.pattern_engine.configure_thresholds(
-                    weights=SystemLimits.PATTERN_TF_WEIGHTS,
-                    bull_thresh=SystemLimits.PATTERN_THRESH_BULLISH,
-                    bear_thresh=SystemLimits.PATTERN_THRESH_BEARISH
-                )
-                pattern_res = await self.pattern_engine.detect_chart_patterns(ohlcv_data)
-                score_patterns = pattern_res.get('pattern_confidence', 0.5)
-                pattern_name = pattern_res.get('pattern_detected', 'Neutral')
-
-            # 3. Monte Carlo (Light)
-            mc_score = 0.5
-            if self.mc_analyzer and '1h' in ohlcv_data:
-                closes = [c[4] for c in ohlcv_data['1h']]
-                raw_mc = self.mc_analyzer.run_light_check(closes)
-                mc_score = 0.5 + (raw_mc * 5.0)
-                mc_score = max(0.0, min(1.0, mc_score))
-
-            # 4. Hybrid Calculation (Using Dynamic Weights from SystemLimits)
-            # هذه الأوزان يتم تحديثها بواسطة AdaptiveHub
-            w_titan = SystemLimits.L2_WEIGHT_TITAN
-            w_patt = SystemLimits.L2_WEIGHT_PATTERNS
-            w_mc = SystemLimits.L2_WEIGHT_MC
-            
-            # تطبيع الأوزان (لضمان أن مجموعها 1.0 تقريباً)
-            total_w = w_titan + w_patt + w_mc
-            if total_w <= 0: total_w = 1.0
-            
-            hybrid_score = ((score_titan * w_titan) + (score_patterns * w_patt) + (mc_score * w_mc)) / total_w
-            
-            return {
-                'symbol': symbol, 
-                'current_price': current_price, 
-                'enhanced_final_score': hybrid_score, 
-                'titan_score': score_titan, 
-                'patterns_score': score_patterns, 
-                'mc_score': mc_score, 
-                'components': {
-                    'titan_score': score_titan, 
-                    'patterns_score': score_patterns, 
-                    'mc_score': mc_score
-                }, 
-                'pattern_name': pattern_name, 
-                'ohlcv': ohlcv_data, 
-                'titan_details': titan_res, 
-                'pattern_details': pattern_res.get('details', {})
-            }
-        except Exception as e: 
-            print(f"❌ [Processor] Error processing {symbol}: {e}")
-            return None
-
-    async def consult_oracle(self, symbol_data: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        L3 Processing:
-        Oracle يستخدم العتبة الحالية من SystemLimits.
-        """
-        if not self.initialized: await self.initialize()
+        # A. TITAN
+        global_titan_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
+        if titan_model:
+            titan_cols = [
+                '5m_open', '5m_high', '5m_low', '5m_close', '5m_volume', '5m_RSI', '5m_MACD', '5m_MACD_h', 
+                '5m_CCI', '5m_ADX', '5m_EMA_9_dist', '5m_EMA_21_dist', '5m_EMA_50_dist', '5m_EMA_200_dist', 
+                '5m_BB_w', '5m_BB_p', '5m_MFI', '5m_VWAP_dist', '15m_timestamp', '15m_RSI', '15m_MACD', 
+                '15m_MACD_h', '15m_CCI', '15m_ADX', '15m_EMA_9_dist', '15m_EMA_21_dist', '15m_EMA_50_dist', 
+                '15m_EMA_200_dist', '15m_BB_w', '15m_BB_p', '15m_MFI', '15m_VWAP_dist', '1h_timestamp', 
+                '1h_RSI', '1h_MACD_h', '1h_EMA_50_dist', '1h_EMA_200_dist', '1h_ATR_pct', '4h_timestamp', 
+                '4h_RSI', '4h_MACD_h', '4h_EMA_50_dist', '4h_EMA_200_dist', '4h_ATR_pct', '1d_timestamp', 
+                '1d_RSI', '1d_EMA_200_dist', '1d_Trend_Strong'
+            ]
+            print("     🚀 Running Global Titan...", flush=True)
+            try:
+                t_vecs = []
+                for col in titan_cols:
+                    parts = col.split('_', 1); tf = parts[0]; feat = parts[1]
+                    target_arr = numpy_htf.get(tf, None)
+                    target_map = locals().get(f"map_{tf}", None)
+                    if target_arr and feat in target_arr: t_vecs.append(target_arr[feat][target_map])
+                    elif target_arr and feat == 'timestamp': t_vecs.append(target_arr['timestamp'][target_map])
+                    elif target_arr and feat in ['open','high','low','close','volume']: t_vecs.append(target_arr[feat][target_map])
+                    else: t_vecs.append(np.zeros(len(arr_ts_1m)))
+                X_TITAN = np.column_stack(t_vecs)
+                global_titan_scores = _revive_score_distribution(titan_model.predict(xgb.DMatrix(X_TITAN, feature_names=titan_cols)))
+            except: pass
+
+        # B. SNIPER
+        global_sniper_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
+        if sniper_models:
+            print("     🚀 Running Global Sniper...", flush=True)
+            try:
+                s_vecs = []
+                for col in sniper_cols:
+                    if col in fast_1m: s_vecs.append(fast_1m[col])
+                    elif col == 'atr' and 'atr_z' in fast_1m: s_vecs.append(fast_1m['atr_z'])
+                    else: s_vecs.append(np.zeros(len(arr_ts_1m)))
+                X_SNIPER = np.column_stack(s_vecs)
+                preds = [m.predict(X_SNIPER) for m in sniper_models]
+                global_sniper_scores = _revive_score_distribution(np.mean(preds, axis=0))
+            except: pass
+
+        # C. ORACLE
+        global_oracle_scores = np.full(len(arr_ts_1m), 0.5, dtype=np.float32)
+        if oracle_dir:
+            print("     🚀 Running Global Oracle...", flush=True)
+            try:
+                o_vecs = []
+                for col in oracle_cols:
+                    if col.startswith('1h_'): o_vecs.append(numpy_htf['1h'].get(col[3:], np.zeros(len(arr_ts_1m)))[map_1h])
+                    elif col.startswith('15m_'): o_vecs.append(numpy_htf['15m'].get(col[4:], np.zeros(len(arr_ts_1m)))[map_15m])
+                    elif col.startswith('4h_'): o_vecs.append(numpy_htf['4h'].get(col[3:], np.zeros(len(arr_ts_1m)))[map_4h])
+                    elif col == 'sim_titan_score': o_vecs.append(global_titan_scores)
+                    elif col == 'sim_mc_score': o_vecs.append(np.full(len(arr_ts_1m), 0.5))
+                    elif col == 'sim_pattern_score': o_vecs.append(np.full(len(arr_ts_1m), 0.5))
+                    else: o_vecs.append(np.zeros(len(arr_ts_1m)))
+                X_ORACLE = np.column_stack(o_vecs)
+                preds_o = oracle_dir.predict(X_ORACLE)
+                preds_o = preds_o if isinstance(preds_o, np.ndarray) and len(preds_o.shape)==1 else preds_o[:, 0]
+                global_oracle_scores = _revive_score_distribution(preds_o)
+            except: pass
+
+        # D. LEGACY V2 & V3
+        global_v2_scores = np.zeros(len(arr_ts_1m), dtype=np.float32)
+        global_v3_scores = np.zeros(len(arr_ts_1m), dtype=np.float32)
         
-        if self.oracle:
-            # تحديث العتبة ديناميكياً
-            if hasattr(self.oracle, 'set_threshold'):
-                self.oracle.set_threshold(SystemLimits.L3_CONFIDENCE_THRESHOLD)
-                
-            decision = await self.oracle.predict(symbol_data)
-            conf = decision.get('confidence', 0.0)
-            
-            # طبقة أمان إضافية (Redundant Safety)
-            if decision.get('action') in ['WATCH', 'BUY'] and conf < SystemLimits.L3_CONFIDENCE_THRESHOLD:
-                decision['action'] = 'WAIT'
-                decision['reason'] = f"Processor Veto: Conf {conf:.2f} < Limit {SystemLimits.L3_CONFIDENCE_THRESHOLD}"
+        if legacy_v2:
+            try:
+                l_log = fast_1m['log_ret']; l_rsi = fast_1m['RSI'] / 100.0; l_fib = fast_1m['fib_pos']; l_vol = fast_1m['volatility']
+                l5_log = numpy_htf['5m']['log_ret'][map_5m]; l5_rsi = numpy_htf['5m']['RSI'][map_5m] / 100.0; l5_fib = numpy_htf['5m']['fib_pos'][map_5m]; l5_trd = numpy_htf['5m']['trend_slope'][map_5m]
+                l15_log = numpy_htf['15m']['log_ret'][map_15m]; l15_rsi = numpy_htf['15m']['RSI'][map_15m] / 100.0; l15_fib618 = numpy_htf['15m']['dist_fib618'][map_15m]; l15_trd = numpy_htf['15m']['trend_slope'][map_15m]
+                lags = []
+                for lag in [1, 2, 3, 5, 10, 20]:
+                    lags.extend([fast_1m[f'log_ret_lag_{lag}'], fast_1m[f'rsi_lag_{lag}'], fast_1m[f'fib_pos_lag_{lag}'], fast_1m[f'volatility_lag_{lag}']])
+                X_V2 = np.column_stack([l_log, l_rsi, l_fib, l_vol, l5_log, l5_rsi, l5_fib, l5_trd, l15_log, l15_rsi, l15_fib618, l15_trd, *lags])
+                preds = legacy_v2.predict(xgb.DMatrix(X_V2))
+                global_v2_scores = preds[:, 2] if len(preds.shape) > 1 else preds
+                global_v2_scores = global_v2_scores.flatten() 
+            except: pass
             
-            return decision
-        return {'action': 'WAIT', 'reason': 'Oracle Engine Missing'}
-
-    async def check_sniper_entry(self, ohlcv_1m_data: List, order_book_data: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        L4 Processing:
-        Sniper يستلم إعدادات دفتر الطلبات (DNA) قبل اتخاذ القرار.
-        """
-        if not self.initialized: await self.initialize()
+        if legacy_v3 and v3_feat_names:
+            try:
+                v3_dict = {}
+                v3_dict['rsi'] = fast_1m['RSI']; v3_dict['dist_ema50'] = fast_1m['dist_ema50']
+                v3_dict['dist_ema200'] = fast_1m['dist_ema200']; v3_dict['log_ret'] = fast_1m['log_ret']
+                v3_dict['rsi_5m'] = numpy_htf['5m']['RSI'][map_5m]; v3_dict['dist_ema50_5m'] = numpy_htf['5m']['dist_ema50'][map_5m]
+                v3_dict['dist_ema200_5m'] = numpy_htf['5m']['dist_ema200'][map_5m]; v3_dict['log_ret_5m'] = numpy_htf['5m']['log_ret'][map_5m]
+                v3_dict['rsi_15m'] = numpy_htf['15m']['RSI'][map_15m]; v3_dict['dist_ema50_15m'] = numpy_htf['15m']['dist_ema50'][map_15m]
+                v3_dict['dist_ema200_15m'] = numpy_htf['15m']['dist_ema200'][map_15m]; v3_dict['log_ret_15m'] = numpy_htf['15m']['log_ret'][map_15m]
+                df_v3_data = {k: v3_dict.get(k, np.zeros(len(arr_ts_1m))) for k in v3_feat_names}
+                df_v3 = pd.DataFrame(df_v3_data)
+                preds_v3 = legacy_v3.predict(xgb.DMatrix(df_v3))
+                global_v3_scores = preds_v3.flatten()
+            except: pass
+
+        # Filter
+        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]
+        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]
+        candidate_indices = candidate_indices[candidate_indices < (len(arr_ts_1m) - 245)]
+        print(f"     🎯 Candidates: {len(candidate_indices)}. Running Vectorized Hydra...", flush=True)
+
+        ai_results = []
+        if hydra_models and len(candidate_indices) > 0:
+            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']
+            ])
+            chunk_size = 5000
+            for i in range(0, len(candidate_indices), chunk_size):
+                chunk_idxs = candidate_indices[i:i+chunk_size]
+                for idx in chunk_idxs:
+                    # ✅ PATTERNS ON DEMAND
+                    s_pattern = 0.5
+                    if pattern_models and '15m' in pattern_models:
+                        try:
+                            idx_15m_entry = map_15m[idx]
+                            if idx_15m_entry > 200:
+                                p_win = pd.DataFrame({
+                                    'open': frames['15m']['open'].values[idx_15m_entry-200:idx_15m_entry],
+                                    'high': frames['15m']['high'].values[idx_15m_entry-200:idx_15m_entry],
+                                    'low': frames['15m']['low'].values[idx_15m_entry-200:idx_15m_entry],
+                                    'close': frames['15m']['close'].values[idx_15m_entry-200:idx_15m_entry],
+                                    'volume': frames['15m']['volume'].values[idx_15m_entry-200:idx_15m_entry]
+                                })
+                                vec = _transform_window_for_pattern(p_win)
+                                if vec is not None:
+                                    s_pattern = pattern_models['15m'].predict(xgb.DMatrix(vec.reshape(1,-1)))[0]
+                        except: pass
+
+                    sl_st = h_static[idx:idx+240] 
+                    sl_close = sl_st[:, 6]; sl_atr = sl_st[:, 5]
+                    entry = float(fast_1m['close'][idx]) 
+                    dist = np.maximum(1.5 * sl_atr, entry * 0.015)
+                    pnl = sl_close - entry
+                    norm_pnl = pnl / dist
+                    max_pnl_r = (np.maximum.accumulate(sl_close) - entry) / dist
+                    atr_pct = sl_atr / sl_close
+                    zeros = np.zeros(240); time_vec = np.arange(1, 241); s_oracle = float(global_oracle_scores[idx])
+                    
+                    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,
+                        np.full(240, s_oracle), np.full(240, 0.7), np.full(240, 3.0)
+                    ])
+                    max_hydra = 0.0; max_giveback = 0.0; hydra_time = 0
+                    try:
+                        probs = hydra_models['crash'].predict_proba(X_H)[:, 1]
+                        max_hydra = float(np.max(probs)) 
+                        if max_hydra > 0.6:
+                            t = np.argmax(probs)
+                            hydra_time = int(fast_1m['timestamp'][idx + t])
+                    except: pass
+                    try:
+                        probs_g = hydra_models['giveback'].predict_proba(X_H)[:, 1]
+                        max_giveback = float(np.max(probs_g))
+                    except: pass
+                    
+                    max_v2 = float(np.max(global_v2_scores[idx:idx+240])) 
+                    max_v3 = float(np.max(global_v3_scores[idx:idx+240]))
+                    v2_time = 0
+                    if max_v2 > 0.8:
+                        t2 = np.argmax(global_v2_scores[idx:idx+240])
+                        v2_time = int(fast_1m['timestamp'][idx + t2])
+                        
+                    ai_results.append({
+                        'timestamp': int(fast_1m['timestamp'][idx]), 
+                        'symbol': sym, 'close': entry,
+                        'real_titan': float(global_titan_scores[idx]), 
+                        'oracle_conf': s_oracle, 
+                        'sniper_score': float(global_sniper_scores[idx]), 
+                        'pattern_score': float(s_pattern), # ✅ SAVED PATTERN
+                        'risk_hydra_crash': max_hydra, 
+                        'risk_hydra_giveback': max_giveback, 
+                        'time_hydra_crash': hydra_time,
+                        'risk_legacy_v2': max_v2, 
+                        'risk_legacy_v3': max_v3, 
+                        'time_legacy_panic': v2_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)
+        gc.collect()
+
+    async def generate_truth_data(self):
+        if self.force_start_date:
+            dt_s = datetime.strptime(self.force_start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
+            dt_e = datetime.strptime(self.force_end_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
+            ms_s = int(dt_s.timestamp()*1000); ms_e = int(dt_e.timestamp()*1000)
+            print(f"\n🚜 [Phase 1] Processing Era: {self.force_start_date} -> {self.force_end_date}")
+            for sym in self.TARGET_COINS:
+                c = await self._fetch_all_data_fast(sym, ms_s, ms_e)
+                if c: await self._process_data_in_memory(sym, c, ms_s, ms_e)
+
+    @staticmethod
+    def _worker_optimize(combinations_batch, scores_files, initial_capital, fees_pct, max_slots):
+        print(f"     ⏳ [System] Loading {len(scores_files)} datasets...", flush=True)
+        data = []
+        for f in scores_files:
+            try: data.append(pd.read_pickle(f))
+            except: pass
+        if not data: return []
+        df = pd.concat(data).sort_values('timestamp').reset_index(drop=True)
         
-        if self.sniper: 
-            # ✅ الحقن الديناميكي لإعدادات Sniper قبل التنفيذ
-            if hasattr(self.sniper, 'configure_settings'):
-                self.sniper.configure_settings(
-                    threshold=SystemLimits.L4_ENTRY_THRESHOLD,
-                    wall_ratio=SystemLimits.L4_OB_WALL_RATIO,
-                    w_ml=SystemLimits.L4_WEIGHT_ML,
-                    w_ob=SystemLimits.L4_WEIGHT_OB
-                )
-            
-            return await self.sniper.check_entry_signal_async(ohlcv_1m_data, order_book_data)
-            
-        return {'signal': 'WAIT', 'reason': 'Sniper Engine Missing'}
-
-    def consult_dual_guardians(self, symbol, ohlcv_1m, ohlcv_5m, ohlcv_15m, trade_context, order_book_snapshot=None):
-        """
-        L0 Guardians:
-        الحراس يستخدمون العتبات الديناميكية (Panic Thresholds) التي قد تختلف بين Bull و Bear.
-        """
-        response = {'action': 'HOLD', 'detailed_log': '', 'probs': {}}
+        ts = df['timestamp'].values
+        close = df['close'].values.astype(float)
+        sym = df['symbol'].values
+        u_syms = np.unique(sym); sym_map = {s: i for i, s in enumerate(u_syms)}; sym_id = np.array([sym_map[s] for s in sym])
         
-        # 1. Hydra
-        hydra_result = {'action': 'HOLD', 'reason': 'Disabled', 'probs': {}}
-        if self.guardian_hydra and self.guardian_hydra.initialized:
-            hydra_result = self.guardian_hydra.analyze_position(symbol, ohlcv_1m, ohlcv_5m, ohlcv_15m, trade_context)
-            h_probs = hydra_result.get('probs', {})
-            p_crash = h_probs.get('crash', 0.0)
-            p_giveback = h_probs.get('giveback', 0.0)
-            
-            # استخدام العتبات الحية
-            if hydra_result['action'] == 'HOLD':
-                if p_crash >= SystemLimits.HYDRA_CRASH_THRESH:
-                    hydra_result['action'] = 'EXIT_HARD'
-                    hydra_result['reason'] = f"Hydra Crash Risk {p_crash:.2f}"
-                elif p_giveback >= SystemLimits.HYDRA_GIVEBACK_THRESH:
-                    hydra_result['action'] = 'EXIT_SOFT'
-                    hydra_result['reason'] = f"Hydra Giveback Risk {p_giveback:.2f}"
+        oracle = df['oracle_conf'].values.astype(float)
+        sniper = df['sniper_score'].values.astype(float)
+        hydra = df['risk_hydra_crash'].values.astype(float)
+        if 'risk_hydra_giveback' in df.columns: hydra_give = df['risk_hydra_giveback'].values.astype(float)
+        else: hydra_give = np.zeros(len(df))
+        titan = df['real_titan'].values.astype(float)
+        
+        # ✅ Load Pattern Score
+        if 'pattern_score' in df.columns: pattern = df['pattern_score'].values.astype(float)
+        else: pattern = np.full(len(df), 0.5)
+
+        l1 = df['l1_score'].values.astype(float)
+        legacy_v2 = df['risk_legacy_v2'].values.astype(float) if 'risk_legacy_v2' in df else np.zeros(len(df))
+        legacy_v3 = df['risk_legacy_v3'].values.astype(float) if 'risk_legacy_v3' in df else np.zeros(len(df))
+        h_times = df['time_hydra_crash'].values.astype(np.int64)
         
-        # 2. Legacy (Volume-Aware Veto)
-        legacy_result = {'action': 'HOLD', 'reason': 'Disabled', 'scores': {}}
-        if self.guardian_legacy and self.guardian_legacy.initialized:
-            # تحديث العتبات قبل التحليل
-            self.guardian_legacy.configure_thresholds(
-                v2_panic=SystemLimits.LEGACY_V2_PANIC_THRESH,
-                v3_hard=SystemLimits.LEGACY_V3_HARD_THRESH,
-                v3_soft=SystemLimits.LEGACY_V3_SOFT_THRESH,
-                v3_ultra=SystemLimits.LEGACY_V3_ULTRA_THRESH
-            )
+        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 = float(initial_capital); alloc = 0.0
             
-            entry_price = float(trade_context.get('entry_price', 0.0))
-            vol_30m = trade_context.get('volume_30m_usd', 0.0)
+            titan_thresh = cfg['TITAN']
+            # ✅ Add Pattern Check
+            pattern_thresh = cfg.get('PATTERN', 0.10) 
             
-            legacy_result = self.guardian_legacy.analyze_position(
-                ohlcv_1m, ohlcv_5m, ohlcv_15m, entry_price, 
-                order_book=order_book_snapshot,
-                volume_30m_usd=vol_30m
-            )
-
-        # 3. Final Arbitration & Display
-        h_probs = hydra_result.get('probs', {})
-        l_scores = legacy_result.get('scores', {})
-        
-        h_c = h_probs.get('crash', 0.0)
-        h_g = h_probs.get('giveback', 0.0)
-        h_s = h_probs.get('stagnation', 0.0) 
-        l_v2 = l_scores.get('v2', 0.0)
-        l_v3 = l_scores.get('v3', 0.0)
+            mask = (l1 >= cfg['L1_SCORE']) & \
+                   (oracle >= cfg['ORACLE']) & \
+                   (sniper >= cfg['SNIPER']) & \
+                   (titan >= titan_thresh) & \
+                   (pattern >= pattern_thresh) # ✅ MASKED
+            
+            for i in range(N):
+                s = sym_id[i]; p = float(close[i]); curr_t = ts[i]
+
+                if s in pos:
+                    entry_p, h_risk_val, h_give_val, size_val, h_time_val = pos[s]
+                    crash_hydra = bool(h_risk_val > cfg['HYDRA_CRASH'])
+                    giveback_hydra = bool(h_give_val > cfg['HYDRA_GIVEBACK']) 
+                    time_match = bool(h_time_val > 0 and curr_t >= h_time_val)
+                    panic_legacy_v2 = bool(legacy_v2[i] > cfg['LEGACY_V2'])
+                    panic_legacy_v3 = bool(legacy_v3[i] > cfg['LEGACY_V3'])
+                    pnl = (p - entry_p) / entry_p
+                    should_exit = (crash_hydra and time_match) or giveback_hydra or panic_legacy_v2 or panic_legacy_v3 or (pnl > 0.04) or (pnl < -0.02)
+                    if should_exit:
+                        realized = pnl - (fees_pct * 2)
+                        bal += size_val * (1.0 + realized)
+                        alloc -= size_val
+                        del pos[s]
+                        log.append({'pnl': realized})
+
+                if len(pos) < max_slots and bool(mask[i]):
+                    if s not in pos and bal >= 5.0:
+                        size = min(10.0, bal * 0.98)
+                        pos[s] = (p, hydra[i], hydra_give[i], size, h_times[i])
+                        bal -= size; alloc += size
+            
+            final_bal = bal + alloc
+            profit = final_bal - initial_capital
+            tot = len(log)
+            winning = [x for x in log if x['pnl'] > 0]
+            losing = [x for x in log if x['pnl'] <= 0]
+            win_count = len(winning); loss_count = len(losing)
+            win_rate = (win_count/tot*100) if tot > 0 else 0.0
+            avg_win = np.mean([x['pnl'] for x in winning]) if winning else 0.0
+            avg_loss = np.mean([x['pnl'] for x in losing]) if losing else 0.0
+            gross_p = sum([x['pnl'] for x in winning])
+            gross_l = abs(sum([x['pnl'] for x in losing]))
+            profit_factor = (gross_p / gross_l) if gross_l > 0 else 99.9
+            max_win_s = 0; max_loss_s = 0; curr_w = 0; curr_l = 0
+            for t in log:
+                if t['pnl'] > 0:
+                    curr_w += 1; curr_l = 0
+                    if curr_w > max_win_s: max_win_s = curr_w
+                else:
+                    curr_l += 1; curr_w = 0
+                    if curr_l > max_loss_s: max_loss_s = curr_l
+
+            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': 0.0, 'high_consensus_win_rate': 0.0, 'high_consensus_avg_pnl': 0.0
+            })
+        return res
+
+    async def run_optimization(self, target_regime="RANGE"):
+        await self.generate_truth_data()
         
-        stamp_str = f"🐲[C:{h_c:.0%}|G:{h_g:.0%}|S:{h_s:.0%}] 🕸️[V2:{l_v2:.0%}|V3:{l_v3:.0%}]"
+        keys = list(self.GRID_RANGES.keys())
+        values = list(self.GRID_RANGES.values())
+        combos = [dict(zip(keys, c)) for c in itertools.product(*values)]
+            
+        files = glob.glob(os.path.join(CACHE_DIR, "*.pkl"))
+        results_list = self._worker_optimize(combos, files, self.INITIAL_CAPITAL, self.TRADING_FEES, self.MAX_SLOTS)
+        if not results_list: return None, {'net_profit': 0.0, 'win_rate': 0.0}
+
+        results_list.sort(key=lambda x: x['net_profit'], reverse=True)
+        best = results_list[0]
+
+        diag = []
+        if best['total_trades'] > 2000 and best['net_profit'] < 10: diag.append("⚠️ Overtrading")
+        if best['win_rate'] > 55 and best['net_profit'] < 0: diag.append("⚠️ Fee Burn")
+        if abs(best['avg_loss']) > best['avg_win']: diag.append("⚠️ Risk/Reward Inversion")
+        if best['max_loss_streak'] > 10: diag.append("⚠️ Consecutive Loss Risk")
+        if not diag: diag.append("✅ System Healthy")
 
-        final_action = 'HOLD'
-        final_reason = f"Safe. {stamp_str}"
+        print("\n" + "="*60)
+        print(f"🏆 CHAMPION REPORT [{target_regime}]:")
+        print(f"   💰 Final Balance:   ${best['final_balance']:,.2f}")
+        print(f"   🚀 Net PnL:         ${best['net_profit']:,.2f}")
+        print("-" * 60)
+        print(f"   📊 Total Trades:    {best['total_trades']}")
+        print(f"   📈 Win Rate:        {best['win_rate']:.1f}%")
+        print(f"   ✅ Winning Trades:  {best['win_count']} (Avg: {best['avg_win']*100:.2f}%)")
+        print(f"   ❌ Losing Trades:   {best['loss_count']} (Avg: {best['avg_loss']*100:.2f}%)")
+        print(f"   🌊 Max Streaks:     Win {best['max_win_streak']} | Loss {best['max_loss_streak']}")
+        print(f"   ⚖️ Profit Factor:   {best['profit_factor']:.2f}")
+        print("-" * 60)
+        print(f"   🩺 DIAGNOSIS: {' '.join(diag)}")
+        print(f"   ⚙️ Oracle={best['config']['ORACLE']:.2f} | Sniper={best['config']['SNIPER']:.2f} | Pattern={best['config'].get('PATTERN',0):.2f}")
+        print("="*60)
+        return best['config'], best
+
+async def run_strategic_optimization_task():
+    print("\n🧪 [STRATEGIC BACKTEST] Configurable Grid Mode...")
+    r2 = R2Service(); dm = DataManager(None, None, r2); proc = MLProcessor(dm)
+    try:
+        await dm.initialize(); await proc.initialize()
+        if proc.guardian_hydra: proc.guardian_hydra.set_silent_mode(True)
+        hub = AdaptiveHub(r2); await hub.initialize()
+        opt = HeavyDutyBacktester(dm, proc)
         
-        if hydra_result['action'] in ['EXIT_HARD', 'EXIT_SOFT', 'TIGHTEN_SL', 'TRAIL_SL']:
-             final_action = hydra_result['action']
-             final_reason = f"🐲 HYDRA: {hydra_result['reason']} | {stamp_str}"
-        elif legacy_result['action'] in ['EXIT_HARD', 'EXIT_SOFT']:
-             final_action = legacy_result['action']
-             final_reason = f"🕸️ LEGACY: {legacy_result['reason']} | {stamp_str}"
-
-        return {
-            'action': final_action, 
-            'reason': final_reason, 
-            'detailed_log': f"{final_action} | {stamp_str}", 
-            'probs': h_probs, 
-            'scores': l_scores
-        }
+        scenarios = [
+                {"regime": "BULL", "start": "2024-01-01", "end": "2024-03-30"},
+                {"regime": "BEAR", "start": "2023-08-01", "end": "2023-09-15"},
+                {"regime": "DEAD", "start": "2023-06-01", "end": "2023-08-01"},
+                {"regime": "RANGE", "start": "2024-07-01", "end": "2024-09-30"}
+            ]
+        
+        for s in scenarios:
+            opt.set_date_range(s["start"], s["end"])
+            best_cfg, best_stats = await opt.run_optimization(s["regime"])
+            if best_cfg: hub.submit_challenger(s["regime"], best_cfg, best_stats)
+        await hub._save_state_to_r2()
+        print("✅ [System] DNA Updated.")
+    finally:
+        print("🔌 [System] Closing connections...")
+        await dm.close()
 
-    async def run_advanced_monte_carlo(self, symbol, timeframe='1h'):
-        if self.mc_analyzer and self.data_manager:
-            try:
-                ohlcv = await self.data_manager.get_latest_ohlcv(symbol, timeframe, limit=300)
-                if ohlcv: return self.mc_analyzer.run_advanced_simulation([c[4] for c in ohlcv])
-            except Exception: pass
-        return 0.0
+if __name__ == "__main__":
+    asyncio.run(run_strategic_optimization_task())