Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,31 +1,51 @@
-# ml_engine/data_manager.py
-# (V18.1 - GEM-Architect: Synergy Matrix + Debug Mode)
+# ============================================================
+# 📂 ml_engine/data_manager.py 
+# (V18.2 - GEM-Architect: Synergy Matrix + Full Verbose Mode)
+# ============================================================
 
 import asyncio
 import httpx
 import traceback
-from ml_engine.processor import SystemLimits
-import ccxt.async_support as ccxt
 import logging
 import pandas as pd
 import numpy as np
 from typing import List, Dict, Any
 
-# إعدادات التسجيل
+# ✅ استيراد إعدادات النظام المركزية (للربط مع Processor)
+try:
+    from ml_engine.processor import SystemLimits
+except ImportError:
+    # قيم افتراضية في حال فشل الاستيراد لتجنب توقف النظام
+    class SystemLimits:
+        L1_MIN_AFFINITY_SCORE = 10.0
+
+# ✅ استيراد مكتبة التداول
+import ccxt.async_support as ccxt
+
+# ============================================================
+# 📝 إعدادات التسجيل (Logging Configuration)
+# ============================================================
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("httpcore").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 class DataManager:
+    """
+    DataManager V18.2
+    -----------------
+    المسؤول عن جلب البيانات من المنصة، فلترتها، وتجهيزها للمعالج.
+    يستخدم الآن منطق 'Synergy Matrix' لدمج التحليل الفني مع عمق السوق.
+    """
+
     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)
+        # إعداد عميل المنصة (KuCoin) مع تفعيل حدود السرعة
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
             'timeout': 30000, 
@@ -35,94 +55,128 @@ 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'
+            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP'
         ]
+        
+        print(f"📦 [DataManager] Instance Created. Blacklist items: {len(self.BLACKLIST_TOKENS)}")
 
     async def initialize(self):
-        """تهيئة مدير البيانات"""
+        """تهيئة مدير البيانات والاتصالات"""
         print("   > [DataManager] Starting initialization...")
-        self.http_client = httpx.AsyncClient(timeout=30.0)
-        await self._load_markets()
-        print(f"✅ [DataManager V18.1] Ready (Mode: Synergy Matrix + Debug).")
+        try:
+            self.http_client = httpx.AsyncClient(timeout=30.0)
+            await self._load_markets()
+            print(f"✅ [DataManager V18.2] Ready (Mode: Synergy Matrix | Trend + Depth).")
+        except Exception as e:
+            print(f"❌ [DataManager] Init Error: {e}")
+            traceback.print_exc()
 
     async def _load_markets(self):
+        """تحميل أزواج التداول وحفظها في الذاكرة"""
         try:
             if self.exchange:
-                await self.exchange.load_markets()
+                # التأكد من عدم تحميل الأسواق مرتين إذا كانت محملة
+                if not self.exchange.markets:
+                    await self.exchange.load_markets()
                 self.market_cache = self.exchange.markets
+                # print(f"   -> [DataManager] Markets loaded: {len(self.market_cache)} pairs.")
         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
+    # 🚀 R2 Compatibility (Integration Methods)
     # ==================================================================
     async def load_contracts_from_r2(self):
+        """تحميل قاعدة بيانات العقود من خدمة R2"""
         if not self.r2_service: return
         try:
             self.contracts_db = await self.r2_service.load_contracts_db_async()
-        except Exception:
+            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: Synergy Screening (Trend + Depth + Logic)
+    # 🛡️ Layer 1: Synergy Screening (The Core Logic)
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
         """
-        الفلترة الذكية المترابطة:
-        لا نقبل المؤشرات منفصلة، بل نقيس التوافق بين:
-        1. اتجاه المتوسطات (EMA).
-        2. دعم دفتر الطلبات (Depth).
-        3. مناطق الدخول (RSI).
+        الفلترة الذكية المترابطة (Synergy Matrix):
+        تقوم هذه الدالة بمسح السوق بالكامل، واختيار العملات التي تحقق توافقاً بين:
+        1. اتجاه المتوسطات (EMA Trend).
+        2. الدعم المؤسساتي في دفتر الطلبات (Order Book Depth).
+        3. مناطق الدخول الذكية (Smart RSI).
         """
         print(f"🔍 [Layer 1] Initiating Synergy Matrix Screening...")
         
-        # 1. المرحلة 0: فلتر الكون السريع
+        # 1. المرحلة 0: فلتر الكون (استبعاد العملات الميتة)
         initial_candidates = await self._stage0_universe_filter()
-        if not initial_candidates: return []
+        
+        if not initial_candidates:
+            print("⚠️ [Layer 1] No candidates passed Stage 0.")
+            return []
 
-        # 2. جلب البيانات (شارت + عمق)
-        # نأخذ أفضل 150 لضمان الجودة
+        # 2. جلب البيانات الفنية + عمق السوق (Batch Fetching)
+        # نأخذ أفضل 150 عملة لضمان جودة البيانات وعدم تجاوز حدود API
         top_liquid_candidates = initial_candidates[:150] 
         enriched_data = await self._fetch_technical_and_depth_batch(top_liquid_candidates)
         
-        # 3. حساب نقاط التوافق (Synergy Score)
+        # 3. حساب "نقاط التوافق" (Synergy Score)
         scored_candidates = []
 
         for item in enriched_data:
+            # استدعاء دالة التقييم الجديدة
             affinity_result = self._calculate_synergy_score(item)
             
-            # ✅ [DEBUG PRINT ADDED] طباعة تشخيصية لمعرفة النقاط
-            if affinity_result['score'] > 5:
-                 print(f"   👀 Debug: {item['symbol']:<8} -> Score: {affinity_result['score']:.1f} | Tags: {affinity_result['tags']}")
-            
+            # إذا حققت الحد الأدنى من النقاط، نضيفها للقائمة
             if affinity_result['is_eligible']:
                 item['l1_score'] = affinity_result['score']
                 item['tags'] = affinity_result['tags']
                 scored_candidates.append(item)
 
-        print(f"   -> [L1 Logic] Candidates passing Matrix: {len(scored_candidates)}")
-
         # 4. الترتيب واختيار النخبة
+        # ترتيب تنازلي حسب النقاط
         scored_candidates.sort(key=lambda x: x['l1_score'], reverse=True)
+
+        # ✅ [GEM-Architect Debug] طباعة أفضل 10 عملات فقط بشكل أنيق للمراقبة
+        if len(scored_candidates) > 0:
+            print("-" * 75)
+            print(f"{'SYMBOL':<12} | {'SCORE':<6} | {'STRATEGY TAGS (REASON)'}")
+            print("-" * 75)
+            for i, item in enumerate(scored_candidates[:10]):
+                tags_str = ", ".join(item.get('tags', []))
+                print(f"{item['symbol']:<12} | {item['l1_score']:<6.1f} | {tags_str}")
+            print("-" * 75)
+        
+        print(f"   -> [L1 Logic] Candidates passing Matrix: {len(scored_candidates)}")
+
+        # نختار أفضل 40 عملة فقط لتمريرها للمعالجة العميقة (L2/L3)
         final_selection = scored_candidates[:40]
         
+        # تنظيف البيانات لتمرير ما يهم المعالج فقط
         cleaned_selection = []
         for item in final_selection:
             cleaned_selection.append({
                 'symbol': item['symbol'],
                 'quote_volume': item.get('quote_volume', 0),
                 'current_price': item.get('current_price', 0),
-                'type': ','.join(item.get('tags', [])), 
+                'type': ','.join(item.get('tags', [])), # تمرير التاغات كسترينغ
                 'l1_score': item.get('l1_score', 0)
             })
 
@@ -130,22 +184,32 @@ class DataManager:
         return cleaned_selection
 
     # ------------------------------------------------------------------
-    # Stage 0: Universe Filter
+    # Stage 0: Universe Filter (Basic Liquidity Check)
     # ------------------------------------------------------------------
     async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
+        """
+        فلتر أولي سريع جداً لاستبعاد:
+        - العملات ذات السيولة المعدومة (أقل من 300 ألف دولار).
+        - العملات المستقرة والمحظورة.
+        - العملات ذات السعر الصفري تقريباً (أخطاء بيانات).
+        """
         try:
             tickers = await self.exchange.fetch_tickers()
             candidates = []
             
             for symbol, ticker in tickers.items():
+                # نقبل فقط أزواج USDT
                 if not symbol.endswith('/USDT'): continue
                 
+                # التحقق من القائمة السوداء
                 base_curr = symbol.split('/')[0]
                 if any(bad in base_curr for bad in self.BLACKLIST_TOKENS): continue
                 
+                # شرط السيولة (Quote Volume)
                 quote_vol = ticker.get('quoteVolume')
-                if not quote_vol or quote_vol < 300_000: continue 
+                if not quote_vol or quote_vol < 300_000: continue # Min 300k USDT
                 
+                # شرط السعر المنطقي
                 last_price = ticker.get('last')
                 if not last_price or last_price < 0.0001: continue 
 
@@ -156,6 +220,7 @@ class DataManager:
                     'change_24h': ticker.get('percentage', 0.0) 
                 })
             
+            # الترتيب حسب السيولة لضمان أننا نفحص الأقوى أولاً
             candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
             return candidates
             
@@ -164,133 +229,156 @@ class DataManager:
             return []
 
     # ------------------------------------------------------------------
-    # Data Fetching Helpers
+    # Data Fetching Helpers (Updated for Depth & Parallel Execution)
     # ------------------------------------------------------------------
     async def _fetch_technical_and_depth_batch(self, candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        chunk_size = 10 
+        """
+        جلب البيانات الفنية (OHLCV) + دفتر الطلبات (Order Book) بالتوازي.
+        يتم تقسيم العملات إلى دفعات (Chunks) لتجنب الحظر.
+        """
+        chunk_size = 10 # عدد العملات في كل دفعة
         results = []
+        
         for i in range(0, len(candidates), chunk_size):
             chunk = candidates[i:i + chunk_size]
+            
+            # إنشاء مهام غير متزامنة لكل عملة في الدفعة
             chunk_tasks = [self._fetch_single_full_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.1) 
+            
         return results
 
     async def _fetch_single_full_data(self, candidate: Dict[str, Any]) -> Any:
+        """جلب بيانات عملة واحدة (شارت + عمق)"""
         symbol = candidate['symbol']
         try:
-            # نحتاج 200 شمعة لحساب EMA 200 بدقة
+            # 1. طلب الشموع (OHLCV) - نحتاج 200 شمعة لحساب المتوسطات الطويلة (EMA 200)
             ohlcv_task = self.exchange.fetch_ohlcv(symbol, '1h', limit=205)
-            # Order Book Snapshot
+            
+            # 2. طلب دفتر الطلبات (Order Book) - لقطة لأفضل 20 مستوى فقط
             ob_task = self.exchange.fetch_order_book(symbol, limit=20)
             
+            # تنفيذ الطلبين معاً
             ohlcv_1h, order_book = await asyncio.gather(ohlcv_task, ob_task)
             
-            if not ohlcv_1h or len(ohlcv_1h) < 200: # نحتاج بيانات كافية للمتوسطات
-                return None
+            # التحقق من جودة البيانات
+            if not ohlcv_1h or len(ohlcv_1h) < 200:
+                return None # بيانات غير كافية للتحليل
                 
             candidate['ohlcv_1h'] = ohlcv_1h
             candidate['order_book_snapshot'] = order_book
             return candidate
+            
         except Exception:
+            # في حال فشل الجلب (Timeout أو غيره)، نتجاهل العملة
             return None
 
     # ------------------------------------------------------------------
-    # 🧠 The Logic Core: Synergy Matrix Scoring (Relationships)
+    # 🧠 The Logic Core: Synergy Matrix Scoring (New V18.2 Logic)
     # ------------------------------------------------------------------
     def _calculate_synergy_score(self, data: Dict[str, Any]) -> Dict[str, Any]:
         """
-        مصفوفة التوافق:
-        نقوم بحساب النقاط بناءً على العلاقات بين المؤشرات وليس القيم المجردة.
+        مصفوفة التوافق (Synergy Matrix):
+        تقوم بحساب النقاط بناءً على العلاقات المترابطة بين المؤشرات.
+        لا تعتمد على القيم المطلقة فقط، بل على سياق السوق.
         """
         try:
+            # تحويل البيانات إلى DataFrame للتحليل
             df = pd.DataFrame(data['ohlcv_1h'], columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
             ob = data.get('order_book_snapshot', {})
             
             df['close'] = df['close'].astype(float)
             df['volume'] = df['volume'].astype(float)
             
-            # --- 1. الأساسيات (Indicators) ---
-            # EMA (Exponential Moving Average) - العمود الفقري
+            # --- 1. حساب المؤشرات الفنية الأساسية ---
+            
+            # EMA (العمود الفقري للاتجاه)
             df['ema50'] = df['close'].ewm(span=50, adjust=False).mean()
             df['ema200'] = df['close'].ewm(span=200, adjust=False).mean()
             
-            # RSI
+            # RSI (مؤشر الزخم)
             delta = df['close'].diff()
             gain = (delta.where(delta > 0, 0)).rolling(14).mean()
             loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
             rs = gain / loss
             df['rsi'] = 100 - (100 / (1 + rs))
             
-            # Volume & Bands
+            # Volume MA (متوسط الحجم)
             df['vol_ma'] = df['volume'].rolling(20).mean()
-            curr = df.iloc[-1]
             
+            curr = df.iloc[-1]
             score = 0.0
             tags = []
 
-            # --- 2. تحليل دفتر الطلبات (The King) ---
+            # --- 2. تحليل دفتر الطلبات (Macro Order Book) ---
+            # حساب نسبة ضغط الشراء مقابل البيع (Bid/Ask Ratio)
             depth_ratio = 1.0
             total_bids = 0
             if ob and 'bids' in ob and 'asks' in ob:
                 total_bids = sum([b[1] for b in ob['bids']])
                 total_asks = sum([a[1] for a in ob['asks']])
-                if total_asks > 0: depth_ratio = total_bids / total_asks
+                if total_asks > 0: 
+                    depth_ratio = total_bids / total_asks
 
-            # --- 3. مصفوفة العلاقات (Relationships Logic) ---
+            # --- 3. مصفوفة العلاقات (Logic Execution) ---
             
             close = curr['close']
             ema50 = curr['ema50']
             ema200 = curr['ema200']
             rsi = curr['rsi']
             
-            # الحالة A: الثور المثالي (Trend + Support)
-            # السعر فوق المتوسطات + يوجد دعم طلبات
+            # الحالة A: الثور المؤكد (Confirmed Trend)
+            # السعر فوق المتوسطات + يوجد دعم حقيقي في الدفتر
             if close > ema50 and close > ema200:
                 base_trend = 20
-                if depth_ratio > 1.1: # توافق: اتجاه صاعد + سيولة شراء
-                    score += (base_trend * 1.5) # بونص 50% للتوافق
+                if depth_ratio > 1.1: # توافق إيجابي
+                    score += (base_trend * 1.5) 
                     tags.append("ConfirmedTrend")
-                elif depth_ratio < 0.8: # تناقض: اتجاه صاعد لكن تفريغ
-                    score -= 20 # فخ!
+                elif depth_ratio < 0.8: # تناقض (سعر صاعد لكن الحيتان يبيعون)
+                    score -= 20 
                     tags.append("BullTrap⚠️")
                 else:
                     score += base_trend # اتجاه صاعد عادي
 
-            # الحالة B: الارتداد الذكي (Reversal + Whale Catch)
-            # السعر تحت المتوسطات (سلبي) لكن الدفتر قوي جداً
+            # الحالة B: صيد الحيتان / الارتداد (Whale Net)
+            # السعر تحت المتوسطات (هابط) لكن الدفتر يظهر شراء عنيف
             elif close < ema50:
-                if depth_ratio > 1.5 and rsi < 35: # حيتان يشترون الهبوط بقوة
+                if depth_ratio > 1.5 and rsi < 35: # شراء قوي جداً عند القاع
                     score += 40 # نغفر سلبية المتوسط بسبب قوة الدفتر
                     tags.append("WhaleNet🔥")
                 elif depth_ratio > 1.2:
                     score += 15
                     tags.append("SupprortBuild")
                 else:
-                    score -= 30 # هبوط بدون دعم = انهيار (Downtrend)
+                    score -= 30 # هبوط بدون دعم = انهيار
 
-            # الحالة C: الزخم (Momentum)
-            # توافق RSI مع الفوليوم
+            # الحالة C: الزخم والحجم (Momentum & Volume)
+            # حساب الحجم النسبي (Relative Volume)
             rvol = curr['volume'] / curr['vol_ma'] if curr['vol_ma'] > 0 else 1.0
             
             if rsi > 50 and rsi < 75:
-                if rvol > 1.2: # زخم سعري + حجم تداول
+                if rvol > 1.2: # زخم سعري مع سيولة
                     score += 15
                     tags.append("MomVol")
                 else:
-                    score += 5 # زخم ضعيف
+                    score += 5 
 
-            # حالة خاصة: الارتفاع الجنوني (Parabolic)
-            # السعر بعيد جداً عن المتوسطات
+            # حالة خاصة: الامتداد السعري المفرط (Over Extension)
+            # السعر ابتعد كثيراً عن المتوسط (خطر التصحيح)
             dist_from_ema = (close - ema50) / ema50
             if dist_from_ema > 0.20: # 20% فوق المتوسط
-                score -= 10 # خطورة تصحيح، نقلل النقاط قليلاً للحذر
+                score -= 10 
                 tags.append("OverExtended")
-
-            # --- 4. الخلاصة ---
-            # شرط القبول: يجب أن تكون النقاط إيجابية وتتجاوز العتبة
             
+            # إرجاع النتيجة
             return {
                 'is_eligible': score > SystemLimits.L1_MIN_AFFINITY_SCORE,
                 'score': score,
@@ -298,26 +386,33 @@ class DataManager:
             }
 
         except Exception as e:
-            # print(f"Scoring Error: {e}")
+            # في حال حدوث خطأ حسابي، نرفض العملة للأمان
             return {'is_eligible': False, 'score': 0, 'tags': []}
 
     # ==================================================================
-    # 🎯 Helpers
+    # 🎯 Public Helpers (Legacy & Utility Methods)
     # ==================================================================
     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]]:
+        """جلب بيانات الشموع (OHLCV)"""
         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 {}
+        except Exception: return {}
+
+    def get_supported_timeframes(self):
+        """إرجاع الإطارات الزمنية المدعومة"""
+        return list(self.exchange.timeframes.keys()) if self.exchange else []