a

trading_system/position_manager.py

@@ -1,6 +1,7 @@
 """
 仓位管理模块 - 管理持仓和订单
 """
+import asyncio
 import logging
 from typing import Dict, List, Optional
 try:
@@ -54,6 +55,8 @@ class PositionManager:
         self.client = client
         self.risk_manager = risk_manager
         self.active_positions: Dict[str, Dict] = {}
+        self._monitor_tasks: Dict[str, asyncio.Task] = {}  # WebSocket监控任务字典
+        self._monitoring_enabled = True  # 是否启用实时监控
     
     async def open_position(
         self, 
@@ -191,6 +194,10 @@ class PositionManager:
                     f"(涨跌幅: {change_percent:.2f}%)"
                 )
                 
+                # 启动WebSocket实时监控
+                if self._monitoring_enabled:
+                    await self._start_position_monitoring(symbol)
+                
                 return position_info
             
             return None
@@ -279,6 +286,9 @@ class PositionManager:
                 elif not Trade:
                     logger.warning(f"Trade模型未导入，无法更新 {symbol} 平仓记录")
                 
+                # 停止WebSocket监控
+                await self._stop_position_monitoring(symbol)
+                
                 # 移除持仓记录
                 if symbol in self.active_positions:
                     del self.active_positions[symbol]
@@ -314,6 +324,7 @@ class PositionManager:
                 
                 current_position = position_dict[symbol]
                 entry_price = position_info['entryPrice']
+                quantity = position_info['quantity']  # 修复：获取quantity
                 # 获取当前标记价格
                 current_price = current_position.get('markPrice', 0)
                 if current_price == 0:
@@ -504,3 +515,257 @@ class PositionManager:
         except Exception as e:
             logger.error(f"获取持仓摘要失败: {e}")
             return {}
+    
+    async def start_all_position_monitoring(self):
+        """
+        启动所有持仓的WebSocket实时监控
+        """
+        if not self._monitoring_enabled:
+            logger.info("实时监控已禁用，跳过启动")
+            return
+        
+        if not self.client.socket_manager:
+            logger.warning("WebSocket未初始化，无法启动实时监控")
+            return
+        
+        # 获取当前所有持仓
+        positions = await self.client.get_open_positions()
+        for position in positions:
+            symbol = position['symbol']
+            if symbol not in self._monitor_tasks and symbol in self.active_positions:
+                await self._start_position_monitoring(symbol)
+        
+        logger.info(f"已启动 {len(self._monitor_tasks)} 个持仓的实时监控")
+    
+    async def stop_all_position_monitoring(self):
+        """
+        停止所有持仓的WebSocket监控
+        """
+        symbols = list(self._monitor_tasks.keys())
+        for symbol in symbols:
+            await self._stop_position_monitoring(symbol)
+        logger.info(f"已停止所有持仓监控 ({len(symbols)} 个)")
+    
+    async def _start_position_monitoring(self, symbol: str):
+        """
+        启动单个持仓的WebSocket价格监控
+        
+        Args:
+            symbol: 交易对
+        """
+        if symbol in self._monitor_tasks:
+            logger.debug(f"{symbol} 监控任务已存在，跳过")
+            return
+        
+        if not self.client.socket_manager:
+            logger.warning(f"{symbol} WebSocket未初始化，无法启动监控")
+            return
+        
+        try:
+            task = asyncio.create_task(self._monitor_position_price(symbol))
+            self._monitor_tasks[symbol] = task
+            logger.info(f"✓ 启动 {symbol} WebSocket实时价格监控")
+        except Exception as e:
+            logger.error(f"启动 {symbol} 监控失败: {e}")
+    
+    async def _stop_position_monitoring(self, symbol: str):
+        """
+        停止单个持仓的WebSocket监控
+        
+        Args:
+            symbol: 交易对
+        """
+        if symbol not in self._monitor_tasks:
+            return
+        
+        task = self._monitor_tasks[symbol]
+        if not task.done():
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+        
+        del self._monitor_tasks[symbol]
+        logger.debug(f"已停止 {symbol} 的WebSocket监控")
+    
+    async def _monitor_position_price(self, symbol: str):
+        """
+        监控单个持仓的价格（WebSocket实时推送）
+        
+        Args:
+            symbol: 交易对
+        """
+        retry_count = 0
+        max_retries = 5
+        
+        while retry_count < max_retries:
+            try:
+                if symbol not in self.active_positions:
+                    logger.info(f"{symbol} 持仓已不存在，停止监控")
+                    break
+                
+                # 使用WebSocket订阅价格流
+                async with self.client.socket_manager.futures_socket(symbol.lower()) as stream:
+                    logger.debug(f"{symbol} WebSocket连接已建立，开始接收价格更新")
+                    retry_count = 0  # 连接成功，重置重试计数
+                    
+                    async for msg in stream:
+                        if symbol not in self.active_positions:
+                            logger.info(f"{symbol} 持仓已不存在，停止监控")
+                            break
+                        
+                        if 'data' in msg:
+                            try:
+                                current_price = float(msg['data']['c'])  # 最新价格
+                                # 立即检查止损止盈
+                                await self._check_single_position(symbol, current_price)
+                            except (KeyError, ValueError, TypeError) as e:
+                                logger.debug(f"{symbol} 解析价格数据失败: {e}")
+                                continue
+                
+            except asyncio.CancelledError:
+                logger.info(f"{symbol} 监控任务已取消")
+                break
+            except Exception as e:
+                retry_count += 1
+                logger.warning(f"{symbol} WebSocket监控出错 (重试 {retry_count}/{max_retries}): {e}")
+                
+                if retry_count < max_retries:
+                    # 指数退避重试
+                    wait_time = min(2 ** retry_count, 30)
+                    logger.info(f"{symbol} {wait_time}秒后重试连接...")
+                    await asyncio.sleep(wait_time)
+                else:
+                    logger.error(f"{symbol} WebSocket监控失败，已达到最大重试次数")
+                    # 回退到定时检查模式
+                    logger.info(f"{symbol} 将使用定时检查模式（非实时）")
+                    break
+        
+        # 清理任务
+        if symbol in self._monitor_tasks:
+            del self._monitor_tasks[symbol]
+    
+    async def _check_single_position(self, symbol: str, current_price: float):
+        """
+        检查单个持仓的止损止盈（实时检查）
+        
+        Args:
+            symbol: 交易对
+            current_price: 当前价格
+        """
+        position_info = self.active_positions.get(symbol)
+        if not position_info:
+            return
+        
+        entry_price = position_info['entryPrice']
+        quantity = position_info['quantity']
+        
+        # 计算当前盈亏
+        if position_info['side'] == 'BUY':
+            pnl_percent = ((current_price - entry_price) / entry_price) * 100
+        else:  # SELL
+            pnl_percent = ((entry_price - current_price) / entry_price) * 100
+        
+        # 更新最大盈利
+        if pnl_percent > position_info.get('maxProfit', 0):
+            position_info['maxProfit'] = pnl_percent
+        
+        # 移动止损逻辑（盈利后保护利润）
+        use_trailing = config.TRADING_CONFIG.get('USE_TRAILING_STOP', True)
+        if use_trailing:
+            trailing_activation = config.TRADING_CONFIG.get('TRAILING_STOP_ACTIVATION', 0.01)
+            trailing_protect = config.TRADING_CONFIG.get('TRAILING_STOP_PROTECT', 0.01)
+            
+            if not position_info.get('trailingStopActivated', False):
+                # 盈利超过阈值后，激活移动止损
+                if pnl_percent > trailing_activation * 100:
+                    position_info['trailingStopActivated'] = True
+                    # 将止损移至成本价（保本）
+                    position_info['stopLoss'] = entry_price
+                    logger.info(
+                        f"{symbol} [实时监控] 移动止损激活: 止损移至成本价 {entry_price:.4f} "
+                        f"(盈利: {pnl_percent:.2f}%)"
+                    )
+            else:
+                # 盈利超过2%后，止损移至保护利润位
+                if pnl_percent > 2.0:
+                    if position_info['side'] == 'BUY':
+                        new_stop_loss = entry_price * (1 + trailing_protect)
+                        if new_stop_loss > position_info['stopLoss']:
+                            position_info['stopLoss'] = new_stop_loss
+                            logger.info(
+                                f"{symbol} [实时监控] 移动止损更新: {new_stop_loss:.4f} "
+                                f"(保护{trailing_protect*100:.1f}%利润)"
+                            )
+                    else:  # SELL
+                        new_stop_loss = entry_price * (1 - trailing_protect)
+                        if new_stop_loss < position_info['stopLoss']:
+                            position_info['stopLoss'] = new_stop_loss
+                            logger.info(
+                                f"{symbol} [实时监控] 移动止损更新: {new_stop_loss:.4f} "
+                                f"(保护{trailing_protect*100:.1f}%利润)"
+                            )
+        
+        # 检查止损
+        stop_loss = position_info['stopLoss']
+        should_close = False
+        exit_reason = None
+        
+        if position_info['side'] == 'BUY' and current_price <= stop_loss:
+            should_close = True
+            exit_reason = 'trailing_stop' if position_info.get('trailingStopActivated') else 'stop_loss'
+            logger.warning(
+                f"{symbol} [实时监控] 触发止损: {current_price:.4f} <= {stop_loss:.4f} "
+                f"(盈亏: {pnl_percent:.2f}%)"
+            )
+        elif position_info['side'] == 'SELL' and current_price >= stop_loss:
+            should_close = True
+            exit_reason = 'trailing_stop' if position_info.get('trailingStopActivated') else 'stop_loss'
+            logger.warning(
+                f"{symbol} [实时监控] 触发止损: {current_price:.4f} >= {stop_loss:.4f} "
+                f"(盈亏: {pnl_percent:.2f}%)"
+            )
+        
+        # 检查止盈
+        if not should_close:
+            take_profit = position_info['takeProfit']
+            if position_info['side'] == 'BUY' and current_price >= take_profit:
+                should_close = True
+                exit_reason = 'take_profit'
+                logger.info(
+                    f"{symbol} [实时监控] 触发止盈: {current_price:.4f} >= {take_profit:.4f} "
+                    f"(盈亏: {pnl_percent:.2f}%)"
+                )
+            elif position_info['side'] == 'SELL' and current_price <= take_profit:
+                should_close = True
+                exit_reason = 'take_profit'
+                logger.info(
+                    f"{symbol} [实时监控] 触发止盈: {current_price:.4f} <= {take_profit:.4f} "
+                    f"(盈亏: {pnl_percent:.2f}%)"
+                )
+        
+        # 如果触发止损止盈，执行平仓
+        if should_close:
+            # 更新数据库
+            if DB_AVAILABLE and Trade:
+                trade_id = position_info.get('tradeId')
+                if trade_id:
+                    try:
+                        if position_info['side'] == 'BUY':
+                            pnl = (current_price - entry_price) * quantity
+                        else:  # SELL
+                            pnl = (entry_price - current_price) * quantity
+                        
+                        Trade.update_exit(
+                            trade_id=trade_id,
+                            exit_price=current_price,
+                            exit_reason=exit_reason,
+                            pnl=pnl,
+                            pnl_percent=pnl_percent
+                        )
+                    except Exception as e:
+                        logger.warning(f"更新{exit_reason}记录失败: {e}")
+            
+            # 执行平仓（注意：这里会停止监控，所以先更新数据库）
+            await self.close_position(symbol)

trading_system/strategy.py

@@ -44,6 +44,7 @@ class TradingStrategy:
         self.risk_manager = risk_manager
         self.position_manager = position_manager
         self.running = False
+        self._monitoring_started = False  # 是否已启动监控
     
     async def execute_strategy(self):
         """
@@ -52,6 +53,14 @@ class TradingStrategy:
         self.running = True
         logger.info("交易策略开始执行...")
         
+        # 启动所有现有持仓的WebSocket实时监控
+        if not self._monitoring_started:
+            try:
+                await self.position_manager.start_all_position_monitoring()
+                self._monitoring_started = True
+            except Exception as e:
+                logger.warning(f"启动持仓监控失败: {e}，将使用定时检查模式")
+        
         try:
             while self.running:
                 # 1. 扫描市场，找出涨跌幅最大的前N个货币对
@@ -138,14 +147,18 @@ class TradingStrategy:
                     
                     if position:
                         logger.info(f"{symbol} 开仓成功: {trade_direction} ({entry_reason})")
+                        # 开仓成功后，WebSocket监控会在position_manager中自动启动
                     else:
                         logger.warning(f"{symbol} 开仓失败")
                     
                     # 避免同时处理太多交易对
                     await asyncio.sleep(1)
                 
-                # 3. 检查止损止盈
+                # 3. 检查止损止盈（作为备用检查，WebSocket实时监控是主要方式）
-                await self.position_manager.check_stop_loss_take_profit()
+                # 注意：如果启用了WebSocket实时监控，这里主要是作为备用检查
+                closed = await self.position_manager.check_stop_loss_take_profit()
+                if closed:
+                    logger.info(f"定时检查触发平仓: {', '.join(closed)}")
                 
                 # 4. 打印持仓摘要并记录账户快照
                 summary = await self.position_manager.get_position_summary()
@@ -197,6 +210,11 @@ class TradingStrategy:
             logger.error(f"策略执行出错: {e}", exc_info=True)
         finally:
             self.running = False
+            # 停止所有持仓的WebSocket监控
+            try:
+                await self.position_manager.stop_all_position_monitoring()
+            except Exception as e:
+                logger.warning(f"停止持仓监控时出错: {e}")
             logger.info("交易策略已停止")
     
     async def _check_volume_confirmation(self, symbol_info: Dict) -> bool: