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
@@ -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("   🔄 تهيئة محلل المشاعر...")
@@ -191,12 +191,12 @@ async def monitor_market_async():
         print(f"❌ فشل تشغيل مراقبة السوق: {e}")
 
 #
-# 🔴 تم التعديل: الدالة الآن ترجع قاموساً مفصلاً بدلاً من قائمة واحدة
+# 🔴 هذه الدالة (المستهلك) لم تتغير، لأنها بالفعل تعالج الدفعات
 #
 async def process_batch_parallel(batch, ml_processor, batch_num, total_batches):
     """معالجة دفعة من الرموز بشكل متوازي وإرجاع نتائج مفصلة"""
     try:
-        print(f"   🔄 معالجة الدفعة {batch_num}/{total_batches} ({len(batch)} عملة)...")
+        print(f"   🔄 [المستهلك] بدء معالجة الدفعة {batch_num}/{total_batches} ({len(batch)} عملة)...")
         
         # إنشاء مهام للدفعة الحالية
         batch_tasks = []
@@ -207,30 +207,23 @@ async def process_batch_parallel(batch, ml_processor, batch_num, total_batches):
         # انتظار انتهاء جميع مهام الدفعة الحالية
         batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
         
-        # 
-        # 🔴 تم التعديل: تصفية النتائج إلى ثلاث فئات
-        #
         successful_results = []
         low_score_results = []
         failed_results = []
 
         for i, result in enumerate(batch_results):
-            symbol = batch[i].get('symbol', 'unknown') # جلب الرمز من بيانات الدفعة الأصلية
+            symbol = batch[i].get('symbol', 'unknown') 
             
             if isinstance(result, Exception):
-                # فشل على مستوى المهمة (مثل Timeout)
                 failed_results.append({"symbol": symbol, "error": f"Task Execution Error: {str(result)}"})
             elif result is None:
-                # فشل المعالجة داخل ML.py (سيرجع 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)
             else:
-                # نجاح - درجة منخفضة
                 low_score_results.append(result)
         
-        print(f"   ✅ اكتملت الدفعة {batch_num}: {len(successful_results)} نجاح | {len(low_score_results)} منخفض | {len(failed_results)} فشل")
+        print(f"   ✅ [المستهلك] اكتملت معالجة الدفعة {batch_num}: {len(successful_results)} نجاح | {len(low_score_results)} منخفض | {len(failed_results)} فشل")
         
         # إرجاع قاموس مفصل
         return {
@@ -240,19 +233,20 @@ async def process_batch_parallel(batch, ml_processor, batch_num, total_batches):
         }
         
     except Exception as error:
-        print(f"❌ خطأ في معالجة الدفعة {batch_num}: {error}")
-        # إرجاع هيكل فارغ في حالة فشل الدفعة بالكامل
+        print(f"❌ [المستهلك] خطأ في معالجة الدفعة {batch_num}: {error}")
         return {'success': [], 'low_score': [], 'failures': []}
 
+
+# 🔴 --- بدء التعديل الجوهري --- 🔴
+# تم إعادة هيكلة هذه الدالة بالكامل لاستخدام نموذج المنتج/المستهلك
 async def run_3_layer_analysis():
     """
-    تشغيل النظام الطبقي المكون من 3 طبقات:
+    (معدلة) تشغيل النظام الطبقي (منتج/مستهلك)
     الطبقة 1: data_manager - الفحص السريع
-    الطبقة 2: MLProcessor - التحليل المتقدم  
+    الطبقة 2: MLProcessor - التحليل المتدفق  
     الطبقة 3: LLMService - النموذج الضخم
     """
     
-    # 🔴 تعريف متغيرات السجل في بداية الدالة
     layer1_candidates = []
     layer2_candidates = []
     all_low_score_candidates = []
@@ -261,13 +255,13 @@ async def run_3_layer_analysis():
     final_opportunities = []
     
     try:
-        print("🎯 بدء النظام الطبقي المكون من 3 طبقات...")
+        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()
         
@@ -277,69 +271,98 @@ async def run_3_layer_analysis():
         
         print(f"✅ تم اختيار {len(layer1_candidates)} عملة للطبقة 2")
 
-        # جلب بيانات OHLCV كاملة للمرشحين
-        layer1_symbols = [candidate['symbol'] for candidate in layer1_candidates]
-        ohlcv_data_list = await data_manager_global.get_ohlcv_data_for_symbols(layer1_symbols)
-        
-        if not ohlcv_data_list:
-            print("❌ فشل جلب بيانات OHLCV للمرشحين")
-            return None
+        # 🔴 --- إعداد نموذج المنتج والمستهلك --- 🔴
 
-        print(f"📊 تم جلب بيانات OHLCV لـ {len(ohlcv_data_list)} عملة بنجاح")
+        # 1. إنشاء الطابور (Queue)
+        # maxsize=2 يعني أن المنتج (جلب البيانات) سيتوقف مؤقتاً إذا كان متقدماً
+        # على المستهلك (تحليل ML) بأكثر من دفعتين. هذا يمنع استهلاك ذاكرة مفرط.
+        DATA_QUEUE_MAX_SIZE = 2
+        ohlcv_data_queue = asyncio.Queue(maxsize=DATA_QUEUE_MAX_SIZE)
+        
+        # قائمة لتجميع كل النتائج النهائية من المستهلك
+        ml_results_list = []
 
-        # الطبقة 2: التحليل المتقدم بشكل متوازي حقيقي
-        print(f"\n📈 الطبقة 2: التحليل المتقدم (MLProcessor) بشكل متوازي لـ {len(ohlcv_data_list)} عملة...")
+        # 2. إعداد المستهلك (MLProcessor)
         market_context = await data_manager_global.get_market_context_async()
-        
-        # إنشاء معالج ML
         ml_processor = MLProcessor(market_context, data_manager_global, learning_engine_global)
         
-        # تجهيز البيانات للطبقة 2
-        layer2_data = []
-        for ohlcv_data in ohlcv_data_list:
-            try:
-                # إضافة أسباب الترشيح من الطبقة 1
-                symbol = ohlcv_data['symbol']
-                layer1_candidate = next((c for c in layer1_candidates if c['symbol'] == symbol), None)
-                if layer1_candidate:
-                    ohlcv_data['reasons_for_candidacy'] = layer1_candidate.get('reasons', [])
-                    ohlcv_data['layer1_score'] = layer1_candidate.get('layer1_score', 0)
-                    ohlcv_data['successful_timeframes'] = ohlcv_data.get('successful_timeframes', 0)
-                
-                layer2_data.append(ohlcv_data)
-                    
-            except Exception as e:
-                continue
-        
-        if not layer2_data:
-            print("❌ فشل إعداد بيانات الطبقة 2")
-            return None
-        
-        # تقسيم العمل إلى دفعات للمعالجة المتوازية
-        batch_size = 15
-        batches = [layer2_data[i:i + batch_size] for i in range(0, len(layer2_data), batch_size)]
-        total_batches = len(batches)
+        # حساب إجمالي عدد الدفعات للمستهلك
+        batch_size = 15 # يجب أن يتطابق هذا مع حجم الدفعة في data_manager
+        total_batches = (len(layer1_candidates) + batch_size - 1) // batch_size
         
-        print(f"   🚀 تقسيم العمل إلى {total_batches} دفعة ({batch_size} عملة لكل دفعة)...")
-        
-        # معالجة جميع الدفعات بشكل متوازي
-        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_list = await asyncio.gather(*batch_tasks)
-        
-        # دمج جميع النتائج
-        layer2_candidates = []
-        all_low_score_candidates = []
-        all_failed_candidates = []
-        
-        for batch_result in batch_results_list:
-            layer2_candidates.extend(batch_result['success'])
+        print(f"   🚀 إعداد المنتج/المستهلك: {total_batches} دفعة متوقعة (بحجم {batch_size})")
+
+        # 3. تعريف وظيفة المستهلك (Consumer)
+        async def ml_consumer_task(queue: asyncio.Queue, results_list: list):
+            batch_num = 0
+            while True:
+                try:
+                    # انتظار بيانات من المنتج
+                    batch_data = await queue.get()
+                    
+                    # 🔴 إشارة التوقف (None)
+                    if batch_data is None:
+                        queue.task_done()
+                        print("   🛑 [المستهلك] تلقى إشارة التوقف.")
+                        break
+                    
+                    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
+                    )
+                    
+                    results_list.append(batch_results_dict)
+                    
+                    # إبلاغ الطابور بانتهاء معالجة هذه المهمة
+                    queue.task_done()
+                    
+                except Exception as e:
+                    print(f"❌ [المستهلك] خطأ فادح في مهمة المستهلك: {e}")
+                    traceback.print_exc()
+                    queue.task_done() # يجب استدعاؤها حتى عند الفشل
+
+        # 4. تشغيل مهمة المستهلك
+        print("   ▶️ [المستهلك] بدء تشغيل مهمة المستهلك...")
+        consumer_task = asyncio.create_task(ml_consumer_task(ohlcv_data_queue, ml_results_list))
+
+        # 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)
+        )
+        
+        # 6. انتظار انتهاء المنتج (جلب كل البيانات)
+        await producer_task
+        print("   ✅ [المنتج] أنهى جلب جميع البيانات.")
+
+        # 7. إرسال إشارة التوقف للمستهلك
+        await ohlcv_data_queue.put(None)
+        
+        # 8. انتظار انتهاء المستهلك من معالجة كل شيء في الطابور
+        await ohlcv_data_queue.join() # انتظار اكتمال كل task_done()
+        await consumer_task # انتظار إغلاق مهمة المستهلك نفسها
+        print("   ✅ [المستهلك] أنهى معالجة جميع الدفعات.")
+
+        # 🔴 --- انتهاء نموذج المنتج والمستهلك --- 🔴
+
+        # 9. تجميع النتائج
+        print("🔄 تجميع جميع النتائج...")
+        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)
+                layer2_candidates.append(success_item)
+                
             all_low_score_candidates.extend(batch_result['low_score'])
             all_failed_candidates.extend(batch_result['failures'])
         
@@ -347,19 +370,17 @@ async def run_3_layer_analysis():
         
         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 (الأقوى فقط)")
-
-        # ✅ حفظ المرشحين العشرة في ملف Candidates في R2
+        
         await r2_service_global.save_candidates_async(final_layer2_candidates)
 
-        # عرض أفضل 10 عملات من الطبقة 2
         print("\n🏆 أفضل 10 عملات من الطبقة 2:")
         for i, candidate in enumerate(final_layer2_candidates):
             score = candidate.get('enhanced_final_score', 0)
@@ -374,49 +395,33 @@ async def run_3_layer_analysis():
                 print(f"      🎯 مونت كارلو: {mc_score:.3f}")
             print(f"      🎯 استراتيجية: {strategy} | نمط: {pattern}")
 
-        # الطبقة 3: التحليل بالنموذج الضخم
+        # 11. الطبقة 3: التحليل بالنموذج الضخم (نفس المنطق السابق)
         print("\n🧠 الطبقة 3: التحليل بالنموذج الضخم (LLMService)...")
-        final_opportunities = []
         
         for candidate in final_layer2_candidates:
             try:
                 symbol = candidate['symbol']
                 print(f"   🤔 تحليل {symbol} بالنموذج الضخم...")
                 
-                # ✅ الإصلاح الرئيسي: التأكد من وجود بيانات الشموع في candidate
-                ohlcv_data = candidate.get('ohlcv')  # التغيير هنا: استخدام 'ohlcv' بدلاً من 'raw_ohlcv'
+                ohlcv_data = candidate.get('ohlcv')
                 
                 if not ohlcv_data:
                     print(f"      ⚠️ لا توجد بيانات شموع لـ {symbol}")
-                    # محاولة الحصول على البيانات من المصدر الأصلي
-                    symbol_ohlcv_list = await data_manager_global.get_ohlcv_data_for_symbols([symbol])
-                    if symbol_ohlcv_list and len(symbol_ohlcv_list) > 0:
-                        ohlcv_data = symbol_ohlcv_list[0].get('ohlcv')
-                        candidate['ohlcv'] = ohlcv_data
-                        candidate['raw_ohlcv'] = ohlcv_data
-                
-                if not ohlcv_data:
-                    print(f"      ⚠️ فشل جلب بيانات شموع لـ {symbol}")
                     continue
                 
-                # ✅ التأكد من تمرير البيانات الخام للنموذج - الإصلاح الرئيسي
                 candidate['raw_ohlcv'] = ohlcv_data
-                candidate['ohlcv'] = ohlcv_data
                 
-                # ✅ التحقق من جودة البيانات قبل الإرسال للنموذج
                 timeframes_count = candidate.get('successful_timeframes', 0)
                 total_candles = sum(len(data) for data in ohlcv_data.values()) if ohlcv_data else 0
                 
-                if total_candles < 30:  # تخفيف الشرط من 50 إلى 30 شمعة
+                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']:
                     opportunity = {
                         'symbol': symbol,
@@ -441,20 +446,15 @@ async def run_3_layer_analysis():
                 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. سجل التدقيق (نفس المنطق السابق)
         try:
-            # 1. ملخص الـ 10 الأوائل (لـ LLM)
             top_10_detailed_summary = []
-            for c in final_layer2_candidates: # هذه هي قائمة الـ 10 الأوائل
+            for c in final_layer2_candidates:
                 whale_summary = "Not Available"
                 whale_data = c.get('whale_data')
                 if whale_data and whale_data.get('data_available'):
@@ -473,30 +473,17 @@ async def run_3_layer_analysis():
                     "pattern": c.get('pattern_analysis', {}).get('pattern_detected', 'N/A'),
                 })
 
-            # 2. ملخص باقي الناجحين (الذين لم يتم إرسالهم للنموذج)
             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"
-                } 
+                {"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
             ]
             
-            # 3. ملخص الدرجات المنخفضة (نجاح < 0.4)
             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"
-                } 
+                {"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
             ]
 
-            # 4. تجميع السجل النهائي
             audit_data = {
                 "timestamp": datetime.now().isoformat(),
                 "total_layer1_candidates": len(layer1_candidates),
@@ -507,23 +494,18 @@ async def run_3_layer_analysis():
                     "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, # {"symbol": ..., "error": ...}
+                "failed_candidates": all_failed_candidates,
             }
             
-            # 5. حفظ السجل
             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("❌ لم يتم العثور على فرص تداول مناسبة")
@@ -535,7 +517,6 @@ async def run_3_layer_analysis():
         print(f"❌ خطأ في النظام الطبقي: {error}")
         traceback.print_exc()
         
-        # 🔴 تسجيل السجل حتى في حالة الفشل
         try:
             audit_data = {
                 "timestamp": datetime.now().isoformat(),
@@ -545,7 +526,7 @@ async def run_3_layer_analysis():
                 "total_layer1_candidates": len(layer1_candidates),
                 "counts": {
                     "sent_to_llm": 0,
-                    "success_not_top_10": len(layer2_candidates[target_count:]) if 'target_count' in locals() else 0,
+                    "success_not_top_10": 0,
                     "success_low_score": len(all_low_score_candidates),
                     "failures": len(all_failed_candidates)
                 },
@@ -557,6 +538,8 @@ async def run_3_layer_analysis():
             print(f"❌ فشل حفظ سجل التدقيق أثناء معالجة خطأ آخر: {audit_fail_error}")
             
         return None
+# 🔴 --- نهاية التعديل الجوهري --- 🔴
+
 
 async def re_analyze_open_trade_async(trade_data):
     """إعادة تحليل الصفقة المفتوحة"""
@@ -565,13 +548,23 @@ 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 = await data_manager_global.get_ohlcv_data_for_symbols([symbol])
+            # 🔴 ملاحظة: هذه الدالة معدلة الآن، لكن استدعاؤها لرمز واحد سيعمل كما كان
+            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())
             
             if not ohlcv_data_list:
                 return None
                 
             ohlcv_data = ohlcv_data_list[0]
-            ohlcv_data['reasons_for_candidacy'] = ['re-analysis']
+            
+            # 🔴 دمج بيانات الطبقة 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) # افتراض درجة متوسطة
             
             # استخدام ML للتحليل
             ml_processor = MLProcessor(market_context, data_manager_global, learning_engine_global)
@@ -656,7 +649,7 @@ async def run_bot_cycle_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:
@@ -706,7 +699,7 @@ async def lifespan(application: FastAPI):
         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
         
@@ -725,8 +718,8 @@ async def lifespan(application: FastAPI):
 application = FastAPI(
     lifespan=lifespan,
     title="AI Trading Bot",
-    description="نظام تداول ذكي بثلاث طبقات تحليلية",
-    version="3.0.0"
+    description="نظام تداول ذكي بثلاث طبقات تحليلية (بنموذج التدفق)",
+    version="3.1.0"
 )
 
 @application.get("/")
@@ -734,7 +727,7 @@ async def root():
     """الصفحة الرئيسية"""
     return {
         "message": "مرحباً بك في نظام التداول الذكي",
-        "system": "3-Layer Analysis System",
+        "system": "3-Layer Analysis System (Streaming Pipeline)",
         "status": "running" if state_manager.initialization_complete else "initializing",
         "timestamp": datetime.now().isoformat()
     }
@@ -745,7 +738,7 @@ async def run_cycle_api():
     if not state_manager.initialization_complete:
         raise HTTPException(status_code=503, detail="الخدمات غير مهيأة بالكامل")
     asyncio.create_task(run_bot_cycle_async())
-    return {"message": "Bot cycle initiated", "system": "3-Layer Analysis"}
+    return {"message": "Bot cycle initiated (Streaming)", "system": "3-Layer Analysis"}
 
 @application.get("/health")
 async def health_check():
@@ -756,10 +749,10 @@ async def health_check():
         "services_initialized": state_manager.services_initialized,
         "initialization_error": state_manager.initialization_error,
         "timestamp": datetime.now().isoformat(),
-        "system_architecture": "3-Layer Analysis System",
+        "system_architecture": "3-Layer Analysis System (Streaming Pipeline)",
         "layers": {
             "layer1": "Data Manager - Rapid Screening",
-            "layer2": "ML Processor - Advanced Analysis", 
+            "layer2": "ML Processor - Advanced Analysis (Streaming Consumer)", 
             "layer3": "LLM Service - Deep Analysis"
         }
     }
@@ -859,7 +852,7 @@ 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...")
+    print("🚀 Starting AI Trading Bot with 3-Layer Analysis System (Streaming)...")
     uvicorn.run(
         application, 
         host="0.0.0.0",