Update app.py

app.py

@@ -96,6 +96,35 @@ async def monitor_market_async():
             state.MARKET_STATE_OK = True
             await asyncio.sleep(60)
 
+async def process_batch_parallel(batch, ml_processor, batch_num, total_batches):
+    """معالجة دفعة من الرموز بشكل متوازي"""
+    try:
+        print(f"   🔄 معالجة الدفعة {batch_num}/{total_batches} ({len(batch)} عملة)...")
+        
+        # إنشاء مهام للدفعة الحالية
+        batch_tasks = []
+        for symbol_data in batch:
+            task = asyncio.create_task(ml_processor.process_and_score_symbol_enhanced(symbol_data))
+            batch_tasks.append(task)
+        
+        # انتظار انتهاء جميع مهام الدفعة الحالية
+        batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
+        
+        # تصفية النتائج الناجحة
+        successful_results = []
+        for result in batch_results:
+            if isinstance(result, Exception):
+                continue
+            if result and result.get('enhanced_final_score', 0) > 0.4:
+                successful_results.append(result)
+        
+        print(f"   ✅ اكتملت الدفعة {batch_num}: {len(successful_results)}/{len(batch)} ناجحة")
+        return successful_results
+        
+    except Exception as error:
+        print(f"❌ خطأ في معالجة الدفعة {batch_num}: {error}")
+        return []
+
 async def run_3_layer_analysis():
     """
     تشغيل النظام الطبقي المكون من 3 طبقات:
@@ -128,18 +157,15 @@ async def run_3_layer_analysis():
             print("❌ فشل جلب بيانات OHLCV للمرشحين")
             return None
 
-        # الطبقة 2: التحليل المتقدم بشكل متوازي
-        print("\n📈 الطبقة 2: التحليل المتقدم (MLProcessor) بشكل متوازي...")
+        # الطبقة 2: التحليل المتقدم بشكل متوازي حقيقي
+        print(f"\n📈 الطبقة 2: التحليل المتقدم (MLProcessor) بشكل متوازي لـ {len(ohlcv_data_list)} عملة...")
         market_context = await data_manager_global.get_market_context_async()
         
         # إنشاء معالج ML
         ml_processor = MLProcessor(market_context, data_manager_global, learning_engine_global)
         
-        # معالجة كل المرشحين بشكل متوازي باستخدام asyncio.gather
-        print(f"   🚀 معالجة {len(ohlcv_data_list)} عملة بشكل متوازي...")
-        
-        # تجهيز المهام
-        tasks = []
+        # تجهيز البيانات للطبقة 2
+        layer2_data = []
         for ohlcv_data in ohlcv_data_list:
             try:
                 # إضافة أسباب الترشيح من الطبقة 1
@@ -149,36 +175,34 @@ async def run_3_layer_analysis():
                     ohlcv_data['reasons_for_candidacy'] = layer1_candidate.get('reasons', [])
                     ohlcv_data['layer1_score'] = layer1_candidate.get('layer1_score', 0)
                 
-                # إنشاء مهمة للتحليل المتقدم
-                task = ml_processor.process_and_score_symbol_enhanced(ohlcv_data)
-                tasks.append(task)
+                layer2_data.append(ohlcv_data)
                     
             except Exception as e:
                 print(f"❌ خطأ في إعداد تحليل {ohlcv_data.get('symbol')}: {e}")
                 continue
         
-        # تشغيل جميع المهام بشكل متوازي
-        layer2_results = await asyncio.gather(*tasks, return_exceptions=True)
+        # تقسيم العمل إلى دفعات للمعالجة المتوازية
+        batch_size = 20  # 20 عملة في كل دفعة
+        batches = [layer2_data[i:i + batch_size] for i in range(0, len(layer2_data), batch_size)]
+        total_batches = len(batches)
         
-        # تصفية النتائج الناجحة
-        layer2_candidates = []
-        successful_analyses = 0
+        print(f"   🚀 تقسيم العمل إلى {total_batches} دفعة ({batch_size} عملة لكل دفعة)...")
         
-        for i, result in enumerate(layer2_results):
-            if isinstance(result, Exception):
-                symbol = ohlcv_data_list[i].get('symbol', 'unknown') if i < len(ohlcv_data_list) else 'unknown'
-                print(f"   ❌ فشل تحليل {symbol}: {result}")
-                continue
-            
-            if result and result.get('enhanced_final_score', 0) > 0.4:
-                layer2_candidates.append(result)
-                successful_analyses += 1
-            
-            # عرض التقدم كل 20 عملة
-            if (i + 1) % 20 == 0:
-                print(f"   ✅ تم تحليل {i + 1}/{len(ohlcv_data_list)} عملة ({successful_analyses} ناجحة)")
+        # معالجة جميع الدفعات بشكل متوازي
+        batch_tasks = []
+        for i, batch in enumerate(batches):
+            task = asyncio.create_task(process_batch_parallel(batch, ml_processor, i+1, total_batches))
+            batch_tasks.append(task)
+        
+        # جمع نتائج جميع الدفعات
+        batch_results = await asyncio.gather(*batch_tasks)
+        
+        # دمج جميع النتائج
+        layer2_candidates = []
+        for batch_result in batch_results:
+            layer2_candidates.extend(batch_result)
         
-        print(f"✅ اكتمل التحليل المتقدم: {successful_analyses}/{len(ohlcv_data_list)} عملة تم تحليلها بنجاح")
+        print(f"✅ اكتمل التحليل المتقدم: {len(layer2_candidates)}/{len(ohlcv_data_list)} عملة تم تحليلها بنجاح")
         
         if not layer2_candidates:
             print("❌ لم يتم العثور على مرشحين في الطبقة 2")