Update app.py

app.py

@@ -118,7 +118,7 @@ async def initialize_services():
         llm_service_global = LLMService()
         llm_service_global.r2_service = r2_service_global
         state_manager.set_service_initialized('llm_service')
-        print("   ✅ LLMService مهيأة")
+        print("   ✅ LLMService مهيأ")
 
         # 6. تهيئة محلل المشاعر
         print("   🔄 تهيئة محلل المشاعر...")
@@ -190,22 +190,44 @@ async def monitor_market_async():
     except Exception as e:
         print(f"❌ فشل تشغيل مراقبة السوق: {e}")
 
-#
-# 🔴 هذه الدالة (المستهلك) لم تتغير، لأنها بالفعل تعالج الدفعات
-#
+
+# 🔴 --- بدء التعديل الجوهري --- 🔴
+# تم تعديل هذه الدالة لإضافة Semaphore داخلي
+# هذا للحد من عدد العمليات المتزامنة *داخل* الدفعة الواحدة
 async def process_batch_parallel(batch, ml_processor, batch_num, total_batches):
-    """معالجة دفعة من الرموز بشكل متوازي وإرجاع نتائج مفصلة"""
+    """
+    (معدلة) معالجة دفعة من الرموز بشكل متوازي وإرجاع نتائج مفصلة
+    مع إضافة منظم سرعة (Semaphore) للتحكم في الضغط على الشبكة
+    """
+    
+    # 🔴 تحديد 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)} عملة)...")
+        print(f"   🔄 [المستهلك] بدء معالجة الدفعة {batch_num}/{total_batches} ({len(batch)} عملة)... (بحد أقصى {CONCURRENT_ML_TASKS} معاً)")
         
-        # إنشاء مهام للدفعة الحالية
         batch_tasks = []
         for symbol_data in batch:
-            task = asyncio.create_task(ml_processor.process_and_score_symbol_enhanced(symbol_data))
+            # 🔴 تشغيل المهام عبر الدالة الداخلية المنظمة
+            task = asyncio.create_task(process_symbol_with_semaphore(symbol_data))
             batch_tasks.append(task)
         
         # انتظار انتهاء جميع مهام الدفعة الحالية
-        batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
+        batch_results = await asyncio.gather(*batch_tasks, return_exceptions=False) # return_exceptions=False لأننا نعالجها يدوياً
         
         successful_results = []
         low_score_results = []
@@ -235,10 +257,9 @@ async def process_batch_parallel(batch, ml_processor, batch_num, total_batches):
     except Exception as error:
         print(f"❌ [المستهلك] خطأ في معالجة الدفعة {batch_num}: {error}")
         return {'success': [], 'low_score': [], 'failures': []}
+# 🔴 --- نهاية التعديل الجوهري --- 🔴
 
 
-# 🔴 --- بدء التعديل الجوهري --- 🔴
-# تم إعادة هيكلة هذه الدالة بالكامل لاستخدام نموذج المنتج/المستهلك
 async def run_3_layer_analysis():
     """
     (معدلة) تشغيل النظام الطبقي (منتج/مستهلك)
@@ -274,8 +295,6 @@ async def run_3_layer_analysis():
         # 🔴 --- إعداد نموذج المنتج والمستهلك --- 🔴
 
         # 1. إنشاء الطابور (Queue)
-        # maxsize=2 يعني أن المنتج (جلب البيانات) سيتوقف مؤقتاً إذا كان متقدماً
-        # على المستهلك (تحليل ML) بأكثر من دفعتين. هذا يمنع استهلاك ذاكرة مفرط.
         DATA_QUEUE_MAX_SIZE = 2
         ohlcv_data_queue = asyncio.Queue(maxsize=DATA_QUEUE_MAX_SIZE)
         
@@ -309,7 +328,7 @@ async def run_3_layer_analysis():
                     batch_num += 1
                     print(f"   📬 [المستهلك] استلم الدفعة {batch_num}/{total_batches} ({len(batch_data)} عملة)")
                     
-                    # تشغيل المعالجة المتوازية للدفعة (هذا ما كان يحدث سابقاً)
+                    # 🔴 تشغيل المعالجة المتوازية (المعدلة) للدفعة
                     batch_results_dict = await process_batch_parallel(
                         batch_data, ml_processor, batch_num, total_batches
                     )
@@ -330,8 +349,6 @@ async def run_3_layer_analysis():
 
         # 5. تشغيل مهمة المنتج (Producer)
         layer1_symbols = [candidate['symbol'] for candidate in layer1_candidates]
-        # دمج بيانات الطبقة 1 مع بيانات OHLCV
-        # 🔴 (تعديل بسيط): دمج بيانات الطبقة 1 *بعد* جلب OHLCV لتبسيط التدفق
         print("   ▶️ [المنتج] بدء تشغيل مهمة المنتج (تدفق بيانات OHLCV)...")
         producer_task = asyncio.create_task(
             data_manager_global.stream_ohlcv_data(layer1_symbols, ohlcv_data_queue)
@@ -548,23 +565,29 @@ async def re_analyze_open_trade_async(trade_data):
         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():
-                ohlcv_data_list.extend(await temp_queue.get())
+                batch = await temp_queue.get()
+                if batch:
+                    ohlcv_data_list.extend(batch)
             
             if not ohlcv_data_list:
+                print(f"⚠️ فشل جلب بيانات إعادة التحليل لـ {symbol}")
                 return None
                 
             ohlcv_data = ohlcv_data_list[0]
             
-            # 🔴 دمج بيانات الطبقة 1 الوهمية لإعادة التحليل
+            # دمج بيانات الطبقة 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) # افتراض درجة متوسطة
+                ohlcv_data['layer1_score'] = l1_data.get('layer1_score', 0.5) 
             
             # استخدام ML للتحليل
             ml_processor = MLProcessor(market_context, data_manager_global, learning_engine_global)
@@ -573,14 +596,12 @@ async def re_analyze_open_trade_async(trade_data):
             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,