Update data_manager.py

data_manager.py

@@ -1,5 +1,5 @@
 # ml_engine/data_manager.py
-# (V12.0 - Titan Ready Data Pipeline)
+# (V12.1 - Titan + Hybrid Support Pipeline)
 
 import os
 import asyncio
@@ -23,7 +23,12 @@ except ImportError:
 from ml_engine.indicators import AdvancedTechnicalAnalyzer
 from ml_engine.monte_carlo import MonteCarloAnalyzer
 from ml_engine.ranker import Layer1Ranker
-# لاحظ: لم نعد بحاجة لمحرك الأنماط القديم هنا، Titan سيقوم بالمهمة في Processor
+# 🔥 إعادة استيراد محرك الأنماط لدعم النظام الهجين
+try:
+    from ml_engine.patterns import ChartPatternAnalyzer
+except ImportError:
+    print("⚠️ [DataManager] لم يتم العثور على ml_engine/patterns.py")
+    ChartPatternAnalyzer = None
 
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("httpcore").setLevel(logging.WARNING)
@@ -32,11 +37,10 @@ logging.getLogger("ccxt").setLevel(logging.WARNING)
 class DataManager:
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
         # ==================================================================
-        # ⚙️ إعدادات التحكم المركزية (V12.0 Titan Thresholds)
+        # ⚙️ إعدادات التحكم المركزية (V12 Hybrid Thresholds)
         # ==================================================================
-        self.SCREENING_THRESHOLD = 0.40   # عتبة الغربلة الأولية السريعة (Layer 1)
-        self.TITAN_ENTRY_THRESHOLD = 0.90 # عتبة دخول Titan الصارمة (Layer 2)
-        self.PROFIT_SAVE_THRESHOLD = 0.30 # عتبة الخروج الذكي (إذا انخفضت ثقة Titan عن هذا الحد)
+        self.SCREENING_THRESHOLD = 0.40   # غربلة أولية
+        self.TITAN_ENTRY_THRESHOLD = 0.90 # عتبة Titan الصارمة
         # ==================================================================
 
         self.contracts_db = contracts_db or {}
@@ -50,33 +54,40 @@ class DataManager:
             
         self.http_client = None
         self.market_cache = {}
-        self.last_market_load = None
 
         self.technical_analyzer = AdvancedTechnicalAnalyzer()
         self.mc_analyzer = MonteCarloAnalyzer()
+        
+        # نماذج الطبقة الأولى والثانية المساندة
         self.layer1_ranker = None
+        self.pattern_analyzer = None # 🔥 تمت إعادته لتجنب AttributeError
 
     async def initialize(self):
-        """تهيئة الاتصالات وتحميل نماذج الطبقة الأولى"""
+        """تهيئة الاتصالات وتحميل جميع النماذج المساندة"""
         self.http_client = httpx.AsyncClient(timeout=30.0)
         await self._load_markets()
         
-        print("   > [DataManager V12.0] تهيئة الكاشف المصغر (Layer1 Ranker)...")
+        print("   > [DataManager] تهيئة النماذج المساندة (Ranker + Patterns)...")
         try:
-            # تأكد من وجود ملف النموذج هذا في مساره
+            # 1. الكاشف المصغر (Ranker)
             self.layer1_ranker = Layer1Ranker(model_path="ml_models/layer1_ranker.lgbm")
             await self.layer1_ranker.initialize()
+            
+            # 2. محرك الأنماط (Patterns) - للنظام الهجين
+            if ChartPatternAnalyzer:
+                self.pattern_analyzer = ChartPatternAnalyzer(r2_service=self.r2_service)
+                await self.pattern_analyzer.initialize()
+                
         except Exception as e:
-            print(f"⚠️ [DataManager] تحذير: فشل تهيئة Layer1 Ranker: {e}")
+            print(f"⚠️ [DataManager] تحذير أثناء تهيئة النماذج المساندة: {e}")
 
-        print(f"✅ DataManager V12.0 initialized. Titan Threshold: {self.TITAN_ENTRY_THRESHOLD}")
+        print(f"✅ DataManager V12.1 initialized (Hybrid Ready).")
 
     async def _load_markets(self):
         try:
             if self.exchange:
                 await self.exchange.load_markets()
                 self.market_cache = self.exchange.markets
-                self.last_market_load = datetime.now()
         except Exception as e:
             print(f"❌ [DataManager] فشل تحميل الأسواق: {e}")
 
@@ -84,123 +95,66 @@ class DataManager:
         if self.http_client: await self.http_client.aclose()
         if self.exchange: await self.exchange.close()
 
-    async def get_market_context_async(self):
-        try:
-            sentiment = await self.get_sentiment_live_async()
-            prices = await self._get_prices_live()
-            return {
-                'timestamp': datetime.now().isoformat(),
-                'bitcoin_price_usd': prices.get('bitcoin'),
-                'fear_and_greed_index': sentiment.get('feargreed_value') if sentiment else 50,
-                'market_trend': self._determine_market_trend(prices.get('bitcoin'), sentiment)
-            }
-        except: return {'market_trend': 'UNKNOWN'}
-
-    async def get_sentiment_live_async(self):
-        try:
-            async with httpx.AsyncClient(timeout=10) as client:
-                resp = await client.get("https://api.alternative.me/fng/")
-                if resp.status_code == 200:
-                    data = resp.json()
-                    return {"feargreed_value": int(data['data'][0]['value'])}
-        except: return None
-
-    async def _get_prices_live(self):
-        if not self.exchange: return {}
-        try:
-            btc = await self.exchange.fetch_ticker('BTC/USDT')
-            return {'bitcoin': btc['last']}
-        except: return {}
-
-    def _determine_market_trend(self, btc_price, sentiment):
-        # منطق بسيط لتحديد الاتجاه العام
-        if not btc_price: return "NEUTRAL"
-        score = 0
-        if btc_price > 65000: score += 1 # مثال
-        if sentiment and sentiment.get('feargreed_value', 50) > 60: score += 1
-        return "BULLISH" if score > 0 else "BEARISH"
+        # تنظيف الذاكرة عند الإغلاق
+        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()
 
     # ==================================================================
-    # 🔴 الطبقة 1: الغربلة السريعة (Layer 1 Rapid Screening)
+    # 🔴 الطبقة 1: الغربلة السريعة
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        """غربلة أولية سريعة لتقليل عدد العملات التي يحللها Titan"""
         print(f"🔍 [Layer 1] بدء الغربلة السريعة...")
         volume_data = await self._get_volume_data_live()
         if not volume_data: return []
         
-        # نأخذ أعلى 150 عملة سيولة لتوسيع نطاق البحث لـ Titan
+        # توسيع النطاق لـ 150 عملة لزيادة فرص Titan
         candidates = volume_data[:150]
-        qualified = []
-        
-        # تحليل سريع جداً (فريم 1H فقط)
-        batch_size = 30 # دفعة أكبر قليلاً للسرعة
-        for i in range(0, len(candidates), batch_size):
-            batch = candidates[i:i+batch_size]
-            tasks = [self._fetch_ohlcv_live(c['symbol'], '1h', 150) for c in batch]
-            results = await asyncio.gather(*tasks, return_exceptions=True)
-            
-            for j, candles in enumerate(results):
-                if isinstance(candles, list) and len(candles) >= 100:
-                    # هنا يمكن إضافة منطق فلترة بسيط (مثلاً: فوق متوسط 50)
-                    # حالياً سنمرر معظم العملات الجيدة لـ Titan ليقرر هو
-                    candidates[i+j]['layer1_score'] = 0.5 # درجة مبدئية
-                    qualified.append(candidates[i+j])
-
-        print(f"✅ [Layer 1] تأهل {len(qualified)} عملة للتحليل العميق بواسطة Titan.")
-        return qualified
+        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():
+                # فلترة العملات ذات السيولة الضعيفة جداً (< 100k)
                 if s.endswith('/USDT') and t['quoteVolume'] and t['quoteVolume'] > 100000:
                     data.append({'symbol': s, 'dollar_volume': t['quoteVolume'], 'current_price': t['last']})
             data.sort(key=lambda x: x['dollar_volume'], reverse=True)
             return data
         except: return []
 
-    async def _fetch_ohlcv_live(self, symbol, tf, limit):
-        try: return await self.exchange.fetch_ohlcv(symbol, tf, limit=limit)
-        except: return None
-
     # ==================================================================
-    # 🔵 خط أنابيب بيانات Titan (Layer 2 Data Stream)
+    # 🔵 خط أنابيب البيانات الهجين (Hybrid Data Stream)
     # ==================================================================
     async def stream_ohlcv_data(self, symbols: List[Dict], queue: asyncio.Queue):
-        """
-        يجلب البيانات العميقة (500 شمعة) لجميع الأطر التي يحتاجها Titan.
-        الأطر المطلوبة: 5m, 15m, 1h, 4h, 1d
-        """
-        print(f"🚚 [Titan Data Stream] جلب بيانات عميقة لـ {len(symbols)} عملة...")
-        
-        tfs = ['5m', '15m', '1h', '4h', '1d'] # أطر Titan الكاملة
-        limit = 500 # عمق كافٍ لحساب كل المؤشرات بدقة (مثل EMA200)
+        """جلب بيانات عميقة (500 شمعة) لكافة الأطر المطلوبة"""
+        tfs = ['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]
             results = await asyncio.gather(*tasks, return_exceptions=False)
             
-            ohlcv_package = {}
-            complete_data = True
+            ohlcv = {}
             for i, res in enumerate(results):
-                if res and isinstance(res, list) and len(res) >= 200: # نحتاج 200 شمعة على الأقل
-                    ohlcv_package[tfs[i]] = res
-                else:
-                    # إذا فقدنا إطاراً رئيسياً (مثل 5m)، قد لا يعمل Titan بدقة
-                    if tfs[i] == '5m': complete_data = False
+                if res and isinstance(res, list) and len(res) >= 200:
+                    ohlcv[tfs[i]] = res
             
-            if complete_data and len(ohlcv_package) >= 4: # نتسامح مع فقدان إطار واحد غير رئيسي
-                sym_data['ohlcv'] = ohlcv_package
-                await queue.put([sym_data]) # إرسال العملة للمعالجة فوراً
+            # يجب توفر معظم الأطر لعمل النظام الهجين بدقة
+            if len(ohlcv) >= 4 and '5m' in ohlcv:
+                sym_data['ohlcv'] = ohlcv
+                await queue.put([sym_data])
             
-            # فاصل زمني صغير لتجنب حظر الـ API
-            await asyncio.sleep(0.1)
+            await asyncio.sleep(0.1) # تجنب حظر API
         
-        await queue.put(None) # إشارة انتهاء الدفق
-        print("🏁 [Titan Data Stream] اكتمل النقل.")
+        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
 
-print("✅ DataManager V12.0 (Titan Pipeline) loaded.")
+print("✅ DataManager V12.1 (Hybrid Support) loaded.")