Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,5 +1,5 @@
 # ml_engine/data_manager.py
-# (V15.1 - GEM-Architect: Tuned Logic Tree - Marksman Mode)
+# (V15.2 - GEM-Architect: Pure Liquidity Edition)
 
 import asyncio
 import httpx
@@ -10,7 +10,7 @@ import pandas as pd
 import numpy as np
 from typing import List, Dict, Any
 
-# إعدادات التسجيل
+# إعدادات التسجيل - تقليل الضجيج
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("httpcore").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
@@ -34,10 +34,10 @@ 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'
+            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', '2S', '2L', '5S', '5L'
         ]
 
     async def initialize(self):
@@ -45,7 +45,7 @@ class DataManager:
         print("   > [DataManager] Starting initialization...")
         self.http_client = httpx.AsyncClient(timeout=30.0)
         await self._load_markets()
-        print(f"✅ [DataManager V15.1] Ready (Logic Tree: Tuned/Flexible).")
+        print(f"✅ [DataManager V15.2] Ready (Strategy: Top 100 Volume).")
 
     async def _load_markets(self):
         try:
@@ -74,229 +74,80 @@ class DataManager:
         return self.contracts_db
     
     # ==================================================================
-    # 🛡️ Layer 1: The Tuned Decision Tree Screening
+    # 🛡️ Layer 1: Rapid Screening (Top 100 Volume Only)
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        print(f"🔍 [Layer 1] Initiating Tuned Logic Tree Screening...")
+        """
+        جلب أعلى 100 عملة من حيث حجم التداول (USDT Volume).
+        المنطق: السيولة هي الملك. العملات النشطة هي الأفضل للتداول الآلي.
+        """
+        print(f"🔍 [Layer 1] Screening Top 100 by Volume...")
         
-        # 1. المرحلة 0: فلتر الكون (مخفف)
-        initial_candidates = await self._stage0_universe_filter()
-        
-        if not initial_candidates:
-            return []
-
-        # 2. جلب البيانات الفنية
-        top_liquid_candidates = initial_candidates[:300] 
-        enriched_data = await self._fetch_technical_data_batch(top_liquid_candidates)
-        
-        # 3. تطبيق شجرة القرار (Overbought -> Classify)
-        breakout_list = []
-        reversal_list = []
-
-        for item in enriched_data:
-            classification = self._apply_logic_tree(item)
-            
-            if classification['type'] == 'BREAKOUT':
-                item['l1_sort_score'] = classification['score'] 
-                breakout_list.append(item)
-            elif classification['type'] == 'REVERSAL':
-                item['l1_sort_score'] = classification['score'] 
-                reversal_list.append(item)
-
-        print(f"   -> [L1 Logic] Found: {len(breakout_list)} Breakouts, {len(reversal_list)} Reversals.")
-
-        # 4. الترتيب والدمج النهائي
-        breakout_list.sort(key=lambda x: x['l1_sort_score'], reverse=True)
-        reversal_list.sort(key=lambda x: x['l1_sort_score'], reverse=False)
-
-        final_selection = breakout_list[:80] + reversal_list[:70]
-        
-        cleaned_selection = []
-        for item in final_selection:
-            cleaned_selection.append({
-                'symbol': item['symbol'],
-                'quote_volume': item.get('quote_volume', 0),
-                'current_price': item.get('current_price', 0),
-                'type': item.get('type', 'UNKNOWN'), # نمرر النوع لـ app.py إذا رغب باستخدامه
-                'l1_score': item.get('l1_sort_score', 0)
-            })
-
-        print(f"✅ [Layer 1] Final Selection: {len(cleaned_selection)} candidates passed to models.")
-        return cleaned_selection
-
-    # ------------------------------------------------------------------
-    # Stage 0: Universe Filter (RELAXED)
-    # ------------------------------------------------------------------
-    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
         try:
+            # 1. جلب كل الأسعار والبيانات اللحظية
             tickers = await self.exchange.fetch_tickers()
             candidates = []
             
             for symbol, ticker in tickers.items():
+                # تصفية الأزواج غير المرغوبة
                 if not symbol.endswith('/USDT'): continue
                 
+                # تصفية العملات المحظورة (stablecoins, leveraged tokens)
                 base_curr = symbol.split('/')[0]
                 if any(bad in base_curr for bad in self.BLACKLIST_TOKENS): continue
                 
-                # 👇 [Tuning] خفضنا السيولة المطلوبة لمليون واحد فقط
-                quote_vol = ticker.get('quoteVolume')
-                if not quote_vol or quote_vol < 1_000_000: continue
-                
+                # التأكد من صحة البيانات
+                quote_vol = ticker.get('quoteVolume') # حجم التداول بالدولار
                 last_price = ticker.get('last')
-                if not last_price or last_price < 0.0005: continue 
+                
+                if quote_vol is None or last_price is None: continue
+                if last_price <= 0: continue
+
+                # الحد الأدنى للقبول (مليون دولار حجم تداول يومي) لتجنب العملات الميتة تماماً
+                if quote_vol < 1_000_000: continue
 
                 candidates.append({
                     'symbol': symbol,
                     'quote_volume': quote_vol,
                     'current_price': last_price,
-                    'change_24h': ticker.get('percentage', 0.0) 
+                    'change_24h': ticker.get('percentage', 0.0)
                 })
             
+            # 2. الترتيب التنازلي حسب حجم التداول
             candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
-            return candidates
             
-        except Exception as e:
-            print(f"❌ [L1 Error] Universe filter failed: {e}")
-            return []
-
-    # ------------------------------------------------------------------
-    # Data Fetching Helpers
-    # ------------------------------------------------------------------
-    async def _fetch_technical_data_batch(self, candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        chunk_size = 10 
-        results = []
-        for i in range(0, len(candidates), chunk_size):
-            chunk = candidates[i:i + chunk_size]
-            chunk_tasks = [self._fetch_single_tech_data(c) for c in chunk]
-            chunk_results = await asyncio.gather(*chunk_tasks)
-            results.extend([r for r in chunk_results if r is not None])
-            await asyncio.sleep(0.1) # تسريع قليل
-        return results
-
-    async def _fetch_single_tech_data(self, candidate: Dict[str, Any]) -> Any:
-        symbol = candidate['symbol']
-        try:
-            ohlcv_1h = await self.exchange.fetch_ohlcv(symbol, '1h', limit=60)
-            ohlcv_15m = await self.exchange.fetch_ohlcv(symbol, '15m', limit=60)
+            # 3. اختيار أعلى 100 عملة فقط
+            top_100 = candidates[:100]
             
-            if not ohlcv_1h or len(ohlcv_1h) < 55 or not ohlcv_15m or len(ohlcv_15m) < 55:
-                return None
-                
-            candidate['ohlcv_1h'] = ohlcv_1h
-            candidate['ohlcv_15m'] = ohlcv_15m
-            return candidate
-        except Exception:
-            return None
-
-    # ------------------------------------------------------------------
-    # 🧠 The Logic Core: Math & Decision Tree (RELAXED)
-    # ------------------------------------------------------------------
-    def _apply_logic_tree(self, data: Dict[str, Any]) -> Dict[str, Any]:
-        try:
-            df_1h = self._calc_indicators(data['ohlcv_1h'])
-            df_15m = self._calc_indicators(data['ohlcv_15m'])
-        except:
-            return {'type': 'NONE'}
-
-        curr_1h = df_1h.iloc[-1]
-        curr_15m = df_15m.iloc[-1]
-        
-        # --- Stage 2: Overbought Filter ---
-        try:
-            close_4h_ago = df_1h.iloc[-5]['close']
-            change_4h = ((curr_1h['close'] - close_4h_ago) / close_4h_ago) * 100
-        except: change_4h = 0.0
-
-        if change_4h > 15.0: return {'type': 'NONE'}
-        if data.get('change_24h', 0) > 25.0: return {'type': 'NONE'}
-        if curr_1h['rsi'] > 80: return {'type': 'NONE'} # 👇 [Tuning] سمحنا بـ RSI أعلى قليلاً
-
-        deviation = (curr_1h['close'] - curr_1h['ema20']) / curr_1h['atr'] if curr_1h['atr'] > 0 else 0
-        if deviation > 2.5: return {'type': 'NONE'} # 👇 [Tuning] سمحنا بإنحراف أكبر قليلاً
-
-        # --- Stage 3: Classification ---
-        
-        # === A. Breakout Logic (RELAXED) ===
-        is_breakout = False
-        breakout_score = 0.0
-        
-        # Trend check (EMA Cross OR Price above both EMAs)
-        bullish_structure = (curr_1h['ema20'] > curr_1h['ema50']) or (curr_1h['close'] > curr_1h['ema20'])
-        
-        if bullish_structure:
-            # 👇 [Tuning] RSI range expanded
-            if 40 <= curr_1h['rsi'] <= 70:
-                # 15m bullish
-                if curr_15m['close'] >= curr_15m['ema20']:
-                    # Volatility check (Range)
-                    avg_range = (df_15m['high'] - df_15m['low']).rolling(10).mean().iloc[-1]
-                    # 👇 [Tuning] Less strict squeeze check (1.5x avg range allowed)
-                    if (curr_15m['high'] - curr_15m['low']) <= avg_range * 1.5:
-                        vol_ma20 = df_15m['volume'].rolling(20).mean().iloc[-1]
-                        # 👇 [Tuning] Volume Spike lowered to 1.2x
-                        if curr_15m['volume'] >= 1.2 * vol_ma20:
-                            is_breakout = True
-                            breakout_score = curr_15m['volume'] / vol_ma20 if vol_ma20 > 0 else 1.0
-
-        if is_breakout:
-            data['type'] = 'BREAKOUT'
-            return {'type': 'BREAKOUT', 'score': breakout_score}
-
-        # === B. Reversal Logic (RELAXED) ===
-        is_reversal = False
-        reversal_score = 100.0 
-        
-        # 👇 [Tuning] RSI threshold increased to 40
-        if 10 <= curr_1h['rsi'] <= 40: 
-            # 👇 [Tuning] Drop requirement reduced to -3%
-            if change_4h <= -3.0: 
-                # Rejection check (Any bullish closing in last 3 candles)
-                last_3 = df_15m.iloc[-3:]
-                found_rejection = False
-                for _, row in last_3.iterrows():
-                    # 👇 [Tuning] Simple logic: Green candle OR Close in upper half
-                    rng = row['high'] - row['low']
-                    if rng > 0:
-                        is_green = row['close'] > row['open']
-                        upper_half = row['close'] > (row['low'] + rng * 0.5)
-                        if is_green or upper_half:
-                            found_rejection = True
-                            break
+            # 4. تنسيق البيانات للنظام التالي (Layer 2)
+            final_selection = []
+            
+            if top_100:
+                max_vol = top_100[0]['quote_volume']
                 
-                if found_rejection:
-                    is_reversal = True
-                    reversal_score = curr_1h['rsi']
-
-        if is_reversal:
-            data['type'] = 'REVERSAL'
-            return {'type': 'REVERSAL', 'score': reversal_score}
+                for item in top_100:
+                    # حساب درجة بسيطة (Score) بناءً على الحجم النسبي
+                    # أعلى عملة تأخذ 1.0، والباقي نسبة منها
+                    vol_score = item['quote_volume'] / max_vol if max_vol > 0 else 0.0
+                    
+                    final_selection.append({
+                        'symbol': item['symbol'],
+                        'quote_volume': item['quote_volume'],
+                        'current_price': item['current_price'],
+                        'type': 'TOP_LIQUIDITY', # وسم موحد للجميع
+                        'l1_score': vol_score    # يستخدم للترتيب الأولي في app.py
+                    })
+
+            print(f"✅ [Layer 1] Selected {len(final_selection)} assets based on pure volume.")
+            return final_selection
 
-        return {'type': 'NONE'}
-
-    def _calc_indicators(self, ohlcv_list):
-        df = pd.DataFrame(ohlcv_list, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-        
-        delta = df['close'].diff()
-        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
-        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
-        rs = gain / loss
-        df['rsi'] = 100 - (100 / (1 + rs))
-        
-        df['ema20'] = df['close'].ewm(span=20, adjust=False).mean()
-        df['ema50'] = df['close'].ewm(span=50, adjust=False).mean()
-        
-        high_low = df['high'] - df['low']
-        high_close = np.abs(df['high'] - df['close'].shift())
-        low_close = np.abs(df['low'] - df['close'].shift())
-        ranges = pd.concat([high_low, high_close, low_close], axis=1)
-        true_range = np.max(ranges, axis=1)
-        df['atr'] = true_range.rolling(14).mean()
-        
-        df.fillna(0, inplace=True)
-        return df
+        except Exception as e:
+            print(f"❌ [Layer 1 Error] Screening failed: {e}")
+            traceback.print_exc()
+            return []
 
     # ==================================================================
-    # 🎯 Public Helpers
+    # 🎯 Public Helpers (Used by Processor & TradeManager)
     # ==================================================================
     async def get_latest_price_async(self, symbol: str) -> float:
         try:
@@ -306,6 +157,7 @@ class DataManager:
 
     async def get_latest_ohlcv(self, symbol: str, timeframe: str = '5m', limit: int = 100) -> List[List[float]]:
         try:
+            # دالة مساعدة لجلب الشموع للمحركات الأخرى (Processor/Guardian)
             candles = await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
             return candles or []
         except Exception: return []