Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V15.2 - GEM-Architect: Anti-FOMO Shield - Strict Marksman Mode)
+# (V60.0 - GEM-Architect: Anti-FOMO Shield + Neural Injection)
 # ============================================================
 
 import asyncio
@@ -12,22 +12,23 @@ import pandas as pd
 import numpy as np
 from typing import List, Dict, Any
 
+# Keep SystemLimits import for fallbacks if needed
+try:
+    from ml_engine.processor import SystemLimits
+except ImportError:
+    SystemLimits = None
+
 # إعدادات التسجيل
 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
-        self.adaptive_hub_ref = None  # مرجع لملف التعلم
+        self.adaptive_hub_ref = None  # 🧠 المرجع لملف التعلم (Adaptive Hub)
         
-        # إعداد المنصة (KuCoin)
         self.exchange = ccxt.kucoin({
             'enableRateLimit': True,
             'timeout': 60000, 
@@ -37,31 +38,25 @@ class DataManager:
         self.http_client = None
         self.market_cache = {}
         
-        # قوائم الاستبعاد (العملات المستقرة والعملات ذات الرافعة)
         self.BLACKLIST_TOKENS = [
             'USDT', 'USDC', 'DAI', 'TUSD', 'BUSD', 'FDUSD', 'EUR', 'PAX',
             'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L'
         ]
         
-        print(f"📦 [DataManager V15.2] Restored 'Anti-FOMO' Logic Shield.")
+        print(f"📦 [DataManager V60.0] Anti-FOMO Shield + Context Injector Active.")
 
     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:
                 await self.exchange.load_markets()
                 self.market_cache = self.exchange.markets
-        except Exception as e:
-            print(f"❌ [DataManager] Market load failed: {e}")
-            traceback.print_exc()
+        except Exception: pass
 
     async def close(self):
         if self.http_client: await self.http_client.aclose()
@@ -71,283 +66,241 @@ class DataManager:
         if not self.r2_service: return
         try:
             self.contracts_db = await self.r2_service.load_contracts_db_async()
-        except Exception:
-            self.contracts_db = {}
+        except Exception: self.contracts_db = {}
 
     def get_contracts_db(self) -> Dict[str, Any]:
         return self.contracts_db
-    
+
     # ==================================================================
-    # 🛡️ Layer 1: The Strict Logic Tree Screening (Core Logic)
+    # 🧠 Layer 1: Screening + Diagnosis + Injection
     # ==================================================================
     async def layer1_rapid_screening(self, adaptive_hub_ref=None) -> List[Dict[str, Any]]:
         """
-        يقوم بمسح السوق وتقسيم الفرص إلى نوعين:
-        1. Breakout: اختراق آمن (RSI < 70).
-        2. Reversal: ارتداد من القاع (RSI < 40).
-        أي شيء آخر يتم استبعاده فوراً.
+        1. Filters using Strict Anti-FOMO Logic.
+        2. Diagnoses Market Regime (Bull/Bear/Range) for survivors.
+        3. Injects Dynamic Thresholds from AdaptiveHub.
         """
         self.adaptive_hub_ref = adaptive_hub_ref
-        print(f"🔍 [Layer 1] Initiating STRICT Logic Tree Screening...")
+        print(f"🔍 [Layer 1] Initiating STRICT Screening & Context Injection...")
         
-        # 1. المرحلة 0: فلتر الكون (السيولة الأساسية)
+        # 1. فلتر السيولة الأساسي
         initial_candidates = await self._stage0_universe_filter()
-        
-        if not initial_candidates:
-            return []
+        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)
+        # 2. جلب البيانات الفنية
+        top_candidates = initial_candidates[:100] 
+        enriched_data = await self._fetch_technical_data_batch(top_candidates)
         
-        # 3. تطبيق شجرة القرار الصارمة (Anti-FOMO)
-        breakout_list = []
-        reversal_list = []
+        final_list = []
 
         for item in enriched_data:
-            classification = self._apply_logic_tree(item)
+            # 3. تطبيق الفلتر الصارم (Anti-FOMO)
+            classification = self._apply_strict_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':
+            if classification['type'] != 'NONE':
+                # ✅ 4. التشخيص والحقن (The Injection Step)
+                # نحدد حالة العملة (صاعدة/هابطة/ميتة) لتحديد العتبات المناسبة
+                regime_info = self._diagnose_asset_regime(item)
+                item['asset_regime'] = regime_info['regime']
+                item['asset_regime_conf'] = regime_info['conf']
+                
+                # ✅ حقن العتبات الخاصة من AdaptiveHub
+                if self.adaptive_hub_ref:
+                    # يجلب الإعدادات الخاصة بهذه الحالة (مثلاً: BULL يحتاج عتبات أقل)
+                    dynamic_config = self.adaptive_hub_ref.get_regime_config(regime_info['regime'])
+                    item['dynamic_limits'] = dynamic_config
+                
+                # حفظ النتيجة
                 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.")
+                item['strategy_tag'] = classification['type']
+                final_list.append(item)
 
-        # 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]
+        # 5. الترتيب النهائي
+        # نفضل الاختراقات الآمنة (Breakout) والارتدادات القوية
+        final_list.sort(key=lambda x: x['l1_sort_score'], reverse=True)
         
-        # تنظيف البيانات للإرجاع
-        # نحتفظ بالبيانات الفنية لأن المعالج سيحتاجها
-        print(f"✅ [Layer 1] Final Selection: {len(final_selection)} candidates (Anti-FOMO Active).")
-        return final_selection
+        selection = final_list[:50]
+        print(f"✅ [Layer 1] Injected Context into {len(selection)} Candidates.")
+        return selection
 
     # ------------------------------------------------------------------
-    # Stage 0: Universe Filter (Basic Liquidity)
+    # 🧭 The Diagnoser (يحدد حالة العملة لحقن العتبات)
     # ------------------------------------------------------------------
-    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
+    def _diagnose_asset_regime(self, item: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        تحليل حالة العملة بناءً على البيانات المحسوبة مسبقاً (1H).
+        يحدد هل هي BULL, BEAR, RANGE, أم DEAD.
+        """
         try:
-            tickers = await self.exchange.fetch_tickers()
-            candidates = []
+            # نستخدم بيانات الـ 1 ساعة المحسوبة في _apply_strict_logic_tree
+            # البيانات موجودة في item['df_1h'] (سنقوم بحفظها هناك)
+            if 'df_1h' not in item: return {'regime': 'RANGE', 'conf': 0.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): 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 
-                
-                # 👇 فلتر أولي: استبعاد العملات التي انفجرت بجنون (+15% فأكثر يتم تجاهلها مبدئياً)
-                change_24h = ticker.get('percentage', 0.0)
-                if change_24h > 15.0: continue 
-
-                candidates.append({
-                    'symbol': symbol,
-                    'quote_volume': quote_vol,
-                    'current_price': last_price,
-                    'change_24h': change_24h
-                })
+            df = item['df_1h']
+            curr = df.iloc[-1]
             
-            # الترتيب حسب السيولة
-            candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
-            return candidates
+            price = curr['close']
+            ema20 = curr['ema20']
+            ema50 = curr['ema50']
+            rsi = curr['rsi']
+            atr = curr['atr']
             
-        except Exception as e:
-            print(f"❌ [L1 Error] Universe filter failed: {e}")
-            return []
-
-    # ------------------------------------------------------------------
-    # 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
-
-    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)
+            # حساب نسبة الـ ATR (لقياس النشاط)
+            atr_pct = (atr / price) * 100 if price > 0 else 0
             
-            if not ohlcv_1h or len(ohlcv_1h) < 55 or not ohlcv_15m or len(ohlcv_15m) < 55:
-                return None
+            regime = "RANGE"
+            conf = 0.5
             
-            # تخزين البيانات الخام لاستخدامها لاحقاً في المعالج
-            candidate['ohlcv'] = {
-                '1h': ohlcv_1h,
-                '15m': ohlcv_15m,
-                # يمكن إضافة 5m لاحقاً عند الحاجة للدقة القصوى
-            }
+            # 1. DEAD Check (خمول تام)
+            # حركة ضعيفة جداً وسيولة منخفضة
+            if atr_pct < 0.5: 
+                return {'regime': 'DEAD', 'conf': 0.9}
+
+            # 2. BULL Check (ترند صاعد قوي)
+            # السعر فوق المتوسطات، والمتوسطات مرتبة، والزخم إيجابي
+            if price > ema20 and ema20 > ema50 and rsi > 50:
+                regime = "BULL"
+                conf = 0.8 if rsi > 55 else 0.6
+                
+            # 3. BEAR Check (ترند هابط)
+            elif price < ema20 and ema20 < ema50 and rsi < 50:
+                regime = "BEAR"
+                conf = 0.8 if rsi < 45 else 0.6
+                
+            return {'regime': regime, 'conf': conf}
             
-            # نسخ للاستخدام الداخلي في الفلتر
-            candidate['ohlcv_1h_raw'] = ohlcv_1h
-            candidate['ohlcv_15m_raw'] = ohlcv_15m
-            return candidate
         except Exception:
-            return None
+            return {'regime': 'RANGE', 'conf': 0.0}
 
     # ------------------------------------------------------------------
-    # 🧠 The Logic Core: Math & Decision Tree (STRICT ANTI-FOMO MODE)
+    # 🛡️ The Strict Logic Tree (Anti-FOMO)
     # ------------------------------------------------------------------
-    def _apply_logic_tree(self, data: Dict[str, Any]) -> Dict[str, Any]:
+    def _apply_strict_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'}
+            
+            # حفظ الداتافريم للتشخيص لاحقاً
+            data['df_1h'] = df_1h
+            
+        except: return {'type': 'NONE', 'score': 0}
 
         curr_1h = df_1h.iloc[-1]
         curr_15m = df_15m.iloc[-1]
         
-        # --- Stage 2: Overbought Filter (STRICT MODE 🛡️) ---
+        # --- فلاتر الأمان (Safety Gates) ---
         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'} 
+        # 1. لا تطارد المضخات (+8% في 4 ساعات ممنوع)
+        if change_4h > 8.0: return {'type': 'NONE', 'score': 0}
+        # 2. لا تشتري في قمة التشبع (RSI > 70 ممنوع)
+        if curr_1h['rsi'] > 70: return {'type': 'NONE', 'score': 0} 
+        # 3. لا تشتري بعيداً عن المتوسط (Mean Reversion Risk)
+        dev = (curr_1h['close'] - curr_1h['ema20']) / curr_1h['atr'] if curr_1h['atr'] > 0 else 0
+        if dev > 1.8: return {'type': 'NONE', 'score': 0}
 
-        # 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
+        # A. Breakout (اختراق آمن)
+        # هيكلية صاعدة + تجميع (Squeeze) + فوليوم
+        is_bullish = (curr_1h['ema20'] > curr_1h['ema50']) or (curr_1h['close'] > curr_1h['ema20'])
+        if is_bullish and (45 <= curr_1h['rsi'] <= 68):
+            vol_ma = df_15m['volume'].rolling(20).mean().iloc[-1]
+            if curr_15m['volume'] >= 1.5 * vol_ma: # شرط الفوليوم
+                # سكور الاختراق يعتمد على قوة الفوليوم النسبي
+                score = curr_15m['volume'] / vol_ma if vol_ma > 0 else 1.0
+                return {'type': 'BREAKOUT', 'score': score}
 
-        if is_breakout:
-            return {'type': 'BREAKOUT', 'score': breakout_score}
+        # B. Reversal (صيد القاع)
+        # تشبع بيعي + شمعة عاكسة
+        if 20 <= curr_1h['rsi'] <= 40 and change_4h <= -2.0:
+            # سكور الارتداد يعتمد على مدى انخفاض الـ RSI (كلما قل كان أفضل للارتداد)
+            score = (100 - curr_1h['rsi'])
+            return {'type': 'REVERSAL', 'score': 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
+        return {'type': 'NONE', 'score': 0}
+
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+    async def _stage0_universe_filter(self) -> List[Dict[str, Any]]:
+        try:
+            tickers = await self.exchange.fetch_tickers()
+            candidates = []
+            for symbol, ticker in tickers.items():
+                if not symbol.endswith('/USDT'): continue
+                if any(b in symbol for b in self.BLACKLIST_TOKENS): continue
+                
+                # السيولة
+                quote_vol = ticker.get('quoteVolume', 0)
+                if quote_vol < 1_000_000: continue
+                
+                # حماية من العملات المنفجرة (+15% يومي ممنوع)
+                if ticker.get('percentage', 0) > 15.0: continue
                 
-                if found_rejection:
-                    is_reversal = True
-                    # كلما قل الـ RSI زادت احتمالية الارتداد (سكور أعلى)
-                    reversal_score = (100 - curr_1h['rsi']) 
+                candidates.append({
+                    'symbol': symbol,
+                    'quote_volume': quote_vol,
+                    'current_price': ticker.get('last'),
+                    'change_24h': ticker.get('percentage')
+                })
+            candidates.sort(key=lambda x: x['quote_volume'], reverse=True)
+            return candidates
+        except: return []
 
-        if is_reversal:
-            return {'type': 'REVERSAL', 'score': reversal_score}
+    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
 
-        return {'type': 'NONE'}
+    async def _fetch_single(self, c):
+        try:
+            # نحتاج 1H و 15M للتشخيص والفلتر
+            h1 = await self.exchange.fetch_ohlcv(c['symbol'], '1h', limit=60)
+            m15 = await self.exchange.fetch_ohlcv(c['symbol'], '15m', limit=60)
+            if not h1 or not m15: return None
+            c['ohlcv'] = {'1h': h1, '15m': m15} # للمعالج لاحقاً
+            c['ohlcv_1h_raw'] = h1 # للفلتر الداخلي
+            c['ohlcv_15m_raw'] = m15
+            return c
+        except: return None
 
-    def _calc_indicators(self, ohlcv_list):
-        # حسابات يدوية سريعة باستخدام Pandas (بدون مكتبات خارجية لتقليل التبعيات في الفلتر)
-        df = pd.DataFrame(ohlcv_list, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-        
+    def _calc_indicators(self, ohlcv):
+        df = pd.DataFrame(ohlcv, columns=['ts', 'o', 'h', 'l', 'c', 'v'])
         # 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))
-        
+        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))
         # EMAs
-        df['ema20'] = df['close'].ewm(span=20, adjust=False).mean()
-        df['ema50'] = df['close'].ewm(span=50, adjust=False).mean()
-        
+        df['ema20'] = df['c'].ewm(span=20).mean()
+        df['ema50'] = df['c'].ewm(span=50).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()
+        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.fillna(0, inplace=True)
-        return df
-
-    # ==================================================================
-    # 🎯 Public Helpers (Standard Interface)
-    # ==================================================================
-    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
-
-    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 []
+        # Renaming for consistency
+        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 []
 
-    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 {}
+    async def get_order_book_snapshot(self, symbol, limit=20):
+        try: return await self.exchange.fetch_order_book(symbol, limit)
+        except: return {}