Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V41.3 - GEM-Architect: Math Debugger Edition)
+# (V41.4 - GEM-Architect: Dynamic Matrix Scanner)
 # ============================================================
 
 import asyncio
@@ -28,9 +28,9 @@ logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 class DataManager:
     """
-    DataManager V41.3 (Math Debugger)
-    - Exposes calculation errors in _apply_scanner_strategies.
-    - Inspects DataFrame integrity.
+    DataManager V41.4 (Dynamic Matrix Scanner)
+    - Fixes KeyError by dynamically finding indicator columns.
+    - Prevents crashes from slight library version differences.
     """
 
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
@@ -52,7 +52,7 @@ class DataManager:
             'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP'
         ]
         
-        print(f"📦 [DataManager V41.3] Math Debugger Online.")
+        print(f"📦 [DataManager V41.4] Dynamic Matrix Scanner Online.")
 
     async def initialize(self):
         print("   > [DataManager] Starting initialization...")
@@ -93,31 +93,24 @@ class DataManager:
         
         print(f"🔍 [L1 Matrix] Regime: {current_regime} | Weights: {scanner_weights}")
         
-        # 1. جلب العملات
+        # 1. جلب العملات (Universe)
         all_tickers = await self._fetch_universe_tickers()
         if not all_tickers: 
             print("⚠️ [Layer 1] Universe fetch returned empty.")
             return []
 
-        # 2. الجلب العميق (أخذ عينة أكبر قليلاً للتجربة)
-        top_candidates = all_tickers[:60]
+        # 2. الجلب العميق (Deep Fetch)
+        top_candidates = all_tickers[:60] # نأخذ عينة جيدة
         enriched_data = await self._batch_fetch_ta_data(top_candidates, timeframe='15m', limit=100)
         
         scored_candidates = []
         debug_log_sample = [] 
 
-        # 🔍 فحص سلامة البيانات لأول عملة (Data Integrity Check)
-        if enriched_data:
-            first_coin = enriched_data[0]
-            if 'df' in first_coin:
-                print(f"   -> 📊 [Data Inspect] {first_coin['symbol']} DF Shape: {first_coin['df'].shape}")
-                print(f"   -> 📊 [Data Inspect] Tail:\n{first_coin['df'].tail(3)[['close', 'volume']]}")
-
         for item in enriched_data:
             df = item.get('df')
             if df is None or len(df) < 50: continue
             
-            # 🔥 تطبيق الكاشفات (سيتم طباعة الخطأ إذا حدث)
+            # 🔥 تطبيق الكاشفات
             scores = self._apply_scanner_strategies(df, item['symbol'])
             
             final_score = 0.0
@@ -128,6 +121,7 @@ class DataManager:
                 final_score += (val['score'] * w)
                 if val['active']: tags.append(strategy)
             
+            # Boost بسيط
             if item['change_24h'] > 3.0 and current_regime == "BULL": final_score += 10
             
             item['l1_score'] = final_score
@@ -160,22 +154,20 @@ class DataManager:
         ]
 
     # ==================================================================
-    # 🧩 Scanner Strategies Logic (With Error Exposure)
+    # 🧩 Scanner Strategies Logic (Dynamic Finder)
     # ==================================================================
     def _apply_scanner_strategies(self, df: pd.DataFrame, symbol: str) -> Dict[str, Any]:
-        """تطبيق المؤشرات مع كشف الأخطاء"""
+        """تطبيق المؤشرات مع البحث الديناميكي عن أسماء الأعمدة"""
         results = {}
         try:
-            # التأكد من عدم وجود قيم فارغة (NaN)
-            # ملء الفراغات بآخر قيمة صالحة (Forward Fill)
+            # حماية البيانات
             df = df.ffill().bfill() 
             close = df['close']
             
             # 1. RSI
             rsi = ta.rsi(close, length=14)
-            # قد يعود RSI بقيم NaN في البداية، نأخذ القيمة الأخيرة ونتأكد أنها رقم
             curr_rsi = rsi.iloc[-1] if rsi is not None else 50.0
-            if np.isnan(curr_rsi): curr_rsi = 50.0 # حماية ضد NaN
+            if np.isnan(curr_rsi): curr_rsi = 50.0
             
             score_rsi = 0
             active_rsi = False
@@ -187,39 +179,54 @@ class DataManager:
                 score_rsi = 80  
                 active_rsi = True
             elif 30 < curr_rsi <= 50: 
-                score_rsi = 40  
+                score_rsi = 40 
             
             results["RSI_MOMENTUM"] = {'score': score_rsi, 'active': active_rsi, 'val': curr_rsi}
 
-            # 2. BB
+            # 2. Bollinger Bands (Dynamic Column Finder)
             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
+            score_bb = 0
+            active_bb = False
+            
+            if bb is not None and not bb.empty:
+                # 🔥 البحث الذكي عن الأعمدة: نبحث عن أي عمود يبدأ بـ BBU (Upper) و BBB (Width)
+                # هذا يتجاوز اختلاف الإصدارات (2.0 vs 2)
+                bbu_col = next((c for c in bb.columns if c.startswith('BBU')), None)
+                bbb_col = next((c for c in bb.columns if c.startswith('BBB')), None)
+                
+                if bbu_col and bbb_col:
+                    upper = bb[bbu_col].iloc[-1]
+                    width = bb[bbb_col].iloc[-1]
+                    curr_price = close.iloc[-1]
+                    
+                    if curr_price > upper and width > 0.1:
+                        score_bb = 100
+                        active_bb = True
+                else:
+                    # إذا فشل البحث، نطبع الأعمدة الموجودة للتشخيص المستقبلي
+                    # print(f"⚠️ [BB Warning] {symbol} columns mismatch: {bb.columns.tolist()}")
+                    pass
+                    
             results["BB_BREAKOUT"] = {'score': score_bb, 'active': active_bb}
 
-            # 3. MACD
+            # 3. MACD (Dynamic Column Finder)
             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
+            score_macd = 0
+            active_macd = False
+            
+            if macd is not None and not macd.empty:
+                # نبحث عن عمود الهستوجرام (يبدأ بـ MACDh)
+                hist_col = next((c for c in macd.columns if c.startswith('MACDh')), None)
+                
+                if hist_col:
+                    hist = macd[hist_col].iloc[-1]
+                    if hist > 0: 
+                        score_macd = 100
+                        active_macd = True
+                        
             results["MACD_CROSS"] = {'score': score_macd, 'active': active_macd}
             
-            # 4. Volume
+            # 4. Volume Flow
             vol = df['volume']
             vol_ma = ta.sma(vol, length=20).iloc[-1]
             curr_vol = vol.iloc[-1]
@@ -231,46 +238,32 @@ class DataManager:
             results["VOLUME_FLOW"] = {'score': score_vol, 'active': active_vol}
 
         except Exception as e:
-            # 🔥 طباعة الخطأ الحقيقي هنا 🔥
             print(f"❌ [Scanner Error] {symbol}: {e}")
-            # traceback.print_exc() # قم بتفعيل هذا إذا أردت التفاصيل المملة
             return {k: {'score': 0, 'active': False, 'val': 0} for k in ["RSI_MOMENTUM", "BB_BREAKOUT", "MACD_CROSS", "VOLUME_FLOW"]}
 
         return results
 
     # ==================================================================
-    # 🌍 Universe & Batch Fetch (Fixed)
+    # 🌍 Universe & Batch Fetch
     # ==================================================================
     async def _fetch_universe_tickers(self) -> List[Dict[str, Any]]:
-        print("   -> 📡 [Debug] Contacting Exchange for Tickers...")
+        # print("   -> 📡 [Debug] Contacting Exchange for Tickers...")
         try:
             tickers = await self.exchange.fetch_tickers()
-            print(f"   -> 📡 [Debug] Raw Tickers Received: {len(tickers)}")
-            
             candidates = []
-            skipped_reason = {"pair": 0, "blacklist": 0, "volume": 0}
             
             for symbol, ticker in tickers.items():
-                if not symbol.endswith('/USDT'): 
-                    skipped_reason["pair"] += 1
-                    continue
+                if not symbol.endswith('/USDT'): continue
                 
                 base_currency = symbol.split('/')[0]
-                
-                if any(bad in base_currency for bad in self.BLACKLIST_TOKENS): 
-                    skipped_reason["blacklist"] += 1
-                    continue
+                if any(bad in base_currency for bad in self.BLACKLIST_TOKENS): continue
                 
                 vol = ticker.get('quoteVolume')
-                if vol is None: 
-                    vol = ticker.get('info', {}).get('volValue') 
-                
+                if vol is None: vol = ticker.get('info', {}).get('volValue') 
                 if vol is None: vol = 0.0
                 else: vol = float(vol)
 
-                if vol < 100_000: 
-                    skipped_reason["volume"] += 1
-                    continue 
+                if vol < 100_000: continue 
 
                 candidates.append({
                     'symbol': symbol,
@@ -279,9 +272,6 @@ class DataManager:
                     'change_24h': float(ticker.get('percentage', 0.0)) 
                 })
             
-            print(f"   -> 📊 [Debug] Filter Stats: BadPair={skipped_reason['pair']}, Blacklist={skipped_reason['blacklist']}, LowVol={skipped_reason['volume']}")
-            print(f"   -> ✅ [Debug] Candidates Passed: {len(candidates)}")
-            
             candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
             return candidates