Update app.py

app.py

@@ -1,4 +1,4 @@
-# app.py (محدث)
+# app.py (محدث بإصلاح IndentationError)
 import os
 import traceback
 import signal
@@ -26,8 +26,7 @@ except ImportError as e:
     print(f"❌ خطأ في استيراد الوحدات: {e}")
     sys.exit(1)
 
-# (المتغيرات العالمية و StateManager كما هي)
-# ...
+# المتغيرات العالمية
 r2_service_global = None
 data_manager_global = None
 llm_service_global = None
@@ -68,8 +67,6 @@ class StateManager:
 
 state_manager = StateManager()
 
-# (initialize_services و monitor_market_async كما هي)
-# ...
 async def initialize_services():
     """تهيئة جميع الخدمات بشكل منفصل"""
     global r2_service_global, data_manager_global, llm_service_global
@@ -116,8 +113,6 @@ async def monitor_market_async():
     except Exception as e: print(f"❌ فشل تشغيل مراقبة السوق: {e}")
 
 
-# 🔴 --- تعديل: إزالة Semaphore من هنا --- 🔴
-# منظم السرعة الآن داخل مهمة جلب الحيتان ومهمة معالجة ML
 async def process_batch_parallel(batch, ml_processor, batch_num, total_batches, preloaded_whale_data):
     """
     (معدلة) معالجة دفعة من الرموز بشكل متوازي وإرجاع نتائج مفصلة
@@ -173,8 +168,6 @@ async def process_batch_parallel(batch, ml_processor, batch_num, total_batches,
         return {'success': [], 'low_score': [], 'failures': []}
 
 
-# 🔴 --- بدء التعديل الجوهري --- 🔴
-# تم إعادة هيكلة هذه الدالة بالكامل
 async def run_3_layer_analysis():
     """
     (معدلة) تشغيل النظام الطبقي (مع فصل جلب بيانات الحيتان)
@@ -210,8 +203,7 @@ async def run_3_layer_analysis():
         
         # مهمة جلب بيانات الحيتان
         async def fetch_whale_data_task(symbols, results_dict):
-            # استخدام منظم سرعة خاص لهذه المهمة (يمكن تعديل القيمة حسب الحاجة)
-            WHALE_FETCH_CONCURRENCY = 3 # عدد طلبات الحيتان المتزامنة
+            WHALE_FETCH_CONCURRENCY = 3 
             semaphore = asyncio.Semaphore(WHALE_FETCH_CONCURRENCY)
             tasks = []
 
@@ -221,18 +213,16 @@ async def run_3_layer_analysis():
                         data = await data_manager_global.get_whale_data_for_symbol(symbol)
                         if data:
                            results_dict[symbol] = data
-                        # else: # Optional: Log if data is None
-                        #    print(f"   ⚠️ [Whale Fetch] No whale data returned for {symbol}")
                     except Exception as e:
                         print(f"   ❌ [Whale Fetch] فشل جلب بيانات الحيتان لـ {symbol}: {e}")
-                        results_dict[symbol] = {'data_available': False, 'error': str(e)} # Store error
+                        results_dict[symbol] = {'data_available': False, 'error': str(e)} 
 
             for symbol in symbols:
                 tasks.append(asyncio.create_task(get_data_with_semaphore(symbol)))
             
-            await asyncio.gather(*tasks) # انتظار اكتمال جميع مهام جلب الحيتان
+            await asyncio.gather(*tasks) 
 
-        # تشغيل مهمة جلب الحيتان في الخلفية (لا ننتظرها هنا)
+        # تشغيل مهمة جلب الحيتان في الخلفية
         whale_fetcher_task = asyncio.create_task(fetch_whale_data_task(layer1_symbols, preloaded_whale_data_dict))
         print("   ⏳ مهمة جلب بيانات الحيتان تعمل في الخلفية...")
         # 🔴 --- نهاية الطبقة 1.5 --- 🔴
@@ -301,7 +291,6 @@ async def run_3_layer_analysis():
 
         # 9. تجميع النتائج (كما كان، لكن بيانات الحيتان موجودة الآن)
         print("🔄 تجميع جميع النتائج...")
-        # ... (الكود لتجميع layer2_candidates, all_low_score_candidates, all_failed_candidates كما هو) ...
         for batch_result in ml_results_list:
             for success_item in batch_result['success']:
                 symbol = success_item['symbol']
@@ -309,16 +298,14 @@ 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)
-                # Ensure whale data from preloading is in the final result item
                 if symbol in preloaded_whale_data_dict:
                     success_item['whale_data'] = preloaded_whale_data_dict[symbol]
-                elif 'whale_data' not in success_item: # If ML processor didn't add a default
+                elif 'whale_data' not in success_item: 
                      success_item['whale_data'] = {'data_available': False, 'reason': 'Failed during preload'}
                 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")
@@ -330,7 +317,6 @@ async def run_3_layer_analysis():
         print(f"🎯 تم اختيار {len(final_layer2_candidates)} عملة للطبقة 3 (الأقوى فقط)")
         await r2_service_global.save_candidates_async(final_layer2_candidates)
         print("\n🏆 أفضل 10 عملات من الطبقة 2:")
-        # ... (كود عرض أفضل 10 كما هو، سيشمل بيانات الحيتان الآن) ...
         for i, candidate in enumerate(final_layer2_candidates):
             score = candidate.get('enhanced_final_score', 0)
             strategy = candidate.get('target_strategy', 'GENERIC')
@@ -348,13 +334,9 @@ async def run_3_layer_analysis():
                  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]}...)")
-            # else:
-            #      print(f"      🐋 حيتان: غير متاحة")
-
 
         # 11. الطبقة 3: التحليل بالنموذج الضخم (كما كان)
         print("\n🧠 الطبقة 3: التحليل بالنموذج الضخم (LLMService)...")
-        # ... (كود الطبقة 3 كما هو) ...
         for candidate in final_layer2_candidates:
             try:
                 symbol = candidate['symbol']
@@ -383,7 +365,6 @@ async def run_3_layer_analysis():
                     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)} فرصة تداول")
@@ -392,7 +373,6 @@ async def run_3_layer_analysis():
         
         # 12. سجل التدقيق (كما كان)
         try:
-            # ... (كود سجل التدقيق كما هو، سيستخدم البيانات المجمعة الآن) ...
             top_10_detailed_summary = []
             for c in final_layer2_candidates:
                 whale_summary = "Not Available"
@@ -437,10 +417,7 @@ async def run_3_layer_analysis():
             print("⚠️ تم حفظ سجل تدقيق جزئي بعد الفشل.")
         except Exception as audit_fail_error: print(f"❌ فشل حفظ سجل التدقيق أثناء معالجة خطأ آخر: {audit_fail_error}")
         return None
-# 🔴 --- نهاية التعديل الجوهري --- 🔴
 
-# (بقية الملف: re_analyze_open_trade_async, run_bot_cycle_async, lifespan, endpoints, cleanup, etc. كما هي)
-# ...
 async def re_analyze_open_trade_async(trade_data):
     """إعادة تحليل الصفقة المفتوحة"""
     symbol = trade_data.get('symbol')
@@ -450,21 +427,31 @@ 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)
-            while not temp_queue.empty():
-                batch = await temp_queue.get(); temp_queue.task_done() # Need task_done here too
-                if batch: ohlcv_data_list.extend(batch)
-            
+            # 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
+                    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 not ohlcv_data_list: print(f"⚠️ فشل جلب بيانات إعادة التحليل لـ {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']
             
-            # 🔴 جلب بيانات الحيتان بشكل منفصل لإعادة التحليل
             re_analysis_whale_data = await data_manager_global.get_whale_data_for_symbol(symbol)
 
             ml_processor = MLProcessor(market_context, data_manager_global, learning_engine_global)
-            # 🔴 تمرير بيانات الحيتان للمعالج
             processed_data = await ml_processor.process_and_score_symbol_enhanced(ohlcv_data, {symbol: re_analysis_whale_data} if re_analysis_whale_data else {})
             
             if not processed_data: return None
@@ -478,7 +465,8 @@ async def re_analyze_open_trade_async(trade_data):
                 await r2_service_global.save_system_logs_async({ "trade_reanalyzed": True, "symbol": symbol, "action": re_analysis_decision.get('action'), 'strategy': re_analysis_decision.get('strategy', 'GENERIC') })
                 return {"symbol": symbol, "decision": re_analysis_decision, "current_price": processed_data.get('current_price')}
             else: return None
-    except Exception as error: await r2_service_global.save_system_logs_async({ "reanalysis_error": True, "symbol": symbol, "error": str(error) }); return None
+    except Exception as error: await r2_service_global.save_system_logs_async({ "reanalysis_error": True, "symbol": symbol, "error": str(error) }); print(f"❌ Error in re_analyze_open_trade_async for {symbol}: {error}"); traceback.print_exc(); return None
+
 
 async def run_bot_cycle_async():
     """دورة التداول الرئيسية"""
@@ -486,6 +474,8 @@ async def run_bot_cycle_async():
         if not await state_manager.wait_for_initialization(): print("❌ الخدمات غير مهيأة بالكامل - تخطي الدورة"); return
         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
         try:
             open_trades = await trade_manager_global.get_open_trades(); print(f"📋 الصفقات المفتوحة: {len(open_trades)}")
             should_look_for_new_trade = len(open_trades) == 0
@@ -494,10 +484,29 @@ 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)} صفقة")
-                    for trade in trades_to_reanalyze:
-                        result = await re_analyze_open_trade_async(trade)
-                        if result and result['decision'].get('action') == "CLOSE_TRADE": await trade_manager_global.close_trade(trade, result['current_price'], 'CLOSED_BY_REANALYSIS'); should_look_for_new_trade = True
-                        elif result and result['decision'].get('action') == "UPDATE_TRADE": await trade_manager_global.update_trade(trade, result['decision'])
+                    # 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')} لم تنتج قرارًا.")
+            
+            # 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
+
             if should_look_for_new_trade:
                 portfolio_state = await r2_service_global.get_portfolio_state_async(); current_capital = portfolio_state.get("current_capital_usd", 0)
                 if current_capital > 1:
@@ -506,10 +515,18 @@ 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("ℹ️ يوجد صفقة مفتوحة بالفعل، تخطي البحث عن صفقة جديدة.")
+
         finally:
-            r2_service_global.release_lock(); await r2_service_global.save_system_logs_async({ "cycle_completed": True, "open_trades": len(open_trades) if 'open_trades' in locals() else 0 }); print("✅ اكتملت دورة التداول")
+            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
+            print("✅ اكتملت دورة التداول")
+            
     except Exception as error:
-        print(f"❌ Unhandled error in main cycle: {error}"); await r2_service_global.save_system_logs_async({ "cycle_error": True, "error": str(error) });
+        print(f"❌ Unhandled error in main cycle: {error}"); traceback.print_exc() # Print traceback
+        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
 
 @asynccontextmanager
@@ -524,9 +541,20 @@ async def lifespan(application: FastAPI):
         await r2_service_global.save_system_logs_async({"application_started": True})
         print("🎯 التطبيق جاهز للعمل - نظام الطبقات 3 فعال (فصل الحيتان)")
         yield
-    except Exception as error: print(f"❌ Application startup failed: {error}"); traceback.print_exc();
-        if r2_service_global: await r2_service_global.save_system_logs_async({ "application_startup_failed": True, "error": str(error) }); raise
-    finally: await cleanup_on_shutdown()
+    except Exception as error: 
+        print(f"❌ Application startup failed: {error}"); 
+        traceback.print_exc() # Print traceback here as well
+        if r2_service_global: 
+            await r2_service_global.save_system_logs_async({ 
+                "application_startup_failed": True, 
+                "error": str(error) 
+            })
+        # 🔴 --- الإصلاح: فصل raise إلى سطر جديد --- 🔴
+        raise # Correct indentation
+        # 🔴 --- نهاية الإصلاح --- 🔴
+    finally: 
+        await cleanup_on_shutdown()
+
 
 application = FastAPI(lifespan=lifespan, title="AI Trading Bot", description="نظام تداول ذكي بثلاث طبقات تحليلية (فصل الحيتان)", version="3.2.0")