Update app.py

app.py

@@ -1,4 +1,4 @@
-# app.py (محدث بإصلاح IndentationError)
+# app.py (محدث بإضافة Timeout لجلب الحيتان وإصلاح IndentationError)
 import os
 import traceback
 import signal
@@ -82,7 +82,7 @@ async def initialize_services():
             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
-        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("   🔄 تهيئة 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("   ✅ محرك التعلم مهيأ")
@@ -117,49 +117,36 @@ async def process_batch_parallel(batch, ml_processor, batch_num, total_batches,
     """
     (معدلة) معالجة دفعة من الرموز بشكل متوازي وإرجاع نتائج مفصلة
     - تستخدم بيانات الحيتان المحملة مسبقًا
-    - لا تحتوي على Semaphore خاص بها (يعتمد على Semaphore داخل MLProcessor)
     """
     try:
-        # Reduced logging for clarity
-        # print(f"   🔄 [المستهلك] بدء معالجة الدفعة {batch_num}/{total_batches} ({len(batch)} عملة)...") 
-        
         batch_tasks = []
         for symbol_data in batch:
-            # 🔴 تمرير بيانات الحيتان المحملة مسبقًا
-            task = asyncio.create_task(ml_processor.process_multiple_symbols_parallel([symbol_data], preloaded_whale_data)) # Pass whale data
+            task = asyncio.create_task(ml_processor.process_multiple_symbols_parallel([symbol_data], preloaded_whale_data)) 
             batch_tasks.append(task)
         
-        # انتظار النتائج (process_multiple_symbols_parallel يعيد قائمة)
         batch_results_list_of_lists = await asyncio.gather(*batch_tasks, return_exceptions=True)
         
         successful_results = []
         low_score_results = []
         failed_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
             
-            # 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
+                result = result_list[0] 
+                if isinstance(result, 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:
-                 # 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}
         
@@ -183,51 +170,41 @@ async def run_3_layer_analysis():
     all_failed_candidates = []
     final_layer2_candidates = []
     final_opportunities = []
-    preloaded_whale_data_dict = {} # لتخزين بيانات الحيتان
+    preloaded_whale_data_dict = {} 
     
     try:
         print("🎯 بدء النظام الطبقي المكون من 3 طبقات (مع فصل جلب الحيتان)...")
         
         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
         print(f"✅ تم اختيار {len(layer1_candidates)} عملة للطبقة 2")
         layer1_symbols = [c['symbol'] for c in layer1_candidates]
 
-        # 🔴 --- الطبقة 1.5: جلب بيانات الحيتان بشكل منفصل وغير معرقل --- 🔴
+        # الطبقة 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
+                        if data: results_dict[symbol] = data
                     except Exception as e:
                         print(f"   ❌ [Whale Fetch] فشل جلب بيانات الحيتان لـ {symbol}: {e}")
                         results_dict[symbol] = {'data_available': False, 'error': str(e)} 
-
-            for symbol in symbols:
-                tasks.append(asyncio.create_task(get_data_with_semaphore(symbol)))
-            
+            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 --- 🔴
 
-        # --- إعداد نموذج المنتج والمستهلك لبيانات OHLCV والتحليل ---
+        # إعداد المنتج/المستهلك (OHLCV/ML)
         DATA_QUEUE_MAX_SIZE = 2
         ohlcv_data_queue = asyncio.Queue(maxsize=DATA_QUEUE_MAX_SIZE)
         ml_results_list = []
@@ -237,59 +214,50 @@ async def run_3_layer_analysis():
         total_batches = (len(layer1_candidates) + batch_size - 1) // batch_size
         print(f"   🚀 إعداد المنتج/المستهلك (OHLCV/ML): {total_batches} دفعة متوقعة (بحجم {batch_size})")
 
-        # وظيفة المستهلك (ML Consumer) - معدلة لتمرير بيانات الحيتان
+        # وظيفة المستهلك (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()
                     if batch_data is None: queue.task_done(); print("   🛑 [ML Consumer] تلقى إشارة التوقف."); break
-                    
                     batch_num += 1
                     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, 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"❌ [ML Consumer] خطأ فادح: {e}"); traceback.print_exc(); queue.task_done() 
+                    print(f"   ✅ [ML Consumer] أكمل معالجة ال��فعة {batch_num}/{total_batches}") 
+                except Exception as e: print(f"❌ [ML Consumer] خطأ فادح: {e}"); traceback.print_exc(); queue.task_done() 
 
-        # تشغيل المستهلك (ML Consumer)
+        # تشغيل المستهلك (ML Consumer) والمنتج (OHLCV Producer)
         print("   ▶️ [ML Consumer] بدء تشغيل مهمة المستهلك...")
         consumer_task = asyncio.create_task(ml_consumer_task(ohlcv_data_queue, ml_results_list, preloaded_whale_data_dict))
-
-        # تشغيل المنتج (OHLCV Producer)
         print("   ▶️ [OHLCV Producer] بدء تشغيل مهمة المنتج (تدفق بيانات OHLCV)...")
-        producer_task = asyncio.create_task(
-            data_manager_global.stream_ohlcv_data(layer1_symbols, ohlcv_data_queue)
-        )
+        producer_task = asyncio.create_task(data_manager_global.stream_ohlcv_data(layer1_symbols, ohlcv_data_queue))
         
-        # انتظار انتهاء المنتج (OHLCV Producer)
-        await producer_task
-        print("   ✅ [OHLCV Producer] أنهى جلب جميع بيانات OHLCV.")
-
-        # إرسال إشارة التوقف للمستهلك (ML Consumer)
-        await ohlcv_data_queue.put(None)
-        
-        # انتظار انتهاء المستهلك (ML Consumer)
+        # انتظار انتهاء المنتج والمستهلك
+        await producer_task; print("   ✅ [OHLCV Producer] أنهى جلب جميع بيانات OHLCV.")
+        await ohlcv_data_queue.put(None) 
         await ohlcv_data_queue.join() 
-        await consumer_task 
-        print("   ✅ [ML Consumer] أنهى معالجة جميع الدفعات.")
+        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. تجميع النتائج (كما كان، لكن بيانات الحيتان موجودة الآن)
+        # انتظار اكتمال مهمة جلب الحيتان (مع Timeout)
+        print("   ⏳ انتظار اكتمال مهمة جلب بيانات الحيتان (بحد أقصى للمهلة)...")
+        WHALE_FETCH_TIMEOUT_SECONDS = 180 
+        try:
+            await asyncio.wait_for(whale_fetcher_task, timeout=WHALE_FETCH_TIMEOUT_SECONDS)
+            end_whale_fetch = time.time()
+            print(f"   ✅ اكتمل جلب بيانات الحيتان في {end_whale_fetch - start_whale_fetch:.2f} ثانية. تم جلب/محاولة جلب بيانات لـ {len(preloaded_whale_data_dict)} عملة.")
+        except asyncio.TimeoutError:
+            end_whale_fetch = time.time()
+            print(f"   ⚠️ انتهت مهلة انتظار جلب بيانات الحيتان ({WHALE_FETCH_TIMEOUT_SECONDS} ثانية)! تم جلب/محاولة جلب بيانات لـ {len(preloaded_whale_data_dict)} عملة حتى الآن.")
+        except Exception as whale_task_err:
+             end_whale_fetch = time.time()
+             print(f"   ❌ حدث خطأ غير متوقع أثناء انتظار مهمة جلب الحيتان: {whale_task_err}")
+
+        # تجميع النتائج
         print("🔄 تجميع جميع النتائج...")
         for batch_result in ml_results_list:
             for success_item in batch_result['success']:
@@ -298,19 +266,16 @@ async def run_3_layer_analysis():
                 if l1_data:
                     success_item['reasons_for_candidacy'] = l1_data.get('reasons', [])
                     success_item['layer1_score'] = l1_data.get('layer1_score', 0)
-                if symbol in preloaded_whale_data_dict:
-                    success_item['whale_data'] = preloaded_whale_data_dict[symbol]
-                elif 'whale_data' not in success_item: 
-                     success_item['whale_data'] = {'data_available': False, 'reason': 'Failed during preload'}
+                if symbol in preloaded_whale_data_dict: success_item['whale_data'] = preloaded_whale_data_dict[symbol]
+                elif 'whale_data' not in success_item: success_item['whale_data'] = {'data_available': False, 'reason': 'Fetch timed out or failed'}
                 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")
         
-        # 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]
@@ -318,90 +283,50 @@ async def run_3_layer_analysis():
         await r2_service_global.save_candidates_async(final_layer2_candidates)
         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')
-            pattern = candidate.get('pattern_analysis', {}).get('pattern_detected', 'no_pattern')
-            timeframes = candidate.get('successful_timeframes', 0)
-            symbol = candidate.get('symbol', 'UNKNOWN')
-
+            score=candidate.get('enhanced_final_score',0); strategy=candidate.get('target_strategy','GENERIC'); mc_score=candidate.get('monte_carlo_probability'); pattern=candidate.get('pattern_analysis',{}).get('pattern_detected','no_pattern'); timeframes=candidate.get('successful_timeframes',0); 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]}...)")
+            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]}...)")
 
-        # 11. الطبقة 3: التحليل بالنموذج الضخم (كما كان)
+        # الطبقة 3
         print("\n🧠 الطبقة 3: التحليل بالنموذج الضخم (LLMService)...")
         for candidate in final_layer2_candidates:
             try:
-                symbol = candidate['symbol']
-                print(f"   🤔 تحليل {symbol} بالنموذج الضخم...")
-                ohlcv_data = candidate.get('ohlcv')
+                symbol = candidate['symbol']; print(f"   🤔 تحليل {symbol} بالنموذج الضخم...")
+                ohlcv_data = candidate.get('ohlcv'); 
                 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
+                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
                 print(f"      📊 إرسال {symbol} للنموذج: {total_candles} شمعة في {timeframes_count} إطار زمني")
-                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),
-                        '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
-                    }
+                llm_analysis = await llm_service_global.get_trading_decision(candidate) 
+                if llm_analysis and llm_analysis.get('action') in ['BUY']: 
+                    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)
                     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})")
+                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 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']}")
+            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. سجل التدقيق (كما كان)
+        # سجل التدقيق
         try:
             top_10_detailed_summary = []
             for c in final_layer2_candidates:
-                whale_summary = "Not Available"
-                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),
-                    "timeframes": f"{c.get('successful_timeframes', 'N/A')}/6",
-                    "whale_data_summary": whale_summary, "strategy": c.get('target_strategy', 'N/A'),
-                    "pattern": c.get('pattern_analysis', {}).get('pattern_detected', 'N/A'),
-                })
+                whale_summary = "Not Available"; whale_data = c.get('whale_data') 
+                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), "timeframes": f"{c.get('successful_timeframes', 'N/A')}/6", "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 ("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),
-                "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,
-            }
+            audit_data = { "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, }
             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()
@@ -427,21 +352,15 @@ async def re_analyze_open_trade_async(trade_data):
             ohlcv_data_list = []
             temp_queue = asyncio.Queue()
             await data_manager_global.stream_ohlcv_data([symbol], temp_queue)
-            # Correctly drain the queue
             while True:
                 try:
-                    batch = await asyncio.wait_for(temp_queue.get(), timeout=1.0) # Add timeout
-                    if batch is None: # Check for sentinel
-                         temp_queue.task_done()
-                         break
+                    batch = await asyncio.wait_for(temp_queue.get(), timeout=1.0) 
+                    if batch is None: temp_queue.task_done(); break
                     ohlcv_data_list.extend(batch)
                     temp_queue.task_done()
                 except asyncio.TimeoutError:
-                    # Queue might be empty or producer finished without sentinel
-                    if temp_queue.empty(): break # Exit if empty after timeout
-                except Exception as q_err:
-                     print(f"Error draining queue for re-analysis: {q_err}")
-                     break # Exit on other errors
+                    if temp_queue.empty(): break 
+                except Exception as q_err: print(f"Error draining queue for re-analysis: {q_err}"); break 
 
             if not ohlcv_data_list: print(f"⚠️ فشل جلب بيانات إعادة التحليل لـ {symbol}"); return None
             ohlcv_data = ohlcv_data_list[0]
@@ -475,7 +394,7 @@ async def run_bot_cycle_async():
         print("🔄 بدء دورة التداول..."); await r2_service_global.save_system_logs_async({"cycle_started": True})
         if not r2_service_global.acquire_lock(): print("❌ فشل الحصول على القفل - تخطي الدورة"); return
         
-        open_trades = [] # Define open_trades before try block
+        open_trades = [] 
         try:
             open_trades = await trade_manager_global.get_open_trades(); print(f"📋 الصفقات المفتوحة: {len(open_trades)}")
             should_look_for_new_trade = len(open_trades) == 0
@@ -484,26 +403,15 @@ async def run_bot_cycle_async():
                 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)} صفقة")
-                    # Use gather for concurrent re-analysis, but process results sequentially
                     reanalysis_results = await asyncio.gather(*[re_analyze_open_trade_async(trade) for trade in trades_to_reanalyze], return_exceptions=True)
-
                     for i, result in enumerate(reanalysis_results):
-                         trade = trades_to_reanalyze[i] # Get corresponding trade
-                         if isinstance(result, Exception):
-                              print(f"   ❌ فشل إعادة تحليل {trade.get('symbol')}: {result}")
-                         elif result and result['decision'].get('action') == "CLOSE_TRADE": 
-                              print(f"   ✅ إغلاق {trade.get('symbol')} بناءً على إعادة التحليل.")
-                              await trade_manager_global.close_trade(trade, result['current_price'], 'CLOSED_BY_REANALYSIS'); 
-                              should_look_for_new_trade = True # Now we can look for a new trade
-                         elif result and result['decision'].get('action') == "UPDATE_TRADE": 
-                              print(f"   ✅ تحديث {trade.get('symbol')} بناءً على إعادة التحليل.")
-                              await trade_manager_global.update_trade(trade, result['decision'])
-                         elif result: # Handle HOLD case (log or do nothing)
-                              print(f"   ℹ️ الاحتفاظ بـ {trade.get('symbol')} بناءً على إعادة التحليل.")
-                         else: # Handle case where re-analysis returned None without error
-                              print(f"   ⚠️ إعادة تحليل {trade.get('symbol')} لم تنتج قرارًا.")
+                         trade = trades_to_reanalyze[i] 
+                         if isinstance(result, Exception): print(f"   ❌ فشل إعادة تحليل {trade.get('symbol')}: {result}")
+                         elif result and result['decision'].get('action') == "CLOSE_TRADE": print(f"   ✅ إغلاق {trade.get('symbol')} بناءً على إعادة التحليل."); await trade_manager_global.close_trade(trade, result['current_price'], 'CLOSED_BY_REANALYSIS'); 
+                         elif result and result['decision'].get('action') == "UPDATE_TRADE": print(f"   ✅ تحديث {trade.get('symbol')} بناءً على إعادة التحليل."); await trade_manager_global.update_trade(trade, result['decision'])
+                         elif result: print(f"   ℹ️ الاحتفاظ بـ {trade.get('symbol')} بناءً على إعادة التحليل.")
+                         else: print(f"   ⚠️ إعادة تحليل {trade.get('symbol')} لم تنتج قرارًا.")
             
-            # Recalculate should_look_for_new_trade after potential closures
             current_open_trades_count = len(await trade_manager_global.get_open_trades())
             should_look_for_new_trade = current_open_trades_count == 0
 
@@ -515,19 +423,16 @@ async def run_bot_cycle_async():
                     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("❌ رأس المال غير كافي لفتح صفقات جديدة")
-            else:
-                 print("ℹ️ يوجد صفقة مفتوحة بالفعل، تخطي البحث عن صفقة جديدة.")
-
+            else: print("ℹ️ يوجد صفقة مفتوحة بالفعل، تخطي البحث عن صفقة جديدة.")
         finally:
-            if r2_service_global.lock_acquired: # Check if lock was acquired before releasing
-                r2_service_global.release_lock()
-            await r2_service_global.save_system_logs_async({ "cycle_completed": True, "open_trades": len(open_trades)}) # Use initial count for log consistency
+            if r2_service_global.lock_acquired: r2_service_global.release_lock()
+            await r2_service_global.save_system_logs_async({ "cycle_completed": True, "open_trades": len(open_trades)}) 
             print("✅ اكتملت دورة التداول")
             
     except Exception as error:
-        print(f"❌ Unhandled error in main cycle: {error}"); traceback.print_exc() # Print traceback
+        print(f"❌ Unhandled error in main cycle: {error}"); traceback.print_exc() 
         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
+        if r2_service_global and r2_service_global.lock_acquired: r2_service_global.release_lock() 
 
 @asynccontextmanager
 async def lifespan(application: FastAPI):
@@ -543,15 +448,11 @@ async def lifespan(application: FastAPI):
         yield
     except Exception as error: 
         print(f"❌ Application startup failed: {error}"); 
-        traceback.print_exc() # Print traceback here as well
+        traceback.print_exc() 
         if r2_service_global: 
-            await r2_service_global.save_system_logs_async({ 
-                "application_startup_failed": True, 
-                "error": str(error) 
-            })
-        # 🔴 --- الإصلاح: فصل raise إلى سطر جديد --- 🔴
+            await r2_service_global.save_system_logs_async({ "application_startup_failed": True, "error": str(error) })
+        # الإصلاح: فصل raise إلى سطر جديد 
         raise # Correct indentation
-        # 🔴 --- نهاية الإصلاح --- 🔴
     finally: 
         await cleanup_on_shutdown()