Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V45.0 - GEM-Architect: Anti-FOMO Revival)
+# (V47.5 - GEM-Architect: Flash-Pulse & Hyper-Matrix)
 # ============================================================
 
 import asyncio
@@ -9,12 +9,12 @@ import traceback
 import logging
 import pandas as pd
 import numpy as np
-import pandas_ta as ta  # سنحتاج بعض الدوال المساعدة
-from typing import List, Dict, Any
+import pandas_ta as ta
+from typing import List, Dict, Any, Tuple
 
 import ccxt.async_support as ccxt
 
-# ✅ استيراد الدستور الديناميكي (للحفاظ على توافق النظام)
+# ✅ استيراد الدستور الديناميكي
 try:
     from ml_engine.processor import SystemLimits
 except ImportError:
@@ -27,10 +27,10 @@ logging.getLogger("ccxt").setLevel(logging.WARNING)
 
 class DataManager:
     """
-    DataManager V45.0 (Anti-FOMO Revival)
-    - Restores the STRICT Logic Tree from V15.2.
-    - Filters: 8% Max Pump, 12% Max Daily, RSI < 70 strict limit.
-    - Targets: Clean Breakouts & Oversold Reversals ONLY.
+    DataManager V47.5 (Flash-Pulse Edition)
+    - Hyper-Matrix Sensor: Detects BULL, BEAR, RANGE, DEAD.
+    - Flash-Pulse Filter: Uses 24h Ticker Data (Change, Range Pos, Spread) 
+      to pre-sort candidates BEFORE fetching candles.
     """
 
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
@@ -47,20 +47,24 @@ 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'
+        
         # القائمة السوداء
         self.BLACKLIST_TOKENS = [
             'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX',
             'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP', 'HT', 'KCS'
         ]
         
-        print(f"📦 [DataManager V45.0] Anti-FOMO Shield Active.")
+        print(f"📦 [DataManager V47.5] Flash-Pulse Engine Active.")
 
     async def initialize(self):
-        print("   > [DataManager] Starting initialization...")
+        print("   > [DataManager] Starting Matrix Initialization...")
         try:
             self.http_client = httpx.AsyncClient(timeout=30.0)
             await self._load_markets()
-            print(f"✅ [DataManager] Ready (Logic: STRICT Anti-FOMO).")
+            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()
@@ -81,321 +85,326 @@ class DataManager:
         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
 
     # ==================================================================
-    # 🛡️ Layer 1: The Strict Logic Tree (From V15.2)
+    # 🧠 The Hyper-Sensor (Matrix Logic)
     # ==================================================================
-    async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
-        print(f"🔍 [L1 Anti-FOMO] Filtering Universe...")
-        
-        # 1. المرحلة 0: فلتر الكون (السيولة العالية فقط)
-        # V15.2 كان يطلب مليون دولار سيولة، سنبقيه كما هو للصرامة
-        initial_candidates = await self._stage0_universe_filter()
-        
-        if not initial_candidates:
-            print("⚠️ [Layer 1] Universe empty.")
-            return []
-
-        # 2. جلب البيانات الفنية لأفضل 300 عملة (كما في V15.2)
-        top_liquid_candidates = initial_candidates[:300] 
-        print(f"   -> Analyzing top {len(top_liquid_candidates)} liquid assets...")
-        
-        enriched_data = await self._fetch_technical_data_batch(top_liquid_candidates)
-        
-        # 3. تطبيق شجرة القرار الصارمة
-        breakout_list = []
-        reversal_list = []
-
-        for item in enriched_data:
-            # هنا نستخدم منطق V15.2 الأصلي
-            classification = self._apply_logic_tree(item)
+    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"
+            
+            # Synthesis Logic
+            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" # Capitulation Hunting
+                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
             
-            if classification['type'] == 'BREAKOUT':
-                item['l1_score'] = classification['score'] 
-                item['type'] = 'BREAKOUT'
-                breakout_list.append(item)
-            elif classification['type'] == 'REVERSAL':
-                item['l1_score'] = classification['score'] 
-                item['type'] = 'REVERSAL'
-                reversal_list.append(item)
+            print(f"   🌍 Regime: {regime} | FG: {fg_index} | Breadth: {breadth_score*100:.0f}% | BTC: {btc_data['trend']}")
 
-        print(f"   -> [L1 Logic] Found: {len(breakout_list)} Breakouts, {len(reversal_list)} Reversals.")
+        except Exception as e:
+            print(f"❌ [Matrix Error] {e}")
+            SystemLimits.CURRENT_REGIME = "RANGE"
 
-        # 4. الترتيب والدمج النهائي
-        # الـ Breakout نرتبهم بالأعلى سكور (فوليوم)
-        breakout_list.sort(key=lambda x: x['l1_score'], reverse=True)
-        # الـ Reversal نرتبهم بالأعلى سكور (سكور الارتداد في V15.2 كان 100-RSI، يعني الأعلى أفضل)
-        reversal_list.sort(key=lambda x: x['l1_score'], reverse=True)
+    # --- 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"
 
-        # نختار الأفضل فقط (مزيج متوازن)
-        final_selection = breakout_list[:25] + reversal_list[:15]
-        
-        return [
-            {
-                'symbol': c['symbol'],
-                'quote_volume': c.get('quote_volume', 0),
-                'current_price': c.get('current_price', 0),
-                'type': c.get('type', 'UNKNOWN'),
-                'l1_score': c.get('l1_score', 0)
-            }
-            for c in final_selection
-        ]
+    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)
 
     # ==================================================================
-    # 🔗 Bridge for Backtest Engine Compatibility (IMPORTANT)
+    # ⚡ Stage 0: The Flash-Pulse Filter (New & Improved)
     # ==================================================================
-    def _calculate_structural_score(self, df: pd.DataFrame, symbol: str, regime: str) -> (float, List[str]):
+    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
         """
-        [Compatibility Wrapper]
-        هذه الدالة موجودة لكي لا يتعطل محرك الباكتست (backtest_engine.py).
-        تقوم بتحويل بيانات الباكتست إلى تنسيق يفهمه منطق V15.2.
+        Uses Ticker Snapshot (Percentage, High, Low, Volume, Spread) 
+        to filter candidates BEFORE fetching candles.
         """
-        # محاكاة تنسيق البيانات الذي يطلبه _apply_logic_tree
-        # نحتاج تقسيم الـ DF إلى 1H و 15M تقريبياً
         try:
-            # Resample لإنشاء بيانات 1H و 15M من البيانات المدخلة (التي غالباً تكون 15M)
-            df_15m = df.copy()
-            
-            agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
-            df_1h = df.resample('1H').agg(agg_dict).dropna()
+            regime = SystemLimits.CURRENT_REGIME
+            print(f"   ⚡ [Flash-Pulse] Sorting Universe for {regime}...")
             
-            # تحويلها لقوائم كما يتوقع الكود القديم
-            ohlcv_1h = df_1h.reset_index()[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
-            ohlcv_15m = df_15m.reset_index()[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
+            tickers = await self.exchange.fetch_tickers()
+            candidates = []
             
-            dummy_data = {
-                'ohlcv_1h': ohlcv_1h,
-                'ohlcv_15m': ohlcv_15m,
-                'change_24h': 0.0 # غير متوفر بدقة في الباكتست الجزئي، نتجاوزه
-            }
+            for symbol, ticker in tickers.items():
+                if not symbol.endswith('/USDT'): continue
+                if any(bad in symbol for bad in self.BLACKLIST_TOKENS): continue
+                
+                # Basic Data
+                last = ticker.get('last')
+                if not last or last < 0.0005: continue # Ignore penny dust
+                
+                quote_vol = ticker.get('quoteVolume', 0)
+                if quote_vol < 1_000_000: continue # Liquidity Floor
+                
+                # 1. Spread Check (Quality Control)
+                bid = ticker.get('bid'); ask = ticker.get('ask')
+                if bid and ask and bid > 0:
+                    spread_pct = (ask - bid) / bid
+                    if spread_pct > 0.01: continue # Skip if spread > 1% (Illiquid/Risky)
+                
+                # 2. Advanced Metrics
+                change_24h = float(ticker.get('percentage', 0.0))
+                high_24h = ticker.get('high')
+                low_24h = ticker.get('low')
+                
+                # Position in Daily Range (0.0 = Low, 1.0 = High)
+                range_pos = 0.5
+                if high_24h and low_24h and high_24h > low_24h:
+                    range_pos = (last - low_24h) / (high_24h - low_24h)
+                
+                # 3. Regime-Specific Scoring (The Magic)
+                sort_score = 0.0
+                
+                if regime == "BULL":
+                    # We want leaders: High Volume + Positive Change + Near Highs
+                    # But not too extended (>30% is dangerous FOMO)
+                    if 2.0 <= change_24h <= 25.0:
+                        sort_score = quote_vol * range_pos # Liquidity + Strength
+                    
+                elif regime == "BEAR":
+                    # We want Capitulation: High Volume + Negative Change + Near Lows
+                    if change_24h < -5.0 and range_pos < 0.2:
+                        # High volume on a dump = Panic Selling = Reversal Opportunity
+                        sort_score = quote_vol * (1 - range_pos) * abs(change_24h)
+                    
+                elif regime == "DEAD":
+                    # We want pure Liquidity + Stability (Squeeze Candidates)
+                    if -3.0 < change_24h < 3.0:
+                        sort_score = quote_vol # Just show me where the money is
+                        
+                else: # RANGE (Anti-FOMO)
+                    # We want active coins but not pumped ones
+                    if -8.0 < change_24h < 8.0:
+                        sort_score = quote_vol
+                
+                if sort_score > 0:
+                    candidates.append({
+                        'symbol': symbol,
+                        'quote_volume': quote_vol,
+                        'current_price': last,
+                        'change_24h': change_24h,
+                        'range_pos': range_pos,
+                        'sort_score': sort_score
+                    })
             
-            res = self._apply_logic_tree(dummy_data)
+            # Sort by the Regime-Specific Score
+            candidates.sort(key=lambda x: x['sort_score'], reverse=True)
             
-            score = res.get('score', 0.0)
-            # تحويل السكور ليكون متوافقاً مع الباكتست (حول 20-80)
-            if res['type'] == 'BREAKOUT':
-                return score * 20.0, ["BREAKOUT"] # Breakout score is usually low (ratio), boost it
-            elif res['type'] == 'REVERSAL':
-                return score, ["REVERSAL"] # Reversal score is already 0-100
+            # Return top 300 for Deep Analysis
+            return candidates[:300]
             
-            return 0.0, ["NONE"]
+        except Exception as e:
+            print(f"❌ [Flash-Pulse Error] {e}")
+            return []
+
+    # ==================================================================
+    # 🛡️ Layer 1: Polymorphic Screening
+    # ==================================================================
+    async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
+        # 1. Update Matrix
+        await self.update_hyper_regime()
+        regime = SystemLimits.CURRENT_REGIME
+        
+        # 2. Flash-Pulse Filter (Smart Universe Selection)
+        initial_candidates = await self._stage0_universe_filter()
+        if not initial_candidates: return []
+
+        # 3. Fetch Technicals
+        enriched_data = await self._fetch_technical_data_batch(initial_candidates)
+        filtered_list = []
+
+        # 4. Logic Switcher
+        for item in enriched_data:
+            classification = {'type': 'NONE', 'score': 0.0}
+            if regime == "BULL": classification = self._apply_bull_logic(item)
+            elif regime == "BEAR": classification = self._apply_bear_logic(item)
+            elif regime == "DEAD": classification = self._apply_dead_logic(item)
+            else: classification = self._apply_range_logic(item)
             
-        except Exception:
-            return 0.0, ["ERROR"]
+            if classification['type'] != 'NONE':
+                item['l1_score'] = classification['score']
+                item['type'] = classification['type']
+                filtered_list.append(item)
+
+        filtered_list.sort(key=lambda x: x['l1_score'], reverse=True)
+        return filtered_list[:40]
 
     # ==================================================================
-    # 🏗️ V15.2 Logic Core (Unchanged Logic)
+    # 🐂 Logic: BULL
     # ==================================================================
-    def _apply_logic_tree(self, data: Dict[str, Any]) -> Dict[str, Any]:
+    def _apply_bull_logic(self, data: Dict[str, Any]) -> Dict[str, Any]:
         try:
             df_1h = self._calc_indicators(data['ohlcv_1h'])
-            df_15m = self._calc_indicators(data['ohlcv_15m'])
-        except:
-            return {'type': 'NONE'}
-
-        if df_1h.empty or df_15m.empty: return {'type': 'NONE'}
+            if df_1h.empty: return {'type': 'NONE'}
+            curr = df_1h.iloc[-1]
+            
+            if not (curr['close'] > curr['ema50']): return {'type': 'NONE'}
+            if self._get_change_4h(df_1h) > 20.0: return {'type': 'NONE'} 
+
+            if 50 <= curr['rsi'] <= 80:
+                vol_ma = df_1h['volume'].rolling(20).mean().iloc[-1]
+                if curr['volume'] > vol_ma * 1.3:
+                    return {'type': 'BULL_MOMENTUM', 'score': curr['volume'] / vol_ma}
+        except: pass
+        return {'type': 'NONE'}
 
-        curr_1h = df_1h.iloc[-1]
-        curr_15m = df_15m.iloc[-1]
-        
-        # --- Stage 2: Anti-FOMO Filters (STRICT) ---
+    # ==================================================================
+    # 🐻 Logic: BEAR
+    # ==================================================================
+    def _apply_bear_logic(self, data: Dict[str, Any]) -> Dict[str, Any]:
         try:
-            # حساب التغير في آخر 4 ساعات
-            if len(df_1h) >= 5:
-                close_4h_ago = df_1h.iloc[-5]['close']
-                change_4h = ((curr_1h['close'] - close_4h_ago) / close_4h_ago) * 100
-            else:
-                change_4h = 0.0
-        except: change_4h = 0.0
+            df_1h = self._calc_indicators(data['ohlcv_1h'])
+            if df_1h.empty: return {'type': 'NONE'}
+            curr = df_1h.iloc[-1]
+
+            if self._get_change_4h(df_1h) > 4.0: return {'type': 'NONE'}
+            if curr['rsi'] < 32:
+                dist_ema = (curr['ema20'] - curr['close']) / curr['close']
+                if dist_ema > 0.04: 
+                     return {'type': 'BEAR_OVERSOLD', 'score': (32 - curr['rsi']) * 2}
+        except: pass
+        return {'type': 'NONE'}
 
-        # 1. فلتر المضخات: ممنوع الدخول إذا صعدت أكثر من 8% في 4 ساعات
-        if change_4h > 8.0: return {'type': 'NONE'}
+    # ==================================================================
+    # 🦀 Logic: RANGE
+    # ==================================================================
+    def _apply_range_logic(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            df_1h = self._calc_indicators(data['ohlcv_1h'])
+            df_15m = self._calc_indicators(data['ohlcv_15m'])
+        except: return {'type': 'NONE'}
+        if df_1h.empty or df_15m.empty: return {'type': 'NONE'}
         
-        # 2. فلتر التذبذب اليومي: ممنوع أكثر من 12% (للبعد عن العملات المجنونة)
-        if data.get('change_24h', 0) > 12.0: return {'type': 'NONE'}
+        curr_1h = df_1h.iloc[-1]; curr_15m = df_15m.iloc[-1]
+        change_4h = self._get_change_4h(df_1h)
         
-        # 3. فلتر القمة: ممنوع RSI فوق 70 قطعاً
+        # Strict Filters
+        if change_4h > 8.0 or data.get('change_24h', 0) > 12.0: return {'type': 'NONE'}
         if curr_1h['rsi'] > 70: return {'type': 'NONE'} 
-
-        # 4. فلتر الامتداد: ممنوع الابتعاد عن المتوسط كثيراً
-        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: Setup Classification ---
-        
-        # === A. Breakout Logic ===
-        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 يجب أن يكون فيه مساحة للصعود (ليس منخفضاً جداً ولا مرتفعاً جداً)
-            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]
-                    current_range = curr_15m['high'] - curr_15m['low']
-                    
-                    if current_range <= avg_range * 1.8:
-                        vol_ma20 = df_15m['volume'].rolling(20).mean().iloc[-1]
-                        # شرط الفوليوم: شمعة الحالية فيها سيولة 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 ===
-        is_reversal = False
-        reversal_score = 0.0 
+        # Breakout
+        if (curr_1h['ema20'] > curr_1h['ema50']) and 45 <= curr_1h['rsi'] <= 68:
+            vol_ma20 = df_15m['volume'].rolling(20).mean().iloc[-1]
+            if curr_15m['volume'] >= 1.5 * vol_ma20:
+                return {'type': 'RANGE_BREAKOUT', 'score': curr_15m['volume'] / vol_ma20}
         
-        # تشبع بيعي واضح
-        if 20 <= curr_1h['rsi'] <= 40: 
-            # السعر هبط مؤخراً
-            if change_4h <= -2.0: 
-                # البحث عن شمعة انعكاسية (Hammer / Green Body)
-                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 pattern logic
-                        lower_wick = min(row['open'], row['close']) - row['low']
-                        body = abs(row['close'] - row['open'])
-                        hammer_shape = lower_wick > (body * 1.5)
-                        
-                        if is_green or hammer_shape:
-                            found_rejection = True
-                            break
-                
-                if found_rejection:
-                    is_reversal = True
-                    # السكور كلما قل الـ RSI كان أفضل للارتداد
-                    reversal_score = (100 - curr_1h['rsi']) 
-
-        if is_reversal:
-            return {'type': 'REVERSAL', 'score': reversal_score}
+        # Reversal
+        if 20 <= curr_1h['rsi'] <= 40 and change_4h <= -2.0:
+             return {'type': 'RANGE_REVERSAL', 'score': (100 - curr_1h['rsi'])}
+        return {'type': 'NONE'}
 
+    # ==================================================================
+    # 🧟 Logic: DEAD
+    # ==================================================================
+    def _apply_dead_logic(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            df_1h = self._calc_indicators(data['ohlcv_1h'])
+            if df_1h.empty: return {'type': 'NONE'}
+            curr = df_1h.iloc[-1]
+            std = df_1h['close'].rolling(20).std().iloc[-1]
+            bb_width = (std * 4) / curr['close']
+            if bb_width < 0.04: 
+                vol_ma = df_1h['volume'].rolling(20).mean().iloc[-1]
+                if curr['volume'] > vol_ma * 1.5:
+                    return {'type': 'DEAD_SQUEEZE', 'score': (1/bb_width)}
+        except: pass
         return {'type': 'NONE'}
 
-    # ------------------------------------------------------------------
-    # Manual Indicator Calculation (Pandas pure - Exactly like V15.2)
-    # ------------------------------------------------------------------
+    # ==================================================================
+    # 🏗️ Helpers
+    # ==================================================================
     def _calc_indicators(self, ohlcv_list):
         if not ohlcv_list: return pd.DataFrame()
         df = pd.DataFrame(ohlcv_list, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-        
-        # RSI Calculation
         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))
-        
-        # EMA
         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
 
-    # ==================================================================
-    # 🌌 Stage 0: Universe Filter (V15.2 Logic)
-    # ==================================================================
-    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
+    def _get_change_4h(self, df_1h):
         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
-                
-                # شرط السيولة الصارم: 1 مليون دولار
-                quote_vol = ticker.get('quoteVolume')
-                if not quote_vol or quote_vol < 1_000_000: continue
-                
-                last_price = ticker.get('last')
-                if not last_price or last_price < 0.0005: continue 
-
-                candidates.append({
-                    'symbol': symbol,
-                    'quote_volume': quote_vol,
-                    'current_price': last_price,
-                    '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 []
+            if len(df_1h) >= 5:
+                return ((df_1h.iloc[-1]['close'] - df_1h.iloc[-5]['close']) / df_1h.iloc[-5]['close']) * 100
+            return 0.0
+        except: return 0.0
 
-    # ==================================================================
-    # 🔄 Batch Fetching
-    # ==================================================================
     async def _fetch_technical_data_batch(self, candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
         chunk_size = 15 
         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])
+            results.extend([r for r in await asyncio.gather(*[self._fetch_single_tech_data(c) for c in chunk]) if r])
             await asyncio.sleep(0.05) 
         return results
 
     async def _fetch_single_tech_data(self, candidate: Dict[str, Any]) -> Any:
-        symbol = candidate['symbol']
         try:
-            # V15.2 Requires 1H and 15M
-            ohlcv_1h = await self.exchange.fetch_ohlcv(symbol, '1h', limit=60)
-            ohlcv_15m = await self.exchange.fetch_ohlcv(symbol, '15m', limit=60)
-            
-            if not ohlcv_1h or len(ohlcv_1h) < 55 or not ohlcv_15m or len(ohlcv_15m) < 55:
-                return None
-                
-            candidate['ohlcv_1h'] = ohlcv_1h
-            candidate['ohlcv_15m'] = ohlcv_15m
+            h1 = await self.exchange.fetch_ohlcv(candidate['symbol'], '1h', limit=60)
+            m15 = await self.exchange.fetch_ohlcv(candidate['symbol'], '15m', limit=60)
+            if not h1 or not m15: return None
+            candidate['ohlcv_1h'] = h1; candidate['ohlcv_15m'] = m15
             return candidate
-        except Exception:
-            return None
-
-    # ==================================================================
-    # 🎯 Public Helpers
-    # ==================================================================
+        except: return None
+        
     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
-
+        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:
-            candles = await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
-            return candles or []
-        except Exception: return []
-
+        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:
-            ob = await self.exchange.fetch_order_book(symbol, limit)
-            return ob
-        except Exception: return {}
+        try: return await self.exchange.fetch_order_book(symbol, limit)
+        except: return {}