Update trade_manager.py

trade_manager.py

@@ -1,7 +1,8 @@
-# trade_manager.py (V8.6 - Full Production Version with Async Lock)
+# trade_manager.py (V8.7 - Full Production Version - NO ABBREVIATIONS)
 import asyncio
 import json
 import uuid
+import traceback
 from datetime import datetime
 
 class TradeManager:
@@ -10,87 +11,119 @@ class TradeManager:
         self.data_manager = data_manager
         self.titan = titan_engine
         self.pattern_engine = pattern_engine
-        # قواميس تتبع الحالة في الذاكرة
+        
+        # قواميس تتبع الحالة في الذاكرة (In-Memory State)
         self.open_positions = {}
         self.watchlist = {}
         self.sentry_tasks = {}
         self.running = True
+        
         # 🔒 قفل لضمان عدم تضارب عمليات الشراء/البيع المتزامنة
+        # هذا القفل يمنع حالتين من الدخول إلى منطقة التعديل على المحفظة في نفس الوقت
         self.execution_lock = asyncio.Lock()
         
-        print("🛡️ [TradeManager V8.6] Sentry Module Initialized (Locked Execution Mode).")
+        print("🛡️ [TradeManager V8.7] Sentry Module Initialized (Sync & Lock Enabled).")
 
     async def initialize_sentry_exchanges(self):
         """
         تحميل الصفقات المفتوحة من R2 عند بدء تشغيل النظام لاستكمال إدارتها.
+        تعتمد الآن على وظيفة المزامنة المركزية.
+        """
+        print("🛡️ [TradeManager] Initializing and syncing state with R2...")
+        await self.sync_internal_state()
+
+    async def sync_internal_state(self):
+        """
+        مزامنة الحالة الداخلية مع R2 لضمان عدم فقدان أي صفقة مفتوحة.
+        تستخدم عند بدء النظام وعند بدء كل دورة تحليل جديدة لضمان تطابق البيانات.
         """
         try:
+            # جلب الصفقات المسجلة في R2
             loaded_trades = await self.r2.get_open_trades_async()
+            
             if loaded_trades:
+                # إنشاء مجموعة من الرموز الموجودة في R2 للمقارنة
+                r2_symbols = set(t['symbol'] for t in loaded_trades)
+                
+                # 1. إضافة الصفقات المفقودة من الذاكرة
                 for t in loaded_trades:
                     symbol = t['symbol']
-                    self.open_positions[symbol] = t
-                print(f"🛡️ [TradeManager] Successfully resumed watching {len(self.open_positions)} active positions.")
-            else:
-                print("🛡️ [TradeManager] No active positions found in R2 to resume.")
+                    if symbol not in self.open_positions:
+                        self.open_positions[symbol] = t
+                        print(f"🔄 [TradeManager] SYNC: Resumed watching trade from R2: {symbol}")
+                
+                # 2. (اختياري) تنظيف الذاكرة من الصفقات التي أغلقت في R2 ولكنها علقت هنا
+                # يتم ذلك بمقارنة عكسية إذا تطلب الأمر مستقبلاً
+                
+            # التأكد من أن جميع الصفقات المفتوحة لديها حراس يعملون
+            await self.start_sentry_loops()
+            
         except Exception as e:
-            print(f"❌ [TradeManager] Error initializing trades from R2: {e}")
+            print(f"⚠️ [TradeManager] Critical State Sync Warning: {e}")
+            traceback.print_exc()
 
     async def update_sentry_watchlist(self, approved_candidates):
         """
-        تحديث قائمة المراقبة النشطة للحارس (Sniper) بناءً على ترشيحات الدماغ.
+        تحديث قائمة المراقبة النشطة للحارس (Sniper) بناءً على ترشيحات الدماغ الجديدة.
         """
-        # تفريغ القائمة القديمة وبدء قائمة جديدة
-        self.watchlist.clear()
+        # إيقاف المهام القديمة التي لم تعد في القائمة الجديدة (تنظيف)
+        new_symbols = set(c['symbol'] for c in approved_candidates)
+        for symbol in list(self.watchlist.keys()):
+            if symbol not in new_symbols:
+                if symbol in self.sentry_tasks:
+                    self.sentry_tasks[symbol].cancel()
+                del self.watchlist[symbol]
+
+        # إضافة المرشحين الجدد
         for cand in approved_candidates:
             symbol = cand['symbol']
-            self.watchlist[symbol] = {
-                'data': cand,
-                'added_at': datetime.utcnow().isoformat(),
-                'monitoring_started': False
-            }
-        print(f"📋 [Sniper] Watchlist updated with {len(self.watchlist)} elite targets.")
+            if symbol not in self.watchlist:
+                self.watchlist[symbol] = {
+                    'data': cand,
+                    'added_at': datetime.utcnow().isoformat(),
+                    'monitoring_started': False
+                }
+        
+        print(f"📋 [Sniper] Watchlist updated. Active targets: {len(self.watchlist)}")
         
         # بدء مهام المراقبة الجديدة فوراً
         for symbol in list(self.watchlist.keys()):
-            # إذا لم تكن المهمة تعمل بالفعل
+            # نتأكد من عدم وجود مهمة قائمة بالفعل لنفس الرمز
             if symbol not in self.sentry_tasks or self.sentry_tasks[symbol].done():
-                 print(f"🎯 [Sniper] Locking on target: {symbol}")
+                 print(f"🎯 [Sniper] Locking radar on target: {symbol}")
                  self.sentry_tasks[symbol] = asyncio.create_task(self._sniper_loop(symbol))
 
     async def _sniper_loop(self, symbol):
         """
-        حلقة المراقبة التكتيكية (القناص) لكل عملة في القائمة.
-        تنتظر اللحظة المناسبة للدخول بناءً على تأكيد زخم لحظي (5m).
+        حلقة المراقبة التكتيكية (القناص) - تراقب الرمز لحظة بلحظة لاقتناص فرصة الدخول.
         """
         print(f"👁️ [Sniper] Monitoring {symbol} for tactical entry...")
         consecutive_signals = 0
-        required_signals = 2 # عدد الإشارات المتتالية المطلوبة للتأكيد
+        required_signals = 2 # عدد الشموع المتتالية المطلوبة لتأكيد الزخم
         
         try:
             while self.running and symbol in self.watchlist and symbol not in self.open_positions:
-                # 1. فحص مبدئي سريع: هل وصلنا للحد الأقصى للصفقات؟
-                # (لتخفيف الضغط على الـ API إذا كانت المحفظة ممتلئة)
+                # 1. فحص سريع: هل وصلنا للحد الأقصى للصفقات؟
+                # إذا نعم، نهدئ اللعب لتوفير الموارد
                 if len(self.open_positions) >= 1:
-                     # print(f"💤 [Sniper] {symbol} on hold. Max positions (1) reached.")
-                     await asyncio.sleep(60) # انتظار أطول
+                     await asyncio.sleep(60) # انتظار دقيقة كاملة قبل الفحص التالي
                      continue
 
-                # 2. جلب أحدث بيانات لحظية (شموع 5 دقائق)
+                # 2. جلب بيانات الشموع اللحظية (5 دقائق)
                 ohlcv = await self.data_manager.get_latest_ohlcv(symbol, timeframe='5m', limit=30)
-                if not ohlcv:
+                if not ohlcv or len(ohlcv) < 20:
                     await asyncio.sleep(30)
                     continue
                 
-                # 3. تحليل فوري باستخدام Titan (نسخة سريعة)
+                # 3. تحليل فوري باستخدام محرك تيتان
                 titan_res = self.titan.predict({'5m': ohlcv})
                 current_score = titan_res.get('score', 0.0)
                 
-                # 4. تقييم شروط الدخول التكتيكي
+                # 4. تقييم الإشارة
                 is_signal = False
                 entry_reason = ""
                 
-                # الشرط: زخم قوي جداً على فريم 5 دقائق
+                # شرط الدخول: زخم قوي جداً (فوق 0.80)
                 if current_score > 0.80:
                      is_signal = True
                      entry_reason = f"Titan Surge ({current_score:.2f})"
@@ -100,55 +133,59 @@ class TradeManager:
                     print(f"🔥 [Sniper] {symbol} Signal detected! ({consecutive_signals}/{required_signals}) - {entry_reason}")
                     
                     if consecutive_signals >= required_signals:
-                        # محاولة تنفيذ الشراء (محمية بالقفل)
+                        # محاولة تنفيذ الشراء
                         current_price = ohlcv[-1][4] # آخر سعر إغلاق
+                        # نستخدم await هنا لأن الدالة execute_buy_order تحتوي على قفل async
                         executed = await self.execute_buy_order(symbol, entry_reason, current_price)
                         
                         if executed:
-                             # إذا تم الشراء بنجاح، نخرج من حلقة القناص
+                             # تم الشراء بنجاح، ننهي مهمة القناص هذه
+                             print(f"✅ [Sniper] Mission accomplished for {symbol}. Standing down.")
                              break 
                         else:
-                             # إذا فشل الشراء (مثلاً سبقتنا عملة أخرى)، نعيد تعيين الإشارات
+                             # فشل الشراء (ربما بسبب اكتمال العدد أو نقص الرصيد)، نعيد المحاولة لاحقاً
                              consecutive_signals = 0
                 else:
-                    # تبريد الإشارات إذا انخفض الزخم
+                    # إذا انقطعت سلسلة الإشارات، نعيد العداد للصفر
                     if consecutive_signals > 0:
                         # print(f"❄️ [Sniper] {symbol} signal cooled off.")
                         consecutive_signals = 0
                 
-                await asyncio.sleep(30) # فاصل زمني بين الفحوصات
+                await asyncio.sleep(30) # فاصل زمني 30 ثانية بين كل فحص
 
         except asyncio.CancelledError:
-            pass # إيقاف طبيعي للمهمة
+            # تم إلغاء المهمة يدوياً (مثلاً عند تحديث القائمة)
+            pass
         except Exception as e:
-            print(f"❌ [Sniper Error] Loop failed for {symbol}: {e}")
+            print(f"❌ [Sniper Error] Loop crashed for {symbol}: {e}")
+            traceback.print_exc()
 
     async def execute_buy_order(self, symbol, reason, price):
         """
-        تنفيذ أمر الشراء الفعلي.
-        محمي بـ execution_lock لمنع أي تضارب في حالة وجود إشارات متزامنة.
+        تنفيذ أمر الشراء. هذه الدالة حرجة جداً ومحمية بقفل (Lock).
         """
         async with self.execution_lock:
-            # 🛡️ التحقق المزدوج الحرج داخل القفل
-            # هل امتلأت المحفظة أثناء انتظارنا للقفل؟
+            # --- منطقة حرجة (Critical Section) ---
+            # لا يمكن لأي كود آخر تعديل الصفقات أو المحفظة أثناء وجودنا هنا
+            
+            # 1. التحقق النهائي من عدد الصفقات
             if len(self.open_positions) >= 1:
-                print(f"⚠️ [Execution Blocked] Cannot buy {symbol}. Max positions (1) reached just now by another trade.")
+                print(f"⚠️ [Execution Blocked] Cannot buy {symbol}. Max positions (1) reached.")
                 return False
 
-            # 1. جلب أحدث حالة للمحفظة من R2
+            # 2. جلب وتحديث المحفظة
             portfolio = await self.r2.get_portfolio_state_async()
             capital = portfolio.get('current_capital_usd', 0.0)
             
-            # التحقق من وجود رصيد كافٍ (حد أدنى 10 دولار مثلاً)
-            if capital < 10.0:
+            if capital < 5.0: # حد أدنى للرصيد (مثلاً 5 دولار)
                  print(f"⚠️ [Execution Failed] Insufficient capital for {symbol}: ${capital:.2f}")
                  return False
 
-            # 2. حساب كمية الشراء (استثمار شبه كامل للرصيد المتاح)
-            invest_amount = capital * 0.98 # ترك هامش 2% للرسوم أو فروقات الأسعار
+            # 3. حساب كمية الاستثمار
+            invest_amount = capital * 0.99 # استثمار 99% من الرصيد المتاح
             quantity = invest_amount / price
             
-            # 3. تجهيز سجل الصفقة الجديد
+            # 4. إنشاء سجل الصفقة
             trade_id = str(uuid.uuid4())[:8]
             new_trade = {
                 'trade_id': trade_id,
@@ -159,19 +196,17 @@ class TradeManager:
                 'quantity': quantity,
                 'invested_usd': invest_amount,
                 'entry_reason': reason,
-                # أهداف أولية (سيقوم الدماغ بتحديثها لاحقاً في دورة إعادة التقييم)
-                'tp_price': price * 1.05, # هدف ربح مبدئي +5%
-                'sl_price': price * 0.95, # وقف خسارة مبدئي -5%
-                'highest_price': price    # لتتبع الوقف المتحرك لاحقاً
+                'tp_price': price * 1.05, # هدف أولي +5%
+                'sl_price': price * 0.95, # وقف أولي -5%
+                'highest_price': price    # لتتبع الوقف المتحرك
             }
             
-            # 4. تحديث الحالة الداخلية
+            # 5. تحديث الحالة الداخلية
             self.open_positions[symbol] = new_trade
-            # إزالة من قائمة المراقبة لأنها أصبحت صفقة مفتوحة
             if symbol in self.watchlist:
                 del self.watchlist[symbol]
                 
-            # 5. تحديث المحفظة وحفظ جميع الملفات في R2 (عملية ذرية قدر الإمكان)
+            # 6. تحديث وحفظ البيانات في R2 (دفعة واحدة لضمان التناسق)
             portfolio['current_capital_usd'] -= invest_amount
             portfolio['invested_capital_usd'] += invest_amount
             portfolio['total_trades'] += 1
@@ -181,59 +216,65 @@ class TradeManager:
             
             print(f"🚀 [EXECUTED] BUY {symbol} @ {price:.4f} | Invested: ${invest_amount:.2f} | Reason: {reason}")
             
-            # 6. تشغيل الحارس (Guardian) لحماية الصفقة الجديدة
+            # 7. تشغيل الحارس فوراً
+            if symbol in self.sentry_tasks and not self.sentry_tasks[symbol].done():
+                self.sentry_tasks[symbol].cancel() # إلغاء مهمة القناص القديمة إن وجدت
             self.sentry_tasks[symbol] = asyncio.create_task(self._guardian_loop(symbol))
+            
             return True
 
     async def _guardian_loop(self, symbol):
         """
-        حلقة الحماية (Guardian) للصفقات المفتوحة.
-        تراقب الأسعار وتنفذ أوامر الخروج (TP/SL) تلقائياً.
+        حلقة الحماية (Guardian) - تراقب الصفقة المفتوحة وتدير الخروج.
         """
-        print(f"🛡️ [Guardian] Protecting position: {symbol}")
+        print(f"🛡️ [Guardian] Activated for position: {symbol}")
         try:
             while self.running and symbol in self.open_positions:
                 trade = self.open_positions[symbol]
                 
                 # جلب السعر الحالي
                 current_price = await self.data_manager.get_latest_price_async(symbol)
-                if not current_price:
-                    await asyncio.sleep(30) # في حال فشل جلب السعر ننتظر قليلاً
+                if not current_price or current_price <= 0:
+                    await asyncio.sleep(30)
                     continue
 
-                # تحديث أعلى سعر وصل له (للـ Trailing SL مستقبلاً)
+                # تحديث أعلى سعر وصل له (مهم للوقف المتحرك المستقبلي)
                 if current_price > trade.get('highest_price', 0):
                     trade['highest_price'] = current_price
-                    # لا نحفظ في R2 كل تحديث بسيط لتوفير الموارد، فقط في الذاكرة مؤقتاً
+                    # ملاحظة: لا نحفظ في R2 هنا لتجنب كثرة الكتابة، نعتمد على الذاكرة
 
-                # التحقق من شروط الخروج
+                # التحقق من شروط الخروج (SL / TP)
                 if current_price <= trade['sl_price']:
+                    print(f"🛡️ [Guardian] STOP LOSS triggered for {symbol} @ {current_price:.4f} (SL: {trade['sl_price']:.4f})")
                     await self.execute_sell_order(symbol, "Stop Loss Hit", current_price)
                     break
+                
                 elif current_price >= trade['tp_price']:
+                    print(f"🛡️ [Guardian] TAKE PROFIT triggered for {symbol} @ {current_price:.4f} (TP: {trade['tp_price']:.4f})")
                     await self.execute_sell_order(symbol, "Take Profit Hit", current_price)
                     break
                 
-                await asyncio.sleep(10) # مراقبة كثيفة كل 10 ثواني
+                await asyncio.sleep(10) # مراقبة نشطة كل 10 ثواني
 
         except asyncio.CancelledError:
-            pass
+            print(f"🛡️ [Guardian] Task cancelled for {symbol}")
         except Exception as e:
-             print(f"❌ [Guardian Error] Loop failed for {symbol}: {e}")
+             print(f"❌ [Guardian Error] Loop crashed for {symbol}: {e}")
+             traceback.print_exc()
 
     async def execute_sell_order(self, symbol, reason, price):
         """
-        تنفيذ أمر البيع وإغلاق الصفقة.
-        محمي أيضاً بالقفل لضمان سلامة بيانات المحفظة.
+        تنفيذ أمر البيع. محمي بـ execution_lock.
         """
         async with self.execution_lock:
+            # --- منطقة حرجة ---
             if symbol not in self.open_positions:
+                print(f"⚠️ [Sell Failed] {symbol} not found in open positions during execution.")
                 return
 
-            # استخراج بيانات الصفقة
             trade = self.open_positions.pop(symbol)
             
-            # حسابات الربح والخسارة
+            # حسابات الأرباح
             revenue = trade['quantity'] * price
             profit_usd = revenue - trade['invested_usd']
             profit_pct = (profit_usd / trade['invested_usd']) * 100
@@ -247,59 +288,56 @@ class TradeManager:
             if profit_usd > 0:
                 portfolio['winning_trades'] += 1
             
-            # حفظ التحديثات في R2
+            # حفظ التحديثات النهائية في R2
             await self.r2.save_portfolio_state_async(portfolio)
             await self.r2.save_open_trades_async(list(self.open_positions.values()))
 
             print(f"💰 [SOLD] {symbol} @ {price:.4f} | PnL: ${profit_usd:.2f} ({profit_pct:+.2f}%) | Reason: {reason}")
-            
-            # (اختياري) يمكن هنا إضافة كود لحفظ الصفقة في سجل تاريخي دائم (Trade History)
 
     async def execute_emergency_exit(self, symbol, reason):
-        """
-        دالة مساعدة للخروج الطارئ عند استلام أمر مباشر من الدماغ.
-        """
+        """واجهة للخروج الطارئ اليدوي أو من الدماغ"""
+        print(f"🚨 [Emergency] Requested exit for {symbol} due to: {reason}")
         current_price = await self.data_manager.get_latest_price_async(symbol)
-        if current_price:
-            print(f"🚨 [Emergency Exit] Initiating immediate sell for {symbol}...")
+        if current_price and current_price > 0:
             await self.execute_sell_order(symbol, f"EMERGENCY: {reason}", current_price)
         else:
-            print(f"❌ [Emergency Exit Failed] Could not fetch price for {symbol}")
+            print(f"❌ [Emergency Failed] Could not fetch valid price for {symbol}")
 
     async def update_trade_targets(self, symbol, new_tp, new_sl, reason):
-        """
-        تحديث أهداف جني الأرباح ووقف الخسارة لصفقة مفتوحة بناءً على إعادة تقييم الدماغ.
-        """
+        """تحديث أهداف الصفقة بناءً على أوامر الدماغ"""
         if symbol in self.open_positions:
             trade = self.open_positions[symbol]
+            old_tp = trade['tp_price']
+            old_sl = trade['sl_price']
+            
+            if new_tp is not None: trade['tp_price'] = float(new_tp)
+            if new_sl is not None: trade['sl_price'] = float(new_sl)
             
-            if new_tp is not None:
-                trade['tp_price'] = float(new_tp)
-            if new_sl is not None:
-                trade['sl_price'] = float(new_sl)
-                
             self.open_positions[symbol] = trade
-            # نحفظ التحديث فوراً في R2 لضمان استمراريته
+            # حفظ فوري لضمان عدم ضياع الأهداف الجديدة في حال إعادة التشغيل
             await self.r2.save_open_trades_async(list(self.open_positions.values()))
-            print(f"🎯 [Targets Updated] {symbol}: New TP={trade['tp_price']:.4f}, New SL={trade['sl_price']:.4f} | {reason}")
+            
+            print(f"🎯 [Targets Updated] {symbol} | TP: {old_tp:.4f}->{trade['tp_price']:.4f} | SL: {old_sl:.4f}->{trade['sl_price']:.4f} | Reason: {reason}")
         else:
-            print(f"⚠️ [Target Update Failed] {symbol} is not actively traded.")
+            print(f"⚠️ [Target Update Failed] {symbol} is not currently open.")
 
     async def start_sentry_loops(self):
-        """
-        بدء تشغيل جميع مهام المراقبة (تستخدم عند بدء النظام).
-        """
-        # إعادة تفعيل الحراس لأي صفقات كانت مفتوحة
+        """بدء أو استئناف جميع مهام الحراسة"""
         for symbol in list(self.open_positions.keys()):
+             # إذا لم يكن هناك حارس نشط، نعين واحداً
              if symbol not in self.sentry_tasks or self.sentry_tasks[symbol].done():
+                 print(f"🛡️ [Sentry Restart] Activating guardian for existing trade: {symbol}")
                  self.sentry_tasks[symbol] = asyncio.create_task(self._guardian_loop(symbol))
 
     async def stop_sentry_loops(self):
-        """
-        إيقاف جميع مهام المراقبة (عند إيقاف النظام).
-        """
+        """إيقاف نظيف لجميع المهام الخلفية"""
+        print("🛑 [TradeManager] Stopping all sentry loops...")
         self.running = False
-        for symbol, task in self.sentry_tasks.items():
+        for sym, task in self.sentry_tasks.items():
             task.cancel()
-        await asyncio.gather(*self.sentry_tasks.values(), return_exceptions=True)
-        print("🛡️ [TradeManager] All sentry loops stopped.")
+        
+        # انتظار انتهاء المهام (اختياري لكن أفضل للأمان)
+        if self.sentry_tasks:
+            await asyncio.gather(*self.sentry_tasks.values(), return_exceptions=True)
+        
+        print("✅ [TradeManager] All loops stopped.")