ml_engine/data_manager.py

# ==============================================================================
# 📂 ml_engine/data_manager.py 
# (V77.0 - GEM-Architect: Smart Exchange Switcher)
# ==============================================================================
# - Feature: Auto-Failover (Binance -> Bybit -> OKX -> MEXC).
# - Logic: Automatically detects the best available exchange for the current IP.
# - Stability: Zero-config switch if one exchange is blocked (451 Error).
# ==============================================================================

import asyncio
import httpx
import traceback
import ccxt.async_support as ccxt
import logging
import pandas as pd
import numpy as np
import time
from typing import List, Dict, Any, Optional

# محاولة استيراد حدود النظام
try:
    from ml_engine.processor import SystemLimits
except ImportError:
    SystemLimits = None

# تقليل ضوضاء السجلات
logging.getLogger("httpx").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
        self.adaptive_hub_ref = None 
        
        self.exchange = None
        self.exchange_id = None
        self.http_client = None
        self.market_cache = {}
        
        # القائمة السوداء (العملات المستقرة والرافعة)
        self.BLACKLIST_TOKENS = [
            'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX', 'USDP',
            'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', '5S', '5L'
        ]
        
        # قائمة المنصات المرشحة بالترتيب (الأولوية للأعلى)
        self.EXCHANGE_CANDIDATES = [
            {'id': 'binance', 'name': 'Binance', 'options': {'adjustForTimeDifference': True}},
            {'id': 'bybit',   'name': 'Bybit',   'options': {}},
            {'id': 'okx',     'name': 'OKX',     'options': {}},
            {'id': 'mexc',    'name': 'MEXC',    'options': {}}
        ]
        
        print(f"📦 [DataManager V77.0] Initialized (Smart Switcher Ready).")

    async def initialize(self):
        """تهيئة الاتصال واختيار أفضل منصة متاحة"""
        print("   > [DataManager] Starting initialization & Exchange Discovery...")
        self.http_client = httpx.AsyncClient(timeout=30.0)
        
        # 🔄 تشغيل بروتوكول البحث عن منصة (Exchange Discovery Protocol)
        await self._connect_to_best_exchange()
        
        await self.load_contracts_from_r2()

    async def _connect_to_best_exchange(self):
        """تجربة المنصات بالترتيب واعتماد أول منصة تعمل"""
        for cand in self.EXCHANGE_CANDIDATES:
            ex_id = cand['id']
            ex_name = cand['name']
            print(f"   🔄 Testing connection to {ex_name}...")
            
            try:
                exchange_class = getattr(ccxt, ex_id)
                exchange_instance = exchange_class({
                    'enableRateLimit': True,
                    'timeout': 30000,
                    'options': {
                        'defaultType': 'spot',
                        **cand['options']
                    }
                })
                
                # اختبار الاتصال الفعلي
                await exchange_instance.load_markets()
                
                # اختبار إضافي: التحقق من وجود BTC/USDT لضمان صحة البيانات
                if 'BTC/USDT' in exchange_instance.markets:
                    self.exchange = exchange_instance
                    self.exchange_id = ex_id
                    self.market_cache = self.exchange.markets
                    print(f"   ✅ [SUCCESS] Connected to {ex_name} (Markets: {len(self.market_cache)}).")
                    return # تم العثور على منصة، الخروج من الحلقة
                else:
                    print(f"   ⚠️ [SKIP] {ex_name} connected but BTC/USDT missing (Unusual).")
                    await exchange_instance.close()
                    
            except Exception as e:
                error_msg = str(e)
                if "451" in error_msg:
                    print(f"   ⛔ [BLOCKED] {ex_name} is restricted in this region (Error 451).")
                else:
                    print(f"   ❌ [FAIL] Could not connect to {ex_name}: {error_msg.splitlines()[0]}")
                
                # تنظيف الموارد قبل الانتقال للتالي
                if 'exchange_instance' in locals():
                    await exchange_instance.close()
        
        # إذا فشلت جميع المنصات
        if not self.exchange:
            raise RuntimeError("❌ ALL Exchanges Failed! Please check your internet connection or VPN.")

    async def close(self):
        if self.http_client: await self.http_client.aclose()
        if self.exchange: await self.exchange.close()

    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()
        except Exception: self.contracts_db = {}

    def get_contracts_db(self) -> Dict[str, Any]:
        return self.contracts_db

    # ==================================================================
    # 🌍 Global Market Validator
    # ==================================================================
    async def check_global_market_health(self) -> Dict[str, Any]:
        if not self.exchange: return {'is_safe': False, 'reason': 'No Exchange'}
        try:
            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'])
            daily_change = (df['c'].iloc[-1] - df['c'].iloc[-2]) / df['c'].iloc[-2]
            
            if daily_change < -0.10: 
                return {'is_safe': False, 'reason': f'🚨 BTC CRASHING ({daily_change*100:.2f}%)'}

            return {'is_safe': True, 'reason': '✅ Market Open'}

        except Exception as e:
            return {'is_safe': True, 'reason': 'Error Bypass'}

    # ==================================================================
    # 🧠 Layer 1: The Open Gate (Volume Only)
    # ==================================================================
    async def layer1_rapid_screening(self, limit=200, adaptive_hub_ref=None) -> List[Dict[str, Any]]:
        self.adaptive_hub_ref = adaptive_hub_ref
        
        if not self.exchange:
            print("❌ [Layer 1] No active exchange connection.")
            return []

        print(f"🔍 [Layer 1] Screening {self.exchange_id.upper()} Market (Volume Only > $1M)...")
        
        # 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 []

        # 1. فلتر السيولة الصارم
        initial_candidates = await self._stage0_universe_filter()
        if not initial_candidates:
            print(f"⚠️ [Layer 1] No coins met the criteria on {self.exchange_id}.")
            return []

        # نأخذ أعلى العملات سيولة
        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)
        
        final_list = []
        for item in enriched_data:
            regime_info = self._diagnose_asset_regime(item)
            item['asset_regime'] = regime_info['regime']
            item['asset_regime_conf'] = regime_info['conf']
            item['strategy_type'] = 'NEURAL_SCAN' 
            item['strategy_tag'] = 'NEURAL'
            item['l1_sort_score'] = 0.5 

            if self.adaptive_hub_ref:
                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.get('quote_volume', 0), reverse=True)
        print(f"✅ [Layer 1] Passed {len(final_list)} candidates directly to Neural Layer.")
        return final_list

    # ==================================================================
    # 🧱 Order Book Depth Scanner
    # ==================================================================
    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 (Multi-Exchange Compatible)
    # ==================================================================
    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
        try:
            MIN_VOLUME_THRESHOLD = 1000000.0 # $1M
            
            print(f"   🛡️ [Stage 0] Fetching Tickers from {self.exchange_id.upper()}...")
            tickers = await self.exchange.fetch_tickers()
            candidates = []
            
            SOVEREIGN_COINS = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'BNB/USDT', 'XRP/USDT']
            
            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
                
                # توحيد حساب الفوليوم (لأن المنصات تختلف في التسميات)
                calc_quote_vol = float(ticker.get('quoteVolume') or 0.0)
                if calc_quote_vol == 0.0:
                    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:
                    if calc_quote_vol < MIN_VOLUME_THRESHOLD: continue
                
                change_pct = ticker.get('percentage')
                if change_pct is None: change_pct = 0.0
                
                candidates.append({
                    'symbol': symbol,
                    'quote_volume': calc_quote_vol,
                    'current_price': float(ticker.get('last') or 0.0),
                    'change_24h': change_pct
                })
            
            candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
            print(f"   ℹ️ [Stage 0] Found {len(candidates)} valid candidates.")
            return candidates
            
        except Exception as e:
            print(f"❌ [L1 Error] Universe filter failed: {e}")
            traceback.print_exc()
            return []

    # ------------------------------------------------------------------
    # Helpers & Data Fetching
    # ------------------------------------------------------------------
    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: item['df_1h'] = self._calc_indicators(item['ohlcv_1h_raw'])
                else: return {'regime': 'RANGE', 'conf': 0.0}
            df = item['df_1h']
            if df.empty: return {'regime': 'RANGE', 'conf': 0.0}
            curr = df.iloc[-1]
            price = curr['close']; ema20 = curr['ema20']; ema50 = curr['ema50']; rsi = curr['rsi']
            regime = "RANGE"; conf = 0.5
            if price > ema20 and ema20 > ema50:
                regime = "BULL"; conf = 0.8 if rsi > 55 else 0.6
            elif price < ema20 and ema20 < ema50:
                regime = "BEAR"; conf = 0.8 if rsi < 45 else 0.6
            return {'regime': regime, 'conf': conf}
        except Exception: return {'regime': 'RANGE', 'conf': 0.0}

    async def _fetch_technical_data_batch(self, candidates):
        chunk_size = 10; results = []
        for i in range(0, len(candidates), chunk_size):
            chunk = candidates[i:i+chunk_size]
            tasks = [self._fetch_single(c) for c in chunk]
            res = await asyncio.gather(*tasks)
            results.extend([r for r in res if r])
            await asyncio.sleep(0.05) 
        return results

    async def _fetch_single(self, c):
        try:
            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
            c['ohlcv'] = {'1h': h1, '15m': m15} 
            c['ohlcv_1h_raw'] = h1 
            c['ohlcv_15m_raw'] = m15
            c['df_1h'] = self._calc_indicators(h1)
            return c
        except: return None

    def _calc_indicators(self, ohlcv):
        df = pd.DataFrame(ohlcv, columns=['ts', 'o', 'h', 'l', 'c', 'v'])
        delta = df['c'].diff()
        gain = (delta.where(delta>0, 0)).rolling(14).mean()
        loss = (-delta.where(delta<0, 0)).rolling(14).mean()
        rs = gain/loss
        df['rsi'] = 100 - (100/(1+rs))
        df['ema20'] = df['c'].ewm(span=20).mean()
        df['ema50'] = df['c'].ewm(span=50).mean()
        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()
        df.rename(columns={'o':'open', 'h':'high', 'l':'low', 'c':'close', 'v':'volume'}, inplace=True)
        return df.fillna(0)
        
    async def get_latest_price_async(self, symbol):
        try: return float((await self.exchange.fetch_ticker(symbol))['last'])
        except: return 0.0
    
    async def get_latest_ohlcv(self, symbol, timeframe='5m', limit=100):
        try: return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        except: return []