Update app.py

app.py

@@ -16,7 +16,7 @@ try:
     from r2 import R2Service
     from LLM import LLMService
     from data_manager import DataManager
-    from ml_engine.processor import MLProcessor # ✅ هذا الاستيراد صحيح
+    from ml_engine.processor import MLProcessor 
     from learning_engine import LearningEngine
     from sentiment_news import SentimentAnalyzer
     from trade_manager import TradeManager
@@ -26,7 +26,8 @@ except ImportError as e:
     print(f"❌ خطأ في استيراد الوحدات: {e}")
     sys.exit(1)
 
-# المتغيرات العالمية
+# (المتغيرات العالمية و StateManager كما هي)
+# ...
 r2_service_global = None
 data_manager_global = None
 llm_service_global = None
@@ -42,25 +43,17 @@ class StateManager:
         self.initialization_complete = False
         self.initialization_error = None
         self.services_initialized = {
-            'r2_service': False,
-            'data_manager': False,
-            'llm_service': False,
-            'learning_engine': False,
-            'trade_manager': False,
-            'sentiment_analyzer': False,
+            'r2_service': False, 'data_manager': False, 'llm_service': False,
+            'learning_engine': False, 'trade_manager': False, 'sentiment_analyzer': False,
             'symbol_whale_monitor': False
         }
 
     async def wait_for_initialization(self, timeout=60):
         start_time = time.time()
         while not self.initialization_complete and (time.time() - start_time) < timeout:
-            if self.initialization_error:
-                raise Exception(f"فشل التهيئة: {self.initialization_error}")
+            if self.initialization_error: raise Exception(f"فشل التهيئة: {self.initialization_error}")
             await asyncio.sleep(2)
-        
-        if not self.initialization_complete:
-            raise Exception(f"انتهت مهلة التهيئة ({timeout} ثانية)")
-        
+        if not self.initialization_complete: raise Exception(f"انتهت مهلة التهيئة ({timeout} ثانية)")
         return self.initialization_complete
 
     def set_service_initialized(self, service_name):
@@ -75,196 +68,119 @@ class StateManager:
 
 state_manager = StateManager()
 
+# (initialize_services و monitor_market_async كما هي)
+# ...
 async def initialize_services():
     """تهيئة جميع الخدمات بشكل منفصل"""
     global r2_service_global, data_manager_global, llm_service_global
     global learning_engine_global, trade_manager_global, sentiment_analyzer_global
     global symbol_whale_monitor_global
-    
     try:
         print("🚀 بدء تهيئة الخدمات...")
-        
-        # 1. تهيئة R2Service أولاً
-        print("   🔄 تهيئة R2Service...")
-        r2_service_global = R2Service()
-        state_manager.set_service_initialized('r2_service')
-        print("   ✅ R2Service مهيأة")
-
-        # 2. تحميل قاعدة بيانات العقود
-        print("   🔄 جلب قاعدة بيانات العقود...")
-        contracts_database = await r2_service_global.load_contracts_db_async()
-        print(f"   ✅ تم تحميل {len(contracts_database)} عقد")
-
-        # 3. تهيئة مراقب الحيتان
-        print("   🔄 تهيئة مراقب الحيتان...")
+        print("   🔄 تهيئة R2Service..."); r2_service_global = R2Service(); state_manager.set_service_initialized('r2_service'); print("   ✅ R2Service مهيأة")
+        print("   🔄 جلب قاعدة بيانات العقود..."); contracts_database = await r2_service_global.load_contracts_db_async(); print(f"   ✅ تم تحميل {len(contracts_database)} عقد")
+        print("   🔄 تهيئة مراقب الحيتان..."); 
         try:
             from whale_news_data import EnhancedWhaleMonitor
             symbol_whale_monitor_global = EnhancedWhaleMonitor(contracts_database, r2_service_global)
-            state_manager.set_service_initialized('symbol_whale_monitor')
-            print("   ✅ مراقب الحيتان مهيأ")
-        except Exception as e:
-            print(f"   ⚠️ فشل تهيئة مراقب الحيتان: {e}")
-            symbol_whale_monitor_global = None
-
-        # 4. تهيئة DataManager
-        print("   🔄 تهيئة DataManager...")
-        data_manager_global = DataManager(contracts_database, symbol_whale_monitor_global)
-        await data_manager_global.initialize()
-        state_manager.set_service_initialized('data_manager')
-        print("   ✅ DataManager مهيأ")
-
-        # 5. تهيئة LLMService
-        print("   🔄 تهيئة LLMService...")
-        llm_service_global = LLMService()
-        llm_service_global.r2_service = r2_service_global
-        state_manager.set_service_initialized('llm_service')
-        print("   ✅ LLMService مهيأ")
-
-        # 6. تهيئة محلل المشاعر
-        print("   🔄 تهيئة محلل المشاعر...")
-        sentiment_analyzer_global = SentimentAnalyzer(data_manager_global)
-        state_manager.set_service_initialized('sentiment_analyzer')
-        print("   ✅ محلل المشاعر مهيأ")
-
-        # 7. تهيئة محرك التعلم
-        print("   🔄 تهيئة محرك التعلم...")
-        learning_engine_global = LearningEngine(r2_service_global, data_manager_global)
-        await learning_engine_global.initialize_enhanced()
-        state_manager.set_service_initialized('learning_engine')
-        print("   ✅ محرك التعلم مهيأ")
-
-        # 8. تهيئة مدير الصفقات
-        print("   🔄 تهيئة مدير الصفقات...")
-        trade_manager_global = TradeManager(r2_service_global, learning_engine_global, data_manager_global)
-        state_manager.set_service_initialized('trade_manager')
-        print("   ✅ مدير الصفقات مهيأ")
-
-        print("🎯 اكتملت تهيئة جميع الخدمات بنجاح")
-        return True
-        
-    except Exception as e:
-        error_msg = f"فشل تهيئة الخدمات: {str(e)}"
-        print(f"❌ {error_msg}")
-        state_manager.set_initialization_error(error_msg)
-        return False
+            state_manager.set_service_initialized('symbol_whale_monitor'); print("   ✅ مراقب الحيتان مهيأ")
+        except Exception as e: print(f"   ⚠️ فشل تهيئة مراقب الحيتان: {e}"); symbol_whale_monitor_global = None
+        print("   🔄 تهيئة DataManager..."); data_manager_global = DataManager(contracts_database, symbol_whale_monitor_global); await data_manager_global.initialize(); state_manager.set_service_initialized('data_manager'); print("   ✅ DataManager مهيأة")
+        print("   🔄 تهيئة LLMService..."); llm_service_global = LLMService(); llm_service_global.r2_service = r2_service_global; state_manager.set_service_initialized('llm_service'); print("   ✅ LLMService مهيأة")
+        print("   🔄 تهيئة محلل المشاعر..."); sentiment_analyzer_global = SentimentAnalyzer(data_manager_global); state_manager.set_service_initialized('sentiment_analyzer'); print("   ✅ محلل المشاعر مهيأ")
+        print("   🔄 تهيئة محرك التعلم..."); learning_engine_global = LearningEngine(r2_service_global, data_manager_global); await learning_engine_global.initialize_enhanced(); state_manager.set_service_initialized('learning_engine'); print("   ✅ محرك التعلم مهيأ")
+        print("   🔄 تهيئة مدير الصفقات..."); trade_manager_global = TradeManager(r2_service_global, learning_engine_global, data_manager_global); state_manager.set_service_initialized('trade_manager'); print("   ✅ مدير الصفقات مهيأ")
+        print("🎯 اكتملت تهيئة جميع الخدمات بنجاح"); return True
+    except Exception as e: error_msg = f"فشل تهيئة الخدمات: {str(e)}"; print(f"❌ {error_msg}"); state_manager.set_initialization_error(error_msg); return False
 
 async def monitor_market_async():
     """مراقبة السوق"""
     global data_manager_global, sentiment_analyzer_global
-    
     try:
-        if not await state_manager.wait_for_initialization():
-            print("❌ فشل تهيئة الخدمات - إيقاف مراقبة السوق")
-            return
-            
+        if not await state_manager.wait_for_initialization(): print("❌ فشل تهيئة الخدمات - إيقاف مراقبة السوق"); return
         while True:
             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
-                        
+                    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')
-                    
                     should_halt_trading, halt_reason = False, ""
-                    
-                    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})
+                    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("✅ تحسنت ظروف السوق. استئناف العمليات العادية.")
+                        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)
-    except Exception as e:
-        print(f"❌ فشل تشغيل مراقبة السوق: {e}")
+            except Exception as error: print(f"❌ خطأ أثناء مراقبة السوق: {error}"); state.MARKET_STATE_OK = True; await asyncio.sleep(60)
+    except Exception as e: print(f"❌ فشل تشغيل مراقبة السوق: {e}")
 
 
-# 🔴 --- بدء التعديل الجوهري --- 🔴
-# تم تعديل هذه الدالة لإضافة Semaphore داخلي
-# هذا للحد من عدد العمليات المتزامنة *داخل* الدفعة الواحدة
-async def process_batch_parallel(batch, ml_processor, batch_num, total_batches):
+# 🔴 --- تعديل: إزالة Semaphore من هنا --- 🔴
+# منظم السرعة الآن داخل مهمة جلب الحيتان ومهمة معالجة ML
+async def process_batch_parallel(batch, ml_processor, batch_num, total_batches, preloaded_whale_data):
     """
     (معدلة) معالجة دفعة من الرموز بشكل متوازي وإرجاع نتائج مفصلة
-    مع إضافة منظم سرعة (Semaphore) للتحكم في الضغط على الشبكة
+    - تستخدم بيانات الحيتان المحملة مسبقًا
+    - لا تحتوي على Semaphore خاص بها (يعتمد على Semaphore داخل MLProcessor)
     """
-    
-    # 🔴 تحديد 5 عمليات متزامنة كحد أقصى (من الـ 15 في الدفعة)
-    # هذا هو مفتاح الحل لمنع انهيار الشبكة
-    CONCURRENT_ML_TASKS = 5
-    semaphore = asyncio.Semaphore(CONCURRENT_ML_TASKS)
-    
-    tasks_results = []
-    
-    async def process_symbol_with_semaphore(symbol_data):
-        """دالة داخلية لاستهدافها بالـ Semaphore"""
-        async with semaphore:
-            try:
-                # هذه هي المهمة التي تستهلك الشبكة (جلب بيانات الحيتان)
-                return await ml_processor.process_and_score_symbol_enhanced(symbol_data)
-            except Exception as e:
-                # إرجاع الخطأ ليتم تسجيله
-                return e
-    
     try:
-        print(f"   🔄 [المستهلك] بدء معالجة الدفعة {batch_num}/{total_batches} ({len(batch)} عملة)... (بحد أقصى {CONCURRENT_ML_TASKS} معاً)")
+        # Reduced logging for clarity
+        # print(f"   🔄 [المستهلك] بدء معالجة الدفعة {batch_num}/{total_batches} ({len(batch)} عملة)...") 
         
         batch_tasks = []
         for symbol_data in batch:
-            # 🔴 تشغيل المهام عبر الدالة الداخلية المنظمة
-            task = asyncio.create_task(process_symbol_with_semaphore(symbol_data))
+            # 🔴 تمرير بيانات الحيتان المحملة مسبقًا
+            task = asyncio.create_task(ml_processor.process_multiple_symbols_parallel([symbol_data], preloaded_whale_data)) # Pass whale data
             batch_tasks.append(task)
         
-        # انتظار انتهاء جميع مهام الدفعة الحالية
-        batch_results = await asyncio.gather(*batch_tasks, return_exceptions=False) # return_exceptions=False لأننا نعالجها يدوياً
+        # انتظار النتائج (process_multiple_symbols_parallel يعيد قائمة)
+        batch_results_list_of_lists = await asyncio.gather(*batch_tasks, return_exceptions=True)
         
         successful_results = []
         low_score_results = []
         failed_results = []
 
-        for i, result in enumerate(batch_results):
+        # Flatten the results and handle errors
+        for i, result_list in enumerate(batch_results_list_of_lists):
             symbol = batch[i].get('symbol', 'unknown') 
+            if isinstance(result_list, Exception):
+                 failed_results.append({"symbol": symbol, "error": f"Task Execution Error: {str(result_list)}"})
+                 continue
             
-            if isinstance(result, Exception):
-                failed_results.append({"symbol": symbol, "error": f"Task Execution Error: {str(result)}"})
-            elif result is None:
-                failed_results.append({"symbol": symbol, "error": "ML.py processing returned None (Check logs for internal error)"})
-            elif result.get('enhanced_final_score', 0) > 0.4:
-                successful_results.append(result)
+            # Since we process one symbol at a time in the task, result_list should contain 0 or 1 item
+            if result_list:
+                result = result_list[0] # Get the actual result dictionary
+                if isinstance(result, dict): # Check if it's a valid dict
+                    if result.get('enhanced_final_score', 0) > 0.4:
+                        successful_results.append(result)
+                    else:
+                        low_score_results.append(result)
+                else:
+                    # Handle cases where process_multiple_symbols_parallel might return non-dict
+                     failed_results.append({"symbol": symbol, "error": f"ML processor returned invalid type: {type(result)}"})
             else:
-                low_score_results.append(result)
-        
-        print(f"   ✅ [المستهلك] اكتملت معالجة الدفعة {batch_num}: {len(successful_results)} نجاح | {len(low_score_results)} منخفض | {len(failed_results)} فشل")
+                 # Handle cases where processing returned None or empty list
+                 failed_results.append({"symbol": symbol, "error": "ML processing returned None or empty list"})
+
+
+        # print(f"   ✅ [المستهلك] اكتملت معالجة الدفعة {batch_num}: {len(successful_results)} نجاح | {len(low_score_results)} منخفض | {len(failed_results)} فشل") # Reduced logging
         
-        # إرجاع قاموس مفصل
-        return {
-            'success': successful_results,
-            'low_score': low_score_results,
-            'failures': failed_results
-        }
+        return {'success': successful_results, 'low_score': low_score_results, 'failures': failed_results}
         
     except Exception as error:
         print(f"❌ [المستهلك] خطأ في معالجة الدفعة {batch_num}: {error}")
         return {'success': [], 'low_score': [], 'failures': []}
-# 🔴 --- نهاية التعديل الجوهري --- 🔴
 
 
+# 🔴 --- بدء التعديل الجوهري --- 🔴
+# تم إعادة هيكلة هذه الدالة بالكامل
 async def run_3_layer_analysis():
     """
-    (معدلة) تشغيل النظام الطبقي (منتج/مستهلك)
+    (معدلة) تشغيل النظام الطبقي (مع فصل جلب بيانات الحيتان)
     الطبقة 1: data_manager - الفحص السريع
-    الطبقة 2: MLProcessor - التحليل المتدفق  
+    الطبقة 1.5: جلب بيانات الحيتان بشكل منفصل (غير معرقل)
+    الطبقة 2: MLProcessor - التحليل المتدفق (يستخدم بيانات الحيتان المحملة)
     الطبقة 3: LLMService - النموذج الضخم
     """
     
@@ -274,610 +190,385 @@ async def run_3_layer_analysis():
     all_failed_candidates = []
     final_layer2_candidates = []
     final_opportunities = []
+    preloaded_whale_data_dict = {} # لتخزين بيانات الحيتان
     
     try:
-        print("🎯 بدء النظام الطبقي المكون من 3 طبقات (بنموذج التدفق)...")
+        print("🎯 بدء النظام الطبقي المكون من 3 طبقات (مع فصل جلب الحيتان)...")
         
-        if not await state_manager.wait_for_initialization():
-            print("❌ الخدمات غير مهيأة بالكامل")
-            return None
+        if not await state_manager.wait_for_initialization(): print("❌ الخدمات غير مهيأة بالكامل"); return None
 
-        # الطبقة 1: الفحص السريع لجميع العملات (لا تغيير هنا)
+        # الطبقة 1: الفحص السريع (لا تغيير)
         print("\n🔍 الطبقة 1: الفحص السريع (data_manager)...")
         layer1_candidates = await data_manager_global.layer1_rapid_screening()
-        
-        if not layer1_candidates:
-            print("❌ لم يتم العثور على مرشحين في الطبقة 1")
-            return None
-        
+        if not layer1_candidates: print("❌ لم يتم العثور على مرشحين في الطبقة 1"); return None
         print(f"✅ تم اختيار {len(layer1_candidates)} عملة للطبقة 2")
+        layer1_symbols = [c['symbol'] for c in layer1_candidates]
+
+        # 🔴 --- الطبقة 1.5: جلب بيانات الحيتان بشكل منفصل وغير معرقل --- 🔴
+        start_whale_fetch = time.time()
+        print(f"\n🐋 الطبقة 1.5: بدء جلب بيانات الحيتان لـ {len(layer1_symbols)} عملة (بشكل غير معرقل)...")
+        
+        # مهمة جلب بيانات الحيتان
+        async def fetch_whale_data_task(symbols, results_dict):
+            # استخدام منظم سرعة خاص لهذه المهمة (يمكن تعديل القيمة حسب الحاجة)
+            WHALE_FETCH_CONCURRENCY = 3 # عدد طلبات الح��تان المتزامنة
+            semaphore = asyncio.Semaphore(WHALE_FETCH_CONCURRENCY)
+            tasks = []
+
+            async def get_data_with_semaphore(symbol):
+                async with semaphore:
+                    try:
+                        data = await data_manager_global.get_whale_data_for_symbol(symbol)
+                        if data:
+                           results_dict[symbol] = data
+                        # else: # Optional: Log if data is None
+                        #    print(f"   ⚠️ [Whale Fetch] No whale data returned for {symbol}")
+                    except Exception as e:
+                        print(f"   ❌ [Whale Fetch] فشل جلب بيانات الحيتان لـ {symbol}: {e}")
+                        results_dict[symbol] = {'data_available': False, 'error': str(e)} # Store error
+
+            for symbol in symbols:
+                tasks.append(asyncio.create_task(get_data_with_semaphore(symbol)))
+            
+            await asyncio.gather(*tasks) # انتظار اكتمال جميع مهام جلب الحيتان
 
-        # 🔴 --- إعداد نموذج المنتج والمستهلك --- 🔴
+        # تشغيل مهمة جلب الحيتان في الخلفية (لا ننتظرها هنا)
+        whale_fetcher_task = asyncio.create_task(fetch_whale_data_task(layer1_symbols, preloaded_whale_data_dict))
+        print("   ⏳ مهمة جلب بيانات الحيتان تعمل في الخلفية...")
+        # 🔴 --- نهاية الطبقة 1.5 --- 🔴
 
-        # 1. إنشاء الطابور (Queue)
+        # --- إعداد نموذج المنتج والمستهلك لبيانات OHLCV والتحليل ---
         DATA_QUEUE_MAX_SIZE = 2
         ohlcv_data_queue = asyncio.Queue(maxsize=DATA_QUEUE_MAX_SIZE)
-        
-        # قائمة لتجميع كل النتائج النهائية من المستهلك
         ml_results_list = []
-
-        # 2. إعداد المستهلك (MLProcessor)
         market_context = await data_manager_global.get_market_context_async()
         ml_processor = MLProcessor(market_context, data_manager_global, learning_engine_global)
-        
-        # حساب إجمالي عدد الدفعات للمستهلك
-        batch_size = 15 # يجب أن يتطابق هذا مع حجم الدفعة في data_manager
+        batch_size = 15 
         total_batches = (len(layer1_candidates) + batch_size - 1) // batch_size
-        
-        print(f"   🚀 إعداد المنتج/المستهلك: {total_batches} دفعة متوقعة (بحجم {batch_size})")
+        print(f"   🚀 إعداد المنتج/المستهلك (OHLCV/ML): {total_batches} دفعة متوقعة (بحجم {batch_size})")
 
-        # 3. تعريف وظيفة المستهلك (Consumer)
-        async def ml_consumer_task(queue: asyncio.Queue, results_list: list):
+        # وظيفة المستهلك (ML Consumer) - معدلة لتمرير بيانات الحيتان
+        async def ml_consumer_task(queue: asyncio.Queue, results_list: list, whale_data_store: dict):
             batch_num = 0
             while True:
                 try:
-                    # انتظار بيانات من المنتج
                     batch_data = await queue.get()
-                    
-                    # 🔴 إشارة التوقف (None)
-                    if batch_data is None:
-                        queue.task_done()
-                        print("   🛑 [المستهلك] تلقى إشارة التوقف.")
-                        break
+                    if batch_data is None: queue.task_done(); print("   🛑 [ML Consumer] تلقى إشارة التوقف."); break
                     
                     batch_num += 1
-                    print(f"   📬 [المستهلك] استلم الدفعة {batch_num}/{total_batches} ({len(batch_data)} عملة)")
+                    print(f"   📬 [ML Consumer] استلم دفعة OHLCV {batch_num}/{total_batches} ({len(batch_data)} عملة)")
                     
-                    # 🔴 تشغيل المعالجة المتوازية (المعدلة) للدفعة
+                    # 🔴 تمرير قاموس بيانات الحيتان إلى دالة المعالجة
                     batch_results_dict = await process_batch_parallel(
-                        batch_data, ml_processor, batch_num, total_batches
+                        batch_data, ml_processor, batch_num, total_batches, whale_data_store
                     )
                     
                     results_list.append(batch_results_dict)
-                    
-                    # إبلاغ الطابور بانتهاء معالجة هذه المهمة
                     queue.task_done()
+                    print(f"   ✅ [ML Consumer] أكمل معالجة الدفعة {batch_num}/{total_batches}") # Added completion log
                     
                 except Exception as e:
-                    print(f"❌ [المستهلك] خطأ فادح في مهمة المستهل��: {e}")
-                    traceback.print_exc()
-                    queue.task_done() # يجب استدعاؤها حتى عند الفشل
+                    print(f"❌ [ML Consumer] خطأ فادح: {e}"); traceback.print_exc(); queue.task_done() 
 
-        # 4. تشغيل مهمة المستهلك
-        print("   ▶️ [المستهلك] بدء تشغيل مهمة المستهلك...")
-        consumer_task = asyncio.create_task(ml_consumer_task(ohlcv_data_queue, ml_results_list))
+        # تشغيل المستهلك (ML Consumer)
+        print("   ▶️ [ML Consumer] بدء تشغيل مهمة المستهلك...")
+        consumer_task = asyncio.create_task(ml_consumer_task(ohlcv_data_queue, ml_results_list, preloaded_whale_data_dict))
 
-        # 5. تشغيل مهمة المنتج (Producer)
-        layer1_symbols = [candidate['symbol'] for candidate in layer1_candidates]
-        print("   ▶️ [المنتج] بدء تشغيل مهمة المنتج (تدفق بيانات OHLCV)...")
+        # تشغيل المنتج (OHLCV Producer)
+        print("   ▶️ [OHLCV Producer] بدء تشغيل مهمة المنتج (تدفق بيانات OHLCV)...")
         producer_task = asyncio.create_task(
             data_manager_global.stream_ohlcv_data(layer1_symbols, ohlcv_data_queue)
         )
         
-        # 6. انتظار انتهاء المنتج (جلب كل البيانات)
+        # انتظار انتهاء المنتج (OHLCV Producer)
         await producer_task
-        print("   ✅ [المنتج] أنهى جلب جميع البيانات.")
+        print("   ✅ [OHLCV Producer] أنهى جلب جميع بيانات OHLCV.")
 
-        # 7. إرسال إشارة التوقف للمستهلك
+        # إرسال إشارة التوقف للمستهلك (ML Consumer)
         await ohlcv_data_queue.put(None)
         
-        # 8. انتظار انتهاء المستهلك من معالجة كل شيء في الطابور
-        await ohlcv_data_queue.join() # انتظار اكتمال كل task_done()
-        await consumer_task # انتظار إغلاق مهمة المستهلك نفسها
-        print("   ✅ [المستهلك] أنهى معالجة جميع الدفعات.")
-
-        # 🔴 --- انتهاء نموذج المنتج والمستهلك --- 🔴
+        # انتظار انتهاء المستهلك (ML Consumer)
+        await ohlcv_data_queue.join() 
+        await consumer_task 
+        print("   ✅ [ML Consumer] أنهى معالجة جميع الدفعات.")
+        
+        # 🔴 انتظار انتهاء مهمة جلب بيانات الحيتان (التي كانت تعمل في الخلفية)
+        print("   ⏳ انتظار اكتمال مهمة جلب بيانات الحيتان...")
+        await whale_fetcher_task 
+        end_whale_fetch = time.time()
+        print(f"   ✅ اكتمل جلب بيانات الحيتان في {end_whale_fetch - start_whale_fetch:.2f} ثانية. تم جلب بيانات لـ {len(preloaded_whale_data_dict)} عملة.")
+        # --- نهاية نموذج المنتج والمستهلك ---
 
-        # 9. تجميع النتائج
+        # 9. تجميع النتائج (كما كان، لكن بيانات الحيتان موجودة الآن)
         print("🔄 تجميع جميع النتائج...")
+        # ... (الكود لتجميع layer2_candidates, all_low_score_candidates, all_failed_candidates كما هو) ...
         for batch_result in ml_results_list:
-            # دمج بيانات الطبقة 1 (الدرجة والأسباب) مع نتائج الطبقة 2
             for success_item in batch_result['success']:
                 symbol = success_item['symbol']
                 l1_data = next((c for c in layer1_candidates if c['symbol'] == symbol), None)
                 if l1_data:
                     success_item['reasons_for_candidacy'] = l1_data.get('reasons', [])
                     success_item['layer1_score'] = l1_data.get('layer1_score', 0)
+                # Ensure whale data from preloading is in the final result item
+                if symbol in preloaded_whale_data_dict:
+                    success_item['whale_data'] = preloaded_whale_data_dict[symbol]
+                elif 'whale_data' not in success_item: # If ML processor didn't add a default
+                     success_item['whale_data'] = {'data_available': False, 'reason': 'Failed during preload'}
                 layer2_candidates.append(success_item)
-                
             all_low_score_candidates.extend(batch_result['low_score'])
             all_failed_candidates.extend(batch_result['failures'])
-        
+
+
         print(f"✅ اكتمل التحليل المتقدم: {len(layer2_candidates)} نجاح (عالي) | {len(all_low_score_candidates)} نجاح (منخفض) | {len(all_failed_candidates)} فشل")
         
-        if not layer2_candidates:
-            print("❌ لم يتم العثور على مرشحين في الطبقة 2")
-            # استمرار لتسجيل ��لسجل
+        if not layer2_candidates: print("❌ لم يتم العثور على مرشحين في الطبقة 2")
         
-        # 10. الترتيب والفلترة (نفس المنطق السابق)
+        # 10. الترتيب والفلترة (كما كان)
         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 (الأقوى فقط)")
-        
         await r2_service_global.save_candidates_async(final_layer2_candidates)
-
         print("\n🏆 أفضل 10 عملات من الطبقة 2:")
+        # ... (كود عرض أفضل 10 كما هو، سيشمل بيانات الحيتان الآن) ...
         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)
+            mc_score = candidate.get('monte_carlo_probability')
             pattern = candidate.get('pattern_analysis', {}).get('pattern_detected', 'no_pattern')
             timeframes = candidate.get('successful_timeframes', 0)
-            
-            print(f"   {i+1}. {candidate['symbol']}:")
-            print(f"      📊 النهائي: {score:.3f} | الأطر: {timeframes}/6")
-            if mc_score > 0:
-                print(f"      🎯 مونت كارلو: {mc_score:.3f}")
+            symbol = candidate.get('symbol', 'UNKNOWN')
+
+            print(f"   {i+1}. {symbol}: 📊 {score:.3f} | الأطر: {timeframes}/6")
+            if mc_score is not None: print(f"      🎯 مونت كارلو: {mc_score:.3f}")
             print(f"      🎯 استراتيجية: {strategy} | نمط: {pattern}")
+            whale_data = candidate.get('whale_data')
+            if whale_data and whale_data.get('data_available'):
+                 signal = whale_data.get('trading_signal', {})
+                 print(f"      🐋 حيتان: {signal.get('action', 'HOLD')} (ثقة: {signal.get('confidence', 0):.2f}){' ⚠️' if signal.get('critical_alert') else ''}")
+            elif whale_data and whale_data.get('error'):
+                 print(f"      🐋 حيتان: خطأ ({whale_data.get('error')[:50]}...)")
+            # else:
+            #      print(f"      🐋 حيتان: غير متاحة")
+
 
-        # 11. الطبقة 3: التحليل بالنموذج الضخم (نفس المنطق السابق)
+        # 11. الطبقة 3: التحليل بالنموذج الضخم (كما كان)
         print("\n🧠 الطبقة 3: التحليل بالنموذج الضخم (LLMService)...")
-        
+        # ... (كود الطبقة 3 كما هو) ...
         for candidate in final_layer2_candidates:
             try:
                 symbol = candidate['symbol']
                 print(f"   🤔 تحليل {symbol} بالنموذج الضخم...")
-                
                 ohlcv_data = candidate.get('ohlcv')
-                
-                if not ohlcv_data:
-                    print(f"      ⚠️ لا توجد بيانات شموع لـ {symbol}")
-                    continue
-                
+                if not ohlcv_data: print(f"      ⚠️ لا توجد بيانات شموع لـ {symbol}"); continue
                 candidate['raw_ohlcv'] = ohlcv_data
-                
                 timeframes_count = candidate.get('successful_timeframes', 0)
                 total_candles = sum(len(data) for data in ohlcv_data.values()) if ohlcv_data else 0
-                
-                if total_candles < 30:
-                    print(f"      ⚠️ بيانات شموع غير كافية لـ {symbol}: {total_candles} شمعة فقط")
-                    continue
-                
+                if total_candles < 30: print(f"      ⚠️ بيانات شموع غير كافية لـ {symbol}: {total_candles} شمعة فقط"); continue
                 print(f"      📊 إرسال {symbol} للنموذج: {total_candles} شمعة في {timeframes_count} إطار زمني")
-                
-                llm_analysis = await llm_service_global.get_trading_decision(candidate)
-                
-                if llm_analysis and llm_analysis.get('action') in ['BUY', 'SELL']:
+                llm_analysis = await llm_service_global.get_trading_decision(candidate) # candidate now includes whale_data
+                if llm_analysis and llm_analysis.get('action') in ['BUY']: # Adjusted for SPOT only
                     opportunity = {
-                        'symbol': symbol,
-                        'current_price': candidate.get('current_price', 0),
-                        'decision': llm_analysis,
-                        'enhanced_score': candidate.get('enhanced_final_score', 0),
+                        '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
+                        'timeframes_count': timeframes_count, 'total_candles': total_candles
                     }
                     final_opportunities.append(opportunity)
-                    
                     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
-        
+            except Exception as e: print(f"❌ خطأ في تحليل النموذج الضخم لـ {candidate.get('symbol')}: {e}"); continue
+
+
         if final_opportunities:
             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']}")
         
-        # 12. سجل التدقيق (نفس المنطق السابق)
+        # 12. سجل التدقيق (كما كان)
         try:
+            # ... (كود سجل التدقيق كما هو، سيستخدم البيانات المجمعة الآن) ...
             top_10_detailed_summary = []
             for c in final_layer2_candidates:
                 whale_summary = "Not Available"
-                whale_data = c.get('whale_data')
+                whale_data = c.get('whale_data') # Already preloaded
                 if whale_data and whale_data.get('data_available'):
                     signal = whale_data.get('trading_signal', {})
                     action = signal.get('action', 'HOLD')
                     confidence = signal.get('confidence', 0)
                     reason_preview = signal.get('reason', 'N/A')[:75] + "..." if signal.get('reason') else 'N/A'
                     whale_summary = f"Action: {action}, Conf: {confidence:.2f}, Alert: {signal.get('critical_alert', False)}, Reason: {reason_preview}"
-                
+                elif whale_data and whale_data.get('error'):
+                     whale_summary = f"Error: {whale_data['error'][:50]}..."
+
                 top_10_detailed_summary.append({
-                    "symbol": c.get('symbol'),
-                    "score": c.get('enhanced_final_score', 0),
+                    "symbol": c.get('symbol'), "score": c.get('enhanced_final_score', 0),
                     "timeframes": f"{c.get('successful_timeframes', 'N/A')}/6",
-                    "whale_data_summary": whale_summary,
-                    "strategy": c.get('target_strategy', 'N/A'),
+                    "whale_data_summary": whale_summary, "strategy": c.get('target_strategy', 'N/A'),
                     "pattern": c.get('pattern_analysis', {}).get('pattern_detected', 'N/A'),
                 })
-
             other_successful_candidates = layer2_candidates[target_count:]
-            other_success_summary = [
-                {"symbol": c['symbol'], "score": c.get('enhanced_final_score', 0), "timeframes": f"{c.get('successful_timeframes', 'N/A')}/6", "whale_data": "Available" if c.get('whale_data', {}).get('data_available') else "Not Available"} 
-                for c in other_successful_candidates
-            ]
-            
-            low_score_summary = [
-                {"symbol": c['symbol'], "score": c.get('enhanced_final_score', 0), "timeframes": f"{c.get('successful_timeframes', 'N/A')}/6", "whale_data": "Available" if c.get('whale_data', {}).get('data_available') else "Not Available"} 
-                for c in all_low_score_candidates
-            ]
-
+            other_success_summary = [{"symbol": c['symbol'], "score": c.get('enhanced_final_score', 0), "timeframes": f"{c.get('successful_timeframes', 'N/A')}/6", "whale_data": "Available" if c.get('whale_data', {}).get('data_available') else ("Error" if c.get('whale_data', {}).get('error') else "Not Available")} for c in other_successful_candidates]
+            low_score_summary = [{"symbol": c['symbol'], "score": c.get('enhanced_final_score', 0), "timeframes": f"{c.get('successful_timeframes', 'N/A')}/6", "whale_data": "Available" if c.get('whale_data', {}).get('data_available') else ("Error" if c.get('whale_data', {}).get('error') else "Not Available")} for c in all_low_score_candidates]
             audit_data = {
-                "timestamp": datetime.now().isoformat(),
-                "total_layer1_candidates": len(layer1_candidates),
+                "timestamp": datetime.now().isoformat(), "total_layer1_candidates": len(layer1_candidates),
                 "total_processed_in_layer2": len(layer2_candidates) + len(all_low_score_candidates) + len(all_failed_candidates),
-                "counts": {
-                    "sent_to_llm": len(final_layer2_candidates),
-                    "success_not_top_10": len(other_successful_candidates),
-                    "success_low_score": len(all_low_score_candidates),
-                    "failures": len(all_failed_candidates)
-                },
-                "top_candidates_for_llm": top_10_detailed_summary,
-                "other_successful_candidates": other_success_summary,
-                "low_score_candidates": low_score_summary,
-                "failed_candidates": all_failed_candidates,
+                "counts": {"sent_to_llm": len(final_layer2_candidates), "success_not_top_10": len(other_successful_candidates), "success_low_score": len(all_low_score_candidates), "failures": len(all_failed_candidates)},
+                "top_candidates_for_llm": top_10_detailed_summary, "other_successful_candidates": other_success_summary,
+                "low_score_candidates": low_score_summary, "failed_candidates": all_failed_candidates,
             }
-            
             await r2_service_global.save_analysis_audit_log_async(audit_data)
             print(f"✅ تم حفظ سجل تدقيق التحليل في R2.")
-
-        except Exception as audit_error:
-            print(f"❌ فشل حفظ سجل تدقيق التحليل: {audit_error}")
-            traceback.print_exc()
-        
-        if not final_opportunities:
-            print("❌ لم يتم العثور على فرص تداول مناسبة")
-            return None
+        except Exception as audit_error: print(f"❌ فشل حفظ سجل تدقيق التحليل: {audit_error}"); traceback.print_exc()
         
+        if not final_opportunities: print("❌ لم يتم العثور على فرص تداول مناسبة"); return None
         return final_opportunities[0] if final_opportunities else None
         
     except Exception as error:
-        print(f"❌ خطأ في النظام الطبقي: {error}")
-        traceback.print_exc()
-        
-        try:
-            audit_data = {
-                "timestamp": datetime.now().isoformat(),
-                "status": "FAILED",
-                "error": str(error),
-                "traceback": traceback.format_exc(),
-                "total_layer1_candidates": len(layer1_candidates),
-                "counts": {
-                    "sent_to_llm": 0,
-                    "success_not_top_10": 0,
-                    "success_low_score": len(all_low_score_candidates),
-                    "failures": len(all_failed_candidates)
-                },
-                "failed_candidates": all_failed_candidates
-            }
+        print(f"❌ خطأ فادح في النظام الطبقي: {error}"); traceback.print_exc()
+        try: # Log partial audit on failure
+            audit_data = { "timestamp": datetime.now().isoformat(), "status": "FAILED", "error": str(error), "traceback": traceback.format_exc(), "total_layer1_candidates": len(layer1_candidates), "counts": {"sent_to_llm": 0, "success_not_top_10": 0, "success_low_score": len(all_low_score_candidates), "failures": len(all_failed_candidates)}, "failed_candidates": all_failed_candidates }
             await r2_service_global.save_analysis_audit_log_async(audit_data)
             print("⚠️ تم حفظ سجل تدقيق جزئي بعد الفشل.")
-        except Exception as audit_fail_error:
-            print(f"❌ فشل حفظ سجل التدقيق أثناء معالجة خطأ آخر: {audit_fail_error}")
-            
+        except Exception as audit_fail_error: print(f"❌ فشل حفظ سجل التدقيق أثناء معالجة خطأ آخر: {audit_fail_error}")
         return None
 # 🔴 --- نهاية التعديل الجوهري --- 🔴
 
-
+# (بقية الملف: re_analyze_open_trade_async, run_bot_cycle_async, lifespan, endpoints, cleanup, etc. كما هي)
+# ...
 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 = []
             temp_queue = asyncio.Queue()
-            # استدعاء المنتج لرمز واحد
             await data_manager_global.stream_ohlcv_data([symbol], temp_queue)
-            # استهلاك البيانات من الطابور
             while not temp_queue.empty():
-                batch = await temp_queue.get()
-                if batch:
-                    ohlcv_data_list.extend(batch)
+                batch = await temp_queue.get(); temp_queue.task_done() # Need task_done here too
+                if batch: ohlcv_data_list.extend(batch)
             
-            if not ohlcv_data_list:
-                print(f"⚠️ فشل جلب بيانات إعادة التحليل لـ {symbol}")
-                return None
-                
+            if not ohlcv_data_list: print(f"⚠️ فشل جلب بيانات إعادة التحليل لـ {symbol}"); return None
             ohlcv_data = ohlcv_data_list[0]
             
-            # دمج بيانات الطبقة 1 الوهمية لإعادة التحليل
             l1_data = await data_manager_global._get_detailed_symbol_data(symbol)
-            if l1_data:
-                ohlcv_data['reasons_for_candidacy'] = l1_data.get('reasons', ['re-analysis'])
-                ohlcv_data['layer1_score'] = l1_data.get('layer1_score', 0.5) 
+            if l1_data: ohlcv_data.update(l1_data); ohlcv_data['reasons_for_candidacy'] = ['re-analysis']
             
-            # استخدام ML للتحليل
+            # 🔴 جلب بيانات الحيتان بشكل منفصل لإعادة التحليل
+            re_analysis_whale_data = await data_manager_global.get_whale_data_for_symbol(symbol)
+
             ml_processor = MLProcessor(market_context, data_manager_global, learning_engine_global)
-            processed_data = await ml_processor.process_and_score_symbol_enhanced(ohlcv_data)
+            # 🔴 تمرير بيانات الحيتان للمعالج
+            processed_data = await ml_processor.process_and_score_symbol_enhanced(ohlcv_data, {symbol: re_analysis_whale_data} if re_analysis_whale_data else {})
             
-            if not processed_data:
-                return None
+            if not processed_data: return None
             
             processed_data['raw_ohlcv'] = ohlcv_data.get('raw_ohlcv') or ohlcv_data.get('ohlcv')
             processed_data['ohlcv'] = processed_data['raw_ohlcv']
             
-            # استخدام LLM لإعادة التحليل
             re_analysis_decision = await llm_service_global.re_analyze_trade_async(trade_data, processed_data)
             
             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')
-                })
-                
-                return {
-                    "symbol": symbol, 
-                    "decision": re_analysis_decision,
-                    "current_price": processed_data.get('current_price')
-                }
-            else:
-                return None
-                
-    except Exception as error:
-        await r2_service_global.save_system_logs_async({
-            "reanalysis_error": True, 
-            "symbol": symbol, 
-            "error": str(error)
-        })
-        return None
+                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') })
+                return {"symbol": symbol, "decision": re_analysis_decision, "current_price": processed_data.get('current_price')}
+            else: return None
+    except Exception as 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 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
         try:
-            open_trades = await trade_manager_global.get_open_trades()
-            print(f"📋 الصفقات المفتوحة: {len(open_trades)}")
-            
+            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()))
-                ]
-                
+                trades_to_reanalyze = [t for t in open_trades if now >= datetime.fromisoformat(t.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 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)
-                
+                portfolio_state = await r2_service_global.get_portfolio_state_async(); current_capital = portfolio_state.get("current_capital_usd", 0)
                 if current_capital > 1:
-                    print("🎯 البحث عن فرص تداول جديدة (بنموذج التدفق)...")
+                    print("🎯 البحث عن فرص تداول جديدة (فصل الحيتان)...")
                     best_opportunity = await run_3_layer_analysis()
-                    
-                    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("❌ رأس المال غير كافي لفتح صفقات جديدة")
-                    
+                    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("❌ رأس المال غير كافي لفتح صفقات جديدة")
         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("✅ اكتملت دورة التداول")
-            
+            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()
+        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 and r2_service_global.lock_acquired: r2_service_global.release_lock() # Ensure lock release on error
 
 @asynccontextmanager
 async def lifespan(application: FastAPI):
     """إدارة دورة حياة التطبيق"""
     print("🚀 بدء تهيئة التطبيق...")
-    
     try:
-        # تهيئة الخدمات
         success = await initialize_services()
-        if not success:
-            print("❌ فشل تهيئة التطبيق - إغلاق...")
-            yield
-            return
-
-        # بدء المهام الخلفية
+        if not success: print("❌ فشل تهيئة التطبيق - إغلاق..."); yield; return
         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})
-        print("🎯 التطبيق جاهز للعمل - نظام الطبقات 3 فعال (بنموذج التدفق)")
-        
+        print("🎯 التطبيق جاهز للعمل - نظام الطبقات 3 فعال (فصل الحيتان)")
         yield
-        
-    except Exception as error:
-        print(f"❌ Application startup failed: {error}")
-        traceback.print_exc()
-        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,
-    title="AI Trading Bot",
-    description="نظام تداول ذكي بثلاث طبقات تحليلية (بنموذج التدفق)",
-    version="3.1.0"
-)
+    except Exception as error: print(f"❌ Application startup failed: {error}"); traceback.print_exc();
+        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.get("/")
-async def root():
-    """الصفحة الرئيسية"""
-    return {
-        "message": "مرحباً بك في نظام التداول الذكي",
-        "system": "3-Layer Analysis System (Streaming Pipeline)",
-        "status": "running" if state_manager.initialization_complete else "initializing",
-        "timestamp": datetime.now().isoformat()
-    }
+application = FastAPI(lifespan=lifespan, title="AI Trading Bot", description="نظام تداول ذكي بثلاث طبقات تحليلية (فصل الحيتان)", version="3.2.0")
 
+@application.get("/")
+async def root(): return {"message": "مرحباً بك في نظام التداول الذكي", "system": "3-Layer Analysis System (Whale Fetch Decoupled)", "status": "running" if state_manager.initialization_complete else "initializing", "timestamp": datetime.now().isoformat()}
 @application.get("/run-cycle")
 async def run_cycle_api():
-    """تشغيل دورة التداول"""
-    if not state_manager.initialization_complete:
-        raise HTTPException(status_code=503, detail="الخدمات غير مهيأة بالكامل")
+    if not state_manager.initialization_complete: raise HTTPException(status_code=503, detail="الخدمات غير مهيأة بالكامل")
     asyncio.create_task(run_bot_cycle_async())
-    return {"message": "Bot cycle initiated (Streaming)", "system": "3-Layer Analysis"}
-
+    return {"message": "Bot cycle initiated (Whale Fetch Decoupled)", "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,
-        "initialization_error": state_manager.initialization_error,
-        "timestamp": datetime.now().isoformat(),
-        "system_architecture": "3-Layer Analysis System (Streaming Pipeline)",
-        "layers": {
-            "layer1": "Data Manager - Rapid Screening",
-            "layer2": "ML Processor - Advanced Analysis (Streaming Consumer)", 
-            "layer3": "LLM Service - Deep Analysis"
-        }
-    }
-    return services_status
-
+async def health_check(): return {"status": "healthy" if state_manager.initialization_complete else "initializing", "initialization_complete": state_manager.initialization_complete, "services_initialized": state_manager.services_initialized, "initialization_error": state_manager.initialization_error, "timestamp": datetime.now().isoformat(), "system_architecture": "3-Layer Analysis System (Whale Fetch Decoupled)", "layers": {"layer1": "Data Manager - Rapid Screening", "layer1.5": "Whale Data Fetcher (Async)", "layer2": "ML Processor - Streaming Consumer", "layer3": "LLM Service - Deep Analysis"}}
 @application.get("/analyze-market")
 async def analyze_market_api():
-    """تشغيل التحليل الطبقي فقط"""
-    if not state_manager.initialization_complete:
-        raise HTTPException(status_code=503, detail="الخدمات غير مهيأة بالكامل")
-    
+    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"}
-
+    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)}")
-
+    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)}")
 @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,
-        "initialization_error": state_manager.initialization_error,
-        "market_state_ok": state.MARKET_STATE_OK,
-        "monitoring_status": monitoring_status,
-        "timestamp": datetime.now().isoformat()
-    }
+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, "initialization_error": state_manager.initialization_error, "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 trade_manager_global: trade_manager_global.stop_monitoring(); print("✅ Trade monitoring stopped")
     if learning_engine_global and learning_engine_global.initialized:
-        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")
-        
+        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)
+        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")
 
-# تسجيل معالجات الإشارات
-signal.signal(signal.SIGINT, signal_handler)
-signal.signal(signal.SIGTERM, signal_handler)
+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 (Streaming)...")
-    uvicorn.run(
-        application, 
-        host="0.0.0.0", 
-        port=7860, 
-        log_level="info",
-        access_log=True
-    )
+    print("🚀 Starting AI Trading Bot with 3-Layer Analysis System (Whale Fetch Decoupled)...")
+    uvicorn.run( application, host="0.0.0.0", port=7860, log_level="info", access_log=True )