Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V41.2 - GEM-Architect: The Fixed Matrix & Debugger)
+# (V41.3 - GEM-Architect: Math Debugger Edition)
 # ============================================================
 
 import asyncio
@@ -14,38 +14,30 @@ from typing import List, Dict, Any
 
 import ccxt.async_support as ccxt
 
-# ✅ استيراد الدستور الديناميكي (لقراءة الحالة الحالية)
+# ✅ استيراد الدستور الديناميكي
 try:
     from ml_engine.processor import SystemLimits
 except ImportError:
-    # Fallback إذا لم يتم التحميل بعد
     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("httpcore").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 class DataManager:
     """
-    DataManager V41.2 (The Fixed Matrix)
-    - Fixes the 'Suicidal Blacklist' bug.
-    - Uses Scanner Matrix for L1 filtering.
-    - Includes Deep Debugging logs.
+    DataManager V41.3 (Math Debugger)
+    - Exposes calculation errors in _apply_scanner_strategies.
+    - Inspects DataFrame integrity.
     """
 
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
-        # ==================================================================
-        # ⚙️ إعدادات التحكم والتهيئة
-        # ==================================================================
         self.contracts_db = contracts_db or {}
         self.whale_monitor = whale_monitor
         self.r2_service = r2_service
         
-        # إعداد عميل المنصة (KuCoin) مع تفعيل حدود السرعة
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
             'timeout': 30000, 
@@ -55,16 +47,14 @@ class DataManager:
         self.http_client = None
         self.market_cache = {}
         
-        # 🚫 قوائم الاستبعاد (العملات المستقرة، الرافعة، والعملات المحظورة)
         self.BLACKLIST_TOKENS = [
             'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX',
             'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP'
         ]
         
-        print(f"📦 [DataManager V41.2] Scanner Matrix (Fixed & Debug) Online.")
+        print(f"📦 [DataManager V41.3] Math Debugger Online.")
 
     async def initialize(self):
-        """تهيئة مدير البيانات والاتصالات"""
         print("   > [DataManager] Starting initialization...")
         try:
             self.http_client = httpx.AsyncClient(timeout=60.0)
@@ -75,92 +65,74 @@ class DataManager:
             traceback.print_exc()
 
     async def _load_markets(self):
-        """تحميل أزواج التداول وحفظها في الذاكرة"""
         try:
-            if self.exchange:
-                # التأكد من عدم تحميل الأسواق مرتين إذا كانت محملة
-                if not self.exchange.markets:
-                    await self.exchange.load_markets()
+            if self.exchange and not self.exchange.markets:
+                await self.exchange.load_markets()
                 self.market_cache = self.exchange.markets
-        except Exception as e:
-            print(f"❌ [DataManager] Market load failed: {e}")
-            traceback.print_exc()
+        except Exception: pass
 
     async def close(self):
-        """إغلاق الاتصالات بأمان"""
-        if self.http_client: 
-            await self.http_client.aclose()
-        if self.exchange: 
-            await self.exchange.close()
-        print("🛑 [DataManager] Connections Closed.")
+        if self.http_client: await self.http_client.aclose()
+        if self.exchange: await self.exchange.close()
 
-    # ==================================================================
-    # 🚀 R2 Compatibility
-    # ==================================================================
     async def load_contracts_from_r2(self):
         if not self.r2_service: return
         try:
             self.contracts_db = await self.r2_service.load_contracts_db_async()
-            print(f"   -> [DataManager] Contracts DB updated from R2: {len(self.contracts_db)} records.")
-        except Exception as e:
-            print(f"⚠️ [DataManager] R2 Load Warning: {e}")
-            self.contracts_db = {}
+        except: self.contracts_db = {}
 
-    def get_contracts_db(self) -> Dict[str, Any]:
-        return self.contracts_db
+    def get_contracts_db(self): return self.contracts_db
 
     # ==================================================================
-    # 🛡️ Layer 1: Matrix Screening (The Core)
+    # 🛡️ Layer 1: Matrix Screening
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        """
-        تقوم باختيار استراتيجية المسح بناءً على مصفوفة الكاشفات (Scanner Matrix).
-        """
-        # 1. قراءة الإعدادات الحالية
         current_regime = getattr(SystemLimits, "CURRENT_REGIME", "RANGE")
         scanner_weights = getattr(SystemLimits, "SCANNER_WEIGHTS", {"RSI_MOMENTUM": 1.0})
         min_score = getattr(SystemLimits, "L1_MIN_AFFINITY_SCORE", 15.0)
         
         print(f"🔍 [L1 Matrix] Regime: {current_regime} | Weights: {scanner_weights}")
         
-        # 2. جلب الكون الأولي (Universe) - باستخدام الدالة المصححة
+        # 1. جلب العملات
         all_tickers = await self._fetch_universe_tickers()
         if not all_tickers: 
             print("⚠️ [Layer 1] Universe fetch returned empty.")
             return []
 
-        # 3. الجلب العميق للبيانات (Batch Fetch 15m)
-        # نأخذ أفضل 80 عملة فقط لتجنب استهلاك الـ Rate Limit
-        top_candidates = all_tickers[:80]
+        # 2. الجلب العميق (أخذ عينة أكبر قليلاً للتجربة)
+        top_candidates = all_tickers[:60]
         enriched_data = await self._batch_fetch_ta_data(top_candidates, timeframe='15m', limit=100)
         
         scored_candidates = []
-        debug_log_sample = [] # لعرض عينة من الدرجات للمراقبة
+        debug_log_sample = [] 
+
+        # 🔍 فحص سلامة البيانات لأول عملة (Data Integrity Check)
+        if enriched_data:
+            first_coin = enriched_data[0]
+            if 'df' in first_coin:
+                print(f"   -> 📊 [Data Inspect] {first_coin['symbol']} DF Shape: {first_coin['df'].shape}")
+                print(f"   -> 📊 [Data Inspect] Tail:\n{first_coin['df'].tail(3)[['close', 'volume']]}")
 
-        # 4. تطبيق الكاشفات وحساب النقاط
         for item in enriched_data:
             df = item.get('df')
             if df is None or len(df) < 50: continue
             
-            # تطبيق استراتيجيات الكشف (RSI, BB, MACD, etc.)
-            scores = self._apply_scanner_strategies(df)
+            # 🔥 تطبيق الكاشفات (سيتم طباعة الخطأ إذا حدث)
+            scores = self._apply_scanner_strategies(df, item['symbol'])
             
             final_score = 0.0
             tags = []
             
-            # حساب المجموع الموزون
             for strategy, val in scores.items():
                 w = scanner_weights.get(strategy, 0.0)
                 final_score += (val['score'] * w)
                 if val['active']: tags.append(strategy)
             
-            # Boost: إضافة نقاط بسيطة إذا كان السعر يرتفع بقوة
             if item['change_24h'] > 3.0 and current_regime == "BULL": final_score += 10
             
             item['l1_score'] = final_score
             item['tags'] = tags
             
-            # تسجيل عينة للمراقبة (للتشخيص)
             if len(debug_log_sample) < 3:
                 rsi_val = scores.get('RSI_MOMENTUM', {}).get('val', 0)
                 debug_details = f"{item['symbol']}: {final_score:.1f} (RSI:{rsi_val:.1f})"
@@ -169,16 +141,13 @@ class DataManager:
             if final_score >= min_score:
                 scored_candidates.append(item)
 
-        # ترتيب النتائج
         scored_candidates.sort(key=lambda x: x['l1_score'], reverse=True)
         
-        # طباعة تقرير
         if debug_log_sample:
             print(f"   -> [DEBUG L1] Sample Scores: { ' | '.join(debug_log_sample) }")
         
         print(f"   -> Matrix selected {len(scored_candidates)} candidates (Threshold: {min_score}).")
         
-        # نمرر أفضل 40 للمعالج
         return [
             {
                 'symbol': c['symbol'],
@@ -191,33 +160,38 @@ class DataManager:
         ]
 
     # ==================================================================
-    # 🧩 Scanner Strategies Logic (The Engines)
+    # 🧩 Scanner Strategies Logic (With Error Exposure)
     # ==================================================================
-    def _apply_scanner_strategies(self, df: pd.DataFrame) -> Dict[str, Any]:
-        """تطبيق مؤشرات فنية متعددة على البيانات"""
+    def _apply_scanner_strategies(self, df: pd.DataFrame, symbol: str) -> Dict[str, Any]:
+        """تطبيق المؤشرات مع كشف الأخطاء"""
         results = {}
         try:
+            # التأكد من عدم وجود قيم فارغة (NaN)
+            # ملء الفراغات بآخر قيمة صالحة (Forward Fill)
+            df = df.ffill().bfill() 
             close = df['close']
             
-            # 1. RSI (معالجة المنطقة 30-50)
+            # 1. RSI
             rsi = ta.rsi(close, length=14)
-            curr_rsi = rsi.iloc[-1] if rsi is not None else 50
+            # قد يعود RSI بقيم NaN في البداية، نأخذ القيمة الأخيرة ونتأكد أنها رقم
+            curr_rsi = rsi.iloc[-1] if rsi is not None else 50.0
+            if np.isnan(curr_rsi): curr_rsi = 50.0 # حماية ضد NaN
             
             score_rsi = 0
             active_rsi = False
             
             if 50 < curr_rsi < 75: 
-                score_rsi = 100 # زخم صاعد
+                score_rsi = 100 
                 active_rsi = True
             elif curr_rsi <= 30: 
-                score_rsi = 80  # ارتداد
+                score_rsi = 80  
                 active_rsi = True
             elif 30 < curr_rsi <= 50: 
-                score_rsi = 40  # ✅ منطقة محايدة (نقاط جزئية)
+                score_rsi = 40  
             
             results["RSI_MOMENTUM"] = {'score': score_rsi, 'active': active_rsi, 'val': curr_rsi}
 
-            # 2. Bollinger Band Breakout
+            # 2. BB
             bb = ta.bbands(close, length=20, std=2)
             if bb is not None:
                 upper = bb[f'BBU_20_2.0'].iloc[-1]
@@ -232,20 +206,20 @@ class DataManager:
                 score_bb = 0; active_bb = False
             results["BB_BREAKOUT"] = {'score': score_bb, 'active': active_bb}
 
-            # 3. MACD Cross
+            # 3. MACD
             macd = ta.macd(close)
             if macd is not None:
                 hist = macd[f'MACDh_12_26_9'].iloc[-1]
                 score_macd = 0
                 active_macd = False
-                if hist > 0: # إيجابي
+                if hist > 0: 
                     score_macd = 100
                     active_macd = True
             else:
                 score_macd = 0; active_macd = False
             results["MACD_CROSS"] = {'score': score_macd, 'active': active_macd}
             
-            # 4. Volume Flow
+            # 4. Volume
             vol = df['volume']
             vol_ma = ta.sma(vol, length=20).iloc[-1]
             curr_vol = vol.iloc[-1]
@@ -256,18 +230,18 @@ class DataManager:
                 active_vol = True
             results["VOLUME_FLOW"] = {'score': score_vol, 'active': active_vol}
 
-        except Exception:
+        except Exception as e:
+            # 🔥 طباعة الخطأ الحقيقي هنا 🔥
+            print(f"❌ [Scanner Error] {symbol}: {e}")
+            # traceback.print_exc() # قم بتفعيل هذا إذا أردت التفاصيل المملة
             return {k: {'score': 0, 'active': False, 'val': 0} for k in ["RSI_MOMENTUM", "BB_BREAKOUT", "MACD_CROSS", "VOLUME_FLOW"]}
 
         return results
 
     # ==================================================================
-    # 🌍 Stage 0 & Pre-Filters (FIXED)
+    # 🌍 Universe & Batch Fetch (Fixed)
     # ==================================================================
     async def _fetch_universe_tickers(self) -> List[Dict[str, Any]]:
-        """
-        جلب وتصفية العملات الأولية (مصحح لتجنب حظر USDT بالخطأ).
-        """
         print("   -> 📡 [Debug] Contacting Exchange for Tickers...")
         try:
             tickers = await self.exchange.fetch_tickers()
@@ -277,20 +251,16 @@ class DataManager:
             skipped_reason = {"pair": 0, "blacklist": 0, "volume": 0}
             
             for symbol, ticker in tickers.items():
-                # 1. فلتر الزوج (USDT Only)
                 if not symbol.endswith('/USDT'): 
                     skipped_reason["pair"] += 1
                     continue
                 
-                # 🔥 FIX: نفصل العملة الأساسية للحظر 🔥
                 base_currency = symbol.split('/')[0]
                 
-                # 2. القائمة السوداء (فحص العملة الأساسية فقط)
                 if any(bad in base_currency for bad in self.BLACKLIST_TOKENS): 
                     skipped_reason["blacklist"] += 1
                     continue
                 
-                # 3. فلتر السيولة
                 vol = ticker.get('quoteVolume')
                 if vol is None: 
                     vol = ticker.get('info', {}).get('volValue') 
@@ -298,7 +268,7 @@ class DataManager:
                 if vol is None: vol = 0.0
                 else: vol = float(vol)
 
-                if vol < 100_000: # 100k للتشخيص والتجربة
+                if vol < 100_000: 
                     skipped_reason["volume"] += 1
                     continue 
 
@@ -317,15 +287,11 @@ class DataManager:
 
         except Exception as e:
             print(f"❌ [L1 Error] Fetch Tickers Failed: {e}")
-            traceback.print_exc()
             return []
 
-    # ==================================================================
-    # ⚡ Batch Fetching Utilities
-    # ==================================================================
     async def _batch_fetch_ta_data(self, candidates, timeframe='15m', limit=100):
         results = []
-        chunk_size = 15 # لعدم تجاوز الحدود
+        chunk_size = 15
         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]
@@ -348,23 +314,21 @@ class DataManager:
     # ==================================================================
     # 🎯 Public Helpers
     # ==================================================================
-    async def get_latest_price_async(self, symbol: str) -> float:
+    async def get_latest_price_async(self, symbol):
         try:
             ticker = await self.exchange.fetch_ticker(symbol)
             return float(ticker['last'])
-        except Exception: return 0.0
+        except: return 0.0
 
-    async def get_latest_ohlcv(self, symbol: str, timeframe: str = '5m', limit: int = 100) -> List[List[float]]:
+    async def get_latest_ohlcv(self, symbol, timeframe='5m', limit=100):
         try:
-            candles = await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
-            return candles or []
-        except Exception: return []
+            return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit) or []
+        except: return []
 
-    async def get_order_book_snapshot(self, symbol: str, limit: int = 20) -> Dict[str, Any]:
+    async def get_order_book_snapshot(self, symbol, limit=20):
         try:
-            ob = await self.exchange.fetch_order_book(symbol, limit)
-            return ob
-        except Exception: return {}
+            return await self.exchange.fetch_order_book(symbol, limit)
+        except: return {}
 
     def get_supported_timeframes(self):
         return list(self.exchange.timeframes.keys()) if self.exchange else []