Update trade_manager.py

trade_manager.py

@@ -1,4 +1,4 @@
-# trade_manager.py (Updated to V5.7 - Fixed Tactical Triggers Logic)
+# trade_manager.py (Updated to V5.8 - Added Asyncio Lock for Watchlist)
 import asyncio
 import json
 import time
@@ -133,6 +133,10 @@ class TradeManager:
         self.sentry_tasks = {}
         self.tactical_data_cache = {}
         
+        # 🔴 --- START OF CHANGE (FIX RACE CONDITION) --- 🔴
+        self.sentry_lock = asyncio.Lock() # إضافة قفل لحماية Watchlist
+        # 🔴 --- END OF CHANGE --- 🔴
+        
         self.kucoin_rest = None # (المنصة الأساسية)
         self.confirmation_exchanges = {} # (المنصات الثانوية للتأكيد)
         self.polling_interval = 1.5 # (KuCoin - سريع)
@@ -185,33 +189,54 @@ class TradeManager:
             for ex in self.confirmation_exchanges.values(): await ex.close()
             raise
 
-    async def start_sentry_and_monitoring_loops(self): # (لا تغيير هنا)
+    # 🔴 --- START OF CHANGE (FIX RACE CONDITION) --- 🔴
+    async def start_sentry_and_monitoring_loops(self):
         """الحلقة الرئيسية للحارس (Sentry) ومراقب الخروج (Exit Monitor)."""
         self.is_running = True
         print(f"✅ [Sentry] بدء حلقات المراقبة التكتيكية (Layer 2 - API Polling)...")
         while self.is_running:
             try:
-                # (نفس منطق إدارة المهام)
-                watchlist_symbols = set(self.sentry_watchlist.keys())
+                # الحصول على قائمة الرموز المطلوب مراقبتها (داخل القفل)
+                async with self.sentry_lock:
+                    watchlist_symbols = set(self.sentry_watchlist.keys())
+                
+                # (جلب الصفقات المفتوحة - هذا I/O، نتركه خارج القفل)
                 open_trades = await self.get_open_trades()
                 open_trade_symbols = {t['symbol'] for t in open_trades}
+                
                 symbols_to_monitor = watchlist_symbols.union(open_trade_symbols)
-                for symbol in symbols_to_monitor:
-                    if symbol not in self.sentry_tasks:
-                        print(f"   [Sentry] بدء المراقبة التكتيكية (Polling) لـ {symbol}")
+                current_tasks = set(self.sentry_tasks.keys())
+
+                # الرموز التي يجب إضافتها
+                symbols_to_add = symbols_to_monitor - current_tasks
+                for symbol in symbols_to_add:
+                    print(f"   [Sentry] بدء المراقبة التكتيكية (Polling) لـ {symbol}")
+                    
+                    # (الحصول على التلميح داخل القفل لضمان عدم تغييره أثناء القراءة)
+                    async with self.sentry_lock:
                         strategy_hint = self.sentry_watchlist.get(symbol, {}).get('strategy_hint', 'generic')
-                        if symbol not in self.tactical_data_cache: self.tactical_data_cache[symbol] = TacticalData(symbol)
-                        task = asyncio.create_task(self._monitor_symbol_activity_polling(symbol, strategy_hint))
-                        self.sentry_tasks[symbol] = task
-                for symbol in list(self.sentry_tasks.keys()):
-                    if symbol not in symbols_to_monitor:
-                        print(f"   [Sentry] إيقاف المراقبة التكتيكية (Polling) لـ {symbol}")
-                        task = self.sentry_tasks.pop(symbol, None)
-                        if task: task.cancel()
-                        if symbol in self.tactical_data_cache: del self.tactical_data_cache[symbol]
-                await asyncio.sleep(15)
+                    
+                    if symbol not in self.tactical_data_cache: 
+                        self.tactical_data_cache[symbol] = TacticalData(symbol)
+                    
+                    task = asyncio.create_task(self._monitor_symbol_activity_polling(symbol, strategy_hint))
+                    self.sentry_tasks[symbol] = task
+
+                # الرموز التي يجب إزالتها
+                symbols_to_remove = current_tasks - symbols_to_monitor
+                for symbol in symbols_to_remove:
+                    print(f"   [Sentry] إيقاف المراقبة التكتيكية (Polling) لـ {symbol}")
+                    task = self.sentry_tasks.pop(symbol, None)
+                    if task: 
+                        task.cancel()
+                    if symbol in self.tactical_data_cache: 
+                        del self.tactical_data_cache[symbol]
+
+                await asyncio.sleep(15) # فاصل زمني للحلقة الرئيسية
+                
             except Exception as error:
                 print(f"❌ [Sentry] خطأ في الحلقة الرئيسية: {error}"); traceback.print_exc(); await asyncio.sleep(60)
+    # 🔴 --- END OF CHANGE --- 🔴
 
     async def stop_sentry_loops(self):
         """(محدث) إي��اف جميع مهام المراقبة وإغلاق جميع اتصالات REST"""
@@ -234,17 +259,24 @@ class TradeManager:
             print("✅ [Sentry] تم إغلاق اتصالات التداول (REST).")
         except Exception as e: print(f"⚠️ [Sentry] خطأ أثناء إغلاق الاتصالات: {e}")
 
-    async def update_sentry_watchlist(self, candidates: List[Dict]): # (لا تغيير هنا)
+    # 🔴 --- START OF CHANGE (FIX RACE CONDITION) --- 🔴
+    async def update_sentry_watchlist(self, candidates: List[Dict]):
         """تحديث قائمة المراقبة التي يستخدمها الحارس (Sentry)."""
-        self.sentry_watchlist = {c['symbol']: c for c in candidates}
-        print(f"ℹ️ [Sentry] تم تحديث Watchlist. عدد المرشحين: {len(self.sentry_watchlist)}")
+        async with self.sentry_lock: # (استخدام القفل قبل الكتابة)
+            self.sentry_watchlist = {c['symbol']: c for c in candidates}
+            print(f"ℹ️ [Sentry] تم تحديث Watchlist. عدد المرشحين: {len(self.sentry_watchlist)}")
+    # 🔴 --- END OF CHANGE --- 🔴
 
     def get_sentry_status(self):
         """(محدث) لواجهة برمجة التطبيقات /system-status"""
+        # (لا حاجة للقفل هنا، هذه قراءة سريعة لعدد العناصر فقط)
+        active_monitoring_count = len(self.sentry_tasks)
+        watchlist_symbols_list = list(self.sentry_watchlist.keys())
+        
         return {
             'is_running': self.is_running,
-            'active_monitoring_tasks': len(self.sentry_tasks),
-            'watchlist_symbols': list(self.sentry_watchlist.keys()),
+            'active_monitoring_tasks': active_monitoring_count,
+            'watchlist_symbols': watchlist_symbols_list,
             'monitored_symbols': list(self.sentry_tasks.keys()),
             'confirmation_exchanges_active': list(self.confirmation_exchanges.keys())
         }
@@ -272,8 +304,11 @@ class TradeManager:
             traceback.print_exc()
         finally:
             print(f"🛑 [Sentry] إنهاء جميع مهام (Polling) {symbol}")
-            if symbol in self.sentry_tasks: del self.sentry_tasks[symbol]
-            if symbol in self.tactical_data_cache: del self.tactical_data_cache[symbol]
+            # (التنظيف يحدث بالفعل في الحلقة الرئيسية، ولكن هذا ضمان إضافي)
+            if symbol in self.sentry_tasks: 
+                self.sentry_tasks.pop(symbol, None)
+            if symbol in self.tactical_data_cache: 
+                del self.tactical_data_cache[symbol]
 
     async def _poll_kucoin_data(self, symbol):
         """(لا تغيير) حلقة استقصاء (Polling) لبيانات KuCoin (الأساسية)"""
@@ -302,6 +337,8 @@ class TradeManager:
             except ccxtasync.RateLimitExceeded as e:
                 print(f"⏳ [Sentry Polling] {symbol} KuCoin Rate Limit Exceeded: {e}. زيادة فترة الانتظار...")
                 await asyncio.sleep(10)
+            except asyncio.CancelledError:
+                raise # تمرير الإلغاء
             except Exception as e:
                 print(f"⚠️ [Sentry Polling] خطأ في {symbol} KuCoin data polling: {e}")
                 await asyncio.sleep(5)
@@ -354,6 +391,8 @@ class TradeManager:
         except ccxtasync.RateLimitExceeded:
             print(f"⏳ [Sentry Conf] {ex_id} Rate Limit لـ {symbol}. الانتظار...")
             await asyncio.sleep(15) # انتظار أطول لهذه المنصة
+        except asyncio.CancelledError:
+            raise # تمرير الإلغاء
         except Exception as e:
             # (الفشل في منصة واحدة لا يجب أن يوقف الحلقة، فقط نتجاهل بياناتها هذه المرة)
             pass
@@ -381,17 +420,28 @@ class TradeManager:
                         if current_price: await self.immediate_close_trade(symbol, current_price, f"Tactical Exit: {exit_reason}")
                 else:
                     # (منطق الدخول - محدث)
-                    if symbol in self.sentry_watchlist:
+                    # (التحقق من قائمة المراقبة يتم الآن في الحلقة الرئيسية)
+                    # (لكننا نحتاج إلى التحقق مرة أخرى هنا لضمان عدم حدوث سباق حالات)
+                    async with self.sentry_lock:
+                        is_still_on_watchlist = symbol in self.sentry_watchlist
+
+                    if is_still_on_watchlist:
                         trigger = self._check_entry_trigger(symbol, strategy_hint, snapshot)
                         if trigger:
                             print(f"✅ [Sentry] زناد دخول تكتيكي لـ {symbol} (استراتيجية: {strategy_hint})")
-                            watchlist_entry = self.sentry_watchlist.pop(symbol, None)
+                            
+                            # (إزالة الع��صر من قائمة المراقبة باستخدام القفل)
+                            watchlist_entry = None
+                            async with self.sentry_lock:
+                                watchlist_entry = self.sentry_watchlist.pop(symbol, None)
+                            
                             if watchlist_entry:
                                 explorer_context = watchlist_entry.get('llm_decision_context', {})
                                 await self._execute_smart_entry(symbol, strategy_hint, snapshot, explorer_context)
+            except asyncio.CancelledError:
+                raise # تمرير الإلغاء
             except Exception as e: print(f"❌ [Sentry] خطأ في حلقة التحليل التكتيكي لـ {symbol}: {e}"); traceback.print_exc()
 
-    # 🔴 --- START OF CHANGE (LOGIC FIXED V5.7) --- 🔴
     def _check_entry_trigger(self, symbol: str, strategy_hint: str, data: Dict) -> bool:
         """(محدث V5.7) إصلاح منطق الزناد للسماح باستراتيجيات الانعكاس."""
         
@@ -436,7 +486,7 @@ class TradeManager:
                  return True
                  
         return False
-    # 🔴 --- END OF CHANGE (LOGIC FIXED V5.7) --- 🔴
+    # 🔴 --- END OF LOGIC FIX (V5.7) --- 🔴
 
     def _check_exit_trigger(self, trade: Dict, data: Dict) -> str: # (لا تغيير هنا)
         """يحدد ما إذا كان يجب الخروج من صفقة مفتوحة تكتيكياً."""
@@ -503,10 +553,12 @@ class TradeManager:
                 open_trades_after_cancel = await self.get_open_trades()
                 if not any(t['symbol'] == symbol for t in open_trades_after_cancel):
                     print(f"   [Sentry] إعادة {symbol} إلى Watchlist بعد فشل التنفيذ.")
-                    if explorer_context:
-                        self.sentry_watchlist[symbol] = {"symbol": symbol, "strategy_hint": strategy_hint, "llm_decision_context": explorer_context}
-                    else:
-                        self.sentry_watchlist[symbol] = {"symbol": symbol, "strategy_hint": strategy_hint}
+                    # (استخدام القفل لإعادة الإضافة بأمان)
+                    async with self.sentry_lock:
+                        if explorer_context:
+                            self.sentry_watchlist[symbol] = {"symbol": symbol, "strategy_hint": strategy_hint, "llm_decision_context": explorer_context}
+                        else:
+                            self.sentry_watchlist[symbol] = {"symbol": symbol, "strategy_hint": strategy_hint}
 
 
         except Exception as e: print(f"❌ [Executor] فشل فادح أثناء التنفيذ لـ {symbol}: {e}"); traceback.print_exc()
@@ -573,7 +625,10 @@ class TradeManager:
             open_trades = await self.r2_service.get_open_trades_async()
             trades_to_keep = [t for t in open_trades if t.get('id') != trade_to_close.get('id')]
             await self.r2_service.save_open_trades_async(trades_to_keep)
-            if symbol in self.sentry_tasks: print(f"ℹ️ [Sentry] الصفقة {symbol} أغلقت، ستتوقف المراقبة.")
+            
+            # (لا حاجة لإزالة المهمة يدوياً، الحلقة الرئيسية ستفعل ذلك)
+            # if symbol in self.sentry_tasks: print(f"ℹ️ [Sentry] الصفقة {symbol} أغلقت، ستتوقف المراقبة.")
+            
             await self.r2_service.save_system_logs_async({
                 "trade_closed": True, "symbol": symbol, "pnl_usd": pnl, "pnl_percent": pnl_percent,
                 "new_capital": new_capital, "strategy": strategy, "reason": reason
@@ -660,4 +715,4 @@ class TradeManager:
         except Exception as e: print(f"❌ Failed to get trade by symbol {symbol}: {e}"); return None
 
 
-print(f"✅ Trade Manager loaded - V5.7 (Fixed Tactical Triggers & ccxt.async_support: {CCXT_ASYNC_AVAILABLE})")
+print(f"✅ Trade Manager loaded - V5.8 (Added Watchlist Lock) (ccxt.async_support: {CCXT_ASYNC_AVAILABLE})")