Update trade_manager.py

trade_manager.py

@@ -1,5 +1,7 @@
 import asyncio
 import json
+import time
+import traceback
 from datetime import datetime, timedelta
 from helpers import safe_float_conversion, _apply_patience_logic
 
@@ -10,6 +12,8 @@ class TradeManager:
         self.data_manager = data_manager
         self.monitoring_tasks = {}
         self.is_running = False
+        self.monitoring_errors = {}
+        self.max_consecutive_errors = 5
 
     async def open_trade(self, symbol, decision, current_price):
         try:
@@ -17,6 +21,7 @@ class TradeManager:
             available_capital = portfolio_state.get("current_capital_usd", 0)
 
             if available_capital < 1:
+                print(f"❌ رأس المال غير كافي لفتح صفقة لـ {symbol}: {available_capital}")
                 return None
 
             expected_target_minutes = decision.get('expected_target_minutes', 15)
@@ -42,7 +47,8 @@ class TradeManager:
                 "expected_target_minutes": expected_target_minutes,
                 "expected_target_time": expected_target_time,
                 "is_monitored": True,
-                "strategy": strategy
+                "strategy": strategy,
+                "monitoring_started": False
             }
             
             trades.append(new_trade)
@@ -62,13 +68,21 @@ class TradeManager:
                 "strategy": strategy
             })
 
+            print(f"✅ تم فتح صفقة جديدة لـ {symbol} باستراتيجية {strategy}")
             return new_trade
 
         except Exception as e:
+            print(f"❌ فشل فتح صفقة لـ {symbol}: {e}")
+            await self.r2_service.save_system_logs_async({
+                "trade_open_failed": True,
+                "symbol": symbol,
+                "error": str(e)
+            })
             raise
 
     async def close_trade(self, trade_to_close, close_price, reason="إغلاق بالنظام"):
         try:
+            symbol = trade_to_close.get('symbol')
             trade_to_close['status'] = 'CLOSED'
             trade_to_close['close_price'] = close_price
             trade_to_close['close_timestamp'] = datetime.now().isoformat()
@@ -122,9 +136,15 @@ class TradeManager:
             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.monitoring_tasks:
+                del self.monitoring_tasks[symbol]
+            if symbol in self.monitoring_errors:
+                del self.monitoring_errors[symbol]
+            
             await self.r2_service.save_system_logs_async({
                 "trade_closed": True,
-                "symbol": trade_to_close.get('symbol'),
+                "symbol": symbol,
                 "entry_price": entry_price,
                 "close_price": close_price,
                 "pnl_usd": pnl,
@@ -139,13 +159,21 @@ class TradeManager:
             if self.learning_engine and self.learning_engine.initialized:
                 await self.learning_engine.analyze_trade_outcome(trade_to_close, reason)
 
+            print(f"✅ تم إغلاق صفقة {symbol} - السبب: {reason} - الربح: {pnl_percent:+.2f}%")
             return True
 
         except Exception as e:
+            print(f"❌ فشل إغلاق صفقة {trade_to_close.get('symbol')}: {e}")
+            await self.r2_service.save_system_logs_async({
+                "trade_close_failed": True,
+                "symbol": trade_to_close.get('symbol'),
+                "error": str(e)
+            })
             raise
 
     async def update_trade(self, trade_to_update, re_analysis_decision):
         try:
+            symbol = trade_to_update.get('symbol')
             if re_analysis_decision.get('new_stop_loss'):
                 trade_to_update['stop_loss'] = re_analysis_decision['new_stop_loss']
             if re_analysis_decision.get('new_take_profit'):
@@ -175,15 +203,17 @@ class TradeManager:
 
             await self.r2_service.save_system_logs_async({
                 "trade_updated": True,
-                "symbol": trade_to_update.get('symbol'),
+                "symbol": symbol,
                 "new_expected_minutes": new_expected_minutes,
                 "action": "UPDATE_TRADE",
                 "strategy": original_strategy
             })
 
+            print(f"✅ تم تحديث صفقة {symbol}")
             return True
 
         except Exception as e:
+            print(f"❌ فشل تحديث صفقة {trade_to_update.get('symbol')}: {e}")
             raise
 
     async def immediate_close_trade(self, symbol, close_price, reason="المراقبة الفورية"):
@@ -197,6 +227,7 @@ class TradeManager:
                     break
             
             if not trade_to_close:
+                print(f"⚠️ لم يتم العثور على صفقة مفتوحة لـ {symbol}")
                 return False
             
             await self.close_trade(trade_to_close, close_price, reason)
@@ -204,38 +235,87 @@ class TradeManager:
             return True
             
         except Exception as e:
+            print(f"❌ فشل الإغلاق الفوري لـ {symbol}: {e}")
             return False
 
     async def start_trade_monitoring(self):
         self.is_running = True
+        print("🔍 بدء مراقبة الصفقات...")
+        
         while self.is_running:
             try:
                 open_trades = await self.r2_service.get_open_trades_async()
+                current_time = time.time()
+                
                 for trade in open_trades:
                     symbol = trade['symbol']
+                    
+                    # تخطي الصفقات التي تجاوزت حد الأخطاء
+                    if self.monitoring_errors.get(symbol, 0) >= self.max_consecutive_errors:
+                        if symbol in self.monitoring_tasks:
+                            del self.monitoring_tasks[symbol]
+                        continue
+                    
+                    # بدء المراقبة إذا لم تكن نشطة
                     if symbol not in self.monitoring_tasks:
-                        asyncio.create_task(self._monitor_single_trade(trade))
-                        self.monitoring_tasks[symbol] = trade
+                        task = asyncio.create_task(self._monitor_single_trade(trade))
+                        self.monitoring_tasks[symbol] = {
+                            'task': task,
+                            'start_time': current_time,
+                            'trade': trade
+                        }
+                        trade['monitoring_started'] = True
                 
+                # تنظيف المهام المنتهية
                 current_symbols = {trade['symbol'] for trade in open_trades}
                 for symbol in list(self.monitoring_tasks.keys()):
                     if symbol not in current_symbols:
+                        task_info = self.monitoring_tasks[symbol]
+                        if not task_info['task'].done():
+                            task_info['task'].cancel()
                         del self.monitoring_tasks[symbol]
+                        if symbol in self.monitoring_errors:
+                            del self.monitoring_errors[symbol]
                 
                 await asyncio.sleep(10)
+                
             except Exception as error:
+                print(f"❌ خطأ في مراقبة الصفقات: {error}")
                 await asyncio.sleep(30)
 
     async def _monitor_single_trade(self, trade):
         symbol = trade['symbol']
-        while symbol in self.monitoring_tasks and self.is_running:
+        max_monitoring_time = 3600  # أقصى وقت مراقبة: ساعة واحدة
+        
+        print(f"🔍 بدء مراقبة الصفقة: {symbol}")
+        
+        while (symbol in self.monitoring_tasks and 
+               self.is_running and 
+               self.monitoring_errors.get(symbol, 0) < self.max_consecutive_errors):
+            
             try:
+                start_time = time.time()
+                
                 if not self.data_manager:
+                    print(f"⚠️ DataManager غير متوفر لـ {symbol}")
+                    await asyncio.sleep(15)
+                    continue
+                    
+                # الحصول على السعر الحالي مع مهلة زمنية
+                try:
+                    current_price = await asyncio.wait_for(
+                        self.data_manager.get_latest_price_async(symbol), 
+                        timeout=10
+                    )
+                except asyncio.TimeoutError:
+                    print(f"⏰ مهلة انتظار السعر لـ {symbol}")
+                    self._increment_monitoring_error(symbol)
                     await asyncio.sleep(15)
                     continue
                     
-                current_price = await self.data_manager.get_latest_price_async(symbol)
                 if not current_price:
+                    print(f"⚠️ لم يتم الحصول على سعر لـ {symbol}")
+                    self._increment_monitoring_error(symbol)
                     await asyncio.sleep(15)
                     continue
                     
@@ -244,34 +324,77 @@ class TradeManager:
                 take_profit = trade.get('take_profit')
                 should_close, close_reason = False, ""
                 
+                # التحقق من شروط الإغلاق
                 if stop_loss and current_price <= stop_loss:
                     should_close, close_reason = True, f"وصول وقف الخسارة: {current_price} <= {stop_loss}"
                 elif take_profit and current_price >= take_profit:
                     should_close, close_reason = True, f"وصول جني الأرباح: {current_price} >= {take_profit}"
                 
+                # تحديث وقف الخسارة الديناميكي
                 if not should_close and current_price > entry_price:
                     dynamic_stop = current_price * 0.98
                     if dynamic_stop > (stop_loss or 0):
                         trade['stop_loss'] = dynamic_stop
                 
+                # إغلاق الصفقة إذا لزم الأمر
                 if should_close:
                     if self.r2_service.acquire_lock():
                         try:
                             await self.immediate_close_trade(symbol, current_price, close_reason)
+                        except Exception as close_error:
+                            print(f"❌ فشل الإغلاق التلقائي لـ {symbol}: {close_error}")
                         finally:
                             self.r2_service.release_lock()
                     
-                    if symbol in self.monitoring_tasks:
-                        del self.monitoring_tasks[symbol]
+                    break
+                
+                # إعادة تعيين عداد الأخطاء عند النجاح
+                if symbol in self.monitoring_errors:
+                    self.monitoring_errors[symbol] = 0
+                    
+                # التحقق من وقت المراقبة الطويل
+                monitoring_duration = time.time() - start_time
+                if monitoring_duration > max_monitoring_time:
+                    print(f"🕒 انتهى وقت مراقبة الصفقة {symbol}")
                     break
                     
                 await asyncio.sleep(15)
+                
             except Exception as error:
+                error_count = self._increment_monitoring_error(symbol)
+                print(f"❌ خطأ في مراقبة {symbol} (الخطأ #{error_count}): {error}")
+                
+                if error_count >= self.max_consecutive_errors:
+                    print(f"🚨 إيقاف مراقبة {symbol} بسبب الأخطاء المتتالية")
+                    await self.r2_service.save_system_logs_async({
+                        "monitoring_stopped": True,
+                        "symbol": symbol,
+                        "error_count": error_count,
+                        "error": str(error)
+                    })
+                    break
+                    
                 await asyncio.sleep(30)
 
+        print(f"🛑 توقيف مراقبة الصفقة: {symbol}")
+
+    def _increment_monitoring_error(self, symbol):
+        if symbol not in self.monitoring_errors:
+            self.monitoring_errors[symbol] = 0
+        self.monitoring_errors[symbol] += 1
+        return self.monitoring_errors[symbol]
+
     def stop_monitoring(self):
         self.is_running = False
+        print("🛑 إيقاف جميع مهام المراقبة...")
+        
+        for symbol, task_info in self.monitoring_tasks.items():
+            if not task_info['task'].done():
+                task_info['task'].cancel()
+                print(f"✅ تم إلغاء مهمة مراقبة {symbol}")
+        
         self.monitoring_tasks.clear()
+        self.monitoring_errors.clear()
 
     async def _archive_closed_trade(self, closed_trade):
         try:
@@ -284,12 +407,16 @@ class TradeManager:
             
             history.append(closed_trade)
             
+            # حفظ آخر 1000 صفقة فقط
+            if len(history) > 1000:
+                history = history[-1000:]
+            
             data_json = json.dumps(history, indent=2).encode('utf-8')
             self.r2_service.s3_client.put_object(
                 Bucket="trading", Key=key, Body=data_json, ContentType="application/json"
             )
         except Exception as e:
-            pass
+            print(f"❌ فشل أرشفة الصفقة: {e}")
 
     async def _update_trade_summary(self, closed_trade):
         try:
@@ -302,7 +429,8 @@ class TradeManager:
                     "total_trades": 0, "winning_trades": 0, "losing_trades": 0,
                     "total_profit_usd": 0.0, "total_loss_usd": 0.0, "win_percentage": 0.0,
                     "avg_profit_per_trade": 0.0, "avg_loss_per_trade": 0.0,
-                    "largest_win": 0.0, "largest_loss": 0.0
+                    "largest_win": 0.0, "largest_loss": 0.0,
+                    "last_updated": datetime.now().isoformat()
                 }
             
             pnl = closed_trade.get('pnl_usd', 0.0)
@@ -327,20 +455,38 @@ class TradeManager:
             if summary['losing_trades'] > 0:
                 summary['avg_loss_per_trade'] = summary['total_loss_usd'] / summary['losing_trades']
             
+            summary['last_updated'] = datetime.now().isoformat()
+            
             data_json = json.dumps(summary, indent=2).encode('utf-8')
             self.r2_service.s3_client.put_object(
                 Bucket="trading", Key=key, Body=data_json, ContentType="application/json"
             )
 
         except Exception as e:
-            pass
+            print(f"❌ فشل تحديث ملخص التداول: {e}")
 
     async def get_open_trades(self):
-        return await self.r2_service.get_open_trades_async()
+        try:
+            return await self.r2_service.get_open_trades_async()
+        except Exception as e:
+            print(f"❌ فشل جلب الصفقات المفتوحة: {e}")
+            return []
 
     async def get_trade_by_symbol(self, symbol):
-        open_trades = await self.get_open_trades()
-        for trade in open_trades:
-            if trade['symbol'] == symbol and trade['status'] == 'OPEN':
-                return trade
-        return None
+        try:
+            open_trades = await self.get_open_trades()
+            for trade in open_trades:
+                if trade['symbol'] == symbol and trade['status'] == 'OPEN':
+                    return trade
+            return None
+        except Exception as e:
+            print(f"❌ فشل البحث عن صفقة {symbol}: {e}")
+            return None
+
+    def get_monitoring_status(self):
+        return {
+            'is_running': self.is_running,
+            'active_tasks': len(self.monitoring_tasks),
+            'monitoring_errors': dict(self.monitoring_errors),
+            'max_consecutive_errors': self.max_consecutive_errors
+        }