a

trading_system/position_manager.py

@@ -540,6 +540,32 @@ class PositionManager:
             quantity = abs(position_amt)
             position_side = 'LONG' if position_amt > 0 else 'SHORT'
 
+            # 二次校验：用币安实时持仓数量兜底，避免 reduceOnly 被拒绝（-2022）
+            live_amt = await self._get_live_position_amt(symbol, position_side=position_side)
+            if live_amt is None or abs(live_amt) <= 0:
+                logger.warning(f"{symbol} [平仓] 实时查询到持仓已为0，跳过下单并按已平仓处理")
+                # 复用“币安无持仓”的处理逻辑：走上面的分支
+                position = None
+                # 触发上方逻辑：直接返回 updated/清理
+                # 这里简单调用同步函数路径
+                ticker = await self.client.get_ticker_24h(symbol)
+                exit_price = float(ticker['price']) if ticker else float(position.get('entryPrice', 0) if position else 0)
+                await self._stop_position_monitoring(symbol)
+                if symbol in self.active_positions:
+                    del self.active_positions[symbol]
+                logger.info(f"{symbol} [平仓] 已清理本地记录（币安无持仓）")
+                return True
+
+            # 以币安实时持仓数量为准（并向下截断到不超过持仓）
+            quantity = min(quantity, abs(live_amt))
+            quantity = await self._adjust_close_quantity(symbol, quantity)
+            if quantity <= 0:
+                logger.warning(f"{symbol} [平仓] 数量调整后为0，跳过下单并清理本地记录")
+                await self._stop_position_monitoring(symbol)
+                if symbol in self.active_positions:
+                    del self.active_positions[symbol]
+                return True
+            
             logger.info(
                 f"{symbol} [平仓] 下单信息: {side} {quantity:.4f} @ MARKET "
                 f"(持仓数量: {position_amt:.4f})"
@@ -739,6 +765,80 @@ class PositionManager:
             
             return False
 
+    async def _get_live_position_amt(self, symbol: str, position_side: Optional[str] = None) -> Optional[float]:
+        """
+        从币安原始接口读取持仓数量（避免本地状态/缓存不一致导致 reduceOnly 被拒绝）。
+        - 单向模式：通常只有一个 net 持仓
+        - 对冲模式：可能同时有 LONG/SHORT，两条腿用 positionSide 区分
+        """
+        try:
+            if not getattr(self.client, "client", None):
+                return None
+            res = await self.client.client.futures_position_information(symbol=symbol)
+            if not isinstance(res, list):
+                return None
+            ps = (position_side or "").upper()
+            nonzero = []
+            for p in res:
+                if not isinstance(p, dict):
+                    continue
+                try:
+                    amt = float(p.get("positionAmt", 0))
+                except Exception:
+                    continue
+                if abs(amt) <= 0:
+                    continue
+                nonzero.append((amt, p))
+            if not nonzero:
+                return 0.0
+            if ps in ("LONG", "SHORT"):
+                for amt, p in nonzero:
+                    pps = (p.get("positionSide") or "").upper()
+                    if pps == ps:
+                        return amt
+                # 如果没匹配到 positionSide，退化为按符号推断
+                if ps == "LONG":
+                    cand = next((amt for amt, _ in nonzero if amt > 0), None)
+                    return cand if cand is not None else 0.0
+                if ps == "SHORT":
+                    cand = next((amt for amt, _ in nonzero if amt < 0), None)
+                    return cand if cand is not None else 0.0
+            # 没提供 position_side：返回净持仓（单向模式）
+            return sum([amt for amt, _ in nonzero])
+        except Exception as e:
+            logger.debug(f"{symbol} 读取实时持仓失败: {e}")
+            return None
+
+    async def _adjust_close_quantity(self, symbol: str, quantity: float) -> float:
+        """
+        平仓数量调整：只允许向下取整（避免超过持仓导致 reduceOnly 被拒绝）。
+        """
+        try:
+            symbol_info = await self.client.get_symbol_info(symbol)
+        except Exception:
+            symbol_info = None
+
+        try:
+            q = float(quantity)
+        except Exception:
+            return 0.0
+
+        if not symbol_info:
+            return max(0.0, q)
+
+        try:
+            step_size = float(symbol_info.get("stepSize", 0) or 0)
+        except Exception:
+            step_size = 0.0
+        qty_precision = int(symbol_info.get("quantityPrecision", 8) or 8)
+
+        if step_size and step_size > 0:
+            q = float(int(q / step_size)) * step_size
+        else:
+            q = round(q, qty_precision)
+        q = round(q, qty_precision)
+        return max(0.0, q)
+    
     async def check_stop_loss_take_profit(self) -> List[str]:
         """
         检查止损止盈
@@ -965,6 +1065,16 @@ class PositionManager:
                             # 部分平仓
                             close_side = 'SELL' if position_info['side'] == 'BUY' else 'BUY'
                             close_position_side = 'LONG' if position_info['side'] == 'BUY' else 'SHORT'
+                            # 二次校验并截断数量，避免 reduceOnly 被拒绝（-2022）
+                            live_amt = await self._get_live_position_amt(symbol, position_side=close_position_side)
+                            if live_amt is None or abs(live_amt) <= 0:
+                                logger.warning(f"{symbol} 部分止盈：实时持仓已为0，跳过部分平仓")
+                                continue
+                            partial_quantity = min(partial_quantity, abs(live_amt))
+                            partial_quantity = await self._adjust_close_quantity(symbol, partial_quantity)
+                            if partial_quantity <= 0:
+                                logger.warning(f"{symbol} 部分止盈：数量调整后为0，跳过")
+                                continue
                             partial_order = await self.client.place_order(
                                 symbol=symbol,
                                 side=close_side,
@@ -1692,18 +1802,17 @@ class PositionManager:
         Args:
             symbol: 交易对
         """
-        if symbol not in self._monitor_tasks:
+        # 幂等：可能会被多处/并发调用，先 pop 再处理，避免 KeyError
+        task = self._monitor_tasks.pop(symbol, None)
+        if task is None:
             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):