Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V42.0 - GEM-Architect: The Titan Mirror Filter)
+# (V43.0 - GEM-Architect: Hot-Flow Sort)
 # ============================================================
 
 import asyncio
@@ -10,6 +10,7 @@ import logging
 import pandas as pd
 import numpy as np
 import pandas_ta as ta
+import math  # ✅ مكتبة مهمة للمعادلة اللوغاريتمية
 from typing import List, Dict, Any
 
 import ccxt.async_support as ccxt
@@ -28,9 +29,9 @@ logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 class DataManager:
     """
-    DataManager V42.0 (Titan Mirror Filter)
-    - L1 Screening is now structurally aligned with L2 Models.
-    - Filters out 'Broken Charts', 'Low Liquidity', and 'Untradable Volatility'.
+    DataManager V43.0 (Hot-Flow Sort)
+    - Replaced 'Raw Volume' sort with 'Volume-Weighted Momentum'.
+    - Prioritizes active mid-caps over stagnant giants (XRP, TRX, BNB).
     """
 
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
@@ -47,13 +48,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', 'USDD', 'USDP', 'HT', 'KCS'
+            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP', 'HT', 'KCS', 'WBTC'
         ]
         
-        print(f"📦 [DataManager V42.0] Titan Mirror Filter Online.")
+        print(f"📦 [DataManager V43.0] Hot-Flow Engine Online.")
 
     async def initialize(self):
         print("   > [DataManager] Starting initialization...")
@@ -85,52 +86,60 @@ class DataManager:
     def get_contracts_db(self): return self.contracts_db
 
     # ==================================================================
-    # 🛡️ Layer 1: The Titan Mirror Filter (Engineered for ML Alignment)
+    # 🛡️ Layer 1: The Titan Mirror Filter (Hot-Flow Edition)
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
         """
-        الفلتر الجديد: يرفض العملات بناءً على "جودة الهيكل" وليس فقط "إشارة المؤشر".
-        يجهز الساحة لنماذج Titan و Sniper.
+        يقوم بفرز العملات بناءً على "السخونة" (Hotness) وليس الحجم فقط.
         """
         current_regime = getattr(SystemLimits, "CURRENT_REGIME", "RANGE")
         min_score = getattr(SystemLimits, "L1_MIN_AFFINITY_SCORE", 15.0)
         
-        print(f"🔍 [L1 Mirror] Scanning for High-Fidelity Structures (Regime: {current_regime})...")
+        print(f"🔍 [L1 Hot-Flow] Scanning for Active Momentum (Regime: {current_regime})...")
         
-        # 1. جلب Universe (تصفية أولية حسب الحجم)
-        # رفعنا حد السيولة لضمان عدم تمرير عملات ميتة للنماذج
-        min_vol = 1000000 if current_regime == "BEAR" else 5000000
-        all_tickers = await self._fetch_universe_tickers(min_volume=min_vol)
+        # 1. جلب Universe (تصفية أولية)
+        # خفضنا الحد الأدنى للحجم للسماح للعملات المتوسطة "الساخنة" بالدخول
+        # في السابق كان الرقم مرتفعاً جداً مما يقتل الفرص
+        min_vol_floor = 1000000 if current_regime == "BEAR" else 5000000
+        
+        all_tickers = await self._fetch_universe_tickers(min_volume=min_vol_floor)
         
         if not all_tickers: 
             print("⚠️ [Layer 1] Universe fetch returned empty.")
             return []
 
-        # 2. الجلب العميق (Deep Fetch) لعدد محدود وعالي الجودة
-        # نركز الموارد على أفضل 50 عملة سيولة بدلاً من فحص العشوائيات
-        top_candidates = all_tickers[:200] 
-        enriched_data = await self._batch_fetch_ta_data(top_candidates, timeframe='15m', limit=200) # نحتاج 200 للمتوسطات
+        # 2. الجلب العميق (Deep Fetch) لأكثر العملات سخونة
+        # نأخذ أفضل 150 عملة بناءً على معادلة (الحجم × الحركة)
+        scan_limit = 150 
+        top_candidates = all_tickers[:scan_limit] 
+        
+        print(f"   -> Deep scanning top {len(top_candidates)} active assets...")
+        
+        enriched_data = await self._batch_fetch_ta_data(top_candidates, timeframe='15m', limit=200)
         
         scored_candidates = []
         
         for item in enriched_data:
             df = item.get('df')
-            if df is None or len(df) < 150: continue # شرط صارم لطول البيانات
+            if df is None or len(df) < 150: continue 
             
-            # 🔥 تطبيق فلتر المرآة (The Mirror Logic)
+            # تطبيق فلتر المرآة الهيكلية
             structural_score, reasons = self._calculate_structural_score(df, item['symbol'], current_regime)
             
+            # 🔥 Bonus: إضافة نقاط إضافية للعملات الساخنة جداً
+            hot_score = item.get('hot_score', 0)
+            if hot_score > 50: structural_score += 5 # Boost for super active coins
+            
             item['l1_score'] = structural_score
             item['type'] = " | ".join(reasons)
             
-            # الفلتر يقبل فقط العملات ذات الهيكلية السليمة
             if structural_score >= min_score:
                 scored_candidates.append(item)
 
-        # الترتيب حسب الجودة الهيكلية
+        # الترتيب النهائي حسب جودة الهيكل
         scored_candidates.sort(key=lambda x: x['l1_score'], reverse=True)
         
-        print(f"   -> Mirror Filter selected {len(scored_candidates)} prime candidates.")
+        print(f"   -> Filter selected {len(scored_candidates)} candidates.")
         
         return [
             {
@@ -140,111 +149,81 @@ class DataManager:
                 'type': c.get('type', 'Structural'),
                 'l1_score': c.get('l1_score', 0)
             }
-            for c in scored_candidates[:25] # تمرير الأفضل فقط للمعالجة الثقيلة
+            for c in scored_candidates[:30] 
         ]
 
     # ==================================================================
-    # 🧬 Structural Alignment Engine (The Core Upgrade)
+    # 🧬 Structural Alignment Engine
     # ==================================================================
     def _calculate_structural_score(self, df: pd.DataFrame, symbol: str, regime: str) -> (float, List[str]):
-        """
-        يحسب درجة توافق العملة مع متطلبات نماذج التعلم الآلي.
-        """
+        # (نفس منطق الفلتر الهيكلي السابق - لم يتغير لأنه ممتاز)
         score = 0.0
         tags = []
-        
         try:
-            # 1. Data Prep
             close = df['close']
             high = df['high']
             low = df['low']
             volume = df['volume']
             current_price = close.iloc[-1]
             
-            # --- Check A: Liquidity Quality (Anti-Sniper Failure) ---
-            # فحص "جودة الشموع": هل هناك الكثير من الشموع الصفرية أو الفجوات؟
             zero_vol_candles = (volume == 0).sum()
-            if zero_vol_candles > 5: return -100.0, ["Illiquid"] # رفض فوري
+            if zero_vol_candles > 5: return -100.0, ["Illiquid"]
             
             avg_vol = volume.rolling(20).mean().iloc[-1]
-            if avg_vol * current_price < 5000: return -100.0, ["Thin Book"] # سيولة دقيقة ضعيفة
+            if avg_vol * current_price < 5000: return -100.0, ["Thin Book"]
 
-            # --- Check B: Trend Alignment (Titan EMA Feature) ---
-            # Titan يعتمد بشكل كبير على المسافة من EMA
             ema200 = ta.ema(close, length=200)
             ema50 = ta.ema(close, length=50)
             
             if ema200 is not None and ema50 is not None:
                 curr_ema200 = ema200.iloc[-1]
-                curr_ema50 = ema50.iloc[-1]
-                
-                # في السوق الصاعد، نريد السعر فوق المتوسطات ولكن ليس بعيداً جداً (Overextended)
                 dist_200 = (current_price - curr_ema200) / curr_ema200
                 
                 if regime == "BULL":
                     if current_price > curr_ema200:
                         score += 20
-                        if current_price > curr_ema50: score += 10
-                        if dist_200 < 0.15: score += 10 # ليس متضخماً جداً
-                        else: score -= 5 # متضخم قد ينعكس
-                    else:
-                        score -= 20 # تحت التريند العام
+                        if current_price > ema50.iloc[-1]: score += 10
+                        if dist_200 < 0.15: score += 10 
+                        else: score -= 5 
+                    else: score -= 20 
                 elif regime == "BEAR":
-                     # في الهبوط نبحث عن الارتدادات المفرطة (Mean Reversion)
-                     if dist_200 < -0.20: score += 30 # Oversold heavily
-                else: # RANGE
-                    if abs(dist_200) < 0.05: score += 20 # يتذبذب حول المتوسط
+                     if dist_200 < -0.20: score += 30 
+                else: 
+                    if abs(dist_200) < 0.05: score += 20
 
-            # --- Check C: Volatility Structure (Titan BB Feature) ---
             bb = ta.bbands(close, length=20, std=2)
             if bb is not None:
-                # نحتاج حساب العرض (Bandwidth) يدوياً لضمان الدقة
-                upper = bb[bb.columns[0]] # عادة Lower في pandas_ta
-                lower = bb[bb.columns[2]] # Upper
-                # تصحيح الاندكس حسب المكتبة، أحيانا تختلف، لذا نستخدم الأسماء لو أمكن
-                # لكن للسرعة سنحسبها يدوياً
-                bb_width = (bb.iloc[:, 2] - bb.iloc[:, 0]) / bb.iloc[:, 1] # (Upper - Lower) / Middle
-                current_width = bb_width.iloc[-1]
-                
-                # Titan يحب "الانفجار من السكون" (Squeeze Breakout)
-                if current_width < 0.05: # نطاق ضيق جداً (Squeeze)
-                    score += 25
-                    tags.append("Squeeze")
-                elif current_width > 0.15: # متقلب جداً
-                    score -= 10 # قد يكون فات الأوان
+                # حساب الـ Bandwidth يدوياً للأمان
+                upper = bb[bb.columns[0]] # Lower band usually index 0 in pandas_ta default
+                lower = bb[bb.columns[2]] # Upper band
+                # نستخدم أسماء الأعمدة إذا أمكن للتأكد، ولكن للسرعة:
+                # pandas_ta returns: Lower, Mid, Upper, Bandwidth, Percent
+                # لنتأكد من الـ Bandwidth مباشرة
+                width_col = next((c for c in bb.columns if c.startswith('BBB')), None)
+                if width_col:
+                    current_width = bb[width_col].iloc[-1] / 100.0 # pandas_ta returns pct
+                    if current_width < 0.05: score += 25; tags.append("Squeeze")
+                    elif current_width > 0.15: score -= 10
 
-            # --- Check D: Momentum Integrity (RSI/ADX) ---
             rsi = ta.rsi(close, length=14).iloc[-1]
             adx = ta.adx(high, low, close, length=14)
             curr_adx = adx.iloc[-1, 0] if adx is not None else 0
             
-            if 40 < rsi < 70:
-                score += 15 # منطقة صحية للتريند
-            elif rsi > 75 and regime == "BULL":
-                score += 10 # زخم قوي
-            elif rsi < 30 and regime == "BEAR":
-                score += 10 # ذروة بيع
+            if 40 < rsi < 70: score += 15 
+            elif rsi > 75 and regime == "BULL": score += 10 
             
-            if curr_adx > 25: 
-                score += 10 # يوجد تريند حقيقي
-                tags.append("Trending")
+            if curr_adx > 25: score += 10; tags.append("Trending")
             
-            # --- Check E: Volume Flow (Sniper Confirmation) ---
             vol_sma = volume.rolling(20).mean().iloc[-1]
-            if volume.iloc[-1] > vol_sma * 1.5:
-                score += 15
-                tags.append("Vol Spike")
-
-        except Exception as e:
-            # print(f"⚠️ [Struct calc error] {symbol}: {e}")
-            return 0.0, ["Error"]
+            if volume.iloc[-1] > vol_sma * 1.5: score += 15; tags.append("Vol Spike")
 
+        except Exception as e: return 0.0, ["Error"]
         return score, tags
 
     # ==================================================================
-    # 🌍 Universe & Batch Fetch (Enhanced)
+    # 🌍 Universe & Batch Fetch (The Hot-Flow Logic)
     # ==================================================================
-    async def _fetch_universe_tickers(self, min_volume=100_000) -> List[Dict[str, Any]]:
+    async def _fetch_universe_tickers(self, min_volume=300_000) -> List[Dict[str, Any]]:
         try:
             tickers = await self.exchange.fetch_tickers()
             candidates = []
@@ -252,34 +231,51 @@ class DataManager:
             for symbol, ticker in tickers.items():
                 if not symbol.endswith('/USDT'): continue
                 
-                # تصفية العملات المحظورة
                 base_currency = symbol.split('/')[0]
                 if any(bad in base_currency for bad in self.BLACKLIST_TOKENS): continue
-                if "3S" in base_currency or "3L" in base_currency: continue # الرموز ذات الرافعة
+                if "3S" in base_currency or "3L" in base_currency: continue 
                 
                 vol = ticker.get('quoteVolume')
                 if vol is None: vol = ticker.get('info', {}).get('volValue') 
                 if vol is None: vol = 0.0
                 else: vol = float(vol)
 
+                # 1. فلتر الحد الأدنى المطلق (لإبعاد العملات الميتة)
                 if vol < min_volume: continue 
 
-                # فلتر السبريد (Spread Filter) - مهم جداً للـ Sniper
+                # 2. فلتر السبريد (للحماية)
                 bid = float(ticker.get('bid', 0) or 0)
                 ask = float(ticker.get('ask', 0) or 0)
                 if bid > 0 and ask > 0:
                     spread_pct = (ask - bid) / bid
-                    if spread_pct > 0.01: continue # رفض السبريد العالي (>1%)
+                    if spread_pct > 0.015: continue # تساهلنا قليلاً (1.5%) للسماح بعملات الـ Meme النشطة
+
+                # 3. حساب درجة السخونة 🔥 (Hot Score)
+                # المعادلة: Log10(Volume) * (1 + Abs(Change%))
+                # هذا يعطي وزناً للحجم، لكن يضرب بقوة في التغير السعري
+                change_pct = float(ticker.get('percentage', 0.0))
+                
+                # نستخدم Log10 لتقليص الفارق بين المليار والمليون
+                # Log10(1B) = 9, Log10(10M) = 7 (الفارق بسيط)
+                # بينما التغير السعري: 1% vs 10% (الفارق 10 أضعاف)
+                # هذا يجعل التغير السعري هو العامل الحاسم في الترتيب
+                
+                log_vol = math.log10(vol + 1)
+                volatility_factor = abs(change_pct) + 1.0 # نضيف 1 لكي لا نضرب في صفر
+                
+                hot_score = log_vol * volatility_factor
 
                 candidates.append({
                     'symbol': symbol,
                     'quote_volume': vol,
                     'current_price': float(ticker['last']) if ticker.get('last') else 0.0,
-                    'change_24h': float(ticker.get('percentage', 0.0)) 
+                    'change_24h': change_pct,
+                    'hot_score': hot_score # الدرجة الجديدة
                 })
             
-            # الترتيب حسب السيولة
-            candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
+            # 🔥 الترتيب حسب درجة السخونة وليس الحجم المطلق
+            candidates.sort(key=lambda x: x['hot_score'], reverse=True)
+            
             return candidates
 
         except Exception as e:
@@ -288,13 +284,12 @@ class DataManager:
 
     async def _batch_fetch_ta_data(self, candidates, timeframe='15m', limit=200):
         results = []
-        chunk_size = 20 # زيادة حجم الدفعة قليلاً
+        chunk_size = 20 
         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]
             chunk_res = await asyncio.gather(*tasks)
             results.extend([r for r in chunk_res if r is not None])
-            # تأخير بسيط جداً لتجنب حظر API
             await asyncio.sleep(0.05)
         return results
 
@@ -303,10 +298,10 @@ class DataManager:
             ohlcv = await self.exchange.fetch_ohlcv(candidate['symbol'], tf, limit=limit)
             if not ohlcv: return None
             df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-            df['close'] = df['close'].astype(float)
-            df['high'] = df['high'].astype(float)
-            df['low'] = df['low'].astype(float)
-            df['volume'] = df['volume'].astype(float)
+            # تحويل البيانات لأرقام
+            cols = ['open', 'high', 'low', 'close', 'volume']
+            df[cols] = df[cols].apply(pd.to_numeric, errors='coerce')
+            
             candidate['df'] = df 
             return candidate
         except: return None