Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,5 +1,5 @@
 # ml_engine/data_manager.py
-# (V15.2 - GEM-Architect: Tuned Logic Tree - Marksman Mode, R2-Compatible)
+# (V15.1 - GEM-Architect: Tuned Logic Tree - Marksman Mode)
 
 import asyncio
 import httpx
@@ -15,66 +15,37 @@ logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("httpcore").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
 
-
 class DataManager:
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
-        """
-        DataManager
-        ----------
-        contracts_db : dict
-            قاعدة بيانات العقود (تُحمّل من R2 عند توفره).
-        whale_monitor : أي كائن حيتان خارجي (EnhancedWhaleMonitor) أو None
-            يتم تمريره من app.py وربطه لاحقاً.
-        r2_service : كائن خدمة R2 أو None
-            يُستخدم لتحميل/حفظ قاعدة العقود.
-        """
-        # إعدادات التحكم
-        self.contracts_db: Dict[str, Any] = contracts_db or {}
+        # ==================================================================
+        # ⚙️ إعدادات التحكم
+        # ==================================================================
+        self.contracts_db = contracts_db or {}
         self.whale_monitor = whale_monitor
         self.r2_service = r2_service
-
-        # إعداد المنصة (KuCoin Spot)
-        self.exchange = ccxt.kucoin(
-            {
-                "enableRateLimit": True,
-                "timeout": 60000,
-                "options": {"defaultType": "spot"},
-            }
-        )
-
-        # HTTP client + كاش الأسواق
+        
+        # إعداد المنصة (KuCoin)
+        self.exchange = ccxt.kucoin({
+            'enableRateLimit': True,
+            'timeout': 60000, 
+            'options': {'defaultType': 'spot'} 
+        })
+            
         self.http_client = None
-        self.market_cache: Dict[str, Any] = {}
-
+        self.market_cache = {}
+        
         # قوائم الاستبعاد
         self.BLACKLIST_TOKENS = [
-            "USDT",
-            "USDC",
-            "DAI",
-            "TUSD",
-            "BUSD",
-            "FDUSD",
-            "EUR",
-            "PAX",
-            "UP",
-            "DOWN",
-            "BEAR",
-            "BULL",
-            "3S",
-            "3L",
+            'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX',
+            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L'
         ]
 
-    # ==================================================================
-    # 🚀 Lifecycle
-    # ==================================================================
     async def initialize(self):
         """تهيئة مدير البيانات والاتصالات"""
         print("   > [DataManager] Starting initialization...")
         self.http_client = httpx.AsyncClient(timeout=30.0)
         await self._load_markets()
-        # تحميل العقود من R2 إن وجد
-        await self.load_contracts_from_r2()
-        print("✅ [DataManager] Ready (Logic Tree: Tuned/Flexible).")
+        print(f"✅ [DataManager V15.1] Ready (Logic Tree: Tuned/Flexible).")
 
     async def _load_markets(self):
         try:
@@ -86,150 +57,104 @@ class DataManager:
             traceback.print_exc()
 
     async def close(self):
-        try:
-            if self.http_client:
-                await self.http_client.aclose()
-        finally:
-            self.http_client = None
-
-        try:
-            if self.exchange:
-                await self.exchange.close()
-        except Exception:
-            pass
-
-    def get_contracts_db(self) -> Dict[str, Any]:
-        return self.contracts_db
+        if self.http_client: await self.http_client.aclose()
+        if self.exchange: await self.exchange.close()
 
     # ==================================================================
     # 🚀 R2 Compatibility
     # ==================================================================
     async def load_contracts_from_r2(self):
-        """تحميل contracts_db من R2 إن كانت الخدمة متاحة"""
-        if not self.r2_service:
-            return
+        if not self.r2_service: return
         try:
             self.contracts_db = await self.r2_service.load_contracts_db_async()
-        except Exception as e:
-            print(f"❌ [DataManager] load_contracts_from_r2 failed: {e}")
+        except Exception:
+            self.contracts_db = {}
 
+    def get_contracts_db(self) -> Dict[str, Any]:
+        return self.contracts_db
+    
     # ==================================================================
-    # 🛡️ Layer 1: Tuned Decision Tree Screening
+    # 🛡️ Layer 1: The Tuned Decision Tree Screening
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        print("🔍 [Layer 1] Initiating Tuned Logic Tree Screening...")
-
+        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]
+        top_liquid_candidates = initial_candidates[:300] 
         enriched_data = await self._fetch_technical_data_batch(top_liquid_candidates)
-
+        
         # 3. تطبيق شجرة القرار (Overbought -> Classify)
-        breakout_list: List[Dict[str, Any]] = []
-        reversal_list: List[Dict[str, Any]] = []
-        neutral_list: List[Dict[str, Any]] = []
+        breakout_list = []
+        reversal_list = []
 
         for item in enriched_data:
             classification = self._apply_logic_tree(item)
-            ctype = classification.get("type")
-
-            # Stage 2 hard filter: تخلص فقط من الحالات المرفوضة تماماً
-            if ctype == "FILTERED":
-                continue
-
-            if ctype == "BREAKOUT":
-                item["l1_sort_score"] = float(classification.get("score", 0.0))
+            
+            if classification['type'] == 'BREAKOUT':
+                item['l1_sort_score'] = classification['score'] 
                 breakout_list.append(item)
-            elif ctype == "REVERSAL":
-                item["l1_sort_score"] = float(classification.get("score", 0.0))
+            elif classification['type'] == 'REVERSAL':
+                item['l1_sort_score'] = classification['score'] 
                 reversal_list.append(item)
-            else:
-                # عملات "عادية" تمر كفلتر أولي فقط – نرتبها بالسيولة
-                item["l1_sort_score"] = float(item.get("quote_volume", 0.0) or 0.0)
-                neutral_list.append(item)
 
-        print(
-            f"   -> [L1 Logic] Found: {len(breakout_list)} Breakouts, "
-            f"{len(reversal_list)} Reversals, {len(neutral_list)} Neutral."
-        )
+        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)
-        neutral_list.sort(key=lambda x: x["l1_sort_score"], reverse=True)
-
-        MAX_L1_CANDIDATES = 150
-        top_breakouts = breakout_list[:80]
-        top_reversals = reversal_list[:70]
+        breakout_list.sort(key=lambda x: x['l1_sort_score'], reverse=True)
+        reversal_list.sort(key=lambda x: x['l1_sort_score'], reverse=False)
 
-        remaining_slots = max(
-            0, MAX_L1_CANDIDATES - len(top_breakouts) - len(top_reversals)
-        )
-        top_neutrals = neutral_list[:remaining_slots] if remaining_slots > 0 else []
-
-        final_selection = top_breakouts + top_reversals + top_neutrals
-
-        cleaned_selection: List[Dict[str, Any]] = []
+        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.0),
-                    "current_price": item.get("current_price", 0.0),
-                    "type": item.get("type", "UNKNOWN"),
-                    "l1_score": item.get("l1_sort_score", 0.0),
-                }
-            )
-
-        print(
-            f"✅ [Layer 1] Final Selection: {len(cleaned_selection)} candidates passed to models."
-        )
+            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)  +  إزالة شرط الحد الأدنى للسعر
+    # Stage 0: Universe Filter (RELAXED)
     # ------------------------------------------------------------------
     async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
         try:
             tickers = await self.exchange.fetch_tickers()
-            candidates: List[Dict[str, Any]] = []
-
+            candidates = []
+            
             for symbol, ticker in tickers.items():
-                # نعمل فقط على أزواج USDT
-                if not symbol.endswith("/USDT"):
-                    continue
-
-                base_curr = symbol.split("/")[0]
-                if any(bad in base_curr for bad in self.BLACKLIST_TOKENS):
-                    continue
-
-                # سيولة دنيا
-                quote_vol = ticker.get("quoteVolume")
-                if not quote_vol or quote_vol < 1_000_000:
-                    continue
-
-                last_price = ticker.get("last")
-                # تم إلغاء شرط 0.0005 بالكامل – نسمح بأي سعر > 0 ما دام موجوداً
-                if last_price is None:
-                    continue
-
-                candidates.append(
-                    {
-                        "symbol": symbol,
-                        "quote_volume": quote_vol,
-                        "current_price": last_price,
-                        "change_24h": ticker.get("percentage", 0.0),
-                    }
-                )
-
-            candidates.sort(key=lambda x: x["quote_volume"], reverse=True)
-            print(f"   -> [Stage0] Universe Filter found {len(candidates)} USDT pairs.")
+                if not symbol.endswith('/USDT'): continue
+                
+                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
+                
+                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) 
+                })
+            
+            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 []
@@ -237,199 +162,156 @@ class DataManager:
     # ------------------------------------------------------------------
     # Data Fetching Helpers
     # ------------------------------------------------------------------
-    async def _fetch_technical_data_batch(
-        self, candidates: List[Dict[str, Any]]
-    ) -> List[Dict[str, Any]]:
-        results: List[Dict[str, Any]] = []
-
-        async def process_symbol(item: Dict[str, Any]):
-            symbol = item["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 not ohlcv_1h or len(ohlcv_1h) < 55:
-                    return None
-                if not ohlcv_15m or len(ohlcv_15m) < 55:
-                    return None
+    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
 
-                item["ohlcv_1h"] = ohlcv_1h
-                item["ohlcv_15m"] = ohlcv_15m
-                return item
-            except Exception as e:
-                print(f"   -> [Stage1.5] Error fetching OHLCV for {symbol}: {e}")
+    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 not ohlcv_1h or len(ohlcv_1h) < 55 or not ohlcv_15m or len(ohlcv_15m) < 55:
                 return None
-
-        batch_size = 25
-        for i in range(0, len(candidates), batch_size):
-            batch = candidates[i : i + batch_size]
-            tasks = [process_symbol(c) for c in batch]
-            batch_results = await asyncio.gather(*tasks, return_exceptions=False)
-            for r in batch_results:
-                if r is not None:
-                    results.append(r)
-            await asyncio.sleep(0.1)
-
-        print(f"   -> [Stage1.5] Enriched OHLCV for {len(results)} symbols.")
-        return results
+                
+            candidate['ohlcv_1h'] = ohlcv_1h
+            candidate['ohlcv_15m'] = ohlcv_15m
+            return candidate
+        except Exception:
+            return None
 
     # ------------------------------------------------------------------
-    # Stage 2 + 3: Overbought Filter + Classification
+    # 🧠 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 Exception as e:
-            print(f"❌ [LogicTree] Indicator calc failed: {e}")
-            return {"type": "FILTERED"}
+            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 / Extreme Pump Filter (RELAXED HARD FILTER) ---
+        
+        # --- 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.0
-        except Exception:
-            change_4h = 0.0
+            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 > 25.0:
-            return {"type": "FILTERED"}
-        if data.get("change_24h", 0.0) > 40.0:
-            return {"type": "FILTERED"}
-        if curr_1h["rsi"] > 88.0:
-            return {"type": "FILTERED"}
+        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"] and curr_1h["atr"] > 0
-            else 0.0
-        )
-        if deviation > 3.5:
-            return {"type": "FILTERED"}
+        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 EMA20)
-        bullish_structure = (curr_1h["ema20"] > curr_1h["ema50"]) or (
-            curr_1h["close"] > curr_1h["ema20"]
-        )
-
+        
+        # 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:
-            # RSI صحي للزخم
-            if 40.0 <= curr_1h["rsi"] <= 70.0:
+            # 👇 [Tuning] RSI range expanded
+            if 40 <= curr_1h['rsi'] <= 70:
                 # 15m bullish
-                if curr_15m["close"] >= curr_15m["ema20"]:
+                if curr_15m['close'] >= curr_15m['ema20']:
                     # Volatility check (Range)
-                    avg_range = (
-                        (df_15m["high"] - df_15m["low"])
-                        .rolling(10)
-                        .mean()
-                        .iloc[-1]
-                    )
-                    # Less strict squeeze check
-                    if (curr_15m["high"] - curr_15m["low"]) <= avg_range * 1.5:
-                        vol_ma20 = df_15m["volume"].rolling(20).mean().iloc[-1]
-                        if vol_ma20 and vol_ma20 > 0:
-                            if curr_15m["volume"] >= 1.2 * vol_ma20:
-                                is_breakout = True
-                                breakout_score = curr_15m["volume"] / vol_ma20
-                        else:
-                            # حجم تداول بدون MA موثوق
+                    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 = 1.0
+                            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}
+            data['type'] = 'BREAKOUT'
+            return {'type': 'BREAKOUT', 'score': breakout_score}
 
         # === B. Reversal Logic (RELAXED) ===
         is_reversal = False
-        reversal_score = 100.0
-
-        if 10.0 <= curr_1h["rsi"] <= 40.0:
-            if change_4h <= -3.0:
+        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():
-                    rng = row["high"] - row["low"]
+                    # 👇 [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)
+                        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 = float(curr_1h["rsi"])
+                    reversal_score = curr_1h['rsi']
 
         if is_reversal:
-            data["type"] = "REVERSAL"
-            return {"type": "REVERSAL", "score": reversal_score}
+            data['type'] = 'REVERSAL'
+            return {'type': 'REVERSAL', 'score': reversal_score}
 
-        return {"type": "NONE"}
-
-    # ------------------------------------------------------------------
-    # Indicator Helper
-    # ------------------------------------------------------------------
-    def _calc_indicators(self, ohlcv_list: List[List[float]]) -> pd.DataFrame:
-        df = pd.DataFrame(
-            ohlcv_list,
-            columns=["timestamp", "open", "high", "low", "close", "volume"],
-        )
+        return {'type': 'NONE'}
 
-        # RSI
-        delta = df["close"].diff()
-        gain = delta.where(delta > 0, 0.0).rolling(window=14).mean()
-        loss = (-delta.where(delta < 0, 0.0)).rolling(window=14).mean()
+    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.0 - (100.0 / (1.0 + rs))
-
-        # EMAs
-        df["ema20"] = df["close"].ewm(span=20, adjust=False).mean()
-        df["ema50"] = df["close"].ewm(span=50, adjust=False).mean()
-
-        # ATR
-        high_low = df["high"] - df["low"]
-        high_close = (df["high"] - df["close"].shift()).abs()
-        low_close = (df["low"] - df["close"].shift()).abs()
+        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 = ranges.max(axis=1)
-        df["atr"] = true_range.rolling(14).mean()
-
+        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 TradeManager / Processor / app.py)
+    # 🎯 Public Helpers
     # ==================================================================
     async def get_latest_price_async(self, symbol: str) -> float:
         try:
             ticker = await self.exchange.fetch_ticker(symbol)
-            return float(ticker["last"])
-        except Exception:
-            return 0.0
+            return float(ticker['last'])
+        except Exception: return 0.0
 
-    async def get_latest_ohlcv(
-        self, symbol: str, timeframe: str = "5m", limit: int = 100
-    ) -> List[List[float]]:
+    async def get_latest_ohlcv(self, symbol: str, timeframe: str = '5m', limit: int = 100) -> List[List[float]]:
         try:
             candles = await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
             return candles or []
-        except Exception:
-            return []
+        except Exception: return []
 
-    async def get_order_book_snapshot(
-        self, symbol: str, limit: int = 20
-    ) -> Dict[str, Any]:
+    async def get_order_book_snapshot(self, symbol: str, limit: int = 20) -> Dict[str, Any]:
         try:
             ob = await self.exchange.fetch_order_book(symbol, limit)
             return ob
-        except Exception:
-            return {}
+        except Exception: return {}