Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,251 +1,353 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V58.1 - GEM-Architect: Asset-Context Edition + Fixed)
+# (V15.2 - GEM-Architect: Anti-FOMO Shield - Strict Marksman Mode)
 # ============================================================
 
 import asyncio
 import httpx
 import traceback
+import ccxt.async_support as ccxt
 import logging
 import pandas as pd
 import numpy as np
-import pandas_ta as ta
-from typing import List, Dict, Any, Tuple
-from datetime import datetime
-import ccxt.async_support as ccxt
-
-# Keep SystemLimits import for fallbacks
-try:
-    from ml_engine.processor import SystemLimits
-except ImportError:
-    SystemLimits = None
+from typing import List, Dict, Any
 
+# إعدادات التسجيل
 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):
+        # ==================================================================
+        # ⚙️ إعدادات التحكم
+        # ==================================================================
         self.contracts_db = contracts_db or {}
         self.whale_monitor = whale_monitor
         self.r2_service = r2_service
-        # Pass the hub instance later via method arguments
-        self.adaptive_hub_ref = None 
-
+        self.adaptive_hub_ref = None  # مرجع لملف التعلم
+        
+        # إعداد المنصة (KuCoin)
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
-            'timeout': 30000, 
+            'timeout': 60000, 
             'options': {'defaultType': 'spot'} 
         })
             
         self.http_client = None
         self.market_cache = {}
         
-        # Core Blacklist (Stablecoins & Leveraged Tokens)
+        # قوائم الاستبعاد (العملات المستقرة والعملات ذات الرافعة)
         self.BLACKLIST_TOKENS = [
             'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX',
-            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP', 'HT', 'KCS'
+            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L'
         ]
         
-        print(f"📦 [DataManager V58.1] Quality Gate & Context Engine Active.")
+        print(f"📦 [DataManager V15.2] Restored 'Anti-FOMO' Logic Shield.")
 
     async def initialize(self):
+        """تهيئة مدير البيانات والاتصالات"""
+        print("   > [DataManager] Starting initialization...")
         self.http_client = httpx.AsyncClient(timeout=30.0)
         await self._load_markets()
+        # تحميل العقود إذا وجدت
+        await self.load_contracts_from_r2()
+        print(f"✅ [DataManager] Ready. Logic: STRICT/Anti-FOMO.")
 
     async def _load_markets(self):
         try:
-            if self.exchange and not self.exchange.markets:
+            if self.exchange:
                 await self.exchange.load_markets()
                 self.market_cache = self.exchange.markets
-        except Exception: pass
+        except Exception as e:
+            print(f"❌ [DataManager] Market load failed: {e}")
+            traceback.print_exc()
 
     async def close(self):
         if self.http_client: await self.http_client.aclose()
         if self.exchange: await self.exchange.close()
 
-    # ✅ FIXED: Restored Missing Method for Startup Sequence
     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 loaded: {len(self.contracts_db)} items.")
-        except Exception as e: 
-            print(f"   ⚠️ [DataManager] Failed to load contracts: {e}")
+        except Exception:
             self.contracts_db = {}
-            
-    def get_contracts_db(self): return self.contracts_db
 
+    def get_contracts_db(self) -> Dict[str, Any]:
+        return self.contracts_db
+    
     # ==================================================================
-    # 🛡️ Stage 0: The "Anti-Junk" Gate (Quality Control)
+    # 🛡️ Layer 1: The Strict Logic Tree Screening (Core Logic)
     # ==================================================================
-    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
+    async def layer1_rapid_screening(self, adaptive_hub_ref=None) -> List[Dict[str, Any]]:
         """
-        Strict Quality Control:
-        1. Liquidity: > $2M Quote Volume.
-        2. Integrity: Spread < 2.0%.
-        3. Sanity: 24h Change < 30% (Avoid chasing massive pumps).
-        4. Safety: No Blacklisted tokens.
+        يقوم بمسح السوق وتقسيم الفرص إلى نوعين:
+        1. Breakout: اختراق آمن (RSI < 70).
+        2. Reversal: ارتداد من القاع (RSI < 40).
+        أي شيء آخر يتم استبعاده فوراً.
         """
+        self.adaptive_hub_ref = adaptive_hub_ref
+        print(f"🔍 [Layer 1] Initiating STRICT Logic Tree Screening...")
+        
+        # 1. المرحلة 0: فلتر الكون (السيولة الأساسية)
+        initial_candidates = await self._stage0_universe_filter()
+        
+        if not initial_candidates:
+            return []
+
+        # 2. جلب البيانات الفنية لأفضل العملات سيولة (Top 100 لتقليل الضغط)
+        top_liquid_candidates = initial_candidates[:100] 
+        enriched_data = await self._fetch_technical_data_batch(top_liquid_candidates)
+        
+        # 3. تطبيق شجرة القرار الصارمة (Anti-FOMO)
+        breakout_list = []
+        reversal_list = []
+
+        for item in enriched_data:
+            classification = self._apply_logic_tree(item)
+            
+            # حقن التكوين الديناميكي (إذا وجد Hub)
+            if self.adaptive_hub_ref:
+                regime = "BULL" if classification['type'] == 'BREAKOUT' else "RANGE"
+                item['dynamic_limits'] = self.adaptive_hub_ref.get_regime_config(regime)
+
+            if classification['type'] == 'BREAKOUT':
+                item['l1_sort_score'] = classification['score'] 
+                item['strategy_tag'] = 'Safe_Breakout'
+                breakout_list.append(item)
+            elif classification['type'] == 'REVERSAL':
+                item['l1_sort_score'] = classification['score'] 
+                item['strategy_tag'] = 'Dip_Sniper'
+                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)
+        # الارتداد: نرتب بالأعلى سكور (كلما كان الـ RSI أقل كان السكور أعلى في منطقنا)
+        reversal_list.sort(key=lambda x: x['l1_sort_score'], reverse=True)
+
+        # نختار صفوة الصفوة
+        final_selection = breakout_list[:25] + reversal_list[:25]
+        
+        # تنظيف البيانات للإرجاع
+        # نحتفظ بالبيانات الفنية لأن المعالج سيحتاجها
+        print(f"✅ [Layer 1] Final Selection: {len(final_selection)} candidates (Anti-FOMO Active).")
+        return final_selection
+
+    # ------------------------------------------------------------------
+    # Stage 0: Universe Filter (Basic Liquidity)
+    # ------------------------------------------------------------------
+    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
         try:
-            print(f"   🛡️ [Stage 0] Filtering Junk (Vol>2M, Spread<2%, No Pumps)...")
             tickers = await self.exchange.fetch_tickers()
-            
-            valid_candidates = []
+            candidates = []
             
             for symbol, ticker in tickers.items():
                 if not symbol.endswith('/USDT'): continue
                 
-                # 1. Blacklist Check
                 base_curr = symbol.split('/')[0]
                 if any(bad in base_curr for bad in self.BLACKLIST_TOKENS): continue
                 
-                # 2. Volume Check (Strict > 2,000,000 USD)
+                # 👇 الحد الأدنى للسيولة (1 مليون دولار لضمان التنفيذ السريع)
                 quote_vol = ticker.get('quoteVolume')
-                if quote_vol is None or quote_vol == 0:
-                    base_vol = ticker.get('baseVolume')
-                    last_p = ticker.get('last')
-                    if base_vol and last_p:
-                        quote_vol = float(base_vol) * float(last_p)
-                    else: 
-                        quote_vol = 0.0
+                if not quote_vol or quote_vol < 1_000_000: continue
                 
-                if quote_vol < 2_000_000: continue 
-
-                # 3. Spread Check (Avoid Orderbook manipulation/illiquidity)
-                bid = ticker.get('bid')
-                ask = ticker.get('ask')
-                if bid and ask and ask > 0:
-                    spread_pct = ((ask - bid) / ask) * 100
-                    if spread_pct > 2.0: continue # Skip if spread > 2%
-                else:
-                    continue # Broken book
-
-                # 4. Sanity Check (Avoid extreme FOMO/Dumps)
-                change_24h = ticker.get('percentage')
-                if change_24h is not None:
-                    if abs(change_24h) > 30.0: continue # Skip huge pumps/dumps
-
-                valid_candidates.append({
+                last_price = ticker.get('last')
+                if not last_price or last_price < 0.0005: continue 
+                
+                # 👇 فلتر أولي: استبعاد العملات التي انفجرت بجنون (+15% فأكثر يتم تجاهلها مبدئياً)
+                change_24h = ticker.get('percentage', 0.0)
+                if change_24h > 15.0: continue 
+
+                candidates.append({
                     'symbol': symbol,
                     'quote_volume': quote_vol,
-                    'current_price': float(ticker.get('last', 0)),
-                    'change_24h': change_24h,
-                    'spread': spread_pct if 'spread_pct' in locals() else 0.0
+                    'current_price': last_price,
+                    'change_24h': change_24h
                 })
             
-            # Sort by Volume (Liquidity King)
-            valid_candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
+            # الترتيب حسب السيولة
+            candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
+            return candidates
             
-            # Cap at Top 80 to ensure we have rate-limit room for 4H analysis
-            final_list = valid_candidates[:80]
-            
-            print(f"   -> [Stage 0] Passed {len(final_list)} High-Quality Assets.")
-            return final_list
-
         except Exception as e:
-            print(f"❌ [Stage 0 Error] {e}")
-            traceback.print_exc()
+            print(f"❌ [L1 Error] Universe filter failed: {e}")
             return []
 
-    # ==================================================================
-    # 🧭 Stage 1: Context Diagnosis (4H Regime)
-    # ==================================================================
-    async def _determine_4h_regime(self, symbol: str) -> Dict[str, Any]:
-        """
-        Diagnose the Asset's specific regime using 4H data.
-        Returns: BULL, BEAR, DEAD, or RANGE + Tech Data.
-        """
-        try:
-            ohlcv = await self.exchange.fetch_ohlcv(symbol, '4h', limit=50)
-            if not ohlcv or len(ohlcv) < 50: return {'regime': 'RANGE', 'conf': 0.0}
+    # ------------------------------------------------------------------
+    # Data Fetching Helpers (Batch Processing)
+    # ------------------------------------------------------------------
+    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])
+            # تأخير بسيط جداً لتجنب حظر API
+            await asyncio.sleep(0.05) 
+        return results
 
-            df = pd.DataFrame(ohlcv, columns=['ts', 'o', 'h', 'l', 'c', 'v'])
-            c = df['c']
-            
-            # Indicators
-            ema50 = ta.ema(c, length=50).iloc[-1]
-            ema200 = ta.ema(c, length=200).iloc[-1]
-            rsi = ta.rsi(c, length=14).iloc[-1]
-            atr = ta.atr(df['h'], df['l'], c, length=14).iloc[-1]
-            price = c.iloc[-1]
-
-            # Logic
-            regime = "RANGE"
-            conf = 0.5
-
-            # 1. Check DEAD (Low Volatility)
-            atr_pct = (atr / price) * 100
-            # Calculate Range of last 20 candles
-            high_20 = df['h'].iloc[-20:].max()
-            low_20 = df['l'].iloc[-20:].min()
-            range_pct = ((high_20 - low_20) / low_20) * 100
-            
-            if atr_pct < 0.8 and range_pct < 4.0:
-                regime = "DEAD"
-                conf = 0.9
+    async def _fetch_single_tech_data(self, candidate: Dict[str, Any]) -> Any:
+        symbol = candidate['symbol']
+        try:
+            # نحتاج 1H للاتجاه العام و 15M للدخول الدقيق
+            ohlcv_1h = await self.exchange.fetch_ohlcv(symbol, '1h', limit=60)
+            ohlcv_15m = await self.exchange.fetch_ohlcv(symbol, '15m', limit=60)
             
-            # 2. Check BULL
-            elif price > ema50 and ema50 > ema200 and rsi > 50:
-                regime = "BULL"
-                conf = 0.8 if rsi > 55 else 0.6
-                
-            # 3. Check BEAR
-            elif price < ema50 and ema50 < ema200 and rsi < 50:
-                regime = "BEAR"
-                conf = 0.8 if rsi < 45 else 0.6
+            if not ohlcv_1h or len(ohlcv_1h) < 55 or not ohlcv_15m or len(ohlcv_15m) < 55:
+                return None
             
-            return {
-                'regime': regime,
-                'conf': conf,
-                'tech': {
-                    'ema50': ema50, 'ema200': ema200, 'rsi': rsi, 'atr_pct': atr_pct
-                }
+            # تخزين البيانات الخام لاستخدامها لاحقاً في المعالج
+            candidate['ohlcv'] = {
+                '1h': ohlcv_1h,
+                '15m': ohlcv_15m,
+                # يمكن إضافة 5m لاحقاً عند الحاجة للدقة القصوى
             }
+            
+            # نسخ للاستخدام الداخلي في الفلتر
+            candidate['ohlcv_1h_raw'] = ohlcv_1h
+            candidate['ohlcv_15m_raw'] = ohlcv_15m
+            return candidate
         except Exception:
-            return {'regime': 'RANGE', 'conf': 0.0}
+            return None
 
-    async def layer1_rapid_screening(self, adaptive_hub_ref=None) -> List[Dict[str, Any]]:
-        """
-        Orchestrates Stage 0 (Filter) -> Stage 1 (Diagnose & Inject Limits).
-        """
-        # 1. Get High Quality Candidates
-        candidates = await self._stage0_universe_filter()
-        if not candidates: return []
-
-        print(f"   🧬 [Stage 1] Diagnosing 4H Regime for {len(candidates)} assets...")
-        
-        # 2. Parallel Diagnosis
-        # Fetching 4H data for all candidates
-        tasks = [self._determine_4h_regime(c['symbol']) for c in candidates]
-        regime_results = await asyncio.gather(*tasks, return_exceptions=True)
-        
-        final_list = []
-        for i, res in enumerate(regime_results):
-            if isinstance(res, dict):
-                cand = candidates[i]
-                cand['asset_regime'] = res['regime']
-                cand['asset_regime_conf'] = res['conf']
-                cand['type'] = 'CANDIDATE'
-                
-                # 3. INJECT DYNAMIC LIMITS
-                # This is where we break the "Global SystemLimits" dependency.
-                if adaptive_hub_ref:
-                    # Get DNA for this specific asset's regime
-                    dynamic_config = adaptive_hub_ref.get_regime_config(res['regime'])
-                    cand['dynamic_limits'] = dynamic_config
+    # ------------------------------------------------------------------
+    # 🧠 The Logic Core: Math & Decision Tree (STRICT ANTI-FOMO MODE)
+    # ------------------------------------------------------------------
+    def _apply_logic_tree(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            df_1h = self._calc_indicators(data['ohlcv_1h_raw'])
+            df_15m = self._calc_indicators(data['ohlcv_15m_raw'])
+        except:
+            return {'type': 'NONE'}
+
+        curr_1h = df_1h.iloc[-1]
+        curr_15m = df_15m.iloc[-1]
+        
+        # --- Stage 2: Overbought Filter (STRICT MODE 🛡️) ---
+        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
+
+        # 1. ⛔ Anti-Pump: إذا صعدت أكثر من 8% في 4 ساعات، اتركها
+        if change_4h > 8.0: return {'type': 'NONE'}
+        
+        # 2. ⛔ Anti-FOMO: إذا صعدت أكثر من 12% في اليوم، اتركها
+        if data.get('change_24h', 0) > 12.0: return {'type': 'NONE'}
+        
+        # 3. ⛔ RSI Ceiling: ممنوع الدخول إذا RSI فوق 70 (منطقة تشبع)
+        if curr_1h['rsi'] > 70: return {'type': 'NONE'} 
+
+        # 4. ⛔ Mean Reversion Risk: السعر بعيد جداً عن المتوسط
+        deviation = (curr_1h['close'] - curr_1h['ema20']) / curr_1h['atr'] if curr_1h['atr'] > 0 else 0
+        if deviation > 1.8: return {'type': 'NONE'} 
+
+        # --- Stage 3: Classification ---
+        
+        # === A. Breakout Logic (Safe & Early) ===
+        is_breakout = False
+        breakout_score = 0.0
+        
+        # شروط الترند الصاعد الصحي
+        bullish_structure = (curr_1h['ema20'] > curr_1h['ema50']) or (curr_1h['close'] > curr_1h['ema20'])
+        
+        if bullish_structure:
+            # RSI يجب أن يكون لديه مساحة للصعود (بين 45 و 68)
+            if 45 <= curr_1h['rsi'] <= 68:
+                if curr_15m['close'] >= curr_15m['ema20']:
+                    # Volatility Squeeze: السعر يتحرك في نطاق ضيق
+                    avg_range = (df_15m['high'] - df_15m['low']).rolling(10).mean().iloc[-1]
+                    if (curr_15m['high'] - curr_15m['low']) <= avg_range * 1.8:
+                        vol_ma20 = df_15m['volume'].rolling(20).mean().iloc[-1]
+                        
+                        # Volume Confirmation: حجم تداول أعلى من المتوسط بـ 1.5 مرة
+                        if curr_15m['volume'] >= 1.5 * vol_ma20:
+                            is_breakout = True
+                            breakout_score = curr_15m['volume'] / vol_ma20 if vol_ma20 > 0 else 1.0
+
+        if is_breakout:
+            return {'type': 'BREAKOUT', 'score': breakout_score}
+
+        # === B. Reversal Logic (Dip Buy / Oversold) ===
+        is_reversal = False
+        reversal_score = 100.0 
+        
+        # شراء التشبع البيعي فقط (RSI بين 20 و 40)
+        if 20 <= curr_1h['rsi'] <= 40: 
+            # السعر هبط مؤخراً
+            if change_4h <= -2.0: 
+                # البحث عن شمعة انعكاسية (مطرقة Hammer أو ابتلاعية Engulfing) في آخر 3 شموع 15m
+                last_3 = df_15m.iloc[-3:]
+                found_rejection = False
+                for _, row in last_3.iterrows():
+                    rng = row['high'] - row['low']
+                    if rng > 0:
+                        is_green = row['close'] > row['open']
+                        # شكل المطرقة: الذيل السفلي طويل
+                        hammer_shape = (min(row['open'], row['close']) - row['low']) > (rng * 0.6)
+                        if is_green or hammer_shape:
+                            found_rejection = True
+                            break
                 
-                final_list.append(cand)
+                if found_rejection:
+                    is_reversal = True
+                    # كلما قل الـ RSI زادت احتمالية الارتداد (سكور أعلى)
+                    reversal_score = (100 - curr_1h['rsi']) 
 
-        print(f"   -> [Stage 1] Prepared {len(final_list)} Context-Aware Candidates.")
-        return final_list
+        if is_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
+        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
+        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
 
-    # Keep helper methods...
+    # ==================================================================
+    # 🎯 Public Helpers (Standard Interface)
+    # ==================================================================
     async def get_latest_price_async(self, symbol: str) -> float:
-        try: return float((await self.exchange.fetch_ticker(symbol))['last'])
-        except: return 0.0
+        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: return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
-        except: return []
+        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: return await self.exchange.fetch_order_book(symbol, limit)
-        except: return {}
+        try:
+            ob = await self.exchange.fetch_order_book(symbol, limit)
+            return ob
+        except Exception: return {}