Fix SHFE close by retrying alternate today/yesterday offset on CTP reject.

Split mixed positions and auto-fallback when vnpy td/yd volumes disagree with the exchange.

Co-authored-by: Cursor <cursoragent@cursor.com>

modules/ctp/vnpy_bridge.py

@@ -986,48 +986,183 @@ class CtpBridge:
             logger.debug("find position: %s", exc)
         return None
 
-    def _resolve_close_offset(self, sym: str, ex_name: str, hold_direction: str, lots: int) -> Any:
+    def _position_td_yd(
+        self, sym: str, ex_name: str, hold_direction: str,
+    ) -> tuple[int, int, int, int]:
+        """返回 (今仓, 昨仓, 冻结, 总持仓)。"""
+        sym_l = (sym or "").lower()
+        hold = (hold_direction or "long").strip().lower()
+        for p in self._collect_positions():
+            ps = (p.get("symbol") or "").lower()
+            if ps != sym_l:
+                continue
+            if (p.get("direction") or "long").strip().lower() != hold:
+                continue
+            vol = int(p.get("lots") or 0)
+            td = int(p.get("td_volume") or 0)
+            yd = int(p.get("yd_volume") or 0)
+            frozen = int(p.get("frozen") or 0)
+            if td + yd <= 0 and vol > 0:
+                td = max(0, vol - yd)
+            return td, yd, frozen, vol
+        pos = self._find_position(sym, ex_name, hold_direction)
+        if not pos:
+            return 0, 0, 0, 0
+        vol = int(getattr(pos, "volume", 0) or 0)
+        yd = int(getattr(pos, "yd_volume", 0) or 0)
+        td_attr = getattr(pos, "today_volume", None)
+        if td_attr is not None:
+            td = int(td_attr or 0)
+            yd = max(0, vol - td)
+        else:
+            td = max(0, vol - yd)
+        frozen = int(getattr(pos, "frozen", 0) or 0)
+        return td, yd, frozen, vol
+
+    def _resolve_close_legs(
+        self, sym: str, ex_name: str, hold_direction: str, lots: int,
+    ) -> list[tuple[Any, int]]:
         from vnpy.trader.constant import Offset
 
         ex_u = (ex_name or "").upper()
-        # 上期所/能源中心/郑商所/中金所须区分平今/平昨；大商所等可用通用 CLOSE
+        lots = max(1, int(lots or 1))
         if ex_u not in ("CZCE", "CFFEX", "SHFE", "INE"):
-            return Offset.CLOSE
+            return [(Offset.CLOSE, lots)]
-        pos = self._find_position(sym, ex_u, hold_direction)
+        td, yd, frozen, vol = self._position_td_yd(sym, ex_name, hold_direction)
-        if not pos:
+        td_close = max(0, td)
-            for p in self._collect_positions():
+        yd_close = max(0, yd)
-                ps = (p.get("symbol") or "").lower()
+        if frozen > 0:
-                if ps != sym.lower():
+            cut = min(frozen, td_close)
+            td_close -= cut
+            frozen -= cut
+            yd_close = max(0, yd_close - frozen)
+        logger.debug(
+            "close legs %s %s hold=%s lots=%s td=%s yd=%s frozen=%s vol=%s",
+            sym, ex_u, hold_direction, lots, td_close, yd_close, frozen, vol,
+        )
+        legs: list[tuple[Any, int]] = []
+        remain = lots
+        if td_close > 0 and remain > 0:
+            take = min(remain, td_close)
+            legs.append((Offset.CLOSETODAY, take))
+            remain -= take
+        if yd_close > 0 and remain > 0:
+            take = min(remain, yd_close)
+            legs.append((Offset.CLOSEYESTERDAY, take))
+            remain -= take
+        if remain > 0:
+            # vnpy 今昨拆分不可靠时，先按今仓尝试，拒单后自动改平昨
+            legs.append((Offset.CLOSETODAY, remain))
+        return legs if legs else [(Offset.CLOSETODAY, lots)]
+
+    def _resolve_close_offset(self, sym: str, ex_name: str, hold_direction: str, lots: int) -> Any:
+        legs = self._resolve_close_legs(sym, ex_name, hold_direction, lots)
+        return legs[0][0]
+
+    def _wait_order_terminal(self, vt_orderid: str, *, timeout: float = 2.5) -> dict[str, Any]:
+        if not self._engine or not vt_orderid:
+            return {"pending": True}
+        deadline = time.time() + max(0.5, float(timeout))
+        while time.time() < deadline:
+            try:
+                order = self._engine.get_order(vt_orderid)
+            except Exception:
+                order = None
+            if order is None:
+                time.sleep(0.08)
                 continue
-                if (p.get("direction") or "long") != hold_direction:
+            status_s = str(getattr(order, "status", "") or "")
+            traded = int(getattr(order, "traded", 0) or 0)
+            vol = int(getattr(order, "volume", 0) or 0)
+            if traded > 0 and (
+                "ALLTRADED" in status_s or "全部成交" in status_s or traded >= vol
+            ):
+                return {"filled": True, "traded": traded, "status": status_s}
+            if traded > 0:
+                return {"partial": True, "traded": traded, "status": status_s}
+            if any(x in status_s for x in ("REJECTED", "拒单")):
+                return {"rejected": True, "status": status_s}
+            if any(x in status_s for x in ("CANCELLED", "已撤销")):
+                return {"cancelled": True, "status": status_s}
+            time.sleep(0.1)
+        return {"pending": True}
+
+    def _send_close_leg(
+        self,
+        *,
+        ths_code: str,
+        sym: str,
+        ex_name: str,
+        exchange: Any,
+        direction: Any,
+        lots: int,
+        primary_off: Any,
+        order_type: Any,
+        price: float,
+        tick: float,
+        use_market: bool,
+    ) -> str:
+        from vnpy.trader.constant import Offset, OrderType
+        from vnpy.trader.object import OrderRequest
+
+        lots = max(1, int(lots))
+        alt_off = (
+            Offset.CLOSEYESTERDAY
+            if primary_off == Offset.CLOSETODAY
+            else Offset.CLOSETODAY
+        )
+        ex_u = (ex_name or "").upper()
+        candidates = [primary_off]
+        if ex_u in ("SHFE", "INE", "CZCE", "CFFEX") and alt_off != primary_off:
+            candidates.append(alt_off)
+        last_vt = ""
+        for idx, off in enumerate(candidates):
+            lp = float(price)
+            if use_market:
+                lp = self._aggressive_limit_price(
+                    ths_code, sym, ex_name, direction, tick, lp,
+                )
+            else:
+                lp = round_to_tick(lp, tick)
+            if lp <= 0:
+                raise ValueError("委托价格无效，请检查行情或手动填写价格")
+            req = OrderRequest(
+                symbol=sym,
+                exchange=exchange,
+                direction=direction,
+                type=order_type,
+                volume=lots,
+                price=lp,
+                offset=off,
+            )
+            logger.info(
+                "CTP 报单 %s %s %s %s手 @%s offset=%s type=%s",
+                sym, ex_name, direction, lots, lp, off, order_type,
+            )
+            with _ctp_td_lock:
+                vt_orderid = self._engine.send_order(req, GATEWAY_NAME)
+            if not vt_orderid:
+                if idx + 1 < len(candidates):
+                    logger.warning("CTP close no order id offset=%s, try alternate", off)
                     continue
-                td = int(p.get("td_volume") or 0)
+                raise RuntimeError(
-                yd = int(p.get("yd_volume") or 0)
+                    "CTP 拒单或未返回委托号（请检查合约代码、价格是否为最小变动价位整数倍）"
-                if td >= lots:
+                )
-                    return Offset.CLOSETODAY
+            last_vt = str(vt_orderid)
-                if yd >= lots:
+            state = self._wait_order_terminal(last_vt)
-                    return Offset.CLOSEYESTERDAY
+            if state.get("filled") or state.get("partial"):
-                if td + yd >= lots:
+                return last_vt
-                    return Offset.CLOSETODAY
+            if state.get("rejected") and idx + 1 < len(candidates):
-                break
+                logger.warning(
-            if ex_u in ("SHFE", "INE", "CZCE"):
+                    "CTP close rejected offset=%s status=%s, retry alternate",
-                return Offset.CLOSETODAY
+                    off, state.get("status"),
-            return Offset.CLOSE
+                )
-        vol = int(getattr(pos, "volume", 0) or 0)
+                continue
-        yd = int(getattr(pos, "yd_volume", 0) or 0)
+            if state.get("pending"):
-        td = max(0, vol - yd)
+                return last_vt
-        if td >= lots:
+            if state.get("rejected"):
-            return Offset.CLOSETODAY
+                raise RuntimeError(f"CTP 平仓被拒 offset={off} status={state.get('status')}")
-        if yd >= lots:
+        return last_vt
-            # 夜盘仓在 vnpy 常记在 yd，日盘平昨易报「平昨仓位不足」→ 上期所/能源优先平今
-            if ex_u in ("SHFE", "INE"):
-                return Offset.CLOSETODAY
-            return Offset.CLOSEYESTERDAY
-        if td + yd >= lots:
-            return Offset.CLOSETODAY
-        if ex_u in ("SHFE", "INE", "CZCE"):
-            return Offset.CLOSETODAY
-        return Offset.CLOSE
 
     def _aggressive_limit_price(
         self,
@@ -2256,16 +2391,6 @@ class CtpBridge:
         if offset in ("open", "open_long", "open_short"):
             d = Direction.LONG if direction == "long" or offset == "open_long" else Direction.SHORT
             off = Offset.OPEN
-        elif offset in ("close", "close_long", "close_short"):
-            hold = "long" if direction == "long" or offset == "close_long" else "short"
-            if hold == "long":
-                d = Direction.SHORT
-            else:
-                d = Direction.LONG
-            off = self._resolve_close_offset(sym, ex_name, hold, lots)
-        else:
-            raise ValueError(f"未知开平: {offset}")
-
             use_market = (order_type or "limit").lower() == "market"
             if use_market:
                 ot = OrderType.FAK
@@ -2275,7 +2400,6 @@ class CtpBridge:
                 price = round_to_tick(float(price), tick)
             if price <= 0:
                 raise ValueError("委托价格无效，请检查行情或手动填写价格")
-
             req = OrderRequest(
                 symbol=sym,
                 exchange=exchange,
@@ -2292,8 +2416,44 @@ class CtpBridge:
             with _ctp_td_lock:
                 vt_orderid = self._engine.send_order(req, GATEWAY_NAME)
             if not vt_orderid:
-            raise RuntimeError("CTP 拒单或未返回委托号（请检查合约代码、价格是否为最小变动价位整数倍）")
+                raise RuntimeError(
+                    "CTP 拒单或未返回委托号（请检查合约代码、价格是否为最小变动价位整数倍）"
+                )
             return str(vt_orderid)
+        elif offset in ("close", "close_long", "close_short"):
+            hold = "long" if direction == "long" or offset == "close_long" else "short"
+            if hold == "long":
+                d = Direction.SHORT
+            else:
+                d = Direction.LONG
+            use_market = (order_type or "limit").lower() == "market"
+            if use_market:
+                ot = OrderType.FAK
+            else:
+                ot = OrderType.LIMIT
+                price = round_to_tick(float(price), tick)
+            if use_market:
+                price = self._aggressive_limit_price(ths_code, sym, ex_name, d, tick, price)
+            if price <= 0:
+                raise ValueError("委托价格无效，请检查行情或手动填写价格")
+            close_legs = self._resolve_close_legs(sym, ex_name, hold, lots)
+            last_vt = ""
+            for off, leg_lots in close_legs:
+                last_vt = self._send_close_leg(
+                    ths_code=ths_code,
+                    sym=sym,
+                    ex_name=ex_name,
+                    exchange=exchange,
+                    direction=d,
+                    lots=leg_lots,
+                    primary_off=off,
+                    order_type=ot,
+                    price=price,
+                    tick=tick,
+                    use_market=use_market,
+                )
+            return last_vt
+        raise ValueError(f"未知开平: {offset}")
 
     def cancel_order(self, vt_orderid: str) -> bool:
         if not self._engine or not vt_orderid: