Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,5 +1,5 @@
 # ml_engine/data_manager.py
-# (V15.0 - GEM-Architect: Advanced L1 Decision Tree Filter)
+# (V15.1 - GEM-Architect: Tuned Logic Tree - Marksman Mode)
 
 import asyncio
 import httpx
@@ -8,7 +8,7 @@ import ccxt.async_support as ccxt
 import logging
 import pandas as pd
 import numpy as np
-from typing import List, Dict, Any, Tuple
+from typing import List, Dict, Any
 
 # إعدادات التسجيل
 logging.getLogger("httpx").setLevel(logging.WARNING)
@@ -25,7 +25,6 @@ class DataManager:
         self.r2_service = r2_service
         
         # إعداد المنصة (KuCoin)
-        # نرفع التايم أوت ونفعل Rate Limit لضمان استقرار جلب الشموع الكثيف
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
             'timeout': 60000, 
@@ -35,7 +34,7 @@ class DataManager:
         self.http_client = None
         self.market_cache = {}
         
-        # قوائم الاستبعاد (Stablecoins & Leverage Tokens)
+        # قوائم الاستبعاد
         self.BLACKLIST_TOKENS = [
             'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX',
             'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L'
@@ -46,10 +45,9 @@ class DataManager:
         print("   > [DataManager] Starting initialization...")
         self.http_client = httpx.AsyncClient(timeout=30.0)
         await self._load_markets()
-        print(f"✅ [DataManager V15.0] Ready (Advanced L1 Tree Filter).")
+        print(f"✅ [DataManager V15.1] Ready (Logic Tree: Tuned/Flexible).")
 
     async def _load_markets(self):
-        """تحميل بيانات الأسواق وتخزينها مؤقتاً"""
         try:
             if self.exchange:
                 await self.exchange.load_markets()
@@ -59,8 +57,6 @@ class DataManager:
             traceback.print_exc()
 
     async def close(self):
-        """إغلاق جميع الاتصالات بأمان"""
-        print("   > [DataManager] Closing connections...")
         if self.http_client: await self.http_client.aclose()
         if self.exchange: await self.exchange.close()
 
@@ -71,7 +67,6 @@ class DataManager:
         if not self.r2_service: return
         try:
             self.contracts_db = await self.r2_service.load_contracts_db_async()
-            print(f"✅ [DataManager] Contracts loaded: {len(self.contracts_db)}")
         except Exception:
             self.contracts_db = {}
 
@@ -79,58 +74,58 @@ class DataManager:
         return self.contracts_db
     
     # ==================================================================
-    # 🛡️ Layer 1: The Advanced Decision Tree Screening
+    # 🛡️ Layer 1: The Tuned Decision Tree Screening
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        """
-        تنفيذ شجرة القرار (Universe -> Overbought -> Breakout/Reversal).
-        الهدف: استخراج 150 عملة نقية فنيًا.
-        """
-        print(f"🔍 [Layer 1] Initiating Advanced Logic Tree Screening...")
+        print(f"🔍 [Layer 1] Initiating Tuned Logic Tree Screening...")
         
-        # 1. المرحلة 0: فلتر الكون (Universe Filter) - تصفية أولية سريعة
+        # 1. المرحلة 0: فلتر الكون (مخفف)
         initial_candidates = await self._stage0_universe_filter()
-        print(f"   -> [L1 Stage 0] Universe Passed: {len(initial_candidates)} tickers.")
-
+        
         if not initial_candidates:
             return []
 
-        # 2. جلب البيانات الفنية (OHLCV) للناجين فقط
-        # نحدد عددًا أقصى للمعالجة لتجنب بطء شديد (مثلاً أفضل 300 سيولة)
+        # 2. جلب البيانات الفنية
         top_liquid_candidates = initial_candidates[:300] 
         enriched_data = await self._fetch_technical_data_batch(top_liquid_candidates)
-        print(f"   -> [L1 Data] Fetched technicals for {len(enriched_data)} tickers.")
-
+        
         # 3. تطبيق شجرة القرار (Overbought -> Classify)
         breakout_list = []
         reversal_list = []
 
         for item in enriched_data:
             classification = self._apply_logic_tree(item)
+            
             if classification['type'] == 'BREAKOUT':
-                item['l1_score'] = classification['score'] # Volume Ratio
+                item['l1_sort_score'] = classification['score'] 
                 breakout_list.append(item)
             elif classification['type'] == 'REVERSAL':
-                item['l1_score'] = classification['score'] # RSI (Low is better)
+                item['l1_sort_score'] = classification['score'] 
                 reversal_list.append(item)
 
         print(f"   -> [L1 Logic] Found: {len(breakout_list)} Breakouts, {len(reversal_list)} Reversals.")
 
-        # 4. الترتيب والدمج النهائي (Selection)
-        # ترتيب الاختراق: الأكبر في نسبة انفجار الحجم ��ولاً
-        breakout_list.sort(key=lambda x: x['l1_score'], reverse=True)
-        
-        # ترتيب الانعكاس: الـ RSI الأقل أولاً (لأن السكور هو RSI الخام)
-        reversal_list.sort(key=lambda x: x['l1_score'], reverse=False)
+        # 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)
 
-        # الدمج: 80 اختراق + 70 انعكاس = 150
         final_selection = breakout_list[:80] + reversal_list[:70]
         
-        print(f"✅ [Layer 1] Final Selection: {len(final_selection)} candidates ready for models.")
-        return final_selection
+        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 (Liquidity & Valid Pairs)
+    # Stage 0: Universe Filter (RELAXED)
     # ------------------------------------------------------------------
     async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
         try:
@@ -138,18 +133,15 @@ class DataManager:
             candidates = []
             
             for symbol, ticker in tickers.items():
-                # 1. فلتر الزوج (USDT Only)
                 if not symbol.endswith('/USDT'): continue
                 
-                # 2. فلتر العملات المحظورة والمستقرة
                 base_curr = symbol.split('/')[0]
                 if any(bad in base_curr for bad in self.BLACKLIST_TOKENS): continue
                 
-                # 3. فلتر السيولة (2M USDT)
+                # 👇 [Tuning] خفضنا السيولة المطلوبة لمليون واحد فقط
                 quote_vol = ticker.get('quoteVolume')
-                if not quote_vol or quote_vol < 2_000_000: continue
+                if not quote_vol or quote_vol < 1_000_000: continue
                 
-                # 4. فلتر السعر الميت (Dust)
                 last_price = ticker.get('last')
                 if not last_price or last_price < 0.0005: continue 
 
@@ -157,11 +149,9 @@ class DataManager:
                     'symbol': symbol,
                     'quote_volume': quote_vol,
                     'current_price': last_price,
-                    # نحتفظ بهذه القيم للفلتر الأولي
                     'change_24h': ticker.get('percentage', 0.0) 
                 })
             
-            # ترتيب مبدئي بالسيولة لضمان أننا نختار أفضل 300 للمعالجة العميقة
             candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
             return candidates
             
@@ -170,30 +160,22 @@ class DataManager:
             return []
 
     # ------------------------------------------------------------------
-    # Data Fetching Helpers (Parallel)
+    # Data Fetching Helpers
     # ------------------------------------------------------------------
     async def _fetch_technical_data_batch(self, candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """جلب بيانات 1h و 15m بالتوازي مع Rate Limiting"""
-        tasks = []
-        # نقسم العمل إلى دفعات صغيرة لتجنب إغراق الـ Rate Limit
         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.2) 
-            
+            await asyncio.sleep(0.1) # تسريع قليل
         return results
 
     async def _fetch_single_tech_data(self, candidate: Dict[str, Any]) -> Any:
         symbol = candidate['symbol']
         try:
-            # نحتاج 1h (للترند والـ RSI) و 15m (للدخول والانفجار)
-            # نطلب عدد شموع يكفي لحساب EMA50 و RSI14
             ohlcv_1h = await self.exchange.fetch_ohlcv(symbol, '1h', limit=60)
             ohlcv_15m = await self.exchange.fetch_ohlcv(symbol, '15m', limit=60)
             
@@ -207,115 +189,102 @@ class DataManager:
             return None
 
     # ------------------------------------------------------------------
-    # 🧠 The Logic Core: Math & Decision Tree
+    # 🧠 The Logic Core: Math & Decision Tree (RELAXED)
     # ------------------------------------------------------------------
     def _apply_logic_tree(self, data: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        تطبيق شروط الفلتر:
-        Returns: {'type': 'BREAKOUT'|'REVERSAL'|'NONE', 'score': float}
-        """
-        # 1. Prepare DataFrames & Indicators
         try:
             df_1h = self._calc_indicators(data['ohlcv_1h'])
             df_15m = self._calc_indicators(data['ohlcv_15m'])
         except:
             return {'type': 'NONE'}
 
-        # Last Candles
         curr_1h = df_1h.iloc[-1]
         curr_15m = df_15m.iloc[-1]
         
-        # --- Stage 2: Overbought Filter (Reject Pumped Coins) ---
-        # أ) التغير السعري المبالغ فيه (4h)
-        # نحسب تغير 4 ساعات تقريبياً من إغلاق الشمعة الحالية وإغلاق الشمعة قبل 4
+        # --- 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', 'reason': 'Pumped 4h'}
-        if data.get('change_24h', 0) > 25.0: return {'type': 'NONE', 'reason': 'Pumped 24h'}
-
-        # ب) RSI المبالغ فيه
-        if curr_1h['rsi'] > 75: return {'type': 'NONE', 'reason': 'Overbought RSI'}
+        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)
-        deviation = (curr_1h['close'] - curr_1h['ema20']) / curr_1h['atr']
-        if deviation > 2.0: return {'type': 'NONE', 'reason': 'Overextended'}
+        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 Candidate Logic ===
+        # === A. Breakout Logic (RELAXED) ===
         is_breakout = False
         breakout_score = 0.0
         
-        # 1. Trend Filter (1h)
-        if (curr_1h['ema20'] > curr_1h['ema50']) and (curr_1h['close'] > curr_1h['ema20']):
-            # 2. RSI Check (Healthy Zone)
-            if 40 <= curr_1h['rsi'] <= 65:
-                # 3. 15m Confirmation
-                if (curr_15m['ema20'] >= curr_15m['ema50']) and (curr_15m['close'] >= curr_15m['ema20']):
-                    # 4. Squeeze / Low Volatility Check (Range Compression)
-                    # بديل BBW: نقارن ATR الحالي بمتوسط الـ ATR القديم أو مدى الشموع
-                    avg_range_last_10 = (df_15m['high'] - df_15m['low']).rolling(10).mean().iloc[-1]
-                    if curr_15m['high'] - curr_15m['low'] <= avg_range_last_10 * 1.2: # ليس شرطاً صارماً جداً
-                        # 5. Volume Spike (Trigger)
+        # 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]
-                        if curr_15m['volume'] >= 1.5 * vol_ma20:
+                        # 👇 [Tuning] Volume Spike lowered to 1.2x
+                        if curr_15m['volume'] >= 1.2 * vol_ma20:
                             is_breakout = True
-                            # Score = Volume Ratio
                             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 Candidate Logic ===
+        # === B. Reversal Logic (RELAXED) ===
         is_reversal = False
-        reversal_score = 100.0 # Default high RSI
+        reversal_score = 100.0 
         
-        # 1. RSI Oversold (1h)
-        if 10 <= curr_1h['rsi'] <= 35: # وسعنا النطاق قليلاً لـ 35
-            # 2. Strong Drop Logic (4h Change is significantly negative)
-            if change_4h <= -5.0: # هبوط 5% على ا��أقل في 4 ساعات
-                # 3. Rejection Candle on 15m (Wick Check)
-                # نبحث في آخر 3 شموع
-                last_3_candles = df_15m.iloc[-3:]
+        # 👇 [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 idx, row in last_3_candles.iterrows():
-                    body_size = abs(row['close'] - row['open'])
-                    lower_wick = min(row['close'], row['open']) - row['low']
-                    candle_range = row['high'] - row['low']
-                    
-                    # شرط الذيل: الذيل السفلي أطول من الجسم، أو يمثل نسبة معتبرة من الشمعة
-                    if (lower_wick > body_size) and (row['close'] > row['low'] + (candle_range * 0.3)):
-                        found_rejection = True
-                        break
+                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'] # Lower is better
+                    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):
-        """حساب المؤشرات الأساسية باستخدام Pandas"""
         df = pd.DataFrame(ohlcv_list, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
         
-        # RSI 14
         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))
         
-        # EMAs
         df['ema20'] = df['close'].ewm(span=20, adjust=False).mean()
         df['ema50'] = df['close'].ewm(span=50, adjust=False).mean()
         
-        # ATR 14
         high_low = df['high'] - df['low']
         high_close = np.abs(df['high'] - df['close'].shift())
         low_close = np.abs(df['low'] - df['close'].shift())
@@ -323,12 +292,11 @@ class DataManager:
         true_range = np.max(ranges, axis=1)
         df['atr'] = true_range.rolling(14).mean()
         
-        # تنظيف NaN
         df.fillna(0, inplace=True)
         return df
 
     # ==================================================================
-    # 🎯 Data Fetching Helpers (Public)
+    # 🎯 Public Helpers
     # ==================================================================
     async def get_latest_price_async(self, symbol: str) -> float:
         try:
@@ -346,6 +314,4 @@ class DataManager:
         try:
             ob = await self.exchange.fetch_order_book(symbol, limit)
             return ob
-        except Exception: return {}
-
-print("✅ [DataManager V15.0] Loaded with Advanced Decision Tree Filter.")
+        except Exception: return {}