Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V36.0 - GEM-Architect: Regime-Adaptive Vision Full)
+# (V40.0 - GEM-Architect: The Scanner Matrix)
 # ============================================================
 
 import asyncio
@@ -14,355 +14,238 @@ from typing import List, Dict, Any
 
 import ccxt.async_support as ccxt
 
-# ✅ استيراد الدستور الديناميكي (لقراءة الحالة الحالية)
 try:
     from ml_engine.processor import SystemLimits
 except ImportError:
-    # Fallback إذا لم يتم التحميل بعد
     class SystemLimits:
         L1_MIN_AFFINITY_SCORE = 15.0
         CURRENT_REGIME = "RANGE"
+        # أوزان الكاشفات الجديدة (تتغير بالباكتست)
+        SCANNER_WEIGHTS = {
+            "RSI_MOMENTUM": 0.3,
+            "BB_BREAKOUT": 0.3,
+            "MACD_CROSS": 0.2,
+            "VOLUME_FLOW": 0.2
+        }
 
-# إعدادات التسجيل لإسكات الإزعاج
 logging.getLogger("httpx").setLevel(logging.WARNING)
-logging.getLogger("httpcore").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 class DataManager:
     """
-    DataManager V36.0 (The Chameleon Eye)
-    - تتغير خوارزمية الفلترة (L1) تماماً بناءً على حالة السوق (Regime).
-    - تعتمد على SystemLimits.CURRENT_REGIME لتحديد التكتيك.
+    DataManager V40.0 (The Scanner Matrix)
+    - L1 Screening uses a multi-strategy ensemble approach.
+    - Optimized for speed using batch processing.
     """
 
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
-        # ==================================================================
-        # ⚙️ إعدادات التحكم والتهيئة
-        # ==================================================================
         self.contracts_db = contracts_db or {}
         self.whale_monitor = whale_monitor
         self.r2_service = r2_service
-        
-        # إعداد عميل المنصة (KuCoin) مع تفعيل حدود السرعة
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
             'timeout': 30000, 
             'options': {'defaultType': 'spot'} 
         })
-            
         self.http_client = None
-        self.market_cache = {}
-        
-        # 🚫 قوائم الاستبعاد (العملات المستقرة، الرافعة، والعملات المحظورة)
-        self.BLACKLIST_TOKENS = [
-            'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX',
-            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP'
-        ]
-        
-        print(f"📦 [DataManager V36.0] Adaptive Vision Online.")
+        self.BLACKLIST_TOKENS = ['USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L']
+        print(f"📦 [DataManager V40.0] Scanner Matrix Online.")
 
     async def initialize(self):
-        """تهيئة مدير البيانات والاتصالات"""
-        print("   > [DataManager] Starting initialization...")
-        try:
-            self.http_client = httpx.AsyncClient(timeout=60.0)
-            await self._load_markets()
-            print(f"✅ [DataManager] Ready (Mode: {getattr(SystemLimits, 'CURRENT_REGIME', 'UNKNOWN')}).")
-        except Exception as e:
-            print(f"❌ [DataManager] Init Error: {e}")
-            traceback.print_exc()
+        self.http_client = httpx.AsyncClient(timeout=60.0)
+        await self._load_markets()
 
     async def _load_markets(self):
-        """تحميل أزواج التداول وحفظها في الذاكرة"""
-        try:
-            if self.exchange:
-                # التأكد من عدم تحميل الأسواق مرتين إذا كانت محملة
-                if not self.exchange.markets:
-                    await self.exchange.load_markets()
-                self.market_cache = self.exchange.markets
-        except Exception as e:
-            print(f"❌ [DataManager] Market load failed: {e}")
-            traceback.print_exc()
+        if self.exchange and not self.exchange.markets:
+            await self.exchange.load_markets()
 
     async def close(self):
-        """إغلاق الاتصالات بأمان"""
-        if self.http_client: 
-            await self.http_client.aclose()
-        if self.exchange: 
-            await self.exchange.close()
-        print("🛑 [DataManager] Connections Closed.")
+        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):
-        if not self.r2_service: return
-        try:
-            self.contracts_db = await self.r2_service.load_contracts_db_async()
-            print(f"   -> [DataManager] Contracts DB updated from R2: {len(self.contracts_db)} records.")
-        except Exception as e:
-            print(f"⚠️ [DataManager] R2 Load Warning: {e}")
-            self.contracts_db = {}
-
-    def get_contracts_db(self) -> Dict[str, Any]:
-        return self.contracts_db
-
-    # ==================================================================
-    # 🛡️ Layer 1: Regime-Adaptive Screening (The Core Update)
+    # 🛡️ Layer 1: The Scanner Matrix (New Logic)
     # ==================================================================
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
         """
-        تقوم باختيار استراتيجية المسح بناءً على حالة السوق الحالية.
+        تنفيذ الفحص المتعدد (Matrix Scan).
+        1. جلب أفضل 80 عملة سيولة.
+        2. جلب شموع 15m لهذه العملات.
+        3. تطبيق 4 استراتيجيات كشف مختلفة.
+        4. حساب النتيجة الموزونة.
         """
-        # 1. تحديد الحالة من الدستور
-        # ملاحظة: AdaptiveHub هو المسؤول عن تحديث هذه القيمة في SystemLimits
         current_regime = getattr(SystemLimits, "CURRENT_REGIME", "RANGE")
+        scanner_weights = getattr(SystemLimits, "SCANNER_WEIGHTS", {"RSI_MOMENTUM": 1.0})
         min_score = getattr(SystemLimits, "L1_MIN_AFFINITY_SCORE", 15.0)
-        
-        print(f"🔍 [Layer 1] Screening Mode: {current_regime} | Min Score: {min_score}")
-        
-        # 2. جلب الكون الأولي (Universe) - استخدام الفلتر الأساسي الكامل
-        all_tickers = await self._stage0_universe_filter()
-        if not all_tickers: 
-            print("⚠️ [Layer 1] No tickers passed Stage 0.")
-            return []
 
-        # 3. توجيه المسح حسب الاستراتيجية
-        candidates = []
+        print(f"🔍 [L1 Matrix] Regime: {current_regime} | Weights: {scanner_weights}")
+
+        # 1. تصفية الكون الأولي (High Volume Universe)
+        tickers = await self._fetch_universe_tickers()
+        if not tickers: return []
         
-        if current_regime == "BULL":
-            candidates = self._pre_filter_bull(all_tickers)
-        elif current_regime == "BEAR":
-            candidates = self._pre_filter_bear(all_tickers)
-        elif current_regime == "DEAD":
-            candidates = self._pre_filter_dead(all_tickers)
-        else: # RANGE or Default
-            candidates = self._pre_filter_range(all_tickers)
-            
-        print(f"   -> Pre-filter selected {len(candidates)} candidates for Deep Scan.")
+        # نأخذ أفضل 80 عملة فقط لتجنب قتل الـ API Rate Limits
+        top_candidates = tickers[:80]
         
-        # 4. الجلب العميق للبيانات (Deep Data Fetch)
-        # نأخذ أفضل 60 عملة فقط لتوفير الموارد
-        top_candidates = candidates[:60]
-        enriched_data = await self._fetch_technical_and_depth_batch(top_candidates)
+        # 2. جلب البيانات الفنية دفعة واحدة (Batch Fetch 15m)
+        enriched_data = await self._batch_fetch_ta_data(top_candidates, timeframe='15m', limit=100)
         
-        if not enriched_data:
-            print("❌ [Layer 1] Failed to fetch deep data.")
-            return []
-
-        # 5. حساب النقاط (Synergy Score) بناءً على الحالة
-        final_selection = []
+        scored_candidates = []
         for item in enriched_data:
-            synergy = self._calculate_adaptive_score(item, current_regime)
-            item['l1_score'] = synergy['score']
-            item['tags'] = synergy['tags']
+            df = item.get('df')
+            if df is None or len(df) < 50: continue
+            
+            # 3. تطبيق الكاشفات (Scanners)
+            scores = self._apply_scanner_strategies(df)
+            
+            # 4. حساب النتيجة النهائية الموزونة
+            final_score = 0.0
+            tags = []
+            
+            for strategy, val in scores.items():
+                w = scanner_weights.get(strategy, 0.0)
+                final_score += (val['score'] * w)
+                if val['active']: tags.append(strategy)
+            
+            # إضافة نقاط إضافية بناءً على النظام القديم (Volume/Price Action)
+            # للحفاظ على التوافقية
+            if item['change_24h'] > 5.0 and current_regime == "BULL": final_score += 10
             
-            if item['l1_score'] >= min_score:
-                final_selection.append(item)
+            item['l1_score'] = final_score
+            item['tags'] = tags
+            
+            if final_score >= min_score:
+                scored_candidates.append(item)
 
-        # ترتيب نهائي وتنظيف
-        final_selection.sort(key=lambda x: x['l1_score'], reverse=True)
-        
-        # طباعة عي��ة للمراقبة
-        if final_selection:
-            print(f"   -> Top pick: {final_selection[0]['symbol']} (Score: {final_selection[0]['l1_score']:.1f})")
-        else:
-            print("⚠️ [Layer 1] No candidates passed the adaptive score threshold.")
+        scored_candidates.sort(key=lambda x: x['l1_score'], reverse=True)
         
-        # نمرر أفضل 40 للمعالج
+        print(f"   -> Matrix selected {len(scored_candidates)} candidates.")
+        if scored_candidates:
+             print(f"   -> Top Pick: {scored_candidates[0]['symbol']} (Score: {scored_candidates[0]['l1_score']:.1f})")
+
         return [
             {
                 'symbol': c['symbol'],
-                'quote_volume': c.get('quote_volume', 0),
-                'current_price': c.get('current_price', 0),
-                'type': ','.join(c.get('tags', [])),
-                'l1_score': c.get('l1_score', 0)
+                'quote_volume': c['quote_volume'],
+                'current_price': c['current_price'],
+                'type': ','.join(c['tags']),
+                'l1_score': c['l1_score']
             }
-            for c in final_selection[:40] 
+            for c in scored_candidates[:40] # تمرير أفضل 40
         ]
 
-    # ==================================================================
-    # 🌍 Stage 0 & Pre-Filters
-    # ==================================================================
-    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
-        """
-        فلتر أولي سريع لاستبعاد العملات الميتة أو المحظورة.
-        """
+    # ------------------------------------------------------------------
+    # 🧩 Scanner Strategies Logic
+    # ------------------------------------------------------------------
+    def _apply_scanner_strategies(self, df: pd.DataFrame) -> Dict[str, Any]:
+        """تطبيق مؤشرات فنية متعددة على البيانات"""
+        results = {}
+        close = df['close']
+        
+        # Strategy A: RSI Momentum (زخم)
+        rsi = ta.rsi(close, length=14)
+        curr_rsi = rsi.iloc[-1]
+        score_rsi = 0
+        active_rsi = False
+        if 50 < curr_rsi < 75: 
+            score_rsi = 100 
+            active_rsi = True
+        elif curr_rsi <= 30: # Oversold Bounce
+            score_rsi = 80
+            active_rsi = True
+        results["RSI_MOMENTUM"] = {'score': score_rsi, 'active': active_rsi}
+
+        # Strategy B: Bollinger Band Breakout (انفجار سعري)
+        bb = ta.bbands(close, length=20, std=2)
+        upper = bb['BBU_20_2.0'].iloc[-1]
+        width = bb['BBB_20_2.0'].iloc[-1]
+        curr_price = close.iloc[-1]
+        score_bb = 0
+        active_bb = False
+        if curr_price > upper and width > 0.1: # اختراق حقيقي
+            score_bb = 100
+            active_bb = True
+        results["BB_BREAKOUT"] = {'score': score_bb, 'active': active_bb}
+
+        # Strategy C: MACD Cross (تغير اتجاه)
+        macd = ta.macd(close)
+        macd_line = macd['MACD_12_26_9'].iloc[-1]
+        signal_line = macd['MACDS_12_26_9'].iloc[-1]
+        hist = macd['MACDh_12_26_9'].iloc[-1]
+        score_macd = 0
+        active_macd = False
+        if macd_line > signal_line and hist > 0:
+            score_macd = 100
+            active_macd = True
+        results["MACD_CROSS"] = {'score': score_macd, 'active': active_macd}
+        
+        # Strategy D: Volume Flow (تدفق سيولة)
+        vol = df['volume']
+        vol_ma = ta.sma(vol, length=20).iloc[-1]
+        curr_vol = vol.iloc[-1]
+        score_vol = 0
+        active_vol = False
+        if curr_vol > (vol_ma * 1.5): # حجم تداول ضخم مفاجئ
+            score_vol = 100
+            active_vol = True
+        results["VOLUME_FLOW"] = {'score': score_vol, 'active': active_vol}
+
+        return results
+
+    # ------------------------------------------------------------------
+    # ⚡ Batch & Async Helpers
+    # ------------------------------------------------------------------
+    async def _fetch_universe_tickers(self):
         try:
             tickers = await self.exchange.fetch_tickers()
             candidates = []
-            
             for symbol, ticker in tickers.items():
                 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 < 300_000: continue # Min 300k USDT
-                
-                last_price = ticker.get('last')
-                if not last_price or last_price < 0.0001: continue 
-
+                if any(bad in symbol for bad in self.BLACKLIST_TOKENS): continue
+                if not ticker.get('quoteVolume') or ticker['quoteVolume'] < 500_000: continue # Min 500k Vol
                 candidates.append({
                     'symbol': symbol,
-                    'quote_volume': quote_vol,
-                    'current_price': last_price,
-                    'change_24h': float(ticker.get('percentage', 0.0)) 
+                    'quote_volume': ticker['quoteVolume'],
+                    'current_price': ticker['last'],
+                    'change_24h': float(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 []
-
-    def _pre_filter_bull(self, tickers):
-        """🐂 Bull Mode: السيولة العالية والارتفاع في السعر"""
-        filtered = [t for t in tickers if t['change_24h'] > -2.0 and t['quote_volume'] > 500_000]
-        filtered.sort(key=lambda x: (x['change_24h'], x['quote_volume']), reverse=True)
-        return filtered
-
-    def _pre_filter_bear(self, tickers):
-        """🐻 Bear Mode: الارتدادات من القاع (Panic Selling)"""
-        filtered = [t for t in tickers if t['change_24h'] < -5.0 and t['quote_volume'] > 1_000_000]
-        filtered.sort(key=lambda x: x['change_24h'], reverse=False) # تصاعدي (الأكثر سلبية)
-        return filtered
-
-    def _pre_filter_range(self, tickers):
-        """↔️ Range Mode: الاستقرار والسيولة المتوسطة"""
-        filtered = [t for t in tickers if -5.0 < t['change_24h'] < 5.0 and t['quote_volume'] > 300_000]
-        filtered.sort(key=lambda x: x['quote_volume'], reverse=True)
-        return filtered
-        
-    def _pre_filter_dead(self, tickers):
-        """💤 Dead/Accumulation Mode: سيولة منخفضة وتحرك مفاجئ"""
-        filtered = [t for t in tickers if t['quote_volume'] > 100_000]
-        import random
-        random.shuffle(filtered) # عشوائية لاستكشاف الجواهر
-        return filtered
-
-    # ==================================================================
-    # 🧠 Adaptive Scoring Matrix
-    # ==================================================================
-    def _calculate_adaptive_score(self, item: Dict[str, Any], regime: str) -> Dict[str, Any]:
-        """
-        حساب النقاط بناءً على السياق (Regime-Specific Scoring).
-        """
-        try:
-            df = pd.DataFrame(item['ohlcv_1h'], columns=['ts', 'o', 'h', 'l', 'c', 'v'])
-            df['c'] = df['c'].astype(float)
-            
-            # المؤشرات الأساسية
-            curr_close = df['c'].iloc[-1]
-            rsi = ta.rsi(df['c'], 14).iloc[-1]
-            ema50 = ta.ema(df['c'], 50).iloc[-1]
-            
-            score = 0.0
-            tags = []
-
-            # 🐂 BULL LOGIC
-            if regime == "BULL":
-                # نحب الـ RSI العالي (زخم) لكن ليس المفرط جداً
-                if 55 < rsi < 80: score += 20
-                if curr_close > ema50: score += 20
-                tags.append("TrendFollow")
-            
-            # 🐻 BEAR LOGIC
-            elif regime == "BEAR":
-                # نحب الـ RSI المنخفض (تشبع بيعي)
-                if rsi < 30: 
-                    score += 30
-                    tags.append("Oversold")
-                elif rsi > 60: 
-                    score -= 20 # خطر
-                
-                # في السوق الهابط، السعر غالباً تحت EMA، نبحث عن الابتعاد الشديد عنه
-                dist = (curr_close - ema50) / ema50
-                if dist < -0.15: 
-                    score += 15
-                    tags.append("DeepValue")
-
-            # ↔️ RANGE LOGIC
-            elif regime == "RANGE":
-                # نحب الـ RSI في المنتصف للارتداد
-                if rsi < 35: 
-                    score += 25; tags.append("RangeBot")
-                elif rsi > 65: 
-                    score -= 10; tags.append("RangeTop")
-                else:
-                    score += 10 # منطقة آمنة
-            
-            # دفتر الطلبات (مشترك لكن بأوزان مختلفة ضمنياً)
-            ob = item.get('order_book_snapshot', {})
-            if ob:
-                bids = sum([float(x[1]) for x in ob.get('bids', [])[:10]])
-                asks = sum([float(x[1]) for x in ob.get('asks', [])[:10]])
-                if asks > 0:
-                    ratio = bids / asks
-                    if ratio > 1.5: score += 15; tags.append("BidWall")
-                    if regime == "BEAR" and ratio > 2.0: score += 20 # جدار الشراء في الهبوط إشارة قوية
-
-            return {'score': score, 'tags': tags}
-
-        except Exception:
-            return {'score': 0, 'tags': ['Error']}
+        except Exception: return []
 
-    # ==================================================================
-    # ⚡ Batch Fetching Utilities (Parallel Execution)
-    # ==================================================================
-    async def _fetch_technical_and_depth_batch(self, candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """جلب البيانات الفنية + دفتر الطلبات بالتوازي"""
-        chunk_size = 10 
+    async def _batch_fetch_ta_data(self, candidates, timeframe='15m', limit=100):
         results = []
-        
+        chunk_size = 15 # لعدم تجاوز الحدود
         for i in range(0, len(candidates), chunk_size):
-            chunk = candidates[i:i + chunk_size]
-            chunk_tasks = [self._fetch_single_full_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) 
-            
+            chunk = candidates[i:i+chunk_size]
+            tasks = [self._fetch_ohlcv_safe(c, timeframe, limit) for c in chunk]
+            chunk_res = await asyncio.gather(*tasks)
+            results.extend([r for r in chunk_res if r is not None])
+            await asyncio.sleep(0.1)
         return results
 
-    async def _fetch_single_full_data(self, candidate: Dict[str, Any]) -> Any:
-        """جلب بيانات عملة واحدة (شارت + عمق)"""
-        symbol = candidate['symbol']
+    async def _fetch_ohlcv_safe(self, candidate, tf, limit):
         try:
-            ohlcv_task = self.exchange.fetch_ohlcv(symbol, '1h', limit=100)
-            ob_task = self.exchange.fetch_order_book(symbol, limit=20)
-            
-            ohlcv_1h, order_book = await asyncio.gather(ohlcv_task, ob_task)
-            
-            if not ohlcv_1h or len(ohlcv_1h) < 50:
-                return None 
-                
-            candidate['ohlcv_1h'] = ohlcv_1h
-            candidate['order_book_snapshot'] = order_book
+            ohlcv = await self.exchange.fetch_ohlcv(candidate['symbol'], tf, limit=limit)
+            if not ohlcv: return None
+            df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+            df['close'] = df['close'].astype(float)
+            df['volume'] = df['volume'].astype(float)
+            candidate['df'] = df 
             return candidate
-            
-        except Exception:
-            return None
-
-    # ==================================================================
-    # 🎯 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
-
-    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 []
-
-    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 {}
-
-    def get_supported_timeframes(self):
-        return list(self.exchange.timeframes.keys()) if self.exchange else []
+        except: return None
+    
+    # Helpers needed for Processor/TradeManager
+    async def get_latest_price_async(self, symbol):
+        t = await self.exchange.fetch_ticker(symbol)
+        return float(t['last'])
+    
+    async def get_latest_ohlcv(self, symbol, tf, limit=100):
+        return await self.exchange.fetch_ohlcv(symbol, tf, limit=limit)
+        
+    async def get_order_book_snapshot(self, symbol, limit=20):
+        return await self.exchange.fetch_order_book(symbol, limit)
+    
+    # R2 Placeholder
+    async def load_contracts_from_r2(self): pass
+    def get_contracts_db(self): return self.contracts_db