Fix LIVE SoT P0/P1: closing state machine, OO exchange fills, BN balances.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-08 15:54:45 +08:00
parent 99e58910d3
commit 8d67f3fc6c
10 changed files with 864 additions and 167 deletions
@@ -546,30 +546,52 @@ class OkxLiveExecutor(Matcher):
        option_inst_id = str(pos.get("option_inst_id") or "")
        opt_contracts = float(pos.get("option_qty_contracts") or 0)
        opt_qty = float(pos.get("option_qty_eth") or 0)
-        if not option_inst_id or opt_contracts <= 0:
+        if not option_inst_id:
            return CloseResult(ok=False, detail="half_open 缺期权合约信息")
        client = self._client()
-        try:
-            opt_live = client.place_market(
-                inst_id=option_inst_id,
-                side="sell",
-                sz=str(int(round(opt_contracts))),
-                td_mode="cash",
-                reduce_only=True,
+        ex_sz = exchange_option_abs_size(client, option_inst_id)
+        if ex_sz is None:
+            return CloseResult(ok=False, detail="half_open：无法核对交易所期权仓位")
+        if ex_sz <= 1e-8:
+            of_px, of_fee, of_notional = 0.0, 0.0, 0.0
+            opt_contracts = 0.0
+            opt_qty = 0.0
+            opt_entry = float(pos.get("option_entry_px") or 0)
+            # 已空：不发明现金
+        else:
+            opt_contracts = float(ex_sz)
+            opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id))
+            try:
+                opt_live = client.place_market(
+                    inst_id=option_inst_id,
+                    side="sell",
+                    sz=str(max(1, int(round(opt_contracts)))),
+                    td_mode="cash",
+                    reduce_only=True,
+                )
+            except Exception as e:
+                return CloseResult(ok=False, detail=f"half_open 平期权失败: {e}")
+            of_px = float(opt_live.avg_px)
+            of_fee = float(opt_live.fee)
+            filled = float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0
+            if filled > 0:
+                opt_contracts = filled
+                opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id))
+            of_notional = of_px * opt_qty
+            ex_left = exchange_option_abs_size(client, option_inst_id)
+            if ex_left is None or ex_left > 1e-8:
+                return CloseResult(
+                    ok=False,
+                    detail=f"half_open：卖后仍有仓或无法核对 left={ex_left}",
+                )
+            opt_entry = float(pos.get("option_entry_px") or of_px)
+            self.ledger.apply_cash(
+                of_notional - of_fee,
+                kind="close_option",
+                group_id=group_id or None,
+                note="LIVE repair half_open",
+                allow_negative=True,
            )
-        except Exception as e:
-            return CloseResult(ok=False, detail=f"half_open 平期权失败: {e}")
-        of_px = float(opt_live.avg_px)
-        of_fee = float(opt_live.fee)
-        of_notional = of_px * opt_qty
-        opt_entry = float(pos.get("option_entry_px") or of_px)
-        self.ledger.apply_cash(
-            of_notional - of_fee,
-            kind="close_option",
-            group_id=group_id or None,
-            note="LIVE repair half_open",
-            allow_negative=True,
-        )
        now = int(time.time() * 1000)
        with self.db._lock:
            if group_id:
@@ -1007,7 +1029,8 @@ class OkxLiveExecutor(Matcher):
        if err:
            return CloseResult(ok=False, detail=err)
        pos = self.current_position()
-        if str(pos.get("status") or "") != "open" or not pos.get("group_id"):
+        st = str(pos.get("status") or "")
+        if st not in ("open", "closing") or not pos.get("group_id"):
            return CloseResult(ok=False, detail="无期期持仓可平")
        if not (
            str(pos.get("hedge_mode") or "") == "option_option"
@@ -1026,7 +1049,7 @@ class OkxLiveExecutor(Matcher):
            except Exception as e:
                return CloseResult(ok=False, detail=f"期期全平卖腿失败: {e}")
        # 必须以交易所两腿皆空才落本地 flat
-        for leg, inst, _qty, _c in legs:
+        for _leg, inst, _qty, _c in legs:
            if not inst:
                continue
            ex_sz = exchange_option_abs_size(client, inst)
@@ -1069,6 +1092,13 @@ class OkxLiveExecutor(Matcher):
                    ),
                )
                self.db._conn.commit()
+        from ..sim.pnl import summarize_fills_pnl
+
+        fill_rows = self.db.fetchall(
+            "SELECT * FROM fills WHERE group_id=? ORDER BY id ASC", (group_id,)
+        )
+        summary = summarize_fills_pnl(list(fill_rows))
+        net = float(summary.get("net_pnl") or 0.0)
        with self.db._lock:
            self.db._conn.execute(
                """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?,
@@ -1077,7 +1107,7 @@ class OkxLiveExecutor(Matcher):
                    "closed",
                    int(time.time() * 1000),
                    reason,
-                    0.0,
+                    net,
                    f"oo full close {reason} exchange_flat_mirror",
                    group_id,
                ),
@@ -1097,47 +1127,54 @@ class OkxLiveExecutor(Matcher):
        return CloseResult(
            ok=True,
            detail="oo_full_closed_live",
-            data={"group_id": group_id, "reason": reason, "net": 0.0},
+            data={"group_id": group_id, "reason": reason, "net": net},
        )

    def close_winning_oo_leave_residual(
        self, *, reason: str = "target_oo_win"
    ) -> CloseResult:
-        """期期达标：先标记 closing，再交易所卖掉盈利腿，再落库。"""
+        """期期达标：先标记 closing，再按交易所张数卖掉盈利腿，验空后落库。"""
        err = self._guard_live()
        if err:
            return CloseResult(ok=False, detail=err)
        pos = self.current_position()
-        if str(pos.get("status") or "") != "open" or not pos.get("option2_inst_id"):
+        st = str(pos.get("status") or "")
+        # 防重入：已在 closing 则优先收尾（须在 open 判断之前）
+        if st == "closing":
+            return self._finish_oo_win_after_exchange(reason=reason, pos=pos)
+        if st != "open" or not pos.get("option2_inst_id"):
            return CloseResult(ok=False, detail="无期期持仓")
-        # 防重入：已在 closing 则只做账本收尾
-        if str(pos.get("status") or "") == "closing":
-            return super().close_winning_oo_leave_residual(
-                reason=reason, skip_market=True
-            )
        upl = self.unrealized()
        call_upl = float(upl.get("option_upl") or 0)
        put_upl = float(upl.get("option2_upl") or 0)
        if call_upl >= put_upl and call_upl > 0:
+            win_leg = "option"
            win_id = str(pos["option_inst_id"])
-            win_contracts = float(pos.get("option_qty_contracts") or 0)
        elif put_upl > 0:
+            win_leg = "option2"
            win_id = str(pos["option2_inst_id"])
-            win_contracts = float(pos.get("option2_qty_contracts") or 0)
        else:
            return CloseResult(ok=False, detail="无明确盈利腿")
+        client = self._client()
+        ex_sz = exchange_option_abs_size(client, win_id)
+        if ex_sz is None:
+            return CloseResult(ok=False, detail="期期平盈利腿：无法核对交易所仓位")
+        if ex_sz <= 1e-8:
+            return self._finish_oo_win_after_exchange(
+                reason=reason, pos=pos, win_leg=win_leg, fill_px=0.0, fill_fee=0.0, fill_c=0.0
+            )
        with self.db._lock:
            self.db._conn.execute(
                "UPDATE positions SET status='closing' WHERE id=1 AND status='open'"
            )
            self.db._conn.commit()
-        client = self._client()
        try:
-            client.place_market(
+            live = client.place_market(
                inst_id=win_id,
                side="sell",
-                sz=str(int(round(win_contracts))),
+                sz=str(max(1, int(round(float(ex_sz))))),
                td_mode="cash",
+                reduce_only=True,
            )
        except Exception as e:
            with self.db._lock:
@@ -1146,8 +1183,65 @@ class OkxLiveExecutor(Matcher):
                )
                self.db._conn.commit()
            return CloseResult(ok=False, detail=f"期期平盈利腿失败: {e}")
+        fill_c = float(live.sz) if live.sz and float(live.sz) > 0 else 0.0
+        return self._finish_oo_win_after_exchange(
+            reason=reason,
+            pos=self.current_position(),
+            win_leg=win_leg,
+            fill_px=float(live.avg_px),
+            fill_fee=float(live.fee),
+            fill_c=fill_c,
+        )
+
+    def _finish_oo_win_after_exchange(
+        self,
+        *,
+        reason: str,
+        pos: dict,
+        win_leg: str | None = None,
+        fill_px: float | None = None,
+        fill_fee: float | None = None,
+        fill_c: float | None = None,
+    ) -> CloseResult:
+        """盈利腿卖后：确认交易所已空，再用真实成交落库。"""
+        client = self._client()
+        call_id = str(pos.get("option_inst_id") or "")
+        put_id = str(pos.get("option2_inst_id") or "")
+        if not win_leg:
+            # 崩溃恢复：哪条腿已空且另一条仍有仓 → 已空者为赢腿
+            c_sz = exchange_option_abs_size(client, call_id) if call_id else None
+            p_sz = exchange_option_abs_size(client, put_id) if put_id else None
+            if c_sz is None or p_sz is None:
+                return CloseResult(
+                    ok=False, detail="closing 收尾：无法核对交易所两腿仓位"
+                )
+            if c_sz <= 1e-8 and p_sz > 1e-8:
+                win_leg = "option"
+            elif p_sz <= 1e-8 and c_sz > 1e-8:
+                win_leg = "option2"
+            elif c_sz <= 1e-8 and p_sz <= 1e-8:
+                # 两腿皆空：走全平镜像
+                return self.close_oo_full(reason=reason, bypass_liquidity=True)
+            else:
+                return CloseResult(
+                    ok=False, detail="closing 收尾：盈利腿仍在交易所，请重试卖出"
+                )
+        win_id = call_id if win_leg == "option" else put_id
+        ex_win = exchange_option_abs_size(client, win_id)
+        if ex_win is None:
+            return CloseResult(ok=False, detail="closing 收尾：无法核对盈利腿仓位")
+        if ex_win > 1e-8:
+            return CloseResult(
+                ok=False,
+                detail=f"closing 收尾：盈利腿仍有仓 {ex_win}，禁止本地清仓",
+            )
        return super().close_winning_oo_leave_residual(
-            reason=reason, skip_market=True
+            reason=reason,
+            skip_market=True,
+            live_fill_px=0.0 if fill_px is None else float(fill_px),
+            live_fill_fee=0.0 if fill_fee is None else float(fill_fee),
+            live_fill_contracts=fill_c,
+            live_win_leg=win_leg,
        )

    def close_group(self, *, reason: str, bypass_liquidity: bool = False) -> CloseResult:
@@ -1162,6 +1256,17 @@ class OkxLiveExecutor(Matcher):
            return self.recover_opening()
        if st == "half_open":
            return self.repair_half_open()
+        if st == "closing":
+            # 期期盈利腿收尾
+            if pos.get("option2_inst_id") or str(pos.get("hedge_mode") or "") == "option_option":
+                return self.close_winning_oo_leave_residual(reason=reason or "closing_retry")
+            return CloseResult(ok=False, detail="closing 非期期状态，请人工核对")
+        is_oo = (
+            str(pos.get("hedge_mode") or "") == "option_option"
+            or bool(pos.get("option2_inst_id"))
+        )
+        if is_oo and st == "open":
+            return self.close_oo_full(reason=reason, bypass_liquidity=bypass_liquidity)
        if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"):
            return CloseResult(ok=False, detail="无持仓可平")

@@ -1213,9 +1318,9 @@ class OkxLiveExecutor(Matcher):
                of_fee = float(prev["fee"] or 0)
                of_notional = float(prev["notional"] or (of_px * opt_qty))
            else:
-                of_px = float(intrinsic) if intrinsic is not None else 0.0
+                of_px = 0.0
                of_fee = 0.0
-                of_notional = of_px * opt_qty
+                of_notional = 0.0
                self._ensure_option_closed_perp_pending(
                    group_id=group_id,
                    option_inst_id=option_inst_id,
@@ -1231,10 +1336,18 @@ class OkxLiveExecutor(Matcher):
            of_slip = 0.0
            option_apply_cash = False
        elif is_expiry:
-            # 到期：交易所自动结算期权，本地只平永续，不卖期权、不本地发明结算现金
-            of_px = float(intrinsic) if intrinsic is not None else 0.0
+            # 到期：交易所自动结算期权，本地只平永续；不卖期权、不用 intrinsic 发明 fill
+            ex_opt = exchange_option_abs_size(client, option_inst_id)
+            if ex_opt is not None and ex_opt > 1e-8:
+                logger.warning(
+                    "expiry: option still on exchange sz=%.4f group=%s; "
+                    "skip option, close perp only",
+                    ex_opt,
+                    group_id,
+                )
+            of_px = 0.0
            of_fee = 0.0
-            of_notional = of_px * opt_qty
+            of_notional = 0.0
            of_slip = 0.0
            option_apply_cash = False
            logger.info(
@@ -2025,40 +2138,54 @@ class OkxLiveExecutor(Matcher):
                ct_val=ct_val,
                allow_db_fallback=False,
            )
-            if perp_sz is None or perp_sz <= 0:
+            if perp_sz is None:
                return CloseResult(
                    ok=False,
-                    detail="弃期权平永续失败: 无法取得有效永续仓位数量",
+                    detail="弃期权平永续失败: 无法核对交易所永续仓位",
                )
-            if perp_side == "long":
-                side, pos_side = "sell", "long"
+            if perp_sz > 0:
+                if perp_side == "long":
+                    side, pos_side = "sell", "long"
+                else:
+                    side, pos_side = "buy", "short"
+                perp_live = client.place_market(
+                    inst_id=perp_inst,
+                    side=side,
+                    sz=str(perp_sz),
+                    td_mode=self._perp_margin_mode_for_group(group_id),
+                    pos_side=pos_side,
+                    reduce_only=True,
+                )
+                pf_px = float(perp_live.avg_px)
+                pf_fee = float(perp_live.fee)
+                try:
+                    perp_qty = float(perp_sz) * float(ct_val)
+                except Exception:
+                    pass
            else:
-                side, pos_side = "buy", "short"
-            perp_live = client.place_market(
-                inst_id=perp_inst,
-                side=side,
-                sz=str(perp_sz),
-                td_mode=self._perp_margin_mode_for_group(group_id),
-                pos_side=pos_side,
-                reduce_only=True,
-            )
+                # 永续已空：仍归档期权 residual
+                pf_px = 0.0
+                pf_fee = 0.0
+                logger.warning(
+                    "abandon: perp already flat on exchange; archive option residual group=%s",
+                    group_id,
+                )
        except Exception as e:
            return CloseResult(ok=False, detail=f"实盘平永续失败: {e}")

-        pf_px = float(perp_live.avg_px)
-        pf_fee = float(perp_live.fee)
        if perp_side == "long":
-            perp_pnl = (pf_px - perp_entry) * perp_qty
+            perp_pnl = (pf_px - perp_entry) * perp_qty if pf_px else 0.0
        else:
-            perp_pnl = (perp_entry - pf_px) * perp_qty
+            perp_pnl = (perp_entry - pf_px) * perp_qty if pf_px else 0.0

-        self.ledger.apply_cash(
-            perp_pnl - pf_fee,
-            kind="close_perp",
-            group_id=group_id,
-            note=f"LIVE close perp abandon option {reason}",
-            allow_negative=True,
-        )
+        if abs(perp_pnl) + abs(pf_fee) > 1e-12:
+            self.ledger.apply_cash(
+                perp_pnl - pf_fee,
+                kind="close_perp",
+                group_id=group_id,
+                note=f"LIVE close perp abandon option {reason}",
+                allow_negative=True,
+            )

        # 复用父类归档写入：临时改 fill 路径太重，直接调用父类会再平一次本地假价。
        # 因此把实盘价写入后走父类结构——这里内联父类 abandon 的 DB 段。