Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V57.0 - GEM-Architect: Pure Volume Edition)
+# (V58.0 - GEM-Architect: Asset-Context Edition)
 # ============================================================
 
 import asyncio
@@ -12,33 +12,26 @@ 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:
-    class SystemLimits:
-        L1_MIN_AFFINITY_SCORE = 15.0
-        CURRENT_REGIME = "RANGE"
+    SystemLimits = None
 
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 class DataManager:
-    """
-    DataManager V57.0 (Pure Volume Edition)
-    - Logic: NO FILTERS. NO CONDITIONS.
-    - Action: Fetches Top 150 coins by Quote Volume and passes them ALL.
-    - Speed: Maximum (No secondary inspection requests).
-    """
-
     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 or via global context if strictly necessary, 
+        # but better to handle regime logic internally or pass config.
+        self.adaptive_hub_ref = None 
+
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
             'timeout': 30000, 
@@ -48,27 +41,17 @@ class DataManager:
         self.http_client = None
         self.market_cache = {}
         
-        self.NETWORK_HEADS = ['ETH/USDT', 'SOL/USDT', 'BNB/USDT', 'AVAX/USDT']
-        self.BENCHMARK_SYMBOL = 'BTC/USDT'
-        
-        # القائمة السوداء (للعملات الأساسية فقط)
+        # 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'
         ]
         
-        print(f"📦 [DataManager V57.0] Pure Volume Engine Active.")
+        print(f"📦 [DataManager V58.0] Quality Gate & Context Engine Active.")
 
     async def initialize(self):
-        print("   > [DataManager] Starting Matrix Initialization...")
-        try:
-            self.http_client = httpx.AsyncClient(timeout=30.0)
-            await self._load_markets()
-            await self.update_hyper_regime()
-            print(f"✅ [DataManager] Ready | Regime: {SystemLimits.CURRENT_REGIME}")
-        except Exception as e:
-            print(f"❌ [DataManager] Init Error: {e}")
-            traceback.print_exc()
+        self.http_client = httpx.AsyncClient(timeout=30.0)
+        await self._load_markets()
 
     async def _load_markets(self):
         try:
@@ -81,167 +64,178 @@ class DataManager:
         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: self.contracts_db = {}
-        
-    def get_contracts_db(self): return self.contracts_db
-
     # ==================================================================
-    # 🧠 The Matrix (Regime Sensor) - JUST FOR THRESHOLDS
-    # ==================================================================
-    async def update_hyper_regime(self):
-        print("   🧠 [Matrix] Scanning Market Dimensions...")
-        try:
-            fg_index, fg_label = await self._fetch_fear_greed()
-            btc_data = await self._analyze_single_asset(self.BENCHMARK_SYMBOL)
-            breadth_score, heads_details = await self._analyze_market_breadth()
-            
-            regime = "RANGE"
-            
-            if btc_data['trend'] == 'BULL':
-                if breadth_score >= 0.50: regime = "BULL"
-                else: regime = "RANGE"
-            elif btc_data['trend'] == 'BEAR':
-                if fg_index < 20: regime = "BEAR" 
-                else: regime = "BEAR"
-            elif btc_data['trend'] == 'NEUTRAL':
-                if btc_data['volatility_state'] == 'LOW' and breadth_score < 0.3: regime = "DEAD"
-                else: regime = "RANGE"
-
-            SystemLimits.CURRENT_REGIME = regime
-            print(f"   🌍 Regime Updated: {regime} (FG:{fg_index} | Breadth:{breadth_score*100:.0f}%)")
-
-        except Exception as e:
-            print(f"❌ [Matrix Error] {e}")
-            SystemLimits.CURRENT_REGIME = "RANGE"
-
-    # --- Sensor Helpers ---
-    async def _fetch_fear_greed(self) -> Tuple[int, str]:
-        try:
-            resp = await self.http_client.get("https://api.alternative.me/fng/?limit=1")
-            data = resp.json()
-            return int(data['data'][0]['value']), data['data'][0]['value_classification']
-        except: return 50, "Neutral"
-
-    async def _analyze_single_asset(self, symbol) -> Dict[str, Any]:
-        try:
-            ohlcv = await self.exchange.fetch_ohlcv(symbol, '1d', limit=100)
-            df = pd.DataFrame(ohlcv, columns=['ts', 'o', 'h', 'l', 'c', 'v'])
-            c = df['c']
-            ema50 = ta.ema(c, length=50).iloc[-1]
-            ema200 = ta.ema(c, length=200).iloc[-1]
-            atr = ta.atr(df['h'], df['l'], c, length=14).iloc[-1]
-            price = c.iloc[-1]
-            
-            trend = "NEUTRAL"
-            if price > ema50 and ema50 > ema200: trend = "BULL"
-            elif price < ema50 and price < ema200: trend = "BEAR"
-            vol_state = "NORMAL"
-            if (atr / price) < 0.025: vol_state = "LOW"
-            return {'trend': trend, 'volatility_state': vol_state}
-        except: return {'trend': 'NEUTRAL', 'volatility_state': 'NORMAL'}
-
-    async def _analyze_market_breadth(self) -> Tuple[float, str]:
-        tasks = [self._analyze_single_asset(sym) for sym in self.NETWORK_HEADS]
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-        bull_count = 0; valid = 0; details = []
-        for i, res in enumerate(results):
-            if isinstance(res, dict):
-                valid += 1
-                if res['trend'] == 'BULL': bull_count += 1
-                details.append(f"{self.NETWORK_HEADS[i].split('/')[0]}:{res['trend'][0]}")
-        return (bull_count / valid) if valid > 0 else 0.0, "|".join(details)
-
-    # ==================================================================
-    # 🌊 Stage 0: Pure Volume Flood (The Open Gate)
+    # 🛡️ Stage 0: The "Anti-Junk" Gate (Quality Control)
     # ==================================================================
     async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
         """
-        FETCH -> SORT BY VOLUME -> RETURN TOP 150.
-        NO structural checks. NO spread checks. NO wall checks.
+        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.
         """
         try:
-            print(f"   🌊 [Volume Gate] Opening floodgates...")
+            print(f"   🛡️ [Stage 0] Filtering Junk (Vol>2M, Spread<2%, No Pumps)...")
             tickers = await self.exchange.fetch_tickers()
             
-            pre_candidates = []
+            valid_candidates = []
+            
             for symbol, ticker in tickers.items():
                 if not symbol.endswith('/USDT'): continue
                 
-                # Check Blacklist (Base only)
+                # 1. Blacklist Check
                 base_curr = symbol.split('/')[0]
                 if any(bad in base_curr for bad in self.BLACKLIST_TOKENS): continue
                 
-                # Handle Volume
+                # 2. Volume Check (Strict > 2,000,000 USD)
                 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
-                else: quote_vol = float(quote_vol)
-
-                # Minimal floor to avoid bugs ($10k)
-                if quote_vol < 10_000: continue 
-
-                pre_candidates.append({
+                    else: 
+                        quote_vol = 0.0
+                
+                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({
                     'symbol': symbol,
                     'quote_volume': quote_vol,
                     'current_price': float(ticker.get('last', 0)),
-                    'sort_score': quote_vol # Score is just volume
+                    'change_24h': change_24h,
+                    'spread': spread_pct if 'spread_pct' in locals() else 0.0
                 })
             
-            # Sort Descending
-            pre_candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
+            # Sort by Volume (Liquidity King)
+            valid_candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
             
-            # Take Top 150
-            final_candidates = pre_candidates[:150]
+            # Cap at Top 80 to ensure we have rate-limit room for 4H analysis
+            final_list = valid_candidates[:80]
             
-            print(f"   -> [Volume Gate] Passed Top {len(final_candidates)} coins by Volume directly to Processor.")
-            return final_candidates
+            print(f"   -> [Stage 0] Passed {len(final_list)} High-Quality Assets.")
+            return final_list
 
         except Exception as e:
-            print(f"❌ [Filter Error] {e}")
+            print(f"❌ [Stage 0 Error] {e}")
             traceback.print_exc()
             return []
 
     # ==================================================================
-    # 🛡️ Layer 1: Pass-Through
+    # 🧭 Stage 1: Context Diagnosis (4H Regime)
     # ==================================================================
-    async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        # Update thresholds only
-        await self.update_hyper_regime()
-        
-        # Get raw candidates
+    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}
+
+            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
+            
+            # 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
+            
+            return {
+                'regime': regime,
+                'conf': conf,
+                'tech': {
+                    'ema50': ema50, 'ema200': ema200, 'rsi': rsi, 'atr_pct': atr_pct
+                }
+            }
+        except Exception:
+            return {'regime': 'RANGE', 'conf': 0.0}
+
+    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 []
 
-        # Format for Layer 2
+        print(f"   🧬 [Stage 1] Diagnosing 4H Regime for {len(candidates)} assets...")
+        
+        # 2. Parallel Diagnosis
+        # We need to fetch 4H data for all of them. 
+        # To avoid rate limits, we might batch them or use asyncio.gather with care.
+        # Assuming KuCoin handles ~80 requests quickly enough or ccxt throttles automatically.
+        
+        tasks = [self._determine_4h_regime(c['symbol']) for c in candidates]
+        regime_results = await asyncio.gather(*tasks, return_exceptions=True)
+        
         final_list = []
-        for c in candidates:
-            final_list.append({
-                'symbol': c['symbol'],
-                'quote_volume': c.get('quote_volume', 0),
-                'current_price': c.get('current_price', 0),
-                'type': 'CANDIDATE',
-                'l1_score': c.get('sort_score', 0)
-            })
-            
+        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
+                
+                final_list.append(cand)
+
+        print(f"   -> [Stage 1] Prepared {len(final_list)} Context-Aware Candidates.")
         return final_list
 
-    # ==================================================================
-    # 🏗️ Helpers (Minimal)
-    # ==================================================================
+    # Keep helper methods...
     async def get_latest_price_async(self, symbol: str) -> float:
         try: return float((await self.exchange.fetch_ticker(symbol))['last'])
         except: 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 []
-    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 {}
+    def get_contracts_db(self): return self.contracts_db