Update data_manager.py

data_manager.py

@@ -1,5 +1,5 @@
 # ml_engine/data_manager.py
-# (V12.3 - Hybrid Data Pipeline with Correct Naming)
+# (V12.4 - Hybrid Data Pipeline with 'get_latest_ohlcv' Fix)
 
 import os
 import asyncio
@@ -45,6 +45,7 @@ class DataManager:
         self.whale_monitor = whale_monitor
         self.r2_service = r2_service
         
+        # تهيئة منصة التبادل (KuCoin كمثال)
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
             'timeout': 30000,
@@ -60,6 +61,7 @@ class DataManager:
         self.pattern_analyzer = None
 
     async def initialize(self):
+        """تهيئة مدير البيانات والاتصالات"""
         self.http_client = httpx.AsyncClient(timeout=30.0)
         await self._load_markets()
         
@@ -75,9 +77,10 @@ class DataManager:
         except Exception as e:
             print(f"⚠️ [DataManager] تحذير أثناء تهيئة النماذج المساندة: {e}")
 
-        print(f"✅ DataManager V12.3 initialized (Hybrid Threshold: {self.HYBRID_ENTRY_THRESHOLD})")
+        print(f"✅ DataManager V12.4 initialized (Hybrid Threshold: {self.HYBRID_ENTRY_THRESHOLD})")
 
     async def _load_markets(self):
+        """تحميل بيانات الأسواق وتخزينها مؤقتاً"""
         try:
             if self.exchange:
                 await self.exchange.load_markets()
@@ -86,49 +89,125 @@ class DataManager:
             print(f"❌ [DataManager] فشل تحميل الأسواق: {e}")
 
     async def close(self):
+        """إغلاق جميع الاتصالات بأمان"""
         if self.http_client: await self.http_client.aclose()
         if self.exchange: await self.exchange.close()
+        # تنظيف ذاكرة النماذج الفرعية إذا لزم الأمر
         if self.pattern_analyzer and hasattr(self.pattern_analyzer, 'clear_memory'):
             self.pattern_analyzer.clear_memory()
         if self.layer1_ranker and hasattr(self.layer1_ranker, 'clear_memory'):
             self.layer1_ranker.clear_memory()
 
+    # ==================================================================
+    # 🛡️ دوال الطبقة الأولى (Layer 1 Screening)
+    # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        print(f"🔍 [Layer 1] بدء الغربلة السريعة...")
+        """
+        الغربلة الأولية السريعة جداً بناءً على الحجم فقط.
+        تختار أعلى 150 عملة سيولةً لتمريرها للتحليل الهجين المعمق.
+        """
+        print(f"🔍 [Layer 1] بدء الغربلة السريعة (Top Liquid Assets)...")
         volume_data = await self._get_volume_data_live()
-        if not volume_data: return []
+        
+        if not volume_data:
+            print("⚠️ [Layer 1 Warning] لم يتم العثور على بيانات حجم تداول.")
+            return []
+
+        # اختيار أعلى 150 عملة من حيث حجم التداول الدولاري
         candidates = volume_data[:150]
         print(f"✅ [Layer 1] تم تمرير {len(candidates)} عملة للتحليل الهجين.")
         return candidates
 
     async def _get_volume_data_live(self):
+        """جلب بيانات الحجم الحية لجميع الأزواج"""
         try:
             tickers = await self.exchange.fetch_tickers()
             data = []
-            for s, t in tickers.items():
-                if s.endswith('/USDT') and t['quoteVolume'] and t['quoteVolume'] > 100000:
-                    data.append({'symbol': s, 'dollar_volume': t['quoteVolume'], 'current_price': t['last']})
+            for symbol, ticker in tickers.items():
+                # تصفية الأزواج: USDT فقط، وحجم تداول معقول (> 100k$) لتجنب العملات الميتة
+                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: return []
+        except Exception as e:
+            print(f"❌ [DataManager] خطأ في جلب بيانات الحجم: {e}")
+            return []
 
+    # ==================================================================
+    # 📊 دوال جلب البيانات (Data Fetching Pipeline)
+    # ==================================================================
     async def stream_ohlcv_data(self, symbols: List[Dict], queue: asyncio.Queue):
-        tfs = ['5m', '15m', '1h', '4h', '1d']
-        limit = 500
+        """
+        مولد بيانات متدفق (Streaming Generator) يجلب شموع OHLCV
+        لعدة إطارات زمنية لكل عملة، ويضع النتائج في طابور المعالجة.
+        """
+        timeframes = ['5m', '15m', '1h', '4h', '1d']
+        limit = 500 # نحتاج تاريخ كافي للمؤشرات المعقدة
+
         for sym_data in symbols:
-            sym = sym_data['symbol']
-            tasks = [self._fetch_ohlcv_live(sym, tf, limit) for tf in tfs]
+            symbol = sym_data['symbol']
+            # جلب جميع الإطارات الزمنية بالتوازي للسرعة القصوى
+            tasks = [self._fetch_ohlcv_live(symbol, tf, limit) for tf in timeframes]
             results = await asyncio.gather(*tasks, return_exceptions=False)
-            ohlcv = {}
+            
+            ohlcv_packet = {}
+            valid_packet = True
             for i, res in enumerate(results):
+                tf = timeframes[i]
+                # التحقق من جودة البيانات (على الأقل 200 شمعة للتحليل الدقيق)
                 if res and isinstance(res, list) and len(res) >= 200:
-                    ohlcv[tfs[i]] = res
-            if len(ohlcv) >= 4 and '5m' in ohlcv:
-                sym_data['ohlcv'] = ohlcv
-                await queue.put([sym_data])
-            await asyncio.sleep(0.1)
+                    ohlcv_packet[tf] = res
+                else:
+                    # إذا فقدنا إطاراً زمنياً حيوياً (مثل 5m أو 1h)، قد نعتبر الحزمة غير صالحة
+                    if tf in ['5m', '1h']: valid_packet = False
+
+            # إذا كانت الحزمة مكتملة بما يكفي، نرسلها للمعالجة
+            if valid_packet and len(ohlcv_packet) >= 4:
+                sym_data['ohlcv'] = ohlcv_packet
+                await queue.put([sym_data]) # نرسل كقائمة لتوافق المعالج
+            
+            # فاصل زمني صغير جداً لتجنب تجاوز حدود الـ API بعنف
+            await asyncio.sleep(0.05)
+
+        # علامة نهاية التدفق
         await queue.put(None)
 
-    async def _fetch_ohlcv_live(self, symbol, tf, limit):
-        try: return await self.exchange.fetch_ohlcv(symbol, tf, limit=limit)
-        except: return None
+    async def _fetch_ohlcv_live(self, symbol, timeframe, limit):
+        """دالة مساعدة لجلب الشموع مع معالجة الأخطاء البسيطة"""
+        try:
+            return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
+        except Exception:
+            # نتجاهل الأخطاء الفردية لعدم إيقاف التدفق الكامل
+            return None
+
+    # ==================================================================
+    # 🎯 دوال مساعدة للحارس والدماغ (Sentry & Brain 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 as e:
+            print(f"⚠️ [DataManager] Failed to fetch price for {symbol}: {e}")
+            return 0.0
+
+    # [NEW FIX V12.4] الدالة التي كانت مفقودة وتمت إضافتها
+    async def get_latest_ohlcv(self, symbol: str, timeframe: str = '5m', limit: int = 100) -> List[List[float]]:
+        """
+        جلب عدد محدود من الشموع الأخيرة بسرعة.
+        يستخدمها 'القناص' (Sniper) للمراقبة اللحظية الخفيفة.
+        """
+        try:
+            candles = await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
+            if candles and len(candles) > 0:
+                return candles
+            return []
+        except Exception as e:
+            # print(f"⚠️ [DataManager] Failed rapid OHLCV fetch for {symbol}: {e}")
+            return []