Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,5 +1,5 @@
 # ml_engine/data_manager.py
-# (V13.1 - Cleaned & Fixed - Removed Obsolete Ranker)
+# (V13.2 - GEM-Architect: Added Order Book Capability)
 
 import os
 import asyncio
@@ -9,7 +9,7 @@ import ccxt.async_support as ccxt
 import logging
 from typing import List, Dict, Any
 
-# إعدادات التسجيل لتقليل الضجيج
+# إعدادات التسجيل
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("httpcore").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
@@ -23,7 +23,7 @@ class DataManager:
         self.whale_monitor = whale_monitor
         self.r2_service = r2_service
         
-        # إعداد المنصة (KuCoin كمثال)
+        # إعداد المنصة (KuCoin)
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
             'timeout': 30000,
@@ -36,11 +36,8 @@ class DataManager:
         """تهيئة مدير البيانات والاتصالات"""
         print("   > [DataManager] Starting initialization...")
         self.http_client = httpx.AsyncClient(timeout=30.0)
-        
-        # تحميل الأسواق
         await self._load_markets()
-        
-        print(f"✅ [DataManager V13.1] Ready (Titan Architecture).")
+        print(f"✅ [DataManager V13.2] Ready (Order Book Enabled).")
 
     async def _load_markets(self):
         """تحميل بيانات الأسواق وتخزينها مؤقتاً"""
@@ -62,7 +59,6 @@ class DataManager:
     # 🚀 R2 Compatibility
     # ==================================================================
     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()
@@ -77,16 +73,14 @@ class DataManager:
     # 🛡️ Layer 1 Screening (Data Filter)
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        """
-        الغربلة الأولية السريعة بناءً على الحجم والسيولة (بدون Ranker ML).
-        """
+        """الغربلة الأولية السريعة بناءً على الحجم والسيولة."""
         print(f"🔍 [Layer 1] Screening top liquid assets...")
         volume_data = await self._get_volume_data_live()
         
         if not volume_data:
             return []
 
-        # نأخذ أعلى 150 عملة من حيث السيولة لتمريرها للمعالج (Titan)
+        # نأخذ أعلى 150 عملة
         candidates = volume_data[:150]
         return candidates
 
@@ -96,24 +90,21 @@ class DataManager:
             tickers = await self.exchange.fetch_tickers()
             data = []
             for symbol, ticker in tickers.items():
-                # تصفية أزواج USDT ذات الحجم المحترم
                 if symbol.endswith('/USDT') and ticker.get('quoteVolume') and ticker['quoteVolume'] > 100000:
                     data.append({
                         'symbol': symbol,
                         'dollar_volume': ticker['quoteVolume'],
                         'current_price': ticker['last']
                     })
-            # الترتيب تنازلياً حسب الحجم
             data.sort(key=lambda x: x['dollar_volume'], reverse=True)
             return data
         except Exception:
             return []
 
     # ==================================================================
-    # 🎯 Helper Functions
+    # 🎯 Data Fetching Helpers
     # ==================================================================
     async def get_latest_price_async(self, symbol: str) -> float:
-        """جلب السعر الحالي"""
         try:
             ticker = await self.exchange.fetch_ticker(symbol)
             return float(ticker['last'])
@@ -121,10 +112,24 @@ class DataManager:
             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)
             if candles: return candles
             return []
         except Exception:
-            return []
+            return []
+
+    # 🔥 [NEW] Order Book Fetcher
+    async def get_order_book_snapshot(self, symbol: str, limit: int = 20) -> Dict[str, Any]:
+        """
+        جلب لقطة سريعة لدفتر الطلبات (Bids/Asks).
+        يستخدم لتحليل عمق السوق وتأكيد القناص.
+        """
+        try:
+            # ccxt يرجع الدفتر بصيغة {'bids': [[price, size], ...], 'asks': ...}
+            ob = await self.exchange.fetch_order_book(symbol, limit)
+            return ob
+        except Exception as e:
+            # طباعة خطأ خفيف لتجنب إغراق السجلات
+            # print(f"⚠️ [DataManager] Failed to fetch OB for {symbol}: {e}")
+            return {}