Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V43.0 - GEM-Architect: Hot-Flow Sort)
+# (V45.0 - GEM-Architect: Anti-FOMO Revival)
 # ============================================================
 
 import asyncio
@@ -9,29 +9,28 @@ import traceback
 import logging
 import pandas as pd
 import numpy as np
-import pandas_ta as ta
-import math  # ✅ مكتبة مهمة للمعادلة اللوغاريتمية
+import pandas_ta as ta  # سنحتاج بعض الدوال المساعدة
 from typing import List, Dict, Any
 
 import ccxt.async_support as ccxt
 
-# ✅ استيراد الدستور الديناميكي
+# ✅ استيراد الدستور الديناميكي (للحفاظ على توافق النظام)
 try:
     from ml_engine.processor import SystemLimits
 except ImportError:
     class SystemLimits:
         L1_MIN_AFFINITY_SCORE = 15.0
         CURRENT_REGIME = "RANGE"
-        SCANNER_WEIGHTS = {"RSI_MOMENTUM": 0.3, "BB_BREAKOUT": 0.3, "MACD_CROSS": 0.2, "VOLUME_FLOW": 0.2}
 
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 class DataManager:
     """
-    DataManager V43.0 (Hot-Flow Sort)
-    - Replaced 'Raw Volume' sort with 'Volume-Weighted Momentum'.
-    - Prioritizes active mid-caps over stagnant giants (XRP, TRX, BNB).
+    DataManager V45.0 (Anti-FOMO Revival)
+    - Restores the STRICT Logic Tree from V15.2.
+    - Filters: 8% Max Pump, 12% Max Daily, RSI < 70 strict limit.
+    - Targets: Clean Breakouts & Oversold Reversals ONLY.
     """
 
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
@@ -41,7 +40,7 @@ class DataManager:
         
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
-            'timeout': 30000, 
+            'timeout': 60000, 
             'options': {'defaultType': 'spot'} 
         })
             
@@ -51,17 +50,17 @@ class DataManager:
         # القائمة السوداء
         self.BLACKLIST_TOKENS = [
             'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX',
-            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP', 'HT', 'KCS', 'WBTC'
+            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP', 'HT', 'KCS'
         ]
         
-        print(f"📦 [DataManager V43.0] Hot-Flow Engine Online.")
+        print(f"📦 [DataManager V45.0] Anti-FOMO Shield Active.")
 
     async def initialize(self):
         print("   > [DataManager] Starting initialization...")
         try:
-            self.http_client = httpx.AsyncClient(timeout=60.0)
+            self.http_client = httpx.AsyncClient(timeout=30.0)
             await self._load_markets()
-            print(f"✅ [DataManager] Ready (Mode: {getattr(SystemLimits, 'CURRENT_REGIME', 'UNKNOWN')}).")
+            print(f"✅ [DataManager] Ready (Logic: STRICT Anti-FOMO).")
         except Exception as e:
             print(f"❌ [DataManager] Init Error: {e}")
             traceback.print_exc()
@@ -86,144 +85,238 @@ class DataManager:
     def get_contracts_db(self): return self.contracts_db
 
     # ==================================================================
-    # 🛡️ Layer 1: The Titan Mirror Filter (Hot-Flow Edition)
+    # 🛡️ Layer 1: The Strict Logic Tree (From V15.2)
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        """
-        يقوم بفرز العملات بناءً على "السخونة" (Hotness) وليس الحجم فقط.
-        """
-        current_regime = getattr(SystemLimits, "CURRENT_REGIME", "RANGE")
-        min_score = getattr(SystemLimits, "L1_MIN_AFFINITY_SCORE", 15.0)
-        
-        print(f"🔍 [L1 Hot-Flow] Scanning for Active Momentum (Regime: {current_regime})...")
-        
-        # 1. جلب Universe (تصفية أولية)
-        # خفضنا الحد الأدنى للحجم للسماح للعملات المتوسطة "الساخنة" بالدخول
-        # في السابق كان الرقم مرتفعاً جداً مما يقتل الفرص
-        min_vol_floor = 1000000 if current_regime == "BEAR" else 5000000
+        print(f"🔍 [L1 Anti-FOMO] Filtering Universe...")
         
-        all_tickers = await self._fetch_universe_tickers(min_volume=min_vol_floor)
+        # 1. المرحلة 0: فلتر الكون (السيولة العالية فقط)
+        # V15.2 كان يطلب مليون دولار سيولة، سنبقيه كما هو للصرامة
+        initial_candidates = await self._stage0_universe_filter()
         
-        if not all_tickers: 
-            print("⚠️ [Layer 1] Universe fetch returned empty.")
+        if not initial_candidates:
+            print("⚠️ [Layer 1] Universe empty.")
             return []
 
-        # 2. الجلب العميق (Deep Fetch) لأكثر العملات سخونة
-        # نأخذ أفضل 150 عملة بناءً على معادلة (الحجم × الحركة)
-        scan_limit = 150 
-        top_candidates = all_tickers[:scan_limit] 
+        # 2. جلب البيانات الفنية لأفضل 300 عملة (كما في V15.2)
+        top_liquid_candidates = initial_candidates[:300] 
+        print(f"   -> Analyzing top {len(top_liquid_candidates)} liquid assets...")
         
-        print(f"   -> Deep scanning top {len(top_candidates)} active assets...")
-        
-        enriched_data = await self._batch_fetch_ta_data(top_candidates, timeframe='15m', limit=200)
-        
-        scored_candidates = []
+        enriched_data = await self._fetch_technical_data_batch(top_liquid_candidates)
         
+        # 3. تطبيق شجرة القرار الصارمة
+        breakout_list = []
+        reversal_list = []
+
         for item in enriched_data:
-            df = item.get('df')
-            if df is None or len(df) < 150: continue 
+            # هنا نستخدم منطق V15.2 الأصلي
+            classification = self._apply_logic_tree(item)
             
-            # تطبيق فلتر المرآة الهيكلية
-            structural_score, reasons = self._calculate_structural_score(df, item['symbol'], current_regime)
-            
-            # 🔥 Bonus: إضافة نقاط إضافية للعملات الساخنة جداً
-            hot_score = item.get('hot_score', 0)
-            if hot_score > 50: structural_score += 5 # Boost for super active coins
-            
-            item['l1_score'] = structural_score
-            item['type'] = " | ".join(reasons)
-            
-            if structural_score >= min_score:
-                scored_candidates.append(item)
+            if classification['type'] == 'BREAKOUT':
+                item['l1_score'] = classification['score'] 
+                item['type'] = 'BREAKOUT'
+                breakout_list.append(item)
+            elif classification['type'] == 'REVERSAL':
+                item['l1_score'] = classification['score'] 
+                item['type'] = 'REVERSAL'
+                reversal_list.append(item)
 
-        # الترتيب النهائي حسب جودة الهيكل
-        scored_candidates.sort(key=lambda x: x['l1_score'], reverse=True)
-        
-        print(f"   -> Filter selected {len(scored_candidates)} candidates.")
+        print(f"   -> [L1 Logic] Found: {len(breakout_list)} Breakouts, {len(reversal_list)} Reversals.")
+
+        # 4. الترتيب والدمج النهائي
+        # الـ Breakout نرتبهم بالأعلى سكور (فوليوم)
+        breakout_list.sort(key=lambda x: x['l1_score'], reverse=True)
+        # الـ Reversal نرتبهم بالأعلى سكور (سكور الارتداد في V15.2 كان 100-RSI، يعني الأعلى أفضل)
+        reversal_list.sort(key=lambda x: x['l1_score'], reverse=True)
+
+        # نختار الأفضل فقط (مزيج متوازن)
+        final_selection = breakout_list[:25] + reversal_list[:15]
         
         return [
             {
                 'symbol': c['symbol'],
                 'quote_volume': c.get('quote_volume', 0),
                 'current_price': c.get('current_price', 0),
-                'type': c.get('type', 'Structural'),
+                'type': c.get('type', 'UNKNOWN'),
                 'l1_score': c.get('l1_score', 0)
             }
-            for c in scored_candidates[:30] 
+            for c in final_selection
         ]
 
     # ==================================================================
-    # 🧬 Structural Alignment Engine
+    # 🔗 Bridge for Backtest Engine Compatibility (IMPORTANT)
     # ==================================================================
     def _calculate_structural_score(self, df: pd.DataFrame, symbol: str, regime: str) -> (float, List[str]):
-        # (نفس منطق الفلتر الهيكلي السابق - لم يتغير لأنه ممتاز)
-        score = 0.0
-        tags = []
+        """
+        [Compatibility Wrapper]
+        هذه الدالة موجودة لكي لا يتعطل محرك الباكتست (backtest_engine.py).
+        تقوم بتحويل بيانات الباكتست إلى تنسيق يفهمه منطق V15.2.
+        """
+        # محاكاة تنسيق البيانات الذي يطلبه _apply_logic_tree
+        # نحتاج تقسيم الـ DF إلى 1H و 15M تقريبياً
         try:
-            close = df['close']
-            high = df['high']
-            low = df['low']
-            volume = df['volume']
-            current_price = close.iloc[-1]
+            # Resample لإنشاء بيانات 1H و 15M من البيانات المدخلة (التي غالباً تكون 15M)
+            df_15m = df.copy()
             
-            zero_vol_candles = (volume == 0).sum()
-            if zero_vol_candles > 5: return -100.0, ["Illiquid"]
+            agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
+            df_1h = df.resample('1H').agg(agg_dict).dropna()
             
-            avg_vol = volume.rolling(20).mean().iloc[-1]
-            if avg_vol * current_price < 5000: return -100.0, ["Thin Book"]
-
-            ema200 = ta.ema(close, length=200)
-            ema50 = ta.ema(close, length=50)
+            # تحويلها لقوائم كما يتوقع الكود القديم
+            ohlcv_1h = df_1h.reset_index()[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
+            ohlcv_15m = df_15m.reset_index()[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
             
-            if ema200 is not None and ema50 is not None:
-                curr_ema200 = ema200.iloc[-1]
-                dist_200 = (current_price - curr_ema200) / curr_ema200
-                
-                if regime == "BULL":
-                    if current_price > curr_ema200:
-                        score += 20
-                        if current_price > ema50.iloc[-1]: score += 10
-                        if dist_200 < 0.15: score += 10 
-                        else: score -= 5 
-                    else: score -= 20 
-                elif regime == "BEAR":
-                     if dist_200 < -0.20: score += 30 
-                else: 
-                    if abs(dist_200) < 0.05: score += 20
-
-            bb = ta.bbands(close, length=20, std=2)
-            if bb is not None:
-                # حساب الـ Bandwidth يدوياً للأمان
-                upper = bb[bb.columns[0]] # Lower band usually index 0 in pandas_ta default
-                lower = bb[bb.columns[2]] # Upper band
-                # نستخدم أسماء الأعمدة إذا أمكن للتأكد، ولكن للسرعة:
-                # pandas_ta returns: Lower, Mid, Upper, Bandwidth, Percent
-                # لنتأكد من الـ Bandwidth مباشرة
-                width_col = next((c for c in bb.columns if c.startswith('BBB')), None)
-                if width_col:
-                    current_width = bb[width_col].iloc[-1] / 100.0 # pandas_ta returns pct
-                    if current_width < 0.05: score += 25; tags.append("Squeeze")
-                    elif current_width > 0.15: score -= 10
-
-            rsi = ta.rsi(close, length=14).iloc[-1]
-            adx = ta.adx(high, low, close, length=14)
-            curr_adx = adx.iloc[-1, 0] if adx is not None else 0
+            dummy_data = {
+                'ohlcv_1h': ohlcv_1h,
+                'ohlcv_15m': ohlcv_15m,
+                'change_24h': 0.0 # غير متوفر بدقة في الباكتست الجزئي، نتجاوزه
+            }
             
-            if 40 < rsi < 70: score += 15 
-            elif rsi > 75 and regime == "BULL": score += 10 
+            res = self._apply_logic_tree(dummy_data)
             
-            if curr_adx > 25: score += 10; tags.append("Trending")
+            score = res.get('score', 0.0)
+            # تحويل السكور ليكون متوافقاً مع الباكتست (حول 20-80)
+            if res['type'] == 'BREAKOUT':
+                return score * 20.0, ["BREAKOUT"] # Breakout score is usually low (ratio), boost it
+            elif res['type'] == 'REVERSAL':
+                return score, ["REVERSAL"] # Reversal score is already 0-100
             
-            vol_sma = volume.rolling(20).mean().iloc[-1]
-            if volume.iloc[-1] > vol_sma * 1.5: score += 15; tags.append("Vol Spike")
+            return 0.0, ["NONE"]
+            
+        except Exception:
+            return 0.0, ["ERROR"]
 
-        except Exception as e: return 0.0, ["Error"]
-        return score, tags
+    # ==================================================================
+    # 🏗️ V15.2 Logic Core (Unchanged Logic)
+    # ==================================================================
+    def _apply_logic_tree(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            df_1h = self._calc_indicators(data['ohlcv_1h'])
+            df_15m = self._calc_indicators(data['ohlcv_15m'])
+        except:
+            return {'type': 'NONE'}
+
+        if df_1h.empty or df_15m.empty: return {'type': 'NONE'}
+
+        curr_1h = df_1h.iloc[-1]
+        curr_15m = df_15m.iloc[-1]
+        
+        # --- Stage 2: Anti-FOMO Filters (STRICT) ---
+        try:
+            # حساب التغير في آخر 4 ساعات
+            if len(df_1h) >= 5:
+                close_4h_ago = df_1h.iloc[-5]['close']
+                change_4h = ((curr_1h['close'] - close_4h_ago) / close_4h_ago) * 100
+            else:
+                change_4h = 0.0
+        except: change_4h = 0.0
+
+        # 1. فلتر المضخات: ممنوع الدخول إذا صعدت أكثر من 8% في 4 ساعات
+        if change_4h > 8.0: return {'type': 'NONE'}
+        
+        # 2. فلتر التذبذب اليومي: ممنوع أكثر من 12% (للبعد عن العملات المجنونة)
+        if data.get('change_24h', 0) > 12.0: return {'type': 'NONE'}
+        
+        # 3. فلتر القمة: ممنوع RSI فوق 70 قطعاً
+        if curr_1h['rsi'] > 70: return {'type': 'NONE'} 
+
+        # 4. فلتر الامتداد: ممنوع الابتعاد عن المتوسط كثيراً
+        deviation = (curr_1h['close'] - curr_1h['ema20']) / curr_1h['atr'] if curr_1h['atr'] > 0 else 0
+        if deviation > 1.8: return {'type': 'NONE'} 
+
+        # --- Stage 3: Setup Classification ---
+        
+        # === A. Breakout Logic ===
+        is_breakout = False
+        breakout_score = 0.0
+        
+        # تريند صاعد
+        bullish_structure = (curr_1h['ema20'] > curr_1h['ema50']) or (curr_1h['close'] > curr_1h['ema20'])
+        
+        if bullish_structure:
+            # RSI يجب أن يكون فيه مساحة للصعود (ليس منخفضاً جداً ولا مرتفعاً جداً)
+            if 45 <= curr_1h['rsi'] <= 68:
+                if curr_15m['close'] >= curr_15m['ema20']:
+                    # Volatility Squeeze (هدوء ما قبل العاصفة)
+                    avg_range = (df_15m['high'] - df_15m['low']).rolling(10).mean().iloc[-1]
+                    current_range = curr_15m['high'] - curr_15m['low']
+                    
+                    if current_range <= avg_range * 1.8:
+                        vol_ma20 = df_15m['volume'].rolling(20).mean().iloc[-1]
+                        # شرط الفوليوم: شمعة الحالية فيها سيولة 1.5 ضعف المتوسط
+                        if curr_15m['volume'] >= 1.5 * vol_ma20:
+                            is_breakout = True
+                            breakout_score = curr_15m['volume'] / vol_ma20 if vol_ma20 > 0 else 1.0
+
+        if is_breakout:
+            return {'type': 'BREAKOUT', 'score': breakout_score}
+
+        # === B. Reversal Logic ===
+        is_reversal = False
+        reversal_score = 0.0 
+        
+        # تشبع بيعي واضح
+        if 20 <= curr_1h['rsi'] <= 40: 
+            # السعر هبط مؤخراً
+            if change_4h <= -2.0: 
+                # البحث عن شمعة انعكاسية (Hammer / Green Body)
+                last_3 = df_15m.iloc[-3:]
+                found_rejection = False
+                for _, row in last_3.iterrows():
+                    rng = row['high'] - row['low']
+                    if rng > 0:
+                        is_green = row['close'] > row['open']
+                        # Hammer pattern logic
+                        lower_wick = min(row['open'], row['close']) - row['low']
+                        body = abs(row['close'] - row['open'])
+                        hammer_shape = lower_wick > (body * 1.5)
+                        
+                        if is_green or hammer_shape:
+                            found_rejection = True
+                            break
+                
+                if found_rejection:
+                    is_reversal = True
+                    # السكور كلما قل الـ RSI كان أفضل للارتداد
+                    reversal_score = (100 - curr_1h['rsi']) 
+
+        if is_reversal:
+            return {'type': 'REVERSAL', 'score': reversal_score}
+
+        return {'type': 'NONE'}
+
+    # ------------------------------------------------------------------
+    # Manual Indicator Calculation (Pandas pure - Exactly like V15.2)
+    # ------------------------------------------------------------------
+    def _calc_indicators(self, ohlcv_list):
+        if not ohlcv_list: return pd.DataFrame()
+        df = pd.DataFrame(ohlcv_list, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+        
+        # RSI Calculation
+        delta = df['close'].diff()
+        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
+        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
+        rs = gain / loss
+        df['rsi'] = 100 - (100 / (1 + rs))
+        
+        # EMA
+        df['ema20'] = df['close'].ewm(span=20, adjust=False).mean()
+        df['ema50'] = df['close'].ewm(span=50, adjust=False).mean()
+        
+        # ATR
+        high_low = df['high'] - df['low']
+        high_close = np.abs(df['high'] - df['close'].shift())
+        low_close = np.abs(df['low'] - df['close'].shift())
+        ranges = pd.concat([high_low, high_close, low_close], axis=1)
+        true_range = np.max(ranges, axis=1)
+        df['atr'] = true_range.rolling(14).mean()
+        
+        df.fillna(0, inplace=True)
+        return df
 
     # ==================================================================
-    # 🌍 Universe & Batch Fetch (The Hot-Flow Logic)
+    # 🌌 Stage 0: Universe Filter (V15.2 Logic)
     # ==================================================================
-    async def _fetch_universe_tickers(self, min_volume=300_000) -> List[Dict[str, Any]]:
+    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
         try:
             tickers = await self.exchange.fetch_tickers()
             candidates = []
@@ -231,96 +324,78 @@ class DataManager:
             for symbol, ticker in tickers.items():
                 if not symbol.endswith('/USDT'): continue
                 
-                base_currency = symbol.split('/')[0]
-                if any(bad in base_currency for bad in self.BLACKLIST_TOKENS): continue
-                if "3S" in base_currency or "3L" in base_currency: continue 
-                
-                vol = ticker.get('quoteVolume')
-                if vol is None: vol = ticker.get('info', {}).get('volValue') 
-                if vol is None: vol = 0.0
-                else: vol = float(vol)
-
-                # 1. فلتر الحد الأدنى المطلق (لإبعاد العملات الميتة)
-                if vol < min_volume: continue 
-
-                # 2. فلتر السبريد (للحماية)
-                bid = float(ticker.get('bid', 0) or 0)
-                ask = float(ticker.get('ask', 0) or 0)
-                if bid > 0 and ask > 0:
-                    spread_pct = (ask - bid) / bid
-                    if spread_pct > 0.015: continue # تساهلنا قليلاً (1.5%) للسماح بعملات الـ Meme النشطة
-
-                # 3. حساب درجة السخونة 🔥 (Hot Score)
-                # المعادلة: Log10(Volume) * (1 + Abs(Change%))
-                # هذا يعطي وزناً للحجم، لكن يضرب بقوة في التغير السعري
-                change_pct = float(ticker.get('percentage', 0.0))
-                
-                # نستخدم Log10 لتقليص الفارق بين المليار والمليون
-                # Log10(1B) = 9, Log10(10M) = 7 (الفارق بسيط)
-                # بينما التغير السعري: 1% vs 10% (الفارق 10 أضعاف)
-                # هذا يجعل التغير السعري هو العامل الحاسم في الترتيب
+                base_curr = symbol.split('/')[0]
+                if any(bad in base_curr for bad in self.BLACKLIST_TOKENS): continue
                 
-                log_vol = math.log10(vol + 1)
-                volatility_factor = abs(change_pct) + 1.0 # نضيف 1 لكي لا نضرب في صفر
+                # شرط السيولة الصارم: 1 مليون دولار
+                quote_vol = ticker.get('quoteVolume')
+                if not quote_vol or quote_vol < 1_000_000: continue
                 
-                hot_score = log_vol * volatility_factor
+                last_price = ticker.get('last')
+                if not last_price or last_price < 0.0005: continue 
 
                 candidates.append({
                     'symbol': symbol,
-                    'quote_volume': vol,
-                    'current_price': float(ticker['last']) if ticker.get('last') else 0.0,
-                    'change_24h': change_pct,
-                    'hot_score': hot_score # الدرجة الجديدة
+                    'quote_volume': quote_vol,
+                    'current_price': last_price,
+                    'change_24h': float(ticker.get('percentage', 0.0)) 
                 })
             
-            # 🔥 الترتيب حسب درجة السخونة وليس الحجم المطلق
-            candidates.sort(key=lambda x: x['hot_score'], reverse=True)
-            
+            # ترتيب مبدئي بالحجم
+            candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
             return candidates
-
+            
         except Exception as e:
-            print(f"❌ [L1 Error] Fetch Tickers Failed: {e}")
+            print(f"❌ [L1 Error] Universe filter failed: {e}")
             return []
 
-    async def _batch_fetch_ta_data(self, candidates, timeframe='15m', limit=200):
+    # ==================================================================
+    # 🔄 Batch Fetching
+    # ==================================================================
+    async def _fetch_technical_data_batch(self, candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        chunk_size = 15 
         results = []
-        chunk_size = 20 
         for i in range(0, len(candidates), chunk_size):
-            chunk = candidates[i:i+chunk_size]
-            tasks = [self._fetch_ohlcv_safe(c, timeframe, limit) for c in chunk]
-            chunk_res = await asyncio.gather(*tasks)
-            results.extend([r for r in chunk_res if r is not None])
-            await asyncio.sleep(0.05)
+            chunk = candidates[i:i + chunk_size]
+            chunk_tasks = [self._fetch_single_tech_data(c) for c in chunk]
+            chunk_results = await asyncio.gather(*chunk_tasks)
+            results.extend([r for r in chunk_results if r is not None])
+            await asyncio.sleep(0.05) 
         return results
 
-    async def _fetch_ohlcv_safe(self, candidate, tf, limit):
+    async def _fetch_single_tech_data(self, candidate: Dict[str, Any]) -> Any:
+        symbol = candidate['symbol']
         try:
-            ohlcv = await self.exchange.fetch_ohlcv(candidate['symbol'], tf, limit=limit)
-            if not ohlcv: return None
-            df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-            # تحويل البيانات لأرقام
-            cols = ['open', 'high', 'low', 'close', 'volume']
-            df[cols] = df[cols].apply(pd.to_numeric, errors='coerce')
+            # V15.2 Requires 1H and 15M
+            ohlcv_1h = await self.exchange.fetch_ohlcv(symbol, '1h', limit=60)
+            ohlcv_15m = await self.exchange.fetch_ohlcv(symbol, '15m', limit=60)
             
-            candidate['df'] = df 
+            if not ohlcv_1h or len(ohlcv_1h) < 55 or not ohlcv_15m or len(ohlcv_15m) < 55:
+                return None
+                
+            candidate['ohlcv_1h'] = ohlcv_1h
+            candidate['ohlcv_15m'] = ohlcv_15m
             return candidate
-        except: return None
+        except Exception:
+            return None
 
     # ==================================================================
     # 🎯 Public Helpers
     # ==================================================================
-    async def get_latest_price_async(self, symbol):
+    async def get_latest_price_async(self, symbol: str) -> float:
         try:
             ticker = await self.exchange.fetch_ticker(symbol)
             return float(ticker['last'])
-        except: return 0.0
+        except Exception: return 0.0
 
-    async def get_latest_ohlcv(self, symbol, timeframe='5m', limit=100):
+    async def get_latest_ohlcv(self, symbol: str, timeframe: str = '5m', limit: int = 100) -> List[List[float]]:
         try:
-            return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit) or []
-        except: return []
+            candles = await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
+            return candles or []
+        except Exception: return []
 
-    async def get_order_book_snapshot(self, symbol, limit=20):
+    async def get_order_book_snapshot(self, symbol: str, limit: int = 20) -> Dict[str, Any]:
         try:
-            return await self.exchange.fetch_order_book(symbol, limit)
-        except: return {}
+            ob = await self.exchange.fetch_order_book(symbol, limit)
+            return ob
+        except Exception: return {}