Update app.py

app.py

@@ -1,621 +1,831 @@
-# app.py (محدث)
+# data_manager.py (محدث)
 import os
-import traceback
-import signal
-import sys
-import uvicorn
 import asyncio
-import json
+import httpx
+import traceback
 import time
-from contextlib import asynccontextmanager
-from fastapi import FastAPI, HTTPException
 from datetime import datetime
-from r2 import R2Service
-from LLM import LLMService
-from data_manager import DataManager
-from ML import MLProcessor
-from learning_engine import LearningEngine
-from sentiment_news import SentimentAnalyzer
-from trade_manager import TradeManager
-import state
-from helpers import safe_float_conversion, validate_candidate_data_enhanced
-
-# المتغيرات العالمية
-r2_service_global = None
-data_manager_global = None
-llm_service_global = None
-learning_engine_global = None
-trade_manager_global = None
-sentiment_analyzer_global = None
-symbol_whale_monitor_global = None
-
-class StateManager:
-    def __init__(self):
-        self.market_analysis_lock = asyncio.Lock()
-        self.trade_analysis_lock = asyncio.Lock()
-        self.initialization_complete = False
-        self.services_initialized = {
-            'r2_service': False,
-            'data_manager': False,
-            'llm_service': False,
-            'learning_engine': False,
-            'trade_manager': False,
-            'sentiment_analyzer': False,
-            'symbol_whale_monitor': False
-        }
+import ccxt
+import numpy as np
+import logging
+from typing import List, Dict, Any
 
-    async def wait_for_initialization(self, timeout=30):
-        start_time = time.time()
-        while not self.initialization_complete and (time.time() - start_time) < timeout:
-            await asyncio.sleep(1)
-        return self.initialization_complete
+logging.getLogger("httpx").setLevel(logging.WARNING)
+logging.getLogger("httpcore").setLevel(logging.WARNING)
 
-    def set_service_initialized(self, service_name):
-        self.services_initialized[service_name] = True
-        if all(self.services_initialized.values()):
-            self.initialization_complete = True
+class DataManager:
+    def __init__(self, contracts_db, whale_monitor):
+        self.contracts_db = contracts_db or {}
+        self.whale_monitor = whale_monitor
+        
+        try:
+            self.exchange = ccxt.kucoin({
+                'sandbox': False,
+                'enableRateLimit': True,
+                'timeout': 30000,
+                'verbose': False,
+            })
+            print("✅ تم تهيئة اتصال KuCoin بنجاح")
+        except Exception as e:
+            print(f"❌ فشل تهيئة اتصال KuCoin: {e}")
+            self.exchange = None
+            
+        self.http_client = None
+        self.market_cache = {}
+        self.last_market_load = None
+        
+    async def initialize(self):
+        self.http_client = httpx.AsyncClient(timeout=30.0)
+        await self._load_markets()
+        print("✅ DataManager initialized - Efficient Volume-Based Screening")
 
-state_manager = StateManager()
+    async def _load_markets(self):
+        try:
+            if not self.exchange:
+                return
+                
+            print("🔄 جلب أحدث بيانات الأسواق من KuCoin...")
+            self.exchange.load_markets()
+            self.market_cache = self.exchange.markets
+            self.last_market_load = datetime.now()
+            print(f"✅ تم تحميل {len(self.market_cache)} سوق من KuCoin")
+            
+        except Exception as e:
+            print(f"❌ فشل تحميل بيانات الأسواق: {e}")
 
-async def monitor_market_async():
-    """مراقبة السوق"""
-    global data_manager_global, sentiment_analyzer_global
-    
-    if not await state_manager.wait_for_initialization():
-        print("❌ فشل تهيئة الخدمات - إوقف مراقبة السوق")
-        return
-        
-    while True:
+    async def close(self):
+        if self.http_client: 
+            await self.http_client.aclose()
+
+    async def get_market_context_async(self):
+        """جلب سياق السوق الأساسي فقط"""
         try:
-            async with state_manager.market_analysis_lock:
-                market_context = await sentiment_analyzer_global.get_market_sentiment()
-                if not market_context: 
-                    state.MARKET_STATE_OK = True
-                    await asyncio.sleep(60)
-                    continue
-                    
-                bitcoin_sentiment = market_context.get('btc_sentiment')
-                fear_greed_index = market_context.get('fear_and_greed_index')
+            sentiment_data = await self.get_sentiment_safe_async()
+            price_data = await self._get_prices_with_fallback()
+            
+            bitcoin_price = price_data.get('bitcoin')
+            ethereum_price = price_data.get('ethereum')
+            
+            market_context = {
+                'timestamp': datetime.now().isoformat(),
+                'bitcoin_price_usd': bitcoin_price,
+                'ethereum_price_usd': ethereum_price,
+                'fear_and_greed_index': sentiment_data.get('feargreed_value') if sentiment_data else None,
+                'sentiment_class': sentiment_data.get('feargreed_class') if sentiment_data else 'NEUTRAL',
+                'market_trend': self._determine_market_trend(bitcoin_price, sentiment_data),
+                'btc_sentiment': self._get_btc_sentiment(bitcoin_price),
+                'data_quality': 'HIGH' if bitcoin_price and ethereum_price else 'LOW'
+            }
+            
+            return market_context
                 
-                should_halt_trading, halt_reason = False, ""
+        except Exception as e:
+            print(f"❌ فشل جلب سياق السوق: {e}")
+            return self._get_minimal_market_context()
+
+    async def get_sentiment_safe_async(self):
+        """جلب بيانات المشاعر"""
+        try:
+            async with httpx.AsyncClient(timeout=10) as client:
+                response = await client.get("https://api.alternative.me/fng/")
+                response.raise_for_status()
+                data = response.json()
                 
-                if bitcoin_sentiment == 'BEARISH' and (fear_greed_index is not None and fear_greed_index < 30): 
-                    should_halt_trading, halt_reason = True, "ظروف سوق هابطة"
-                    
-                if should_halt_trading:
-                    state.MARKET_STATE_OK = False
-                    await r2_service_global.save_system_logs_async({"market_halt": True, "reason": halt_reason})
-                else:
-                    if not state.MARKET_STATE_OK: 
-                        print("✅ تحسنت ظروف السوق. استئناف العمليات العادية.")
-                    state.MARKET_STATE_OK = True
-                    
-            await asyncio.sleep(60)
-        except Exception as error:
-            print(f"❌ خطأ أثناء مراقبة السوق: {error}")
-            state.MARKET_STATE_OK = True
-            await asyncio.sleep(60)
-
-async def process_batch_parallel(batch, ml_processor, batch_num, total_batches):
-    """معالجة دفعة من الرموز بشكل متوازي"""
-    try:
-        print(f"   🔄 معالجة الدفعة {batch_num}/{total_batches} ({len(batch)} عملة)...")
-        
-        # إنشاء مهام للدفعة الحالية
-        batch_tasks = []
-        for symbol_data in batch:
-            task = asyncio.create_task(ml_processor.process_and_score_symbol_enhanced(symbol_data))
-            batch_tasks.append(task)
-        
-        # انتظار انتهاء جميع مهام الدفعة الحالية
-        batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
-        
-        # تصفية النتائج الناجحة
-        successful_results = []
-        for result in batch_results:
-            if isinstance(result, Exception):
-                continue
-            if result and result.get('enhanced_final_score', 0) > 0.4:
-                successful_results.append(result)
-        
-        print(f"   ✅ اكتملت الدفعة {batch_num}: {len(successful_results)}/{len(batch)} ناجحة")
-        return successful_results
-        
-    except Exception as error:
-        print(f"❌ خطأ في معالجة الدفعة {batch_num}: {error}")
-        return []
-
-async def run_3_layer_analysis():
-    """
-    تشغيل النظام الطبقي المكون من 3 طبقات:
-    الطبقة 1: data_manager - الفحص السريع
-    الطبقة 2: MLProcessor - التحليل المتقدم  
-    الطبقة 3: LLMService - النموذج الضخم
-    """
-    try:
-        print("🎯 بدء النظام الطبقي المكون من 3 طبقات...")
-        
-        if not await state_manager.wait_for_initialization():
-            print("❌ الخدمات غير مهيأة بالكامل")
+                if 'data' not in data or not data['data']:
+                    raise ValueError("بيانات المشاعر غير متوفرة")
+                
+                latest_data = data['data'][0]
+                return {
+                    "feargreed_value": int(latest_data['value']), 
+                    "feargreed_class": latest_data['value_classification'],
+                    "source": "alternative.me",
+                    "timestamp": datetime.now().isoformat()
+                }
+        except Exception as e:
+            print(f"❌ فشل جلب بيانات المشاعر: {e}")
             return None
 
-        # الطبقة 1: الفحص السريع لجميع العملات
-        print("\n🔍 الطبقة 1: الفحص السريع (data_manager)...")
-        layer1_candidates = await data_manager_global.layer1_rapid_screening()
-        
-        if not layer1_candidates:
-            print("❌ لم يتم العثور على مرشحين في الطبقة 1")
-            return None
-        
-        print(f"✅ تم ��ختيار {len(layer1_candidates)} عملة للطبقة 2")
+    def _determine_market_trend(self, bitcoin_price, sentiment_data):
+        """تحديد اتجاه السوق"""
+        if bitcoin_price is None:
+            return "UNKNOWN"
+        
+        score = 0
+        if bitcoin_price > 60000: 
+            score += 1
+        elif bitcoin_price < 55000: 
+            score -= 1
+        
+        if sentiment_data and sentiment_data.get('feargreed_value') is not None:
+            fear_greed = sentiment_data.get('feargreed_value')
+            if fear_greed > 60: 
+                score += 1
+            elif fear_greed < 40: 
+                score -= 1
+        
+        if score >= 1: 
+            return "bull_market"
+        elif score <= -1: 
+            return "bear_market"
+        else:
+            return "sideways_market"
+
+    def _get_btc_sentiment(self, bitcoin_price):
+        if bitcoin_price is None:
+            return 'UNKNOWN'
+        elif bitcoin_price > 60000:
+            return 'BULLISH'
+        elif bitcoin_price < 55000:
+            return 'BEARISH'
+        else:
+            return 'NEUTRAL'
+
+    async def _get_prices_with_fallback(self):
+        """جلب أسعار البيتكوين والإيثيريوم"""
+        try:
+            prices = await self._get_prices_from_kucoin_safe()
+            if prices.get('bitcoin') and prices.get('ethereum'):
+                return prices
+            return await self._get_prices_from_coingecko()
+        except Exception as e:
+            print(f"❌ فشل جلب الأسعار: {e}")
+            return {'bitcoin': None, 'ethereum': None}
+
+    async def _get_prices_from_kucoin_safe(self):
+        if not self.exchange:
+            return {'bitcoin': None, 'ethereum': None}
+            
+        try:
+            prices = {'bitcoin': None, 'ethereum': None}
+            
+            btc_ticker = self.exchange.fetch_ticker('BTC/USDT')
+            btc_price = float(btc_ticker.get('last', 0)) if btc_ticker.get('last') else None
+            if btc_price and btc_price > 0:
+                prices['bitcoin'] = btc_price
+            
+            eth_ticker = self.exchange.fetch_ticker('ETH/USDT')
+            eth_price = float(eth_ticker.get('last', 0)) if eth_ticker.get('last') else None
+            if eth_price and eth_price > 0:
+                prices['ethereum'] = eth_price
+            
+            return prices
+            
+        except Exception as e:
+            print(f"❌ خطأ في جلب الأسعار من KuCoin: {e}")
+            return {'bitcoin': None, 'ethereum': None}
 
-        # جلب بيانات OHLCV كاملة للمرشحين
-        layer1_symbols = [candidate['symbol'] for candidate in layer1_candidates]
-        ohlcv_data_list = await data_manager_global.get_ohlcv_data_for_symbols(layer1_symbols)
-        
-        if not ohlcv_data_list:
-            print("❌ فشل جلب بيانات OHLCV للمرشحين")
-            return None
+    async def _get_prices_from_coingecko(self):
+        """الاحتياطي: جلب الأسعار من CoinGecko"""
+        try:
+            await asyncio.sleep(0.5)
+            url = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin,ethereum&vs_currencies=usd"
+            
+            # إضافة headers لتجنب rate limiting
+            headers = {
+                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
+                'Accept': 'application/json'
+            }
+            
+            async with httpx.AsyncClient(headers=headers) as client:
+                response = await client.get(url, timeout=10)
+                
+                if response.status_code == 429:
+                    print("⏰ Rate limit من CoinGecko - الانتظار 2 ثانية")
+                    await asyncio.sleep(2)
+                    response = await client.get(url, timeout=10)
+                
+                response.raise_for_status()
+                data = response.json()
+                
+                btc_price = data.get('bitcoin', {}).get('usd')
+                eth_price = data.get('ethereum', {}).get('usd')
+                
+                if btc_price and eth_price:
+                    return {'bitcoin': btc_price, 'ethereum': eth_price}
+                else:
+                    return {'bitcoin': None, 'ethereum': None}
+                    
+        except Exception as e:
+            print(f"❌ فشل جلب الأسعار من CoinGecko: {e}")
+            return {'bitcoin': None, 'ethereum': None}
 
-        print(f"📊 تم جلب بيانات OHLCV لـ {len(ohlcv_data_list)} عملة بنجاح")
+    def _get_minimal_market_context(self):
+        """سياق سوق بدائي عند الفشل"""
+        return {
+            'timestamp': datetime.now().isoformat(),
+            'data_available': False,
+            'market_trend': 'UNKNOWN',
+            'btc_sentiment': 'UNKNOWN',
+            'data_quality': 'LOW'
+        }
 
-        # الطبقة 2: التحليل المتقدم بشكل متوازي حقيقي
-        print(f"\n📈 الطبقة 2: التحليل المتقدم (MLProcessor) بشكل متوازي لـ {len(ohlcv_data_list)} عملة...")
-        market_context = await data_manager_global.get_market_context_async()
+    async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
+        """
+        الطبقة 1: فحص سريع - جلب أفضل 200 عملة حسب الحجم مباشرة
+        """
+        print("📊 الطبقة 1: جلب أفضل 200 عملة حسب حجم التداول...")
         
-        # إنشاء معالج ML
-        ml_processor = MLProcessor(market_context, data_manager_global, learning_engine_global)
+        # المحاولة 1: الطريقة المثلى - استخدام fetch_tickers
+        print("   🔍 المحاولة 1: جلب جميع التاكرز مرة واحدة...")
+        volume_data = await self._get_volume_data_optimal()
         
-        # تجهيز البيانات للطبقة 2
-        layer2_data = []
-        for ohlcv_data in ohlcv_data_list:
-            try:
-                # إضافة أسباب الترشيح من الطبقة 1
-                symbol = ohlcv_data['symbol']
-                layer1_candidate = next((c for c in layer1_candidates if c['symbol'] == symbol), None)
-                if layer1_candidate:
-                    ohlcv_data['reasons_for_candidacy'] = layer1_candidate.get('reasons', [])
-                    ohlcv_data['layer1_score'] = layer1_candidate.get('layer1_score', 0)
-                    ohlcv_data['successful_timeframes'] = ohlcv_data.get('successful_timeframes', 0)
-                
-                layer2_data.append(ohlcv_data)
-                    
-            except Exception as e:
-                print(f"❌ خطأ في إعداد تحليل {ohlcv_data.get('symbol')}: {e}")
-                continue
+        if not volume_data:
+            # المحاولة 2: الطريقة البديلة - استخدام API المباشر
+            print("   🔄 المحاولة 2: استخدام API المباشر...")
+            volume_data = await self._get_volume_data_direct_api()
         
-        if not layer2_data:
-            print("❌ فشل إعداد بيانات الطبقة 2")
-            return None
+        if not volume_data:
+            # المحاولة 3: الطريقة التقليدية (الاحتياطية)
+            print("   ⚠️ المحاولة 3: استخدام الطريقة التقليدية...")
+            volume_data = await self._get_volume_data_traditional()
         
-        # تقسيم العمل إلى دفعات للمعالجة المتوازية
-        batch_size = 15  # تقليل حجم الدفعة
-        batches = [layer2_data[i:i + batch_size] for i in range(0, len(layer2_data), batch_size)]
-        total_batches = len(batches)
+        if not volume_data:
+            print("❌ فشل جميع محاولات جلب بيانات الأحجام")
+            return []
         
-        print(f"   🚀 تقسيم العمل إلى {total_batches} دفعة ({batch_size} عملة لكل دفعة)...")
+        # أخذ أفضل 200 عملة حسب الحجم فقط
+        volume_data.sort(key=lambda x: x['dollar_volume'], reverse=True)
+        top_200_by_volume = volume_data[:200]
         
-        # معالجة جميع الدفعات بشكل متوازي
-        batch_tasks = []
-        for i, batch in enumerate(batches):
-            task = asyncio.create_task(process_batch_parallel(batch, ml_processor, i+1, total_batches))
-            batch_tasks.append(task)
+        print(f"✅ تم اختيار أفضل {len(top_200_by_volume)} عملة حسب الحجم")
         
-        # جمع نتائج جميع الدفعات
-        batch_results = await asyncio.gather(*batch_tasks)
+        # المرحلة 2: تطبيق المؤشرات الأخرى على الـ200 فقط
+        print("   📈 تطبيق المؤشرات المتقدمة على أفضل 200 عملة...")
+        final_candidates = await self._apply_advanced_indicators(top_200_by_volume)
         
-        # دمج جميع النتائج
-        layer2_candidates = []
-        for batch_result in batch_results:
-            layer2_candidates.extend(batch_result)
+        print(f"🎯 تم تحليل {len(final_candidates)} عملة للطبقة 2")
         
-        print(f"✅ اكتمل التحليل المتقدم: {len(layer2_candidates)}/{len(layer2_data)} عملة تم تحليلها بنجاح")
+        # عرض أفضل 15 عملة
+        print("🏆 أفضل 15 عملة من الطبقة 1:")
+        for i, candidate in enumerate(final_candidates[:15]):
+            score = candidate.get('layer1_score', 0)
+            volume = candidate.get('dollar_volume', 0)
+            change = candidate.get('price_change_24h', 0)
+            print(f"   {i+1:2d}. {candidate['symbol']}: {score:.3f} | ${volume:>10,.0f} | {change:>+6.1f}%")
         
-        if not layer2_candidates:
-            print("❌ لم يتم العثور على مرشحين في الطبقة 2")
-            return None
-        
-        # ترتيب المرشحين حسب الدرجة المحسنة وأخذ أقوى 10 مرشحين فقط للطبقة 3
-        layer2_candidates.sort(key=lambda x: x.get('enhanced_final_score', 0), reverse=True)
-        target_count = min(10, len(layer2_candidates))
-        final_layer2_candidates = layer2_candidates[:target_count]
-        
-        print(f"🎯 تم اختيار {len(final_layer2_candidates)} عملة للطبقة 3 (الأقوى فقط)")
+        return final_candidates
 
-        # ✅ حفظ المرشحين العشرة في ملف Candidates في R2
-        await r2_service_global.save_candidates_async(final_layer2_candidates)
+    async def _get_volume_data_optimal(self) -> List[Dict[str, Any]]:
+        """الطريقة المثلى: استخدام fetch_tickers لجميع البيانات مرة واحدة"""
+        try:
+            if not self.exchange:
+                return []
+                
+            print("      📊 جلب جميع بيانات التداول مرة واحدة...")
+            tickers = self.exchange.fetch_tickers()
+            
+            volume_data = []
+            processed = 0
+            
+            for symbol, ticker in tickers.items():
+                # تصفية أزواج USDT النشطة فقط
+                if not symbol.endswith('/USDT'):
+                    continue
+                    
+                if not ticker.get('active', True):
+                    continue
+                
+                # استخدام quoteVolume (الحجم بالدولار) إذا متوفر
+                current_price = ticker.get('last', 0)
+                quote_volume = ticker.get('quoteVolume', 0)
+                
+                if quote_volume > 0:
+                    dollar_volume = quote_volume
+                else:
+                    # fallback: baseVolume * السعر
+                    base_volume = ticker.get('baseVolume', 0)
+                    dollar_volume = base_volume * current_price
+                
+                # فلترة أولية: تجاهل العملات ذات الحجم المنخفض جداً
+                if dollar_volume < 50000:  # أقل من 50K دولار
+                    continue
+                
+                volume_data.append({
+                    'symbol': symbol,
+                    'dollar_volume': dollar_volume,
+                    'current_price': current_price,
+                    'volume_24h': ticker.get('baseVolume', 0),
+                    'price_change_24h': (ticker.get('percentage', 0) or 0) * 100
+                })
+                
+                processed += 1
+            
+            print(f"      ✅ تم معالجة {processed} عملة من أصل {len(tickers)}")
+            return volume_data
+            
+        except Exception as e:
+            print(f"      ❌ فشل الطريقة المثلى: {e}")
+            return []
+
+    async def _get_volume_data_direct_api(self) -> List[Dict[str, Any]]:
+        """الطريقة الثانية: استخدام KuCoin API مباشرة"""
+        try:
+            print("      🌐 الاتصال بـ KuCoin API مباشرة...")
+            url = "https://api.kucoin.com/api/v1/market/allTickers"
+            
+            async with httpx.AsyncClient(timeout=15) as client:
+                response = await client.get(url)
+                response.raise_for_status()
+                data = response.json()
+                
+                if data.get('code') != '200000':
+                    raise ValueError(f"استجابة API غير متوقعة: {data.get('code')}")
+                
+                tickers = data['data']['ticker']
+                volume_data = []
+                
+                for ticker in tickers:
+                    symbol = ticker['symbol']
+                    
+                    # تصفية أزواج USDT فقط
+                    if not symbol.endswith('USDT'):
+                        continue
+                    
+                    # تحويل الرمز للتنسيق القياسي (BTC-USDT → BTC/USDT)
+                    formatted_symbol = symbol.replace('-', '/')
+                    
+                    try:
+                        dollar_volume = float(ticker['volValue'])  # الحجم بالدولار
+                        current_price = float(ticker['last'])
+                        price_change = float(ticker['changeRate']) * 100
+                        
+                        if dollar_volume >= 50000:
+                            volume_data.append({
+                                'symbol': formatted_symbol,
+                                'dollar_volume': dollar_volume,
+                                'current_price': current_price,
+                                'volume_24h': float(ticker['vol']),
+                                'price_change_24h': price_change
+                            })
+                    except (ValueError, KeyError) as e:
+                        continue
+                
+                print(f"      ✅ تم جلب {len(volume_data)} عملة من API المباشر")
+                return volume_data
+                
+        except Exception as e:
+            print(f"      ❌ فشل API المباشر: {e}")
+            return []
 
-        # عرض أفضل 10 عملات من الطبقة 2
-        print("\n🏆 أفضل 10 عملات من الطبقة 2:")
-        for i, candidate in enumerate(final_layer2_candidates):
-            score = candidate.get('enhanced_final_score', 0)
-            strategy = candidate.get('target_strategy', 'GENERIC')
-            mc_score = candidate.get('monte_carlo_probability', 0)
-            pattern = candidate.get('pattern_analysis', {}).get('pattern_detected', 'no_pattern')
-            timeframes = candidate.get('successful_timeframes', 0)
+    async def _get_volume_data_traditional(self) -> List[Dict[str, Any]]:
+        """الطريقة التقليدية: جلب كل رمز على حدة (الاحتياطي)"""
+        try:
+            if not self.exchange or not self.market_cache:
+                return []
+                
+            usdt_symbols = [
+                symbol for symbol in self.market_cache.keys() 
+                if symbol.endswith('/USDT') and self.market_cache[symbol].get('active', False)
+            ]
+            
+            print(f"      🔄 معالجة {len(usdt_symbols)} عملة (طريقة تقليدية)...")
+            
+            volume_data = []
+            processed = 0
+            
+            # معالجة دفعات لتجنب rate limits
+            batch_size = 50
+            for i in range(0, len(usdt_symbols), batch_size):
+                batch = usdt_symbols[i:i + batch_size]
+                batch_tasks = [self._process_single_symbol(sym) for sym in batch]
+                batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
+                
+                for result in batch_results:
+                    if isinstance(result, dict):
+                        volume_data.append(result)
+                
+                processed += len(batch)
+                print(f"         ✅ تم معالجة {processed}/{len(usdt_symbols)} عملة...")
+                
+                # انتظار قصير بين الدفعات
+                if i + batch_size < len(usdt_symbols):
+                    await asyncio.sleep(1)
+            
+            print(f"      ✅ تم جمع بيانات {len(volume_data)} عملة مؤهلة")
+            return volume_data
             
-            print(f"   {i+1}. {candidate['symbol']}:")
-            print(f"      📊 النهائي: {score:.3f} | الأطر: {timeframes}/6")
-            if mc_score > 0:
-                print(f"      🎯 مونت كارلو: {mc_score:.3f}")
-            print(f"      🎯 استراتيجية: {strategy} | نمط: {pattern}")
+        except Exception as e:
+            print(f"      ❌ فشل الطريقة التقليدية: {e}")
+            return []
 
-        # الطبقة 3: التحليل بالنموذج الضخم
-        print("\n🧠 الطبقة 3: التحليل بالنموذج الضخم (LLMService)...")
-        final_opportunities = []
+    async def _process_single_symbol(self, symbol: str) -> Dict[str, Any]:
+        """معالجة رمز واحد لجلب بيانات الحجم"""
+        try:
+            ticker = self.exchange.fetch_ticker(symbol)
+            if not ticker:
+                return None
+            
+            current_price = ticker.get('last', 0)
+            quote_volume = ticker.get('quoteVolume', 0)
+            
+            if quote_volume > 0:
+                dollar_volume = quote_volume
+            else:
+                base_volume = ticker.get('baseVolume', 0)
+                dollar_volume = base_volume * current_price
+            
+            if dollar_volume < 50000:
+                return None
+            
+            return {
+                'symbol': symbol,
+                'dollar_volume': dollar_volume,
+                'current_price': current_price,
+                'volume_24h': ticker.get('baseVolume', 0),
+                'price_change_24h': (ticker.get('percentage', 0) or 0) * 100
+            }
+        except Exception:
+            return None
+
+    async def _apply_advanced_indicators(self, volume_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """تطبيق المؤشرات المتقدمة على أفضل العملات حسب الحجم"""
+        candidates = []
         
-        for candidate in final_layer2_candidates:
+        print("   🔧 تطبيق مؤشرات الزخم والتقلب وقوة السعر...")
+        
+        for i, symbol_data in enumerate(volume_data):
             try:
-                symbol = candidate['symbol']
-                print(f"   🤔 تحليل {symbol} بالنموذج الضخم...")
+                symbol = symbol_data['symbol']
                 
-                # ✅ التأكد من تمرير بيانات OHLCV بشكل صحيح
-                ohlcv_data = candidate.get('ohlcv') or candidate.get('raw_ohlcv')
-                if not ohlcv_data:
-                    print(f"      ⚠️ لا توجد بيانات شموع لـ {symbol}")
+                # جلب بيانات إضافية للرمز
+                detailed_data = await self._get_detailed_symbol_data(symbol)
+                if not detailed_data:
                     continue
                 
-                # ✅ التأكد من تمرير البيانات الخام للنموذج
-                candidate['raw_ohlcv'] = ohlcv_data
-                candidate['ohlcv'] = ohlcv_data
-                
-                timeframes_count = candidate.get('successful_timeframes', 0)
-                total_candles = sum(len(data) for data in ohlcv_data.values())
+                # دمج البيانات
+                symbol_data.update(detailed_data)
                 
-                print(f"      📊 بيانات الشموع: {timeframes_count} إطار زمني, {total_candles} شمعة")
+                # حساب الدرجة النهائية
+                score = self._calculate_advanced_score(symbol_data)
+                symbol_data['layer1_score'] = score
                 
-                # ✅ إرسال كل عملة للنموذج الضخم على حدة
-                llm_analysis = await llm_service_global.get_trading_decision(candidate)
+                candidates.append(symbol_data)
                 
-                # ✅ التحقق من وجود قرار صالح من النموذج
-                if llm_analysis and llm_analysis.get('action') in ['BUY', 'SELL']:
-                    opportunity = {
-                        'symbol': symbol,
-                        'current_price': candidate.get('current_price', 0),
-                        'decision': llm_analysis,
-                        'enhanced_score': candidate.get('enhanced_final_score', 0),
-                        'llm_confidence': llm_analysis.get('confidence_level', 0),
-                        'strategy': llm_analysis.get('strategy', 'GENERIC'),
-                        'analysis_timestamp': datetime.now().isoformat(),
-                        'timeframes_count': timeframes_count,
-                        'total_candles': total_candles
-                    }
-                    final_opportunities.append(opportunity)
+                if (i + 1) % 50 == 0:
+                    print(f"      ✅ تم تحليل {i + 1}/{len(volume_data)} عملة")
                     
-                    print(f"   ✅ {symbol}: {llm_analysis.get('action')} - ثقة: {llm_analysis.get('confidence_level', 0):.2f}")
-                else:
-                    action = llm_analysis.get('action', 'NO_DECISION') if llm_analysis else 'NO_RESPONSE'
-                    print(f"   ⚠️ {symbol}: لا يوجد قرار تداول من النموذج الضخم ({action})")
-                
             except Exception as e:
-                print(f"❌ خطأ في تحليل النموذج الضخم لـ {candidate.get('symbol')}: {e}")
                 continue
         
-        if not final_opportunities:
-            print("❌ لم يتم العثور على فرص تداول مناسبة")
+        # ترتيب المرشحين حسب الدرجة النهائية
+        candidates.sort(key=lambda x: x.get('layer1_score', 0), reverse=True)
+        return candidates
+
+    async def _get_detailed_symbol_data(self, symbol: str) -> Dict[str, Any]:
+        """جلب بيانات تفصيلية للرمز"""
+        try:
+            ticker = self.exchange.fetch_ticker(symbol)
+            if not ticker:
+                return None
+            
+            current_price = ticker.get('last', 0)
+            high_24h = ticker.get('high', 0)
+            low_24h = ticker.get('low', 0)
+            open_price = ticker.get('open', 0)
+            price_change_24h = (ticker.get('percentage', 0) or 0) * 100
+            
+            # حساب المؤشرات المتقدمة
+            volatility = self._calculate_volatility(high_24h, low_24h, current_price)
+            price_strength = self._calculate_price_strength(current_price, open_price, price_change_24h)
+            momentum = self._calculate_momentum(price_change_24h)
+            
+            return {
+                'price_change_24h': price_change_24h,
+                'high_24h': high_24h,
+                'low_24h': low_24h,
+                'open_price': open_price,
+                'volatility': volatility,
+                'price_strength': price_strength,
+                'momentum': momentum,
+                'reasons': []
+            }
+            
+        except Exception as e:
             return None
+
+    def _calculate_advanced_score(self, symbol_data: Dict[str, Any]) -> float:
+        """حساب درجة متقدمة تجمع بين الحجم والمؤشرات الأخرى"""
+        dollar_volume = symbol_data.get('dollar_volume', 0)
+        price_change = symbol_data.get('price_change_24h', 0)
+        volatility = symbol_data.get('volatility', 0)
+        price_strength = symbol_data.get('price_strength', 0)
+        momentum = symbol_data.get('momentum', 0)
+        
+        # 1. درجة الحجم (40%) - الأهم
+        volume_score = self._calculate_volume_score(dollar_volume)
+        
+        # 2. درجة الزخم (25%)
+        momentum_score = momentum
+        
+        # 3. درجة التقلب (20%)
+        volatility_score = self._calculate_volatility_score(volatility)
+        
+        # 4. درجة قوة السعر (15%)
+        strength_score = price_strength
+        
+        # الدرجة النهائية
+        final_score = (
+            volume_score * 0.40 +
+            momentum_score * 0.25 +
+            volatility_score * 0.20 +
+            strength_score * 0.15
+        )
+        
+        # تحديث أسباب الترشيح
+        reasons = []
+        if volume_score >= 0.7:
+            reasons.append('high_volume')
+        if momentum_score >= 0.7:
+            reasons.append('strong_momentum')
+        if volatility_score >= 0.7:
+            reasons.append('good_volatility')
+            
+        symbol_data['reasons'] = reasons
+        
+        return final_score
+
+    def _calculate_volume_score(self, dollar_volume: float) -> float:
+        """حساب درجة الحجم"""
+        if dollar_volume >= 10000000:    # 10M+
+            return 1.0
+        elif dollar_volume >= 5000000:   # 5M+
+            return 0.9
+        elif dollar_volume >= 2000000:   # 2M+
+            return 0.8
+        elif dollar_volume >= 1000000:   # 1M+
+            return 0.7
+        elif dollar_volume >= 500000:    # 500K+
+            return 0.6
+        elif dollar_volume >= 250000:    # 250K+
+            return 0.5
+        elif dollar_volume >= 100000:    # 100K+
+            return 0.4
+        else:
+            return 0.3
+
+    def _calculate_volatility(self, high_24h: float, low_24h: float, current_price: float) -> float:
+        """حساب التقلب"""
+        if current_price == 0:
+            return 0
+        return (high_24h - low_24h) / current_price
+
+    def _calculate_volatility_score(self, volatility: float) -> float:
+        """حساب درجة التقلب"""
+        if 0.02 <= volatility <= 0.15:   # تقلب مثالي 2%-15%
+            return 1.0
+        elif 0.01 <= volatility <= 0.20: # مقبول 1%-20%
+            return 0.8
+        elif volatility <= 0.01:         # قليل جداً
+            return 0.4
+        elif volatility > 0.20:          # عالي جداً
+            return 0.3
+        else:
+            return 0.5
+
+    def _calculate_price_strength(self, current_price: float, open_price: float, price_change: float) -> float:
+        """حساب قوة السعر"""
+        if open_price == 0:
+            return 0.5
+        
+        # قوة السعر تعتمد على المسافة من سعر الافتتاح ونسبة التغير
+        distance_from_open = abs(current_price - open_price) / open_price
+        change_strength = min(abs(price_change) / 50, 1.0)
+        
+        return (distance_from_open * 0.6 + change_strength * 0.4)
+
+    def _calculate_momentum(self, price_change: float) -> float:
+        """حساب الزخم"""
+        if price_change >= 15:           # +15%+
+            return 1.0
+        elif price_change >= 10:         # +10%+
+            return 0.9
+        elif price_change >= 5:          # +5%+
+            return 0.8
+        elif price_change >= 2:          # +2%+
+            return 0.7
+        elif price_change >= 0:          # موجب
+            return 0.6
+        elif price_change >= -5:         # حتى -5%
+            return 0.5
+        elif price_change >= -10:        # حتى -10%
+            return 0.4
+        else:                            # أكثر من -10%
+            return 0.3
+
+    async def get_ohlcv_data_for_symbols(self, symbols: List[str]) -> List[Dict[str, Any]]:
+        """
+        جلب بيانات OHLCV كاملة للرموز المحددة مع جميع الإطارات الزمنية بشكل متوازي
+        """
+        print(f"📊 جلب بيانات OHLCV كاملة لـ {len(symbols)} عملة بشكل متوازي...")
+        
+        # تقسيم الرموز إلى دفعات لتجنب rate limits
+        batch_size = 15  # تقليل حجم الدفعة لتحسين الاستقرار
+        batches = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)]
+        
+        all_results = []
+        
+        for batch_num, batch in enumerate(batches):
+            print(f"   🔄 معالجة الدفعة {batch_num + 1}/{len(batches)} ({len(batch)} عملة)...")
+            
+            # إنشاء مهام للدفعة الحالية بشكل متوازي
+            batch_tasks = []
+            for symbol in batch:
+                task = asyncio.create_task(self._fetch_complete_ohlcv_parallel(symbol))
+                batch_tasks.append(task)
+            
+            # انتظار انتهاء جميع مهام الدفعة الحالية
+            batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
+            
+            # معالجة نتائج الدفعة
+            successful_count = 0
+            for i, result in enumerate(batch_results):
+                symbol = batch[i]
+                if isinstance(result, Exception):
+                    print(f"      ❌ فشل جلب بيانات {symbol}")
+                elif result is not None:
+                    all_results.append(result)
+                    successful_count += 1
+                    timeframes = result.get('successful_timeframes', 0)
+                    print(f"      ✅ {symbol}: {timeframes}/6 أطر زمنية")
+                else:
+                    print(f"      ❌ فشل جلب بيانات {symbol}: لا توجد بيانات كافية")
         
-        # ترتيب الفرص النهائية حسب الثقة والدرجة
-        final_opportunities.sort(key=lambda x: (x['llm_confidence'] + x['enhanced_score']) / 2, reverse=True)
-        
-        print(f"\n🏆 النظام الطبقي اكتمل: {len(final_opportunities)} فرصة تداول")
-        for i, opportunity in enumerate(final_opportunities[:5]):
-            print(f"   {i+1}. {opportunity['symbol']}: {opportunity['decision'].get('action')} - ثقة: {opportunity['llm_confidence']:.2f} - أطر: {opportunity['timeframes_count']}")
-        
-        return final_opportunities[0] if final_opportunities else None
+            print(f"      ✅ اكتملت الدفعة {batch_num + 1}: {successful_count}/{len(batch)} ناجحة")
+            
+            # انتظار قصير بين الدفعات لتجنب rate limits
+            if batch_num < len(batches) - 1:
+                await asyncio.sleep(1)  # زيادة وقت الانتظار
         
-    except Exception as error:
-        print(f"❌ خطأ في النظام الطبقي: {error}")
-        import traceback
-        traceback.print_exc()
-        return None
+        print(f"✅ تم تجميع بيانات OHLCV كاملة لـ {len(all_results)} عملة من أصل {len(symbols)}")
+        return all_results
 
-async def re_analyze_open_trade_async(trade_data):
-    """إعادة تحليل الصفقة المفتوحة"""
-    symbol = trade_data.get('symbol')
-    try:
-        async with state_manager.trade_analysis_lock:
-            # جلب البيانات الحالية
-            market_context = await data_manager_global.get_market_context_async()
-            ohlcv_data_list = await data_manager_global.get_ohlcv_data_for_symbols([symbol])
+    async def _fetch_complete_ohlcv_parallel(self, symbol: str) -> Dict[str, Any]:
+        """جلب بيانات OHLCV كاملة مع جميع الإطارات الزمنية بشكل متوازي"""
+        try:
+            ohlcv_data = {}
             
-            if not ohlcv_data_list:
-                return None
-                
-            ohlcv_data = ohlcv_data_list[0]
-            ohlcv_data['reasons_for_candidacy'] = ['re-analysis']
+            # جلب 200 شمعة لكل إطار زمني مع تحسين التعامل مع الأخطاء
+            timeframes = [
+                ('5m', 200),
+                ('15m', 200),  
+                ('1h', 200),
+                ('4h', 200),
+                ('1d', 200),
+                ('1w', 200),
+            ]
             
-            # استخدام ML للتحليل
-            ml_processor = MLProcessor(market_context, data_manager_global, learning_engine_global)
-            processed_data = await ml_processor.process_and_score_symbol_enhanced(ohlcv_data)
+            # إنشاء مهام لجميع الإطارات الزمنية بشكل متوازي
+            timeframe_tasks = []
+            for timeframe, limit in timeframes:
+                task = asyncio.create_task(self._fetch_single_timeframe_improved(symbol, timeframe, limit))
+                timeframe_tasks.append(task)
             
-            if not processed_data:
-                return None
+            # انتظار جميع المهام
+            timeframe_results = await asyncio.gather(*timeframe_tasks, return_exceptions=True)
             
-            # استخدام LLM لإعادة التحليل
-            re_analysis_decision = await llm_service_global.re_analyze_trade_async(trade_data, processed_data)
+            # تحسين: قبول البيانات حتى لو كانت غير مكتملة
+            successful_timeframes = 0
+            min_required_timeframes = 2  # تخفيف الشرط من 3 إلى 2 أطر زمنية
             
-            # ✅ التحقق من وجود قرار صالح من النموذج
-            if re_analysis_decision:
-                await r2_service_global.save_system_logs_async({
-                    "trade_reanalyzed": True, 
-                    "symbol": symbol, 
-                    "action": re_analysis_decision.get('action'),
-                    "strategy": re_analysis_decision.get('strategy', 'GENERIC')
-                })
+            # معالجة النتائج
+            for i, (timeframe, limit) in enumerate(timeframes):
+                result = timeframe_results[i]
                 
-                return {
-                    "symbol": symbol, 
-                    "decision": re_analysis_decision,
-                    "current_price": processed_data.get('current_price')
-                }
-            else:
-                print(f"⚠️ لا يوجد قرار إعادة تحليل لـ {symbol}")
-                return None
+                if isinstance(result, Exception):
+                    continue
                 
-    except Exception as error:
-        print(f"❌ Error during trade re-analysis: {error}")
-        await r2_service_global.save_system_logs_async({
-            "reanalysis_error": True, 
-            "symbol": symbol, 
-            "error": str(error)
-        })
-        return None
-
-async def run_bot_cycle_async():
-    """دورة التداول الرئيسية"""
-    try:
-        if not await state_manager.wait_for_initialization():
-            print("❌ الخدمات غير مهيأة بالكامل - تخطي الدورة")
-            return
-
-        print("🔄 بدء دورة التداول...")
-        await r2_service_global.save_system_logs_async({"cycle_started": True})
-        
-        if not r2_service_global.acquire_lock(): 
-            print("❌ فشل الحصول على القفل - تخطي الدورة")
-            return
+                if result and len(result) >= 10:  # تخفيف الشرط من 50 إلى 10 شموع
+                    ohlcv_data[timeframe] = result
+                    successful_timeframes += 1
             
-        try:
-            open_trades = await trade_manager_global.get_open_trades()
-            print(f"📋 الصفقات المفتوحة: {len(open_trades)}")
-            
-            should_look_for_new_trade = len(open_trades) == 0
-
-            # إعادة تحليل الصفقات المفتوحة
-            if open_trades:
-                now = datetime.now()
-                trades_to_reanalyze = [
-                    trade for trade in open_trades 
-                    if now >= datetime.fromisoformat(trade.get('expected_target_time', now.isoformat()))
-                ]
-                
-                if trades_to_reanalyze:
-                    print(f"🔄 إعادة تحليل {len(trades_to_reanalyze)} صفقة")
-                    for trade in trades_to_reanalyze:
-                        result = await re_analyze_open_trade_async(trade)
-                        if result and result['decision'].get('action') == "CLOSE_TRADE":
-                            await trade_manager_global.close_trade(trade, result['current_price'], 'CLOSED_BY_REANALYSIS')
-                            should_look_for_new_trade = True
-                        elif result and result['decision'].get('action') == "UPDATE_TRADE": 
-                            await trade_manager_global.update_trade(trade, result['decision'])
-            
-            # البحث عن صفقات جديدة إذا لزم الأمر
-            if should_look_for_new_trade:
-                portfolio_state = await r2_service_global.get_portfolio_state_async()
-                current_capital = portfolio_state.get("current_capital_usd", 0)
-                
-                if current_capital > 1:
-                    print("🎯 البحث عن فرص تداول جديدة...")
-                    best_opportunity = await run_3_layer_analysis()
+            # تحسين: قبول العملة إذا كان لديها عدد كافٍ من الأطر الزمنية
+            if successful_timeframes >= min_required_timeframes and ohlcv_data:
+                try:
+                    ticker = self.exchange.fetch_ticker(symbol)
+                    current_price = ticker.get('last', 0) if ticker else 0
                     
-                    if best_opportunity:
-                        print(f"✅ فتح صفقة جديدة: {best_opportunity['symbol']}")
-                        await trade_manager_global.open_trade(
-                            best_opportunity['symbol'], 
-                            best_opportunity['decision'], 
-                            best_opportunity['current_price']
-                        )
-                    else:
-                        print("❌ لم يتم العثور على فرص تداول مناسبة")
-                else:
-                    print("❌ رأس المال غير كافي لفتح صفقات جديدة")
+                    result_data = {
+                        'symbol': symbol,
+                        'ohlcv': ohlcv_data,
+                        'raw_ohlcv': ohlcv_data,  # ✅ إضافة البيانات الخام مباشرة
+                        'current_price': current_price,
+                        'timestamp': datetime.now().isoformat(),
+                        'successful_timeframes': successful_timeframes
+                    }
                     
-        finally:
-            r2_service_global.release_lock()
-            await r2_service_global.save_system_logs_async({
-                "cycle_completed": True, 
-                "open_trades": len(open_trades) if 'open_trades' in locals() else 0
-            })
-            print("✅ اكتملت دورة التداول")
-            
-    except Exception as error:
-        print(f"❌ Unhandled error in main cycle: {error}")
-        await r2_service_global.save_system_logs_async({
-            "cycle_error": True, 
-            "error": str(error)
-        })
-        if r2_service_global.lock_acquired: 
-            r2_service_global.release_lock()
-
-@asynccontextmanager
-async def lifespan(application: FastAPI):
-    """إدارة دورة حياة التطبيق"""
-    global r2_service_global, data_manager_global, llm_service_global, learning_engine_global
-    global trade_manager_global, sentiment_analyzer_global, symbol_whale_monitor_global
-    
-    initialization_successful = False
-    try:
-        print("🚀 بدء تهيئة التطبيق...")
-        
-        # تهيئة الخدمات
-        r2_service_global = R2Service()
-        state_manager.set_service_initialized('r2_service')
-
-        contracts_database = await r2_service_global.load_contracts_db_async()
-
-        from whale_news_data import EnhancedWhaleMonitor
-        symbol_whale_monitor_global = EnhancedWhaleMonitor(contracts_database, r2_service_global)
-        state_manager.set_service_initialized('symbol_whale_monitor')
+                    return result_data
+                    
+                except Exception as price_error:
+                    return None
+            else:
+                return None
+                
+        except Exception as e:
+            return None
 
-        data_manager_global = DataManager(contracts_database, symbol_whale_monitor_global)
-        await data_manager_global.initialize()
-        state_manager.set_service_initialized('data_manager')
+    async def _fetch_single_timeframe_improved(self, symbol: str, timeframe: str, limit: int):
+        """جلب بيانات إطار زمني واحد مع تحسين التعامل مع الأخطاء"""
+        max_retries = 3
+        retry_delay = 2
+        
+        for attempt in range(max_retries):
+            try:
+                ohlcv_data = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
+                
+                if ohlcv_data and len(ohlcv_data) > 0:
+                    return ohlcv_data
+                else:
+                    return []
+                    
+            except Exception as e:
+                if attempt < max_retries - 1:
+                    await asyncio.sleep(retry_delay * (attempt + 1))
+                else:
+                    return []
 
-        llm_service_global = LLMService()
-        llm_service_global.r2_service = r2_service_global  # ✅ ربط R2Service مع LLMService
-        state_manager.set_service_initialized('llm_service')
+    async def get_latest_price_async(self, symbol):
+        """جلب السعر الحالي لعملة محددة"""
+        try:
+            if not self.exchange:
+                return None
 
-        sentiment_analyzer_global = SentimentAnalyzer(data_manager_global)
-        state_manager.set_service_initialized('sentiment_analyzer')
+            ticker = self.exchange.fetch_ticker(symbol)
+            current_price = ticker.get('last')
+            
+            if current_price:
+                return float(current_price)
+            else:
+                return None
+            
+        except Exception as e:
+            return None
 
-        learning_engine_global = LearningEngine(r2_service_global, data_manager_global)
-        await learning_engine_global.initialize_enhanced()
-        state_manager.set_service_initialized('learning_engine')
+    async def get_available_symbols(self):
+        """الحصول على جميع الرموز المتاحة"""
+        try:
+            if not self.exchange:
+                return []
+                
+            if not self.market_cache:
+                await self._load_markets()
+                
+            usdt_symbols = [
+                symbol for symbol in self.market_cache.keys() 
+                if symbol.endswith('/USDT') and self.market_cache[symbol].get('active', False)
+            ]
+            
+            return usdt_symbols
+            
+        except Exception as e:
+            print(f"❌ خطأ في جلب الرموز المتاحة: {e}")
+            return []
 
-        trade_manager_global = TradeManager(r2_service_global, learning_engine_global, data_manager_global)
-        state_manager.set_service_initialized('trade_manager')
+    async def validate_symbol(self, symbol):
+        """التحقق من صحة الرمز"""
+        try:
+            if not self.exchange:
+                return False
+                
+            if not self.market_cache:
+                await self._load_markets()
+                
+            return symbol in self.market_cache and self.market_cache[symbol].get('active', False)
+            
+        except Exception as e:
+            print(f"❌ خطأ في التحقق من الرمز {symbol}: {e}")
+            return False
 
-        # بدء المهام الخلفية
-        asyncio.create_task(monitor_market_async())
-        asyncio.create_task(trade_manager_global.start_trade_monitoring())
-        
-        await r2_service_global.save_system_logs_async({"application_started": True})
-        initialization_successful = True
-        print("🎯 التطبيق جاهز للعمل - نظام الطبقات 3 فعال")
-        
-        yield
-        
-    except Exception as error:
-        print(f"❌ Application startup failed: {error}")
-        if r2_service_global: 
-            await r2_service_global.save_system_logs_async({
-                "application_startup_failed": True, 
-                "error": str(error)
-            })
-        raise
-    finally: 
-        await cleanup_on_shutdown()
-
-application = FastAPI(lifespan=lifespan)
-
-@application.get("/run-cycle")
-async def run_cycle_api():
-    """تشغيل دورة التداول"""
-    if not state_manager.initialization_complete:
-        raise HTTPException(status_code=503, detail="الخدمات غير مهيأة بالكامل")
-    asyncio.create_task(run_bot_cycle_async())
-    return {"message": "Bot cycle initiated", "system": "3-Layer Analysis"}
-
-@application.get("/health")
-async def health_check():
-    """فحص صحة النظام"""
-    services_status = {
-        "status": "healthy" if state_manager.initialization_complete else "initializing",
-        "initialization_complete": state_manager.initialization_complete,
-        "services_initialized": state_manager.services_initialized,
-        "timestamp": datetime.now().isoformat(),
-        "system_architecture": "3-Layer Analysis System",
-        "layers": {
-            "layer1": "Data Manager - Rapid Screening",
-            "layer2": "ML Processor - Advanced Analysis", 
-            "layer3": "LLM Service - Deep Analysis"
-        }
-    }
-    return services_status
-
-@application.get("/analyze-market")
-async def analyze_market_api():
-    """تشغيل التحليل الطبقي فقط"""
-    if not state_manager.initialization_complete:
-        raise HTTPException(status_code=503, detail="الخدمات غير مهيأة بالكامل")
-    
-    result = await run_3_layer_analysis()
-    if result:
-        return {
-            "opportunity_found": True,
-            "symbol": result['symbol'],
-            "action": result['decision'].get('action'),
-            "confidence": result['llm_confidence'],
-            "strategy": result['strategy']
-        }
-    else:
-        return {"opportunity_found": False, "message": "No suitable opportunities found"}
-
-@application.get("/portfolio")
-async def get_portfolio_api():
-    """الحصول على حالة المحفظة"""
-    if not state_manager.initialization_complete:
-        raise HTTPException(status_code=503, detail="الخدمات غير مهيأة بالكامل")
+    # === الدوال الجديدة لدعم بيانات الحيتان ===
     
-    try:
-        portfolio_state = await r2_service_global.get_portfolio_state_async()
-        open_trades = await trade_manager_global.get_open_trades()
-        
-        return {
-            "portfolio": portfolio_state,
-            "open_trades": open_trades,
-            "timestamp": datetime.now().isoformat()
-        }
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"خطأ في جلب بيانات المحفظة: {str(e)}")
+    async def get_whale_data_for_symbol(self, symbol):
+        """جلب بيانات الحيتان لعملة محددة"""
+        try:
+            if self.whale_monitor:
+                whale_data = await self.whale_monitor.get_symbol_whale_activity(symbol)
+                return whale_data
+            else:
+                return None
+        except Exception as e:
+            return None
 
-@application.get("/system-status")
-async def get_system_status():
-    """الحصول على حالة النظام التفصيلية"""
-    monitoring_status = trade_manager_global.get_monitoring_status() if trade_manager_global else {}
-    
-    return {
-        "initialization_complete": state_manager.initialization_complete,
-        "services_initialized": state_manager.services_initialized,
-        "market_state_ok": state.MARKET_STATE_OK,
-        "monitoring_status": monitoring_status,
-        "timestamp": datetime.now().isoformat()
-    }
-
-async def cleanup_on_shutdown():
-    """تنظيف الموارد عند الإغلاق"""
-    global r2_service_global, data_manager_global, trade_manager_global, learning_engine_global
-    
-    print("🛑 Shutdown signal received. Cleaning up...")
-    
-    if trade_manager_global: 
-        trade_manager_global.stop_monitoring()
-        print("✅ Trade monitoring stopped")
-        
-    if learning_engine_global and learning_engine_global.initialized:
+    async def get_whale_trading_signal(self, symbol, whale_data, market_context):
+        """جلب إشارة التداول بناءً على بيانات الحيتان"""
         try:
-            await learning_engine_global.save_weights_to_r2()
-            await learning_engine_global.save_performance_history()
-            print("✅ Learning engine data saved")
-        except Exception as e: 
-            print(f"❌ Failed to save learning engine data: {e}")
-            
-    if data_manager_global: 
-        await data_manager_global.close()
-        print("✅ Data manager closed")
-        
-    if r2_service_global:
-        try: 
-            await r2_service_global.save_system_logs_async({"application_shutdown": True})
-            print("✅ Shutdown log saved")
-        except Exception as e: 
-            print(f"❌ Failed to save shutdown log: {e}")
-            
-        if r2_service_global.lock_acquired: 
-            r2_service_global.release_lock()
-            print("✅ R2 lock released")
-
-def signal_handler(signum, frame):
-    """معالج إشارات الإغلاق"""
-    print(f"🛑 Received signal {signum}. Initiating shutdown...")
-    asyncio.create_task(cleanup_on_shutdown())
-    sys.exit(0)
-
-# تسجيل معالجات الإشارات
-signal.signal(signal.SIGINT, signal_handler)
-signal.signal(signal.SIGTERM, signal_handler)
-
-if __name__ == "__main__": 
-    print("🚀 Starting AI Trading Bot with 3-Layer Analysis System...")
-    uvicorn.run(application, host="0.0.0.0", port=7860, log_level="info")
+            if self.whale_monitor:
+                return await self.whale_monitor.generate_whale_trading_signal(symbol, whale_data, market_context)
+            else:
+                return {
+                    'action': 'HOLD',
+                    'confidence': 0.3,
+                    'reason': 'Whale monitor not available',
+                    'source': 'whale_analysis'
+                }
+        except Exception as e:
+            return {
+                'action': 'HOLD',
+                'confidence': 0.3,
+                'reason': f'Error: {str(e)}',
+                'source': 'whale_analysis'
+            }
+
+print("✅ DataManager loaded - Parallel OHLCV Fetching System with Whale Data Support & Improved Error Handling")