Use CTP PositionDate from investor position query for SHFE close offset.

Parse official today/history legs in onRspQryInvestorPosition instead of guessing or retrying alternate offsets.

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

docs/CTP_LIVE.md

@@ -119,7 +119,7 @@ CTP_LIVE_ENV=实盘
 |--------|------|
 | **大商所（DCE）** 等 | 使用通用 **CLOSE** |
 | **上期所（SHFE）、能源（INE）、郑商所（CZCE）、中金所（CFFEX）** | 区分 **平今 CLOSETODAY** / **平昨 CLOSEYESTERDAY** |
-| 选择依据 | 查询持仓的今仓 `td_volume`、昨仓 `yd_volume`，优先平今再平昨 |
+| 选择依据 | CTP `OnRspQryInvestorPosition` 的 **PositionDate**（1=今仓、2=昨仓）；上期所今昨为**两条独立回报**，与同花顺/快期一致 |
 
 这与国内期货公司 CTP 客户端、快期等 **标准投机平仓逻辑** 一致。
 

modules/ctp/vnpy_bridge.py

@@ -345,6 +345,8 @@ class CtpBridge:
         self._position_open_avg: dict[str, float] = {}
         self._position_open_cost_acc: dict[str, dict[str, float]] = {}
         self._last_open_cost_query_ts: float = 0.0
+        self._ctp_pos_legs: dict[str, dict[str, Any]] = {}
+        self._ctp_pos_leg_query_active = False
         self._margin_hooked = False
         self._trade_hooked = False
         self._trade_query_results: list[dict[str, Any]] = []
@@ -676,6 +678,8 @@ class CtpBridge:
         self._hooks_td_api_id = None
         self._instrument_hooked = False
         self._margin_rate_hooked = False
+        self._ctp_pos_legs.clear()
+        self._ctp_pos_leg_query_active = False
         self._last_position_query_ts = 0.0
         self._last_instruments_ready_ts = 0.0
         try:
@@ -986,12 +990,80 @@ class CtpBridge:
             logger.debug("find position: %s", exc)
         return None
 
+    @staticmethod
+    def _ctp_hold_from_posi_direction(posi_direction: str) -> str:
+        """CTP PosiDirection: 2=多头持仓 3=空头持仓。"""
+        return "short" if str(posi_direction or "").strip() == "3" else "long"
+
+    @staticmethod
+    def _ctp_is_today_position_date(position_date: str) -> bool:
+        """CTP PositionDate: 1=今仓 2=昨仓（上期所/能源分两条回报）。"""
+        pd = str(position_date or "1").strip()
+        if pd in ("2", "History", "HISTORY", "history", "昨"):
+            return False
+        return True
+
+    def _ingest_ctp_position_leg(self, data: dict) -> None:
+        """按 CTP 官方 PositionDate 缓存今/昨仓（与同花顺/快期一致）。"""
+        sym = str(data.get("InstrumentID") or "").strip()
+        if not sym:
+            return
+        hold = self._ctp_hold_from_posi_direction(data.get("PosiDirection"))
+        position_date = str(data.get("PositionDate") or "1")
+        lots = int(data.get("Position") or 0)
+        key = f"{sym.lower()}|{hold}|{position_date}"
+        if lots <= 0:
+            self._ctp_pos_legs.pop(key, None)
+            return
+        if hold == "short":
+            frozen = int(data.get("LongFrozen") or 0)
+        else:
+            frozen = int(data.get("ShortFrozen") or 0)
+        self._ctp_pos_legs[key] = {
+            "symbol": sym,
+            "direction": hold,
+            "position_date": position_date,
+            "lots": lots,
+            "today_position": int(data.get("TodayPosition") or 0),
+            "yd_position": int(data.get("YdPosition") or 0),
+            "frozen": frozen,
+            "exchange": str(data.get("ExchangeID") or ""),
+        }
+
+    def _position_td_yd_from_ctp_legs(
+        self, sym: str, hold_direction: str,
+    ) -> Optional[tuple[int, int, int, int]]:
+        sym_l = (sym or "").lower()
+        hold = (hold_direction or "long").strip().lower()
+        td = yd = 0
+        frozen = 0
+        found = False
+        for leg in self._ctp_pos_legs.values():
+            if (leg.get("symbol") or "").lower() != sym_l:
+                continue
+            if (leg.get("direction") or "long") != hold:
+                continue
+            found = True
+            lots = int(leg.get("lots") or 0)
+            fz = int(leg.get("frozen") or 0)
+            frozen += fz
+            if self._ctp_is_today_position_date(leg.get("position_date")):
+                td += lots
+            else:
+                yd += lots
+        if not found:
+            return None
+        return td, yd, frozen, td + yd
+
     def _position_td_yd(
         self, sym: str, ex_name: str, hold_direction: str,
     ) -> tuple[int, int, int, int]:
-        """返回 (今仓, 昨仓, 冻结, 总持仓)。"""
+        """返回 (今仓, 昨仓, 冻结, 总持仓)。优先 CTP PositionDate 回报。"""
-        sym_l = (sym or "").lower()
         hold = (hold_direction or "long").strip().lower()
+        ctp = self._position_td_yd_from_ctp_legs(sym, hold)
+        if ctp is not None:
+            return ctp
+        sym_l = (sym or "").lower()
         for p in self._collect_positions():
             ps = (p.get("symbol") or "").lower()
             if ps != sym_l:
@@ -1019,6 +1091,23 @@ class CtpBridge:
         frozen = int(getattr(pos, "frozen", 0) or 0)
         return td, yd, frozen, vol
 
+    def _refresh_ctp_position_legs_for_close(
+        self, sym: str, hold_direction: str, *, timeout: float = 3.0,
+    ) -> bool:
+        """平仓前确保已收到 CTP PositionDate 持仓回报。"""
+        if self._position_td_yd_from_ctp_legs(sym, hold_direction) is not None:
+            return True
+        try:
+            self.request_position_snapshot(force=True)
+        except Exception as exc:
+            logger.debug("position snapshot for close: %s", exc)
+        deadline = time.time() + max(0.5, float(timeout))
+        while time.time() < deadline:
+            if self._position_td_yd_from_ctp_legs(sym, hold_direction) is not None:
+                return True
+            time.sleep(0.1)
+        return False
+
     def _resolve_close_legs(
         self, sym: str, ex_name: str, hold_direction: str, lots: int,
     ) -> list[tuple[Any, int]]:
@@ -1028,6 +1117,10 @@ class CtpBridge:
         lots = max(1, int(lots or 1))
         if ex_u not in ("CZCE", "CFFEX", "SHFE", "INE"):
             return [(Offset.CLOSE, lots)]
+        hold = (hold_direction or "long").strip().lower()
+        if ex_u in ("SHFE", "INE"):
+            self._refresh_ctp_position_legs_for_close(sym, hold)
+        ctp_legs = self._position_td_yd_from_ctp_legs(sym, hold)
         td, yd, frozen, vol = self._position_td_yd(sym, ex_name, hold_direction)
         td_close = max(0, td)
         yd_close = max(0, yd)
@@ -1036,9 +1129,10 @@ class CtpBridge:
             td_close -= cut
             frozen -= cut
             yd_close = max(0, yd_close - frozen)
-        logger.debug(
+        logger.info(
-            "close legs %s %s hold=%s lots=%s td=%s yd=%s frozen=%s vol=%s",
+            "close legs %s %s hold=%s want=%s td=%s yd=%s frozen=%s source=%s",
-            sym, ex_u, hold_direction, lots, td_close, yd_close, frozen, vol,
+            sym, ex_u, hold, lots, td_close, yd_close, frozen,
+            "ctp_position_date" if ctp_legs is not None else "vnpy_fallback",
         )
         legs: list[tuple[Any, int]] = []
         remain = lots
@@ -1051,42 +1145,16 @@ class CtpBridge:
             legs.append((Offset.CLOSEYESTERDAY, take))
             remain -= take
         if remain > 0:
-            # vnpy 今昨拆分不可靠时，先按今仓尝试，拒单后自动改平昨
+            raise ValueError(
-            legs.append((Offset.CLOSETODAY, remain))
+                f"可平仓位不足：今仓可平{td_close}手、昨仓可平{yd_close}手，"
+                f"需平{lots}手（请检查未成交平仓挂单是否占用仓位）"
+            )
         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
-            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,
         *,
@@ -1102,21 +1170,9 @@ class CtpBridge:
         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(
@@ -1133,36 +1189,19 @@ class CtpBridge:
             type=order_type,
             volume=lots,
             price=lp,
-                offset=off,
+            offset=primary_off,
         )
         logger.info(
             "CTP 报单 %s %s %s %s手 @%s offset=%s type=%s",
-                sym, ex_name, direction, lots, lp, off, order_type,
+            sym, ex_name, direction, lots, lp, primary_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
             raise RuntimeError(
                 "CTP 拒单或未返回委托号（请检查合约代码、价格是否为最小变动价位整数倍）"
             )
-            last_vt = str(vt_orderid)
+        return str(vt_orderid)
-            state = self._wait_order_terminal(last_vt)
-            if state.get("filled") or state.get("partial"):
-                return last_vt
-            if state.get("rejected") and idx + 1 < len(candidates):
-                logger.warning(
-                    "CTP close rejected offset=%s status=%s, retry alternate",
-                    off, state.get("status"),
-                )
-                continue
-            if state.get("pending"):
-                return last_vt
-            if state.get("rejected"):
-                raise RuntimeError(f"CTP 平仓被拒 offset={off} status={state.get('status')}")
-        return last_vt
 
     def _aggressive_limit_price(
         self,
@@ -1384,11 +1423,17 @@ class CtpBridge:
                 data: dict, error: dict, reqid: int, last: bool,
             ) -> None:
                 try:
+                    if data:
+                        if not bridge._ctp_pos_leg_query_active:
+                            bridge._ctp_pos_legs.clear()
+                            bridge._ctp_pos_leg_query_active = True
+                        bridge._ingest_ctp_position_leg(data)
                     if data:
                         if not bridge._position_open_cost_acc:
                             bridge._position_open_avg.clear()
                         bridge._ingest_position_open_cost(data)
                     if last:
+                        bridge._ctp_pos_leg_query_active = False
                         bridge._finalize_position_open_cost_acc()
                 except Exception as exc:
                     logger.debug("position open avg cache: %s", exc)