Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,5 +1,5 @@
 # ml_engine/data_manager.py
-# (V15.2 - GEM-Architect: Pure Liquidity Edition)
+# (V15.1 - GEM-Architect: Tuned Logic Tree - Marksman Mode)
 
 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', '2S', '2L', '5S', '5L'
+            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L'
         ]
 
     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.2] Ready (Strategy: Top 100 Volume).")
+        print(f"✅ [DataManager V15.1] Ready (Logic Tree: Tuned/Flexible).")
 
     async def _load_markets(self):
         try:
@@ -74,80 +74,229 @@ class DataManager:
         return self.contracts_db
     
     # ==================================================================
-    # 🛡️ Layer 1: Rapid Screening (Top 100 Volume Only)
+    # 🛡️ Layer 1: The Tuned Decision Tree Screening
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        """
-        جلب أعلى 100 عملة من حيث حجم التداول (USDT Volume).
-        المنطق: السيولة هي الملك. العملات النشطة هي الأفضل للتداول الآلي.
-        """
-        print(f"🔍 [Layer 1] Screening Top 100 by Volume...")
+        print(f"🔍 [Layer 1] Initiating Tuned Logic Tree Screening...")
         
+        # 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
                 
-                # التأكد من صحة البيانات
-                quote_vol = ticker.get('quoteVolume') # حجم التداول بالدولار
-                last_price = ticker.get('last')
+                # 👇 [Tuning] خفضنا السيولة المطلوبة لمليون واحد فقط
+                quote_vol = ticker.get('quoteVolume')
+                if not quote_vol or quote_vol < 1_000_000: continue
                 
-                if quote_vol is None or last_price is None: continue
-                if last_price <= 0: continue
-
-                # الحد الأدنى للقبول (مليون دولار حجم تداول يومي) لتجنب العملات الميتة تماماً
-                if quote_vol < 1_000_000: continue
+                last_price = ticker.get('last')
+                if not last_price or last_price < 0.0005: 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
             
-            # 3. اختيار أعلى 100 عملة فقط
-            top_100 = candidates[:100]
-            
-            # 4. تنسيق البيانات للنظام التالي (Layer 2)
-            final_selection = []
+        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)
             
-            if top_100:
-                max_vol = top_100[0]['quote_volume']
+            if not ohlcv_1h or len(ohlcv_1h) < 55 or not ohlcv_15m or len(ohlcv_15m) < 55:
+                return None
                 
-                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
+            candidate['ohlcv_1h'] = ohlcv_1h
+            candidate['ohlcv_15m'] = ohlcv_15m
+            return candidate
+        except Exception:
+            return None
 
-        except Exception as e:
-            print(f"❌ [Layer 1 Error] Screening failed: {e}")
-            traceback.print_exc()
-            return []
+    # ------------------------------------------------------------------
+    # 🧠 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
+                
+                if found_rejection:
+                    is_reversal = True
+                    reversal_score = curr_1h['rsi']
+
+        if is_reversal:
+            data['type'] = 'REVERSAL'
+            return {'type': 'REVERSAL', 'score': reversal_score}
+
+        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
 
     # ==================================================================
-    # 🎯 Public Helpers (Used by Processor & TradeManager)
+    # 🎯 Public Helpers
     # ==================================================================
     async def get_latest_price_async(self, symbol: str) -> float:
         try:
@@ -157,7 +306,6 @@ 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 []