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: Tuned Logic Tree - Marksman Mode, R2-Compatible)
 
 import asyncio
 import httpx
@@ -17,20 +17,33 @@ logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 
 class DataManager:
-    def __init__(self, contracts_db, whale_tracker, exchange, news_client):
-        self.contracts_db = contracts_db
-        self.whale_tracker = whale_tracker
-        self.exchange = exchange
-        self.news_client = news_client
-
-        # KuCoin spot
-        self.exchange = ccxt.kucoin({
-            "enableRateLimit": True,
-            "timeout": 60000,
-            "options": {"defaultType": "spot"},
-        })
-
-        self.http_client: httpx.AsyncClient | None = None
+    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.whale_monitor = whale_monitor
+        self.r2_service = r2_service
+
+        # إعداد المنصة (KuCoin Spot)
+        self.exchange = ccxt.kucoin(
+            {
+                "enableRateLimit": True,
+                "timeout": 60000,
+                "options": {"defaultType": "spot"},
+            }
+        )
+
+        # HTTP client + كاش الأسواق
+        self.http_client = None
         self.market_cache: Dict[str, Any] = {}
 
         # قوائم الاستبعاد
@@ -51,15 +64,17 @@ class DataManager:
             "3L",
         ]
 
-        if self.contracts_db is None:
-            self.contracts_db = {}
-
+    # ==================================================================
+    # 🚀 Lifecycle
+    # ==================================================================
     async def initialize(self):
         """تهيئة مدير البيانات والاتصالات"""
         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).")
+        # تحميل العقود من R2 إن وجد
+        await self.load_contracts_from_r2()
+        print("✅ [DataManager] Ready (Logic Tree: Tuned/Flexible).")
 
     async def _load_markets(self):
         try:
@@ -71,23 +86,41 @@ class DataManager:
             traceback.print_exc()
 
     async def close(self):
-        if self.http_client:
-            await self.http_client.aclose()
-        if self.exchange:
-            await self.exchange.close()
+        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
 
     # ==================================================================
-    # 🛡️ Layer 1: The Tuned Decision Tree Screening
+    # 🚀 R2 Compatibility
+    # ==================================================================
+    async def load_contracts_from_r2(self):
+        """تحميل contracts_db من R2 إن كانت الخدمة متاحة"""
+        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}")
+
+    # ==================================================================
+    # 🛡️ Layer 1: Tuned Decision Tree Screening
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        print(f"🔍 [Layer 1] Initiating Tuned Logic Tree Screening...")
+        print("🔍 [Layer 1] Initiating Tuned Logic Tree Screening...")
 
         # 1. المرحلة 0: فلتر الكون (مخفف)
         initial_candidates = await self._stage0_universe_filter()
-
         if not initial_candidates:
             return []
 
@@ -96,9 +129,9 @@ class DataManager:
         enriched_data = await self._fetch_technical_data_batch(top_liquid_candidates)
 
         # 3. تطبيق شجرة القرار (Overbought -> Classify)
-        breakout_list = []
-        reversal_list = []
-        neutral_list = []
+        breakout_list: List[Dict[str, Any]] = []
+        reversal_list: List[Dict[str, Any]] = []
+        neutral_list: List[Dict[str, Any]] = []
 
         for item in enriched_data:
             classification = self._apply_logic_tree(item)
@@ -109,14 +142,14 @@ class DataManager:
                 continue
 
             if ctype == "BREAKOUT":
-                item["l1_sort_score"] = classification.get("score", 0.0)
+                item["l1_sort_score"] = float(classification.get("score", 0.0))
                 breakout_list.append(item)
             elif ctype == "REVERSAL":
-                item["l1_sort_score"] = classification.get("score", 0.0)
+                item["l1_sort_score"] = float(classification.get("score", 0.0))
                 reversal_list.append(item)
             else:
                 # عملات "عادية" تمر كفلتر أولي فقط – نرتبها بالسيولة
-                item["l1_sort_score"] = float(item.get("quote_volume", 0)) or 0.0
+                item["l1_sort_score"] = float(item.get("quote_volume", 0.0) or 0.0)
                 neutral_list.append(item)
 
         print(
@@ -140,17 +173,15 @@ class DataManager:
 
         final_selection = top_breakouts + top_reversals + top_neutrals
 
-        cleaned_selection = []
+        cleaned_selection: List[Dict[str, Any]] = []
         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),
+                    "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),
                 }
             )
 
@@ -160,7 +191,7 @@ class DataManager:
         return cleaned_selection
 
     # ------------------------------------------------------------------
-    # Stage 0: Universe Filter (RELAXED)
+    # Stage 0: Universe Filter (RELAXED)  +  إزالة شرط الحد الأدنى للسعر
     # ------------------------------------------------------------------
     async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
         try:
@@ -168,6 +199,7 @@ class DataManager:
             candidates: List[Dict[str, Any]] = []
 
             for symbol, ticker in tickers.items():
+                # نعمل فقط على أزواج USDT
                 if not symbol.endswith("/USDT"):
                     continue
 
@@ -181,6 +213,7 @@ class DataManager:
                     continue
 
                 last_price = ticker.get("last")
+                # تم إلغاء شرط 0.0005 بالكامل – نسمح بأي سعر > 0 ما دام موجوداً
                 if last_price is None:
                     continue
 
@@ -194,6 +227,7 @@ class DataManager:
                 )
 
             candidates.sort(key=lambda x: x["quote_volume"], reverse=True)
+            print(f"   -> [Stage0] Universe Filter found {len(candidates)} USDT pairs.")
             return candidates
 
         except Exception as e:
@@ -206,39 +240,39 @@ class DataManager:
     async def _fetch_technical_data_batch(
         self, candidates: List[Dict[str, Any]]
     ) -> List[Dict[str, Any]]:
-        chunk_size = 10
         results: List[Dict[str, Any]] = []
-        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, return_exceptions=False)
-            for r in chunk_results:
+
+        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
+
+                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}")
+                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
 
-    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
-
-            candidate["ohlcv_1h"] = ohlcv_1h
-            candidate["ohlcv_15m"] = ohlcv_15m
-            return candidate
-        except Exception:
-            return None
-
     # ------------------------------------------------------------------
     # Stage 2 + 3: Overbought Filter + Classification
     # ------------------------------------------------------------------
@@ -246,7 +280,8 @@ class DataManager:
         try:
             df_1h = self._calc_indicators(data["ohlcv_1h"])
             df_15m = self._calc_indicators(data["ohlcv_15m"])
-        except:
+        except Exception as e:
+            print(f"❌ [LogicTree] Indicator calc failed: {e}")
             return {"type": "FILTERED"}
 
         curr_1h = df_1h.iloc[-1]
@@ -255,23 +290,22 @@ class DataManager:
         # --- Stage 2: Overbought / Extreme Pump Filter (RELAXED HARD 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 = ((curr_1h["close"] - close_4h_ago) / close_4h_ago) * 100.0
+        except Exception:
             change_4h = 0.0
 
+        # فلتر "الجنون" فقط
         if change_4h > 25.0:
             return {"type": "FILTERED"}
-        if data.get("change_24h", 0) > 40.0:
+        if data.get("change_24h", 0.0) > 40.0:
             return {"type": "FILTERED"}
-        if curr_1h["rsi"] > 88:
+        if curr_1h["rsi"] > 88.0:
             return {"type": "FILTERED"}
 
         deviation = (
             (curr_1h["close"] - curr_1h["ema20"]) / curr_1h["atr"]
-            if curr_1h["atr"] > 0
-            else 0
+            if curr_1h["atr"] and curr_1h["atr"] > 0
+            else 0.0
         )
         if deviation > 3.5:
             return {"type": "FILTERED"}
@@ -282,14 +316,14 @@ class DataManager:
         is_breakout = False
         breakout_score = 0.0
 
-        # Trend check (EMA Cross OR Price above both EMAs)
+        # Trend check (EMA Cross OR Price above EMA20)
         bullish_structure = (curr_1h["ema20"] > curr_1h["ema50"]) or (
             curr_1h["close"] > curr_1h["ema20"]
         )
 
         if bullish_structure:
-            # RSI range
-            if 40 <= curr_1h["rsi"] <= 70:
+            # RSI صحي للزخم
+            if 40.0 <= curr_1h["rsi"] <= 70.0:
                 # 15m bullish
                 if curr_15m["close"] >= curr_15m["ema20"]:
                     # Volatility check (Range)
@@ -301,17 +335,15 @@ class DataManager:
                     )
                     # 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]
-                        )
-                        # Volume Spike lowered
-                        if curr_15m["volume"] >= 1.2 * vol_ma20:
+                        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 موثوق
                             is_breakout = True
-                            breakout_score = (
-                                curr_15m["volume"] / vol_ma20
-                                if vol_ma20 > 0
-                                else 1.0
-                            )
+                            breakout_score = 1.0
 
         if is_breakout:
             data["type"] = "BREAKOUT"
@@ -321,28 +353,22 @@ class DataManager:
         is_reversal = False
         reversal_score = 100.0
 
-        # RSI threshold
-        if 10 <= curr_1h["rsi"] <= 40:
-            # Drop requirement
+        if 10.0 <= curr_1h["rsi"] <= 40.0:
             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():
-                    # 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
-                        )
+                        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"]
+                    reversal_score = float(curr_1h["rsi"])
 
         if is_reversal:
             data["type"] = "REVERSAL"
@@ -350,7 +376,10 @@ class DataManager:
 
         return {"type": "NONE"}
 
-    def _calc_indicators(self, ohlcv_list):
+    # ------------------------------------------------------------------
+    # 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"],
@@ -358,10 +387,10 @@ class DataManager:
 
         # RSI
         delta = df["close"].diff()
-        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
-        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
+        gain = delta.where(delta > 0, 0.0).rolling(window=14).mean()
+        loss = (-delta.where(delta < 0, 0.0)).rolling(window=14).mean()
         rs = gain / loss
-        df["rsi"] = 100 - (100 / (1 + rs))
+        df["rsi"] = 100.0 - (100.0 / (1.0 + rs))
 
         # EMAs
         df["ema20"] = df["close"].ewm(span=20, adjust=False).mean()
@@ -369,8 +398,8 @@ class DataManager:
 
         # ATR
         high_low = df["high"] - df["low"]
-        high_close = np.abs(df["high"] - df["close"].shift())
-        low_close = np.abs(df["low"] - df["close"].shift())
+        high_close = (df["high"] - df["close"].shift()).abs()
+        low_close = (df["low"] - df["close"].shift()).abs()
         ranges = pd.concat([high_low, high_close, low_close], axis=1)
         true_range = ranges.max(axis=1)
         df["atr"] = true_range.rolling(14).mean()
@@ -378,7 +407,7 @@ class DataManager:
         return df
 
     # ==================================================================
-    # 🎯 Public Helpers
+    # 🎯 Public Helpers (used by TradeManager / Processor / app.py)
     # ==================================================================
     async def get_latest_price_async(self, symbol: str) -> float:
         try:
@@ -391,9 +420,7 @@ class DataManager:
         self, symbol: str, timeframe: str = "5m", limit: int = 100
     ) -> List[List[float]]:
         try:
-            candles = await self.exchange.fetch_ohlcv(
-                symbol, timeframe, limit=limit
-            )
+            candles = await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
             return candles or []
         except Exception:
             return []