Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V41.0 - GEM-Architect: Debugger & Logic Fix)
+# (V41.1 - GEM-Architect: Matrix Scanner + Deep Debugger)
 # ============================================================
 
 import asyncio
@@ -14,113 +14,192 @@ 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.1 (The Scanner Matrix - Debug Edition)
+    - تتغير خوارزمية الفلترة (L1) تماماً بناءً على حالة السوق (Regime).
+    - تعتمد على SystemLimits.SCANNER_WEIGHTS لتحديد التكتيك.
+    - يتضمن سجلات تشخيصية (Debug Logs) لحل مشاكل البيانات.
+    """
+
     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, 
             'options': {'defaultType': 'spot'} 
         })
+            
         self.http_client = None
-        self.BLACKLIST_TOKENS = ['USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L']
-        print(f"📦 [DataManager V41.0] Scanner Matrix (Debug Mode) Online.")
+        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.1] Scanner Matrix (Debug Mode) Online.")
 
     async def initialize(self):
-        self.http_client = httpx.AsyncClient(timeout=60.0)
-        await self._load_markets()
+        """تهيئة مدير البيانات والاتصالات"""
+        print("   > [DataManager] Starting initialization...")
+        try:
+            self.http_client = httpx.AsyncClient(timeout=60.0)
+            await self._load_markets()
+            print(f"✅ [DataManager] Ready (Mode: {getattr(SystemLimits, 'CURRENT_REGIME', 'UNKNOWN')}).")
+        except Exception as e:
+            print(f"❌ [DataManager] Init Error: {e}")
+            traceback.print_exc()
 
     async def _load_markets(self):
-        if self.exchange and not self.exchange.markets:
-            await self.exchange.load_markets()
+        """تحميل أزواج التداول وحفظها في الذاكرة"""
+        try:
+            if self.exchange:
+                # التأكد من عدم تحميل الأسواق مرتين إذا كانت محملة
+                if 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()
 
     async def close(self):
-        if self.http_client: await self.http_client.aclose()
-        if self.exchange: await self.exchange.close()
+        """إغلاق الاتصالات بأمان"""
+        if self.http_client: 
+            await self.http_client.aclose()
+        if self.exchange: 
+            await self.exchange.close()
+        print("🛑 [DataManager] Connections Closed.")
 
+    # ==================================================================
+    # 🚀 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 = {}
+
+    def get_contracts_db(self) -> Dict[str, Any]:
+        return self.contracts_db
+
+    # ==================================================================
+    # 🛡️ Layer 1: Matrix Screening (The Core)
+    # ==================================================================
     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}")
-
-        tickers = await self._fetch_universe_tickers()
-        if not tickers: 
-            print("⚠️ [L1] Universe fetch returned empty.")
-            return []
         
-        top_candidates = tickers[:80]
+        # 2. جلب الكون الأولي (Universe) - باستخدام الدالة التشخيصية الجديدة
+        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]
         enriched_data = await self._batch_fetch_ta_data(top_candidates, timeframe='15m', limit=100)
         
         scored_candidates = []
-        debug_log_sample = [] # لعرض عينة من الدرجات
+        debug_log_sample = [] # لعرض عينة من الدرجات للمراقبة
 
+        # 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)
             
             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)
+            # Boost: إضافة نقاط بسيطة إذا كان السعر يرتفع بقوة (للتوافق مع النظام القديم)
             if item['change_24h'] > 3.0 and current_regime == "BULL": final_score += 10
             
             item['l1_score'] = final_score
             item['tags'] = tags
             
-            # تسجيل عينة للمراقبة (أول 5 عملات فقط لتجنب إغراق السجل)
-            if len(debug_log_sample) < 5:
-                debug_details = f"{item['symbol']}: {final_score:.1f} (RSI:{scores['RSI_MOMENTUM']['val']:.1f}|Score:{scores['RSI_MOMENTUM']['score']})"
+            # تسجيل عينة للمراقبة (للتشخيص)
+            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})"
                 debug_log_sample.append(debug_details)
 
             if final_score >= min_score:
                 scored_candidates.append(item)
 
+        # ترتيب النتائج
         scored_candidates.sort(key=lambda x: x['l1_score'], reverse=True)
         
-        # طباعة تقرير التصحيح
-        print(f"   -> [DEBUG L1] Sample Scores: { ' | '.join(debug_log_sample) }")
+        # طباعة تقرير
+        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'],
-                'quote_volume': c['quote_volume'],
-                'current_price': c['current_price'],
-                'type': ','.join(c['tags']),
-                'l1_score': c['l1_score']
+                'quote_volume': c.get('quote_volume', 0),
+                'current_price': c.get('current_price', 0),
+                'type': ','.join(c.get('tags', [])),
+                'l1_score': c.get('l1_score', 0)
             }
             for c in scored_candidates[:40] 
         ]
 
+    # ==================================================================
+    # 🧩 Scanner Strategies Logic (The Engines)
+    # ==================================================================
     def _apply_scanner_strategies(self, df: pd.DataFrame) -> Dict[str, Any]:
+        """تطبيق مؤشرات فنية متعددة على البيانات"""
         results = {}
         try:
             close = df['close']
             
-            # 1. RSI (تم إصلاح المنطقة العمياء)
+            # 1. RSI (تم إصلاح المنطقة العمياء 30-50)
             rsi = ta.rsi(close, length=14)
             curr_rsi = rsi.iloc[-1] if rsi is not None else 50
             
@@ -128,17 +207,17 @@ class DataManager:
             active_rsi = False
             
             if 50 < curr_rsi < 75: 
-                score_rsi = 100 
+                score_rsi = 100 # زخم صاعد قوي
                 active_rsi = True
-            elif curr_rsi <= 30: # Oversold
-                score_rsi = 80
+            elif curr_rsi <= 30: 
+                score_rsi = 80  # تشبع بيعي (ارتداد محتمل)
                 active_rsi = True
-            elif 30 < curr_rsi <= 50: # ✅ المنطقة المحايدة (تمت إضافتها)
-                score_rsi = 40 # نعطيها بعض النقاط بدلاً من الصفر
+            elif 30 < curr_rsi <= 50: 
+                score_rsi = 40  # ✅ المنطقة المحايدة (لا تعاقب بشدة)
             
             results["RSI_MOMENTUM"] = {'score': score_rsi, 'active': active_rsi, 'val': curr_rsi}
 
-            # 2. BB Breakout
+            # 2. Bollinger Band Breakout
             bb = ta.bbands(close, length=20, std=2)
             if bb is not None:
                 upper = bb[f'BBU_20_2.0'].iloc[-1]
@@ -153,7 +232,7 @@ class DataManager:
                 score_bb = 0; active_bb = False
             results["BB_BREAKOUT"] = {'score': score_bb, 'active': active_bb}
 
-            # 3. MACD
+            # 3. MACD Cross
             macd = ta.macd(close)
             if macd is not None:
                 hist = macd[f'MACDh_12_26_9'].iloc[-1]
@@ -172,40 +251,87 @@ class DataManager:
             curr_vol = vol.iloc[-1]
             score_vol = 0
             active_vol = False
-            if curr_vol > (vol_ma * 1.2): # خففنا الشرط من 1.5 إلى 1.2
+            if curr_vol > (vol_ma * 1.2): # خففنا الشرط قليلاً
                 score_vol = 100
                 active_vol = True
             results["VOLUME_FLOW"] = {'score': score_vol, 'active': active_vol}
 
-        except Exception as e:
-            # في حال حدوث خطأ في الحساب، نعيد قيم صفرية لعدم إيقاف النظام
-            # print(f"Indicator Error: {e}") 
+        except Exception:
+            # في حال الخطأ نعيد أصفار لعدم إيقاف النظام
             return {k: {'score': 0, 'active': False, 'val': 0} for k in ["RSI_MOMENTUM", "BB_BREAKOUT", "MACD_CROSS", "VOLUME_FLOW"]}
 
         return results
 
-    # --- Helpers ---
-    async def _fetch_universe_tickers(self):
+    # ==================================================================
+    # 🌍 Stage 0 & Pre-Filters (With Deep Debugging)
+    # ==================================================================
+    async def _fetch_universe_tickers(self) -> List[Dict[str, Any]]:
+        """
+        جلب وتصفية العملات الأولية مع تسجيلات تشخيصية دقيقة.
+        """
+        print("   -> 📡 [Debug] Contacting Exchange for Tickers...")
         try:
             tickers = await self.exchange.fetch_tickers()
+            print(f"   -> 📡 [Debug] Raw Tickers Received: {len(tickers)}")
+            
             candidates = []
+            skipped_reason = {"pair": 0, "blacklist": 0, "volume": 0}
+            
+            # فحص عينة من البيانات لمعرفة هيكلها (لأول عملة)
+            if tickers:
+                sample_key = list(tickers.keys())[0]
+                # print(f"   -> 📡 [Debug] Sample Data ({sample_key}): {tickers[sample_key]}")
+
             for symbol, ticker in tickers.items():
-                if not symbol.endswith('/USDT'): continue
-                if any(bad in symbol for bad in self.BLACKLIST_TOKENS): continue
-                if not ticker.get('quoteVolume') or ticker['quoteVolume'] < 300_000: continue # تخفيف شرط السيولة قليلاً
+                # 1. فلتر الزوج (USDT Only)
+                if not symbol.endswith('/USDT'): 
+                    skipped_reason["pair"] += 1
+                    continue
+                
+                # 2. القائمة السوداء
+                if any(bad in symbol for bad in self.BLACKLIST_TOKENS): 
+                    skipped_reason["blacklist"] += 1
+                    continue
+                
+                # 3. فلتر السيولة (Quote Volume)
+                # KuCoin قد تستخدم مسميات مختلفة أحياناً
+                vol = ticker.get('quoteVolume')
+                if vol is None: 
+                    # محاولة بديلة لـ KuCoin API V1/V2 differences
+                    vol = ticker.get('info', {}).get('volValue') 
+                
+                if vol is None: vol = 0.0
+                else: vol = float(vol)
+
+                # تخفيف الشرط مؤقتاً للتشخيص (100k بدلاً من 300k)
+                if vol < 100_000: 
+                    skipped_reason["volume"] += 1
+                    continue 
+
                 candidates.append({
                     'symbol': symbol,
-                    'quote_volume': ticker['quoteVolume'],
-                    'current_price': ticker['last'],
-                    'change_24h': float(ticker.get('percentage', 0.0))
+                    'quote_volume': vol,
+                    'current_price': float(ticker['last']) if ticker.get('last') else 0.0,
+                    'change_24h': float(ticker.get('percentage', 0.0)) 
                 })
+            
+            print(f"   -> 📊 [Debug] Filter Stats: BadPair={skipped_reason['pair']}, Blacklist={skipped_reason['blacklist']}, LowVol={skipped_reason['volume']}")
+            print(f"   -> ✅ [Debug] Candidates Passed: {len(candidates)}")
+            
             candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
             return candidates
-        except Exception: return []
 
+        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]
@@ -224,13 +350,27 @@ class DataManager:
             candidate['df'] = df 
             return candidate
         except: return None
-    
-    async def get_latest_price_async(self, symbol):
-        t = await self.exchange.fetch_ticker(symbol)
-        return float(t['last'])
-    async def get_latest_ohlcv(self, symbol, tf, limit=100):
-        return await self.exchange.fetch_ohlcv(symbol, tf, limit=limit)
-    async def get_order_book_snapshot(self, symbol, limit=20):
-        return await self.exchange.fetch_order_book(symbol, limit)
-    async def load_contracts_from_r2(self): pass
-    def get_contracts_db(self): return self.contracts_db
+
+    # ==================================================================
+    # 🎯 Public Helpers
+    # ==================================================================
+    async def get_latest_price_async(self, symbol: str) -> float:
+        try:
+            ticker = await self.exchange.fetch_ticker(symbol)
+            return float(ticker['last'])
+        except Exception: return 0.0
+
+    async def get_latest_ohlcv(self, symbol: str, timeframe: str = '5m', limit: int = 100) -> List[List[float]]:
+        try:
+            candles = await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
+            return candles or []
+        except Exception: return []
+
+    async def get_order_book_snapshot(self, symbol: str, limit: int = 20) -> Dict[str, Any]:
+        try:
+            ob = await self.exchange.fetch_order_book(symbol, limit)
+            return ob
+        except Exception: return {}
+
+    def get_supported_timeframes(self):
+        return list(self.exchange.timeframes.keys()) if self.exchange else []