Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V60.1 - GEM-Architect: Tuned Anti-FOMO Shield)
+# (V60.2 - GEM-Architect: Robust Data Fix & Tuned Shield)
 # ============================================================
 
 import asyncio
@@ -12,11 +12,13 @@ import pandas as pd
 import numpy as np
 from typing import List, Dict, Any
 
+# محاولة استيراد حدود النظام (اختياري)
 try:
     from ml_engine.processor import SystemLimits
 except ImportError:
     SystemLimits = None
 
+# إعدادات التسجيل لتقليل الضوضاء
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
 
@@ -25,7 +27,7 @@ class DataManager:
         self.contracts_db = contracts_db or {}
         self.whale_monitor = whale_monitor
         self.r2_service = r2_service
-        self.adaptive_hub_ref = None 
+        self.adaptive_hub_ref = None  # مرجع لملف التعلم
         
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
@@ -36,12 +38,13 @@ 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'
         ]
         
-        print(f"📦 [DataManager V60.1] Tuned Shield + Context Injector Active.")
+        print(f"📦 [DataManager V60.2] Robust Data Fix + Tuned Shield Active.")
 
     async def initialize(self):
         print("   > [DataManager] Starting initialization...")
@@ -74,23 +77,23 @@ class DataManager:
     # ==================================================================
     async def layer1_rapid_screening(self, adaptive_hub_ref=None) -> List[Dict[str, Any]]:
         self.adaptive_hub_ref = adaptive_hub_ref
-        print(f"🔍 [Layer 1] Initiating Tuned Screening...")
+        print(f"🔍 [Layer 1] Initiating Robust Screening...")
         
-        # 1. فلتر السيولة الأساسي (تم تخفيفه)
+        # 1. فلتر السيولة الأساسي (المصحح)
         initial_candidates = await self._stage0_universe_filter()
         
         if not initial_candidates:
-            print("⚠️ [Layer 1] Stage 0 returned 0 candidates. Check Quote Volume Limit.")
+            print("⚠️ [Layer 1] CRITICAL: Stage 0 returned 0 candidates.")
             return []
 
-        # 2. جلب البيانات الفنية
-        top_candidates = initial_candidates[:120] # وسعنا النطاق قليلاً
+        # 2. جلب البيانات الفنية (لأفضل 120 عملة لضمان التنوع)
+        top_candidates = initial_candidates[:120] 
         enriched_data = await self._fetch_technical_data_batch(top_candidates)
         
         final_list = []
 
         for item in enriched_data:
-            # 3. تطبيق الفلتر (تم تخفيف القيود)
+            # 3. تطبيق الفلتر الفني (Tuned Anti-FOMO)
             classification = self._apply_strict_logic_tree(item)
             
             if classification['type'] != 'NONE':
@@ -99,6 +102,7 @@ class DataManager:
                 item['asset_regime'] = regime_info['regime']
                 item['asset_regime_conf'] = regime_info['conf']
                 
+                # حقن العتبات الديناميكية
                 if self.adaptive_hub_ref:
                     dynamic_config = self.adaptive_hub_ref.get_regime_config(regime_info['regime'])
                     item['dynamic_limits'] = dynamic_config
@@ -114,6 +118,79 @@ class DataManager:
         print(f"✅ [Layer 1] Passed {len(selection)} candidates to Processor.")
         return selection
 
+    # ==================================================================
+    # 🔍 Stage 0: Universe Filter (Debugged & Robust)
+    # ==================================================================
+    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
+        try:
+            print("   🛡️ [Stage 0] Fetching Tickers...")
+            tickers = await self.exchange.fetch_tickers()
+            candidates = []
+            
+            # عداد لمعرفة سبب الرفض (للتشخيص)
+            reject_stats = {"volume": 0, "change": 0, "blacklist": 0, "bad_data": 0}
+            debug_printed = False
+
+            for symbol, ticker in tickers.items():
+                if not symbol.endswith('/USDT'): continue
+                
+                # 1. Blacklist
+                base_curr = symbol.split('/')[0]
+                if any(bad in base_curr for bad in self.BLACKLIST_TOKENS): 
+                    reject_stats["blacklist"] += 1
+                    continue
+                
+                # 2. Robust Volume Calculation (الإصلاح الجذري)
+                quote_vol = ticker.get('quoteVolume')
+                
+                # إذا ��ان غير موجود أو صفر، نحسبه يدوياً
+                if quote_vol is None or float(quote_vol) == 0:
+                    base_vol = ticker.get('baseVolume')
+                    last_price = ticker.get('last')
+                    if base_vol is not None and last_price is not None:
+                        quote_vol = float(base_vol) * float(last_price)
+                    else:
+                        quote_vol = 0.0
+                
+                quote_vol = float(quote_vol)
+
+                # طباعة فحص لعملة معروفة (BTC) للتأكد من صحة البيانات في السجل
+                if "BTC/USDT" in symbol and not debug_printed:
+                    print(f"   🐛 [DEBUG] BTC Data Check: Vol=${quote_vol:,.2f} | Change={ticker.get('percentage')}%")
+                    debug_printed = True
+
+                # فلتر السيولة (500k)
+                if quote_vol < 500_000: 
+                    reject_stats["volume"] += 1
+                    continue
+                
+                # فلتر التغير (20%)
+                change_pct = ticker.get('percentage')
+                if change_pct is None: change_pct = 0.0
+                
+                if abs(change_pct) > 20.0: 
+                    reject_stats["change"] += 1
+                    continue
+                
+                candidates.append({
+                    'symbol': symbol,
+                    'quote_volume': quote_vol,
+                    'current_price': ticker.get('last'),
+                    'change_24h': change_pct
+                })
+            
+            # طباعة تقرير الرفض
+            print(f"   📊 [Filter Stats] Total Tickers: {len(tickers)} | Passed: {len(candidates)}")
+            print(f"      ❌ Rejected: Vol < 500k ({reject_stats['volume']}) | Change > 20% ({reject_stats['change']})")
+
+            candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
+            return candidates
+            
+        except Exception as e:
+            print(f"❌ [L1 Error] Universe filter failed: {e}")
+            traceback.print_exc()
+            return []
+
     # ------------------------------------------------------------------
     # 🧭 The Diagnoser
     # ------------------------------------------------------------------
@@ -144,7 +221,7 @@ class DataManager:
         except Exception: return {'regime': 'RANGE', 'conf': 0.0}
 
     # ------------------------------------------------------------------
-    # 🛡️ The Logic Tree (TUNED for better flow)
+    # 🛡️ The Logic Tree (TUNED)
     # ------------------------------------------------------------------
     def _apply_strict_logic_tree(self, data: Dict[str, Any]) -> Dict[str, Any]:
         try:
@@ -162,13 +239,13 @@ class DataManager:
         except: change_4h = 0.0
 
         # 🔧 TUNED GATES (فتحات التنفس)
-        # 1. السماح بحركة 12% في 4 ساعات (بدلاً من 8%)
+        # 1. السماح بحركة 12% في 4 ساعات
         if change_4h > 12.0: return {'type': 'NONE', 'score': 0}
         
-        # 2. السماح بـ RSI حتى 75 (بدلاً من 70)
+        # 2. السماح بـ RSI حتى 75
         if curr_1h['rsi'] > 75: return {'type': 'NONE', 'score': 0} 
         
-        # 3. السماح بانحراف 2.2 ATR (بدلاً من 1.8)
+        # 3. السماح بانحراف 2.2 ATR
         dev = (curr_1h['close'] - curr_1h['ema20']) / curr_1h['atr'] if curr_1h['atr'] > 0 else 0
         if dev > 2.2: return {'type': 'NONE', 'score': 0}
 
@@ -176,10 +253,10 @@ class DataManager:
         
         # A. Breakout (اختراق آمن)
         is_bullish = (curr_1h['ema20'] > curr_1h['ema50']) or (curr_1h['close'] > curr_1h['ema20'])
-        if is_bullish and (45 <= curr_1h['rsi'] <= 72): # وسعنا نطاق الـ RSI قليلاً
+        if is_bullish and (45 <= curr_1h['rsi'] <= 75): 
             vol_ma = df_15m['volume'].rolling(20).mean().iloc[-1]
             
-            # 🔧 TUNED: خفضنا شرط الفوليوم لـ 1.2x (بدلاً من 1.5x) لالتقاط بداية الحركة
+            # 🔧 TUNED: 1.2x Volume
             if curr_15m['volume'] >= 1.2 * vol_ma: 
                 score = curr_15m['volume'] / vol_ma if vol_ma > 0 else 1.0
                 return {'type': 'BREAKOUT', 'score': score}
@@ -194,31 +271,6 @@ class DataManager:
     # ------------------------------------------------------------------
     # Helpers
     # ------------------------------------------------------------------
-    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
-        try:
-            tickers = await self.exchange.fetch_tickers()
-            candidates = []
-            for symbol, ticker in tickers.items():
-                if not symbol.endswith('/USDT'): continue
-                if any(b in symbol for b in self.BLACKLIST_TOKENS): continue
-                
-                # 🔧 TUNED: خفضنا الحد الأدنى للسيولة لـ 500k
-                quote_vol = ticker.get('quoteVolume', 0)
-                if quote_vol < 500_000: continue
-                
-                # 🔧 TUNED: السماح بـ 20% تغير يومي (بدلاً من 15%)
-                if ticker.get('percentage', 0) > 20.0: continue
-                
-                candidates.append({
-                    'symbol': symbol,
-                    'quote_volume': quote_vol,
-                    'current_price': ticker.get('last'),
-                    'change_24h': ticker.get('percentage')
-                })
-            candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
-            return candidates
-        except: return []
-
     async def _fetch_technical_data_batch(self, candidates):
         chunk_size = 10; results = []
         for i in range(0, len(candidates), chunk_size):