Update app.py

app.py

@@ -1,4 +1,4 @@
-# app.py (Fully updated to Explorer-Sentry-Executor architecture V5.2 - Fixed typing import)
+# app.py (Fully updated to Explorer-Sentry-Executor architecture V5.3 - Fixed Producer Task)
 import os
 import traceback
 import signal
@@ -10,9 +10,8 @@ import time
 from contextlib import asynccontextmanager
 from fastapi import FastAPI, HTTPException
 from datetime import datetime
-from typing import List, Dict, Any # (إضافة الاستيراد الناقص)
+from typing import List, Dict, Any 
 
-# استيراد الخدمات (بما في ذلك TradeManager المحدث)
 try:
     from r2 import R2Service
     from LLM import LLMService
@@ -20,12 +19,8 @@ try:
     from ml_engine.processor import MLProcessor
     from learning_hub.hub_manager import LearningHubManager
     from sentiment_news import SentimentAnalyzer
-    from trade_manager import TradeManager # (الملف المحدث)
-    
-    # 🔴 --- START OF CHANGE --- 🔴
-    # (استيراد الدالة المساعدة للتعقيم من ملف مونت كارلو)
+    from trade_manager import TradeManager
     from ml_engine.monte_carlo import _sanitize_results_for_json
-    # 🔴 --- END OF CHANGE --- 🔴
     
     from helpers import safe_float_conversion, validate_candidate_data_enhanced
 except ImportError as e:
@@ -43,11 +38,9 @@ trade_manager_global = None
 sentiment_analyzer_global = None
 symbol_whale_monitor_global = None
 
-# (تم تعريف حالة السوق هنا بدلاً من استيرادها - إصلاح سابق)
 MARKET_STATE_OK = True
 
 
-# (StateManager - لا تغيير، هذا هو الكلاس الصحيح)
 class StateManager:
     def __init__(self):
         self.market_analysis_lock = asyncio.Lock()
@@ -78,7 +71,6 @@ class StateManager:
         self.initialization_error = error
         print(f"❌ خطأ في التهيئة: {error}")
 
-# (استخدام النسخة المحلية الصحيحة)
 state_manager = StateManager()
 
 async def initialize_services():
@@ -87,24 +79,21 @@ async def initialize_services():
     global learning_hub_global, trade_manager_global, sentiment_analyzer_global
     global symbol_whale_monitor_global
     try:
-        print("🚀 بدء تهيئة الخدمات (بنية Sentry الجديدة V5.2)...")
+        print("🚀 بدء تهيئة الخدمات (بنية Sentry الجديدة V5.3)...")
         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("   🔄 تهيئة مراقب الحيتان (Layer 1 Data)...");
         try:
-            # (استخدام المسار الجديد الذي قمت بإنشائه)
             from whale_monitor.core import EnhancedWhaleMonitor
             
             symbol_whale_monitor_global = EnhancedWhaleMonitor(contracts_database, r2_service_global)
             state_manager.set_service_initialized('symbol_whale_monitor'); print("   ✅ مراقب الحيتان مهيأ")
         
-        # (هذا هو الإصلاح الحاسم لمشكلة الـ Timeout)
         except Exception as e: 
             print(f"   ⚠️ فشل تهيئة مراقب الحيتان: {e}"); 
-            traceback.print_exc() # (إضافة تتبع الخطأ لمعرفة السبب الحقيقي)
+            traceback.print_exc() 
             symbol_whale_monitor_global = None
-            # (إخبار مدير الحالة بأن الخدمة "انتهت" حتى لو فشلت)
             state_manager.set_service_initialized('symbol_whale_monitor'); 
             print("   ℹ️ مراقبة الحيتان معطلة. استمرار التهيئة...")
 
@@ -138,7 +127,7 @@ async def initialize_services():
             r2_service=r2_service_global,
             learning_hub=learning_hub_global,
             data_manager=data_manager_global,
-            state_manager=state_manager # (تمرير النسخة المحلية الصحيحة)
+            state_manager=state_manager 
         )
         await trade_manager_global.initialize_sentry_exchanges()
         state_manager.set_service_initialized('trade_manager');
@@ -174,7 +163,7 @@ async def monitor_market_async():
                 await asyncio.sleep(60)
             except Exception as error: 
                 print(f"❌ خطأ أثناء مراقبة السوق: {error}"); 
-                MARKET_STATE_OK = True; # (الحالة الآمنة عند الفشل)
+                MARKET_STATE_OK = True; 
                 await asyncio.sleep(60)
     except Exception as e: print(f"❌ فشل تشغيل مراقبة السوق: {e}")
 
@@ -200,7 +189,6 @@ async def run_periodic_distillation():
 
 async def process_batch_parallel(batch, ml_processor, batch_num, total_batches, preloaded_whale_data):
     """(لا تغيير) معالج الدفعات لـ Layer 1"""
-    # (ملاحظة: preloaded_whale_data سيكون فارغاً الآن، وهذا طبيعي)
     try:
         batch_tasks = []
         for symbol_data in batch:
@@ -240,17 +228,14 @@ async def process_batch_parallel(batch, ml_processor, batch_num, total_batches,
 
 async def run_3_layer_analysis_explorer() -> List[Dict[str, Any]]:
     """
-    (معدل) - هذا هو "المستكشف" (Layer 1).
-    يجلب بيانات الحيتان *فقط* لأفضل 10 مرشحين ويعيد حساب النقاط.
+    (معدل V5.3) - هذا هو "المستكشف" (Layer 1).
     """
-
     layer1_candidates = []
     layer2_candidates = []
     final_layer2_candidates = []
     watchlist_candidates = []
     
-    # (لم نعد بحاجة لهذا القاموس الكبير مقدماً)
-    preloaded_whale_data_dict = {} # (سيبقى فارغاً)
+    preloaded_whale_data_dict = {}
 
     try:
         print("🎯 Starting Explorer Analysis (Layer 1)...")
@@ -258,15 +243,16 @@ async def run_3_layer_analysis_explorer() -> List[Dict[str, Any]]:
         if not await state_manager.wait_for_initialization():
             print("❌ Services not fully initialized (Explorer)"); return []
 
-        print("\n🔍 Layer 1.1: Rapid Screening (data_manager)...")
+        print("\n🔍 Layer 1.1: Rapid Screening (data_manager V7.1)...")
+        # (هذه الآن ترجع قائمة المرشحين الناجحين من الغربلة الذكية)
         layer1_candidates = await data_manager_global.layer1_rapid_screening()
         if not layer1_candidates: print("❌ No candidates found in Layer 1.1"); return []
         print(f"✅ Selected {len(layer1_candidates)} symbols for Layer 1.2")
-        layer1_symbols = [c['symbol'] for c in layer1_candidates]
-
-        # (تم حذف كتلة 'Layer 1.2: Fetching whale data' بالكامل من هنا)
+        
+        # (لم نعد بحاجة إلى layer1_symbols)
+        # layer1_symbols = [c['symbol'] for c in layer1_candidates]
 
-        print(f"\n📊 Layer 1.2: Fetching OHLCV data for {len(layer1_symbols)} symbols (Streaming)...")
+        print(f"\n📊 Layer 1.2: Fetching OHLCV data for {len(layer1_candidates)} symbols (Streaming)...")
         DATA_QUEUE_MAX_SIZE = 2
         ohlcv_data_queue = asyncio.Queue(maxsize=DATA_QUEUE_MAX_SIZE)
         ml_results_list = []
@@ -283,43 +269,42 @@ async def run_3_layer_analysis_explorer() -> List[Dict[str, Any]]:
                 try:
                     batch_data = await queue.get()
                     if batch_data is None: 
-                        queue.task_done() # (هام: استدعاء task_done() للـ None)
+                        queue.task_done()
                         break
                     batch_num += 1
-                    # (سيتم تمرير whale_data_store الفارغ، وهذا صحيح)
                     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() # (هام: استدعاء task_done() للدفعة)
+                    queue.task_done()
                 except Exception as e: 
                     print(f"❌ [ML Consumer] Fatal Error: {e}"); 
                     traceback.print_exc(); 
-                    queue.task_done() # (هام: استدعاء task_done() عند الخطأ)
+                    queue.task_done()
 
         consumer_task = asyncio.create_task(ml_consumer_task(ohlcv_data_queue, ml_results_list, preloaded_whale_data_dict))
-        producer_task = asyncio.create_task(data_manager_global.stream_ohlcv_data(layer1_symbols, ohlcv_data_queue))
         
-        # (إصلاح الـ Deadlock)
-        await producer_task; # (انتظار انتهاء المنتج - المنتج يرسل None بنفسه)
+        # 🔴 --- START OF CHANGE (V5.3 - FIX) --- 🔴
+        # (تمرير layer1_candidates (List[Dict]) بدلاً من layer1_symbols (List[str]))
+        producer_task = asyncio.create_task(data_manager_global.stream_ohlcv_data(layer1_candidates, ohlcv_data_queue))
+        # 🔴 --- END OF CHANGE --- 🔴
         
-        # (تم حذف السطر المسبب للجمود)
-        
-        await ohlcv_data_queue.join() # (انتظار المستهلك لإنهاء جميع المهام)
-        await consumer_task; # (التأكد من أن مهمة المستهلك قد خرجت)
-
-        # (تم حذف كتلة 'asyncio.wait_for(whale_fetcher_task, ...)' بالكامل من هنا)
+        await producer_task; 
+        await ohlcv_data_queue.join()
+        await consumer_task; 
 
         print("🔄 Aggregating all ML (Layer 1.3) results...")
         for batch_result in ml_results_list:
             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)
+                
+                # (البيانات الأولية موجودة الآن في success_item)
+                l1_data = success_item
+                
                 if l1_data:
-                    success_item['reasons_for_candidacy'] = l1_data.get('reasons', [])
+                    success_item['reasons_for_candidacy'] = l1_data.get('reasons_for_candidacy', [])
                     success_item['layer1_score'] = l1_data.get('layer1_score', 0)
                 
-                # (لم نعد نضيف بيانات الحيتان هنا)
                 success_item['whale_data'] = {'data_available': False, 'reason': 'Not fetched yet'}
                 
                 layer2_candidates.append(success_item)
@@ -330,7 +315,6 @@ async def run_3_layer_analysis_explorer() -> List[Dict[str, Any]]:
         target_count = min(10, len(layer2_candidates))
         final_layer2_candidates = layer2_candidates[:target_count]
 
-        # (المنطق الجديد: جلب بيانات الحيتان فقط لأفضل 10)
         print(f"\n🐋 Layer 1.3 (Optimized): Fetching whale data for top {len(final_layer2_candidates)} candidates...")
         
         whale_tasks = []
@@ -338,7 +322,6 @@ async def run_3_layer_analysis_explorer() -> List[Dict[str, Any]]:
             """دالة مساعدة لجلب وتحديث بيانات الحيتان للمرشح"""
             symbol = candidate.get('symbol', 'UNKNOWN')
             try:
-                # (هذا الاستدعاء هو الآن محدود بـ 10 طلبات فقط)
                 data = await data_manager_global.get_whale_data_for_symbol(symbol)
                 if data:
                     candidate['whale_data'] = data
@@ -348,25 +331,20 @@ async def run_3_layer_analysis_explorer() -> List[Dict[str, Any]]:
                 print(f"   ❌ [Whale Fetch] {symbol} - Error: {e}")
                 candidate['whale_data'] = {'data_available': False, 'error': str(e)}
         
-        # إنشاء المهام
         for candidate in final_layer2_candidates:
             whale_tasks.append(asyncio.create_task(get_whale_data_for_candidate(candidate)))
         
-        # انتظار اكتمال جلب بيانات الحيتان (سيكون هذا سريعاً جداً الآن)
         await asyncio.gather(*whale_tasks)
         print("   ✅ Whale data fetched for top candidates.")
 
-        # (المنطق الجديد: إعادة حساب النقاط وإعادة الفرز)
         print("   🔄 Re-calculating enhanced scores with new whale data...")
         for candidate in final_layer2_candidates:
             try:
-                # (استدعاء دالة حساب النقاط من processor لتحديث النتيجة)
                 new_score = ml_processor._calculate_enhanced_final_score(candidate)
                 candidate['enhanced_final_score'] = new_score
             except Exception as e:
                 print(f"   ❌ [Score Recalc] {candidate.get('symbol')} - Error: {e}")
         
-        # (إعادة الفرز بناءً على النقاط الجديدة والكاملة)
         final_layer2_candidates.sort(key=lambda x: x.get('enhanced_final_score', 0), reverse=True)
         print("   ✅ Top 10 scores updated and re-sorted.")
 
@@ -392,34 +370,28 @@ async def run_3_layer_analysis_explorer() -> List[Dict[str, Any]]:
                 candidate['advanced_mc_run'] = False
                 updated_candidates_for_llm.append(candidate)
         
-        # 🔴 --- START OF CHANGE (FIX JSON BUG) --- 🔴
-        # (تعقيم القائمة *قبل* إرسالها إلى R2 أو LLM)
         print("   🔄 Sanitizing final candidates for JSON serialization...")
         sanitized_candidates = []
         for cand in updated_candidates_for_llm:
             sanitized_candidates.append(_sanitize_results_for_json(cand))
         
         final_layer2_candidates = sanitized_candidates
-        # 🔴 --- END OF CHANGE (FIX JSON BUG) --- 🔴
-
-        # (الآن هذا الاستدعاء آمن ولن يفشل)
+        
         await r2_service_global.save_candidates_async(final_layer2_candidates)
 
         print("\n🧠 Layer 1.5: LLM Strategic Analysis (Explorer Brain)...")
         for candidate in final_layer2_candidates:
             try:
                 symbol = candidate['symbol']
-                ohlcv_data = candidate.get('ohlcv'); # (لا يزال موجوداً حتى بعد التعقيم كـ list)
+                ohlcv_data = candidate.get('ohlcv'); 
                 if not ohlcv_data: continue
                 
-                # (نمرر البيانات الخام المعقمة)
                 candidate['raw_ohlcv'] = ohlcv_data
                 total_candles = sum(len(data) for data in ohlcv_data.values()) if ohlcv_data else 0
                 if total_candles < 30: continue
 
                 candidate['sentiment_data'] = await data_manager_global.get_market_context_async()
 
-                # (الآن، 'candidate' تحتوي على بيانات الحيتان + وهي معقمة JSON)
                 llm_analysis = await llm_service_global.get_trading_decision(candidate)
 
                 if llm_analysis and llm_analysis.get('action') in ['WATCH']:
@@ -434,7 +406,7 @@ async def run_3_layer_analysis_explorer() -> List[Dict[str, Any]]:
                         'analysis_timestamp': datetime.now().isoformat(),
                         'llm_decision_context': {
                             'decision': llm_analysis,
-                            'full_candidate_data': candidate # (هذه آمنة الآن)
+                            'full_candidate_data': candidate 
                         }
                     }
                     watchlist_candidates.append(watchlist_entry)
@@ -469,7 +441,13 @@ async def re_analyze_open_trade_async(trade_data):
             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)
+            
+            # (إصلاح هنا أيضاً: تمرير قائمة Dict)
+            await data_manager_global.stream_ohlcv_data(
+                [{'symbol': symbol, 'layer1_score': 0, 'reasons': ['re-analysis']}], 
+                temp_queue
+            )
+            
             while True:
                 try:
                     batch = await asyncio.wait_for(temp_queue.get(), timeout=1.0)
@@ -483,8 +461,9 @@ async def re_analyze_open_trade_async(trade_data):
             if not ohlcv_data_list: print(f"⚠️ Failed to get re-analysis data for {symbol}"); return None
             ohlcv_data = ohlcv_data_list[0]
 
-            l1_data = await data_manager_global._get_detailed_symbol_data(symbol)
-            if l1_data: ohlcv_data.update(l1_data); ohlcv_data['reasons_for_candidacy'] = ['re-analysis']
+            # (لم نعد بحاجة لهذا، البيانات مدمجة)
+            # l1_data = await data_manager_global._get_detailed_symbol_data(symbol)
+            # if l1_data: ohlcv_data.update(l1_data); ohlcv_data['reasons_for_candidacy'] = ['re-analysis']
 
             re_analysis_whale_data = await data_manager_global.get_whale_data_for_symbol(symbol)
 
@@ -610,7 +589,7 @@ async def lifespan(application: FastAPI):
         await cleanup_on_shutdown()
 
 
-application = FastAPI(lifespan=lifespan, title="AI Trading Bot", description="Explorer-Sentry-Executor Architecture (V5.2)", version="5.2.0")
+application = FastAPI(lifespan=lifespan, title="AI Trading Bot", description="Explorer-Sentry-Executor Architecture (V5.3)", version="5.3.0")
 
 @application.get("/")
 async def root(): return {"message": "Welcome to the AI Trading System", "system": "Explorer-Sentry-Executor", "status": "running" if state_manager.initialization_complete else "initializing", "timestamp": datetime.now().isoformat()}
@@ -622,7 +601,7 @@ async def run_cycle_api():
     return {"message": "Explorer (Layer 1) cycle initiated", "system": "Explorer-Sentry-Executor"}
 
 @application.get("/health")
-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": "Explorer-Sentry-Executor (V5.2)"}
+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": "Explorer-Sentry-Executor (V5.3)"}
 
 @application.get("/analyze-market")
 async def analyze_market_api():
@@ -641,7 +620,6 @@ async def get_portfolio_api():
 async def get_system_status():
     monitoring_status = trade_manager_global.get_sentry_status() if trade_manager_global else {};
     
-    # (يستخدم المتغير المحلي بدلاً من state.MARKET_STATE_OK - إصلاح سابق)
     return {"initialization_complete": state_manager.initialization_complete, "services_initialized": state_manager.services_initialized, "initialization_error": state_manager.initialization_error, "market_state_ok": MARKET_STATE_OK, "sentry_status": monitoring_status, "timestamp": datetime.now().isoformat()}
 
 async def cleanup_on_shutdown():
@@ -659,7 +637,6 @@ async def cleanup_on_shutdown():
             print("✅ Learning hub data saved")
         except Exception as e: print(f"❌ Failed to save learning hub data: {e}")
 
-    # (إضافة إغلاق لخدمة الحيتان)
     if symbol_whale_monitor_global:
         try:
             await symbol_whale_monitor_global.cleanup()
@@ -677,5 +654,5 @@ def signal_handler(signum, frame): print(f"🛑 Received signal {signum}. Initia
 signal.signal(signal.SIGINT, signal_handler); signal.signal(signal.SIGTERM, signal_handler)
 
 if __name__ == "__main__":
-    print("🚀 Starting AI Trading Bot (Explorer-Sentry-Executor V5.2)...")
+    print("🚀 Starting AI Trading Bot (Explorer-Sentry-Executor V5.3)...")
     uvicorn.run( application, host="0.0.0.0", port=7860, log_level="info", access_log=True )