Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,11 +1,11 @@
-# ============================================================
+# ==============================================================================
 # 📂 ml_engine/data_manager.py 
-# (V70.4 - GEM-Architect: Anti-Stagnation & Deep Scanning)
-# ============================================================
-# - Removed "Volatility Trap" to stop repeating bad coins.
-# - Expanded scanning range to 900 coins.
-# - Stricter pattern filtering at Layer 1.
-# ============================================================
+# (V75.0 - GEM-Architect: Open Floodgates Mode)
+# ==============================================================================
+# - Removed Technical Hard Filters (Safe Bottom, Momentum, etc.).
+# - Primary Filter: Volume > $1M ONLY.
+# - Decision Authority: Delegated 100% to Layer 2 (Neural Models).
+# ==============================================================================
 
 import asyncio
 import httpx
@@ -44,13 +44,13 @@ class DataManager:
         self.http_client = None
         self.market_cache = {}
         
-        # القائمة السوداء للعملات غير المرغوبة (Leveraged/Stable/Fiat)
+        # القائمة السوداء (العملات المستقرة والرافعة)
         self.BLACKLIST_TOKENS = [
             'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX',
             'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', '5S', '5L'
         ]
         
-        print(f"📦 [DataManager V70.4] Initialized (Smart Filter Active).")
+        print(f"📦 [DataManager V75.0] Initialized (Filter: Vol > $1M Only).")
 
     async def initialize(self):
         """تهيئة الاتصال والأسواق"""
@@ -82,238 +82,130 @@ class DataManager:
         return self.contracts_db
 
     # ==================================================================
-    # 🌍 Global Market Validator V2 (Smart Breadth Scanner)
+    # 🌍 Global Market Validator (Bypassable)
     # ==================================================================
     async def check_global_market_health(self) -> Dict[str, Any]:
         """
-        يفحص صحة السوق العامة باستخدام منطق مزدوج.
+        فحص سريع لصحة السوق. لن يوقف التداول ولكنه يعطي تحذيراً.
         """
         try:
-            # 1. جلب بيانات البيتكوين الأساسية
-            btc_ohlcv = await self.exchange.fetch_ohlcv('BTC/USDT', '1d', limit=30)
-            if not btc_ohlcv: return {'is_safe': True, 'reason': 'No BTC Data - Bypassed'}
+            btc_ohlcv = await self.exchange.fetch_ohlcv('BTC/USDT', '1d', limit=7)
+            if not btc_ohlcv: return {'is_safe': True, 'reason': 'No BTC Data'}
 
             df = pd.DataFrame(btc_ohlcv, columns=['ts', 'o', 'h', 'l', 'c', 'v'])
-            current_close = df['c'].iloc[-1]
-            prev_close = df['c'].iloc[-2]
+            daily_change = (df['c'].iloc[-1] - df['c'].iloc[-2]) / df['c'].iloc[-2]
             
-            # --- [ CRITICAL CHECK ] ---
-            daily_change = (current_close - prev_close) / prev_close
-            if daily_change < -0.05: # -5% Crash check
+            # فقط الانهيار الكبير جداً يوقف النظام
+            if daily_change < -0.10: 
                 return {'is_safe': False, 'reason': f'🚨 BTC CRASHING ({daily_change*100:.2f}%)'}
 
-            # فحص المتوسطات (Trend Check)
-            sma20 = df['c'].rolling(20).mean().iloc[-1]
-            if current_close < sma20 * 0.90: # Bear Market Deep
-                 return {'is_safe': False, 'reason': '📉 Deep Bear Market (Risk Off)'}
-
-            return {'is_safe': True, 'reason': '✅ Market Healthy'}
+            return {'is_safe': True, 'reason': '✅ Market Open'}
 
         except Exception as e:
-            print(f"⚠️ [Market Validator] Error: {e}")
             return {'is_safe': True, 'reason': 'Error Bypass'}
 
     # ==================================================================
-    # 🧠 Layer 1: Classification (Strict Funnel)
+    # 🧠 Layer 1: The Open Gate (Volume Only)
     # ==================================================================
-    async def layer1_rapid_screening(self, limit=300, adaptive_hub_ref=None) -> List[Dict[str, Any]]:
+    async def layer1_rapid_screening(self, limit=200, adaptive_hub_ref=None) -> List[Dict[str, Any]]:
+        """
+        يقوم بفلترة العملات بناءً على الفوليوم فقط (1 مليون دولار).
+        يمرر أفضل العملات (حسب الفوليوم) إلى الطبقة الثانية للتحليل العصبي.
+        """
         self.adaptive_hub_ref = adaptive_hub_ref
-        print(f"🔍 [Layer 1] Screening Market (Deep Scan)...")
+        print(f"🔍 [Layer 1] Screening Market (Volume Only > $1M)...")
         
-        # 0. فحص صحة السوق
+        # 0. فحص السوق
         market_health = await self.check_global_market_health()
-        
         if not market_health['is_safe']:
             print(f"⛔ [Market Validator] Trading Halted: {market_health['reason']}")
             return []
-        else:
-            print(f"   🌍 [Market Validator] Status: {market_health['reason']}")
 
-        # 1. فلتر السيولة الأساسي
+        # 1. فلتر السيولة الصارم (Stage 0)
         initial_candidates = await self._stage0_universe_filter()
         if not initial_candidates:
-            print("⚠️ [Layer 1] Stage 0 returned 0 candidates.")
+            print("⚠️ [Layer 1] No coins met the $1M volume criteria.")
             return []
 
-        # 2. جلب البيانات الفنية (Batch Fetching)
-        # ✅ زيادة نطاق البحث إلى 900 عملة لزيادة التنوع
-        top_candidates = initial_candidates[:900] 
+        # نأخذ أعلى العملات سيولة لنقوم بجلب بياناتها
+        # نرفع العدد هنا لأننا لن نفلتر بالفنيات، بل سنرسل للنماذج
+        # لكن يجب الحذر من الـ Rate Limits، لذا سنأخذ أفضل 150 عملة
+        top_candidates = initial_candidates[:limit] 
+        print(f"   📥 Fetching data for top {len(top_candidates)} liquid assets...")
+        
         enriched_data = await self._fetch_technical_data_batch(top_candidates)
         
-        semi_final_list = []
+        final_list = []
 
-        # 3. التصنيف الفني
+        # 2. التجهيز للطبقة الثانية (بدون فلترة فنية)
         for item in enriched_data:
-            classification = self._classify_opportunity_type(item)
+            # تشخيص الحالة فقط للعلم (Info) وليس للرفض
+            regime_info = self._diagnose_asset_regime(item)
             
-            if classification['type'] != 'NONE':
-                # تشخيص الحالة
-                regime_info = self._diagnose_asset_regime(item)
-                item['asset_regime'] = regime_info['regime']
-                item['asset_regime_conf'] = regime_info['conf']
-                
-                item['strategy_type'] = classification['type'] 
-                item['l1_sort_score'] = classification['score']
-                item['strategy_tag'] = classification['type']
-                
-                # تخطي العملات "الميتة" إلا إذا كانت في حالة ضغط
-                if regime_info['regime'] == 'DEAD' and classification['type'] == 'MOMENTUM_LAUNCH':
-                    if not classification.get('is_squeeze', False):
-                        continue
-
-                semi_final_list.append(item)
-
-        # 4. فحص العمق وحقن الإعدادات
-        final_list = []
-        # ترتيب حسب السكور الفني وليس الفوليوم
-        semi_final_list.sort(key=lambda x: x['l1_sort_score'], reverse=True)
-        candidates_for_depth = semi_final_list[:limit] 
-        
-        if candidates_for_depth:
-            print(f"   🛡️ [Layer 1.5] Checking Depth for {len(candidates_for_depth)} candidates...")
-        
-        for item in candidates_for_depth:
-            # أ. فحص العمق (Depth Check)
-            if item['strategy_type'] in ['ACCUMULATION_SQUEEZE', 'SAFE_BOTTOM']:
-                try:
-                    atr_val = item.get('atr_value', 0.0)
-                    curr_price = item.get('current_price', 0.0)
-                    
-                    if atr_val > 0 and curr_price > 0:
-                        range_2h = atr_val * 2.0
-                        ob_score = await self._check_ob_pressure(item['symbol'], curr_price, range_2h)
-                        
-                        if ob_score > 0.6: 
-                            item['l1_sort_score'] += 0.15 
-                            item['note'] = f"Strong Depth Support ({ob_score:.2f})"
-                        elif ob_score < 0.4: 
-                            item['l1_sort_score'] -= 0.10 
-                except Exception: pass
+            item['asset_regime'] = regime_info['regime']
+            item['asset_regime_conf'] = regime_info['conf']
             
-            # ب. حقن الإعدادات
+            # ✅ تعيين نوع موحد ليتم التعامل معه في الطبقات التالية
+            item['strategy_type'] = 'NEURAL_SCAN' 
+            item['strategy_tag'] = 'NEURAL'
+            
+            # السكور الأولي يعتمد على السيولة النسبية (لترتيب الأولويات فقط)
+            # يمكن تعديله لاحقاً بواسطة Neural Models
+            item['l1_sort_score'] = 0.5 # درجة حيادية للسماح للكل بالمرور
+
+            # حقن الإعدادات الديناميكية (إن وجدت)
             if self.adaptive_hub_ref:
-                coin_type = item.get('strategy_type', 'SAFE_BOTTOM')
-                dynamic_config = self.adaptive_hub_ref.get_coin_type_config(coin_type)
+                # نستخدم إعدادات SAFE_BOTTOM كإعداد��ت افتراضية محافظة
+                dynamic_config = self.adaptive_hub_ref.get_coin_type_config('SAFE_BOTTOM')
                 item['dynamic_limits'] = dynamic_config
 
             final_list.append(item)
 
-        # الترتيب النهائي
-        final_list.sort(key=lambda x: x['l1_sort_score'], reverse=True)
-        selection = final_list[:limit]
+        # الترتيب حسب الفوليوم (الأكثر سيولة أولاً)
+        # item['quote_volume'] موجودة من Stage 0
+        final_list.sort(key=lambda x: x.get('quote_volume', 0), reverse=True)
         
-        print(f"✅ [Layer 1] Passed {len(selection)} active candidates.")
-        return selection
+        print(f"✅ [Layer 1] Passed {len(final_list)} candidates directly to Neural Layer.")
+        return final_list
 
     # ==================================================================
-    # 🧱 Order Book Depth Scanner
+    # 🧱 Order Book Depth Scanner (Optional Helper)
     # ==================================================================
-    async def _check_ob_pressure(self, symbol: str, current_price: float, price_range: float) -> float:
-        """فحص ضغط الشراء مقابل البيع في عمق السوق"""
-        try:
-            ob = await self.exchange.fetch_order_book(symbol, limit=50)
-            bids = ob['bids']
-            asks = ob['asks']
-            
-            min_price = current_price - price_range
-            max_price = current_price + price_range
-            
-            support_vol = 0.0
-            resistance_vol = 0.0
-            
-            for p, v in bids:
-                if p >= min_price: support_vol += v
-                else: break
-                
-            for p, v in asks:
-                if p <= max_price: resistance_vol += v
-                else: break
-                
-            if (support_vol + resistance_vol) == 0: return 0.5
-            return support_vol / (support_vol + resistance_vol)
-        except Exception:
-            return 0.5
-
-    # ==================================================================
-    # ⚖️ The Dual-Classifier Logic (STRICT MODE)
-    # ==================================================================
-    def _classify_opportunity_type(self, data: Dict[str, Any]) -> Dict[str, Any]:
-        """تصنيف العملة - نسخة منقحة بدون تمرير عشوائي"""
-        try:
-            df_1h = self._calc_indicators(data['ohlcv_1h_raw'])
-            curr = df_1h.iloc[-1]
-            data['atr_value'] = curr['atr']
-        except: return {'type': 'NONE', 'score': 0}
-
-        rsi = curr['rsi']
-        close = curr['close']
-        ema20 = curr['ema20']
-        ema50 = curr['ema50']
-        atr = curr['atr']
-        
-        lower_bb = curr['lower_bb'] if 'lower_bb' in curr else (curr['ema20'] - (2*curr['atr']))
-        upper_bb = curr['upper_bb'] if 'upper_bb' in curr else (curr['ema20'] + (2*curr['atr']))
-        bb_width = (upper_bb - lower_bb) / curr['ema20'] if curr['ema20'] > 0 else 1.0
-
-        # 🔥 1. Dead Coin Filter
-        volatility_pct = (atr / close) * 100 if close > 0 else 0
-        if volatility_pct < 2 : return {'type': 'NONE', 'score': 0}
-
-        # 🛡️ TYPE 1: SAFE_BOTTOM (القيعان الآمنة)
-        # تشبع بيعي حقيقي فقط
-        if rsi < 50: # Raised slightly from 55
-            # يجب أن يكون السعر قريباً جداً من القاع أو تحته
-            if close <= lower_bb * 1.05: 
-                score = (60 - rsi) / 20.0 
-                return {'type': 'SAFE_BOTTOM', 'score': min(score, 1.0)}
-
-        # 🔋 TYPE 2: ACCUMULATION_SQUEEZE (التجميع والضغط)
-        elif 40 <= rsi <= 65:
-            if bb_width < 0.15: # Stricter squeeze
-                score = 1.0 - (bb_width * 3.0) 
-                return {'type': 'ACCUMULATION_SQUEEZE', 'score': max(score, 0.5), 'is_squeeze': True}
-
-        # 🚀 TYPE 3: MOMENTUM_LAUNCH (انطلاق الزخم)
-        elif 50 < rsi < 80:
-            if close > ema50:
-                dist_to_upper = (upper_bb - close) / close
-                if dist_to_upper < 0.10: # قريب جداً من الاختراق
-                    score = rsi / 100.0
-                    return {'type': 'MOMENTUM_LAUNCH', 'score': score}
-        
-        # ❌ REMOVED: "Special Case: High Volatility Catch"
-        # This was causing the "Same Coin" loop. 
-        # If it doesn't fit the patterns above, it's rejected.
-
-        return {'type': 'NONE', 'score': 0}
+    async def get_order_book_snapshot(self, symbol: str, limit=20):
+        try: return await self.exchange.fetch_order_book(symbol, limit)
+        except: return {}
 
     # ==================================================================
-    # 🔍 Stage 0: Universe Filter
+    # 🔍 Stage 0: Universe Filter (1 Million Dollar Rule)
     # ==================================================================
     async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
-        """جلب كل العملات وتصفيتها حسب الحجم"""
+        """جلب كل العملات وتصفيتها حسب شرط المليون دولار"""
         try:
-            MIN_VOLUME_THRESHOLD = 2000000.0  # Reduced to $750k to find hidden gems
+            # ✅ الشرط الوحيد: 1 مليون دولار
+            MIN_VOLUME_THRESHOLD = 1000000.0  
             
             print(f"   🛡️ [Stage 0] Fetching Tickers (Min Vol: ${MIN_VOLUME_THRESHOLD:,.0f})...")
             tickers = await self.exchange.fetch_tickers()
             candidates = []
             
             SOVEREIGN_COINS = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'BNB/USDT', 'XRP/USDT']
-            reject_stats = {"volume": 0, "change": 0, "blacklist": 0}
+            reject_stats = {"volume": 0, "blacklist": 0}
 
             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): 
                     reject_stats["blacklist"] += 1
                     continue
                 
+                # حساب الفوليوم بالدولار
                 base_vol = float(ticker.get('baseVolume') or 0.0)
                 last_price = float(ticker.get('last') or 0.0)
                 calc_quote_vol = base_vol * last_price
                 
+                # استثناء العملات السيادية من شرط الفوليوم (دائماً مقبولة)
                 is_sovereign = symbol in SOVEREIGN_COINS
                 
                 if not is_sovereign:
@@ -324,10 +216,6 @@ class DataManager:
                 change_pct = ticker.get('percentage')
                 if change_pct is None: change_pct = 0.0
                 
-                if abs(change_pct) > 40.0: # Relaxed slightly 
-                    reject_stats["change"] += 1
-                    continue
-                
                 candidates.append({
                     'symbol': symbol,
                     'quote_volume': calc_quote_vol,
@@ -335,8 +223,9 @@ class DataManager:
                     'change_24h': change_pct
                 })
             
+            # ترتيب القائمة حسب الفوليوم لضمان معالجة الأهم أولاً
             candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
-            print(f"   ℹ️ [Stage 0] Ignored {reject_stats['volume']} low-vol coins.")
+            print(f"   ℹ️ [Stage 0] Ignored {reject_stats['volume']} low-vol coins. Found {len(candidates)} candidates.")
             return candidates
             
         except Exception as e:
@@ -345,10 +234,10 @@ class DataManager:
             return []
 
     # ------------------------------------------------------------------
-    # 🧭 The Diagnoser
+    # 🧭 The Diagnoser (For Context Only)
     # ------------------------------------------------------------------
     def _diagnose_asset_regime(self, item: Dict[str, Any]) -> Dict[str, Any]:
-        """تشخيص حالة العملة الفردية"""
+        """تشخيص حالة العملة الفردية (للعلم فقط)"""
         try:
             if 'df_1h' not in item: 
                 if 'ohlcv_1h_raw' in item:
@@ -364,18 +253,14 @@ class DataManager:
             ema20 = curr['ema20']
             ema50 = curr['ema50']
             rsi = curr['rsi']
-            atr = curr['atr']
-            atr_pct = (atr / price) * 100 if price > 0 else 0
             
             regime = "RANGE"
             conf = 0.5
             
-            if atr_pct < 0.4: return {'regime': 'DEAD', 'conf': 0.9}
-            
-            if price > ema20 and ema20 > ema50 and rsi > 50:
+            if price > ema20 and ema20 > ema50:
                 regime = "BULL"
                 conf = 0.8 if rsi > 55 else 0.6
-            elif price < ema20 and ema20 < ema50 and rsi < 50:
+            elif price < ema20 and ema20 < ema50:
                 regime = "BEAR"
                 conf = 0.8 if rsi < 45 else 0.6
                 
@@ -383,7 +268,7 @@ class DataManager:
         except Exception: return {'regime': 'RANGE', 'conf': 0.0}
 
     # ------------------------------------------------------------------
-    # Helpers
+    # Helpers & Data Fetching
     # ------------------------------------------------------------------
     async def _fetch_technical_data_batch(self, candidates):
         """جلب البيانات الفنية (1h, 15m) على دفعات"""
@@ -398,6 +283,7 @@ class DataManager:
 
     async def _fetch_single(self, c):
         try:
+            # نحتاج هذه البيانات للمراحل القادمة (Processor)
             h1 = await self.exchange.fetch_ohlcv(c['symbol'], '1h', limit=100) 
             m15 = await self.exchange.fetch_ohlcv(c['symbol'], '15m', limit=50)
             if not h1 or not m15: return None
@@ -427,11 +313,6 @@ class DataManager:
         tr = np.maximum(df['h']-df['l'], np.maximum(abs(df['h']-df['c'].shift()), abs(df['l']-df['c'].shift())))
         df['atr'] = tr.rolling(14).mean()
         
-        # Bollinger Bands
-        std = df['c'].rolling(20).std()
-        df['upper_bb'] = df['ema20'] + (2 * std)
-        df['lower_bb'] = df['ema20'] - (2 * std)
-        
         df.rename(columns={'o':'open', 'h':'high', 'l':'low', 'c':'close', 'v':'volume'}, inplace=True)
         return df.fillna(0)
         
@@ -441,8 +322,4 @@ class DataManager:
     
     async def get_latest_ohlcv(self, symbol, timeframe='5m', limit=100):
         try: return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
-        except: return []
-
-    async def get_order_book_snapshot(self, symbol, limit=20):
-        try: return await self.exchange.fetch_order_book(symbol, limit)
-        except: return {}
+        except: return []