Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V40.0 - GEM-Architect: The Scanner Matrix)
+# (V41.0 - GEM-Architect: Debugger & Logic Fix)
 # ============================================================
 
 import asyncio
@@ -20,24 +20,12 @@ except ImportError:
     class SystemLimits:
         L1_MIN_AFFINITY_SCORE = 15.0
         CURRENT_REGIME = "RANGE"
-        # أوزان الكاشفات الجديدة (تتغير بالباكتست)
-        SCANNER_WEIGHTS = {
-            "RSI_MOMENTUM": 0.3,
-            "BB_BREAKOUT": 0.3,
-            "MACD_CROSS": 0.2,
-            "VOLUME_FLOW": 0.2
-        }
+        SCANNER_WEIGHTS = {"RSI_MOMENTUM": 0.3, "BB_BREAKOUT": 0.3, "MACD_CROSS": 0.2, "VOLUME_FLOW": 0.2}
 
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 class DataManager:
-    """
-    DataManager V40.0 (The Scanner Matrix)
-    - L1 Screening uses a multi-strategy ensemble approach.
-    - Optimized for speed using batch processing.
-    """
-
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
         self.contracts_db = contracts_db or {}
         self.whale_monitor = whale_monitor
@@ -49,7 +37,7 @@ class DataManager:
         })
         self.http_client = None
         self.BLACKLIST_TOKENS = ['USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L']
-        print(f"📦 [DataManager V40.0] Scanner Matrix Online.")
+        print(f"📦 [DataManager V41.0] Scanner Matrix (Debug Mode) Online.")
 
     async def initialize(self):
         self.http_client = httpx.AsyncClient(timeout=60.0)
@@ -63,42 +51,31 @@ class DataManager:
         if self.http_client: await self.http_client.aclose()
         if self.exchange: await self.exchange.close()
 
-    # ==================================================================
-    # 🛡️ Layer 1: The Scanner Matrix (New Logic)
-    # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        """
-        تنفيذ الفحص المتعدد (Matrix Scan).
-        1. جلب أفضل 80 عملة سيولة.
-        2. جلب شموع 15m لهذه العملات.
-        3. تطبيق 4 استراتيجيات كشف مختلفة.
-        4. حساب النتيجة الموزونة.
-        """
         current_regime = getattr(SystemLimits, "CURRENT_REGIME", "RANGE")
         scanner_weights = getattr(SystemLimits, "SCANNER_WEIGHTS", {"RSI_MOMENTUM": 1.0})
         min_score = getattr(SystemLimits, "L1_MIN_AFFINITY_SCORE", 15.0)
 
         print(f"🔍 [L1 Matrix] Regime: {current_regime} | Weights: {scanner_weights}")
 
-        # 1. تصفية الكون الأولي (High Volume Universe)
         tickers = await self._fetch_universe_tickers()
-        if not tickers: return []
+        if not tickers: 
+            print("⚠️ [L1] Universe fetch returned empty.")
+            return []
         
-        # نأخذ أفضل 80 عملة فقط لتجنب قتل الـ API Rate Limits
         top_candidates = tickers[:80]
-        
-        # 2. جلب البيانات الفنية دفعة واحدة (Batch Fetch 15m)
         enriched_data = await self._batch_fetch_ta_data(top_candidates, timeframe='15m', limit=100)
         
         scored_candidates = []
+        debug_log_sample = [] # لعرض عينة من الدرجات
+
         for item in enriched_data:
             df = item.get('df')
             if df is None or len(df) < 50: continue
             
-            # 3. تطبيق الكاشفات (Scanners)
+            # تطبيق الكاشفات
             scores = self._apply_scanner_strategies(df)
             
-            # 4. حساب النتيجة النهائية الموزونة
             final_score = 0.0
             tags = []
             
@@ -107,22 +84,26 @@ class DataManager:
                 final_score += (val['score'] * w)
                 if val['active']: tags.append(strategy)
             
-            # إضافة نقاط إضافية بناءً على النظام القديم (Volume/Price Action)
-            # للحفاظ على التوافقية
-            if item['change_24h'] > 5.0 and current_regime == "BULL": final_score += 10
+            # توافقية النظام القديم (Boost)
+            if item['change_24h'] > 3.0 and current_regime == "BULL": final_score += 10
             
             item['l1_score'] = final_score
             item['tags'] = tags
             
+            # تسجيل عينة للمراقبة (أول 5 عملات فقط لتجنب إغراق السجل)
+            if len(debug_log_sample) < 5:
+                debug_details = f"{item['symbol']}: {final_score:.1f} (RSI:{scores['RSI_MOMENTUM']['val']:.1f}|Score:{scores['RSI_MOMENTUM']['score']})"
+                debug_log_sample.append(debug_details)
+
             if final_score >= min_score:
                 scored_candidates.append(item)
 
         scored_candidates.sort(key=lambda x: x['l1_score'], reverse=True)
         
-        print(f"   -> Matrix selected {len(scored_candidates)} candidates.")
-        if scored_candidates:
-             print(f"   -> Top Pick: {scored_candidates[0]['symbol']} (Score: {scored_candidates[0]['l1_score']:.1f})")
-
+        # طباعة تقرير التصحيح
+        print(f"   -> [DEBUG L1] Sample Scores: { ' | '.join(debug_log_sample) }")
+        print(f"   -> Matrix selected {len(scored_candidates)} candidates (Threshold: {min_score}).")
+        
         return [
             {
                 'symbol': c['symbol'],
@@ -131,70 +112,79 @@ class DataManager:
                 'type': ','.join(c['tags']),
                 'l1_score': c['l1_score']
             }
-            for c in scored_candidates[:40] # تمرير أفضل 40
+            for c in scored_candidates[:40] 
         ]
 
-    # ------------------------------------------------------------------
-    # 🧩 Scanner Strategies Logic
-    # ------------------------------------------------------------------
     def _apply_scanner_strategies(self, df: pd.DataFrame) -> Dict[str, Any]:
-        """تطبيق مؤشرات فنية متعددة على البيانات"""
         results = {}
-        close = df['close']
-        
-        # Strategy A: RSI Momentum (زخم)
-        rsi = ta.rsi(close, length=14)
-        curr_rsi = rsi.iloc[-1]
-        score_rsi = 0
-        active_rsi = False
-        if 50 < curr_rsi < 75: 
-            score_rsi = 100 
-            active_rsi = True
-        elif curr_rsi <= 30: # Oversold Bounce
-            score_rsi = 80
-            active_rsi = True
-        results["RSI_MOMENTUM"] = {'score': score_rsi, 'active': active_rsi}
-
-        # Strategy B: Bollinger Band Breakout (انفجار سعري)
-        bb = ta.bbands(close, length=20, std=2)
-        upper = bb['BBU_20_2.0'].iloc[-1]
-        width = bb['BBB_20_2.0'].iloc[-1]
-        curr_price = close.iloc[-1]
-        score_bb = 0
-        active_bb = False
-        if curr_price > upper and width > 0.1: # اختراق حقيقي
-            score_bb = 100
-            active_bb = True
-        results["BB_BREAKOUT"] = {'score': score_bb, 'active': active_bb}
-
-        # Strategy C: MACD Cross (تغير اتجاه)
-        macd = ta.macd(close)
-        macd_line = macd['MACD_12_26_9'].iloc[-1]
-        signal_line = macd['MACDS_12_26_9'].iloc[-1]
-        hist = macd['MACDh_12_26_9'].iloc[-1]
-        score_macd = 0
-        active_macd = False
-        if macd_line > signal_line and hist > 0:
-            score_macd = 100
-            active_macd = True
-        results["MACD_CROSS"] = {'score': score_macd, 'active': active_macd}
-        
-        # Strategy D: Volume Flow (تدفق سيولة)
-        vol = df['volume']
-        vol_ma = ta.sma(vol, length=20).iloc[-1]
-        curr_vol = vol.iloc[-1]
-        score_vol = 0
-        active_vol = False
-        if curr_vol > (vol_ma * 1.5): # حجم تداول ضخم مفاجئ
-            score_vol = 100
-            active_vol = True
-        results["VOLUME_FLOW"] = {'score': score_vol, 'active': active_vol}
+        try:
+            close = df['close']
+            
+            # 1. RSI (تم إصلاح المنطقة العمياء)
+            rsi = ta.rsi(close, length=14)
+            curr_rsi = rsi.iloc[-1] if rsi is not None else 50
+            
+            score_rsi = 0
+            active_rsi = False
+            
+            if 50 < curr_rsi < 75: 
+                score_rsi = 100 
+                active_rsi = True
+            elif curr_rsi <= 30: # Oversold
+                score_rsi = 80
+                active_rsi = True
+            elif 30 < curr_rsi <= 50: # ✅ المنطقة المحايدة (تمت إضافتها)
+                score_rsi = 40 # نعطيها بعض النقاط بدلاً من الصفر
+            
+            results["RSI_MOMENTUM"] = {'score': score_rsi, 'active': active_rsi, 'val': curr_rsi}
+
+            # 2. BB Breakout
+            bb = ta.bbands(close, length=20, std=2)
+            if bb is not None:
+                upper = bb[f'BBU_20_2.0'].iloc[-1]
+                width = bb[f'BBB_20_2.0'].iloc[-1]
+                curr_price = close.iloc[-1]
+                score_bb = 0
+                active_bb = False
+                if curr_price > upper and width > 0.1:
+                    score_bb = 100
+                    active_bb = True
+            else:
+                score_bb = 0; active_bb = False
+            results["BB_BREAKOUT"] = {'score': score_bb, 'active': active_bb}
+
+            # 3. MACD
+            macd = ta.macd(close)
+            if macd is not None:
+                hist = macd[f'MACDh_12_26_9'].iloc[-1]
+                score_macd = 0
+                active_macd = False
+                if hist > 0: # الهستوجرام إيجابي
+                    score_macd = 100
+                    active_macd = True
+            else:
+                score_macd = 0; active_macd = False
+            results["MACD_CROSS"] = {'score': score_macd, 'active': active_macd}
+            
+            # 4. Volume Flow
+            vol = df['volume']
+            vol_ma = ta.sma(vol, length=20).iloc[-1]
+            curr_vol = vol.iloc[-1]
+            score_vol = 0
+            active_vol = False
+            if curr_vol > (vol_ma * 1.2): # خففنا الشرط من 1.5 إلى 1.2
+                score_vol = 100
+                active_vol = True
+            results["VOLUME_FLOW"] = {'score': score_vol, 'active': active_vol}
+
+        except Exception as e:
+            # في حال حدوث خطأ في الحساب، نعيد قيم صفرية لعدم إيقاف النظام
+            # print(f"Indicator Error: {e}") 
+            return {k: {'score': 0, 'active': False, 'val': 0} for k in ["RSI_MOMENTUM", "BB_BREAKOUT", "MACD_CROSS", "VOLUME_FLOW"]}
 
         return results
 
-    # ------------------------------------------------------------------
-    # ⚡ Batch & Async Helpers
-    # ------------------------------------------------------------------
+    # --- Helpers ---
     async def _fetch_universe_tickers(self):
         try:
             tickers = await self.exchange.fetch_tickers()
@@ -202,7 +192,7 @@ class DataManager:
             for symbol, ticker in tickers.items():
                 if not symbol.endswith('/USDT'): continue
                 if any(bad in symbol for bad in self.BLACKLIST_TOKENS): continue
-                if not ticker.get('quoteVolume') or ticker['quoteVolume'] < 500_000: continue # Min 500k Vol
+                if not ticker.get('quoteVolume') or ticker['quoteVolume'] < 300_000: continue # تخفيف شرط السيولة قليلاً
                 candidates.append({
                     'symbol': symbol,
                     'quote_volume': ticker['quoteVolume'],
@@ -215,7 +205,7 @@ class DataManager:
 
     async def _batch_fetch_ta_data(self, candidates, timeframe='15m', limit=100):
         results = []
-        chunk_size = 15 # لعدم تجاوز الحدود
+        chunk_size = 15
         for i in range(0, len(candidates), chunk_size):
             chunk = candidates[i:i+chunk_size]
             tasks = [self._fetch_ohlcv_safe(c, timeframe, limit) for c in chunk]
@@ -235,17 +225,12 @@ class DataManager:
             return candidate
         except: return None
     
-    # Helpers needed for Processor/TradeManager
     async def get_latest_price_async(self, symbol):
         t = await self.exchange.fetch_ticker(symbol)
         return float(t['last'])
-    
     async def get_latest_ohlcv(self, symbol, tf, limit=100):
         return await self.exchange.fetch_ohlcv(symbol, tf, limit=limit)
-        
     async def get_order_book_snapshot(self, symbol, limit=20):
         return await self.exchange.fetch_order_book(symbol, limit)
-    
-    # R2 Placeholder
     async def load_contracts_from_r2(self): pass
     def get_contracts_db(self): return self.contracts_db