修复期权分批平仓只记最后一笔收回导致盈亏漏算。

每次部分成交累加权利金与张数，全平后再合计 realized_pnl，并优先匹配交易所全平历史。 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-24 21:13:05 +08:00
parent f5104fe68b
commit a7015ec5c8
4 changed files with 325 additions and 11 deletions
@@ -1595,6 +1595,9 @@ def resolve_option_close_from_history(
        u_i, row = item
        c_ms = _safe_float(row.get("cTime"))
        parts: list[float] = []
+        # 全平/强平/ADL 优先于部分平仓(同合约多条时避免只吃到 type=1)
+        close_type = str(row.get("type") or "").strip()
+        parts.append(0.0 if close_type in ("2", "3", "6") else 1.0)
        # 张数优先:同合约多笔时最稳,且不受本地/交易所时区偏差影响
        if want_sheets is not None:
            hist_sheets = _safe_float(row.get("closeTotalPos"))
@@ -1609,7 +1612,7 @@ def resolve_option_close_from_history(
            parts.append(float(abs(int(c_ms) - int(open_ms))))
        if close_ms is not None:
            parts.append(float(abs(u_i - int(close_ms))))
-        if not parts:
+        if len(parts) <= 1:
            parts.append(float(-u_i))
        # 同距时偏向更晚平仓
        parts.append(float(-u_i))
@@ -306,7 +306,8 @@ def close_option_by_bid1(

    px = float(order.get("px", locked_bid_px))
    oid = str((order.get("data") or {}).get("ordId") or "")
-    prem_recv = round(total_premium(px, level_sheets * ct_mult), 4)
+    # 本批预估收回(按挂单张数);成交后按实际减仓张数重算并入累计
+    prem_recv_submit = round(total_premium(px, level_sheets * ct_mult), 4)
    time.sleep(0.6)
    invalidate_option_positions_cache()
    raw2 = cfg["fetch_option_positions"](ex)
@@ -325,14 +326,47 @@ def close_option_by_bid1(
    reduced = max(0, before_avail - after_avail)
    remaining_pos = after_avail
    fully_closed = remaining_pos < 1
+    filled_sheets = min(reduced, level_sheets) if reduced > 0 else 0
+    # 全平但减仓读数异常时回退挂单张数,避免漏记本批
+    if fully_closed and filled_sheets < 1:
+        filled_sheets = level_sheets
+    batch_recv = (
+        round(total_premium(px, filled_sheets * ct_mult), 4) if filled_sheets > 0 else 0.0
+    )
+
+    total_recv = batch_recv
+    close_px = px
+    merged_ord = oid or None

    if fully_closed:
        clear_close_gate(inst_id)
        conn = cfg["get_db"]()
        try:
-            from lib.options.options_db import init_options_tables
+            from lib.options.options_db import (
+                add_open_close_accum,
+                get_open_close_accum,
+                init_options_tables,
+            )

            init_options_tables(conn)
+            # 先并入本批,再读累计(含此前分批)
+            if filled_sheets > 0:
+                add_open_close_accum(
+                    conn,
+                    inst_id,
+                    recv=batch_recv,
+                    sheets=filled_sheets,
+                    px=px,
+                    ord_id=oid or None,
+                )
+            accum = get_open_close_accum(conn, inst_id)
+            total_recv = round(float(accum.get("recv") or 0), 4)
+            accum_sheets = int(accum.get("sheets") or 0)
+            accum_notional = float(accum.get("notional") or 0)
+            if accum_sheets > 0 and accum_notional > 0:
+                close_px = round(accum_notional / accum_sheets, 8)
+            merged_ord = str(accum.get("close_ord_id") or "").strip() or (oid or None)
+
            open_rows = conn.execute(
                """
                SELECT id, premium_paid FROM options_trades
@@ -346,19 +380,25 @@ def close_option_by_bid1(
            for i, row in enumerate(open_rows):
                paid = float(row["premium_paid"] or 0)
                if i == len(open_rows) - 1:
-                    recv = round(prem_recv - allocated, 4)
+                    recv = round(total_recv - allocated, 4)
                elif total_paid > 0:
-                    recv = round(prem_recv * (paid / total_paid), 4)
+                    recv = round(total_recv * (paid / total_paid), 4)
                    allocated += recv
                else:
-                    recv = round(prem_recv / len(open_rows), 4)
+                    recv = round(total_recv / len(open_rows), 4)
                    allocated += recv
                pnl = round(recv - paid, 4)
                note_sql = ""
                from lib.options.options_time_lib import options_app_now_str

                # 占位顺序必须与 SQL 一致: close_quote/recv/pnl/ord/closed_at/[signal_note]/id
-                params: list[Any] = [px, recv, pnl, oid or None, options_app_now_str()]
+                params: list[Any] = [
+                    close_px,
+                    recv,
+                    pnl,
+                    merged_ord,
+                    options_app_now_str(),
+                ]
                if signal_note and i == len(open_rows) - 1:
                    note_sql = """,
                    signal_note = CASE
@@ -371,7 +411,9 @@ def close_option_by_bid1(
                    f"""
                    UPDATE options_trades
                    SET status = 'closed', close_quote = ?, premium_received = ?,
-                        realized_pnl = ?, close_ord_id = ?, closed_at = ?
+                        realized_pnl = ?, close_ord_id = ?, closed_at = ?,
+                        close_recv_accum = 0, close_sheets_accum = 0,
+                        close_notional_accum = 0
                        {note_sql}
                    WHERE id = ?
                    """,
@@ -380,6 +422,27 @@ def close_option_by_bid1(
            conn.commit()
        finally:
            conn.close()
+    elif filled_sheets > 0:
+        # 部分成交:持久化本批收回,下次全平合计
+        conn = cfg["get_db"]()
+        try:
+            from lib.options.options_db import add_open_close_accum, init_options_tables
+
+            init_options_tables(conn)
+            add_open_close_accum(
+                conn,
+                inst_id,
+                recv=batch_recv,
+                sheets=filled_sheets,
+                px=px,
+                ord_id=oid or None,
+            )
+            conn.commit()
+        finally:
+            conn.close()
+        if require_recycle_gate:
+            # 自动平已挂过单:同仓续批只验流动性
+            mark_close_gate_passed(inst_id)
    elif require_recycle_gate:
        # 自动平已挂过单:同仓续批只验流动性
        mark_close_gate_passed(inst_id)
@@ -391,11 +454,12 @@ def close_option_by_bid1(
        "bid": px,
        "locked_bid_px": locked_bid_px,
        "submitted_sheets": level_sheets,
-        "filled_or_reduced_sheets": min(reduced, level_sheets) if reduced else 0,
+        "filled_or_reduced_sheets": filled_sheets,
        "remaining_sheets": remaining_pos,
-        "premium_received": prem_recv,
+        "premium_received": total_recv if fully_closed else (batch_recv or prem_recv_submit),
+        "batch_premium_received": batch_recv,
        "stopped_reason": None if fully_closed else ("partial_bid1" if reduced > 0 else "order_not_filled"),
-        "close_ord_id": oid or None,
+        "close_ord_id": merged_ord if fully_closed else (oid or None),
        "fully_closed": fully_closed,
        "msg": (
            f"已按买一 {locked_bid_px:g} 提交 {level_sheets} 张"
@@ -105,6 +105,10 @@ def init_options_tables(conn: sqlite3.Connection) -> None:
        "ALTER TABLE options_trades ADD COLUMN premium_ccy TEXT DEFAULT 'USDC'",
        "ALTER TABLE options_trades ADD COLUMN bridge_id INTEGER",
        "ALTER TABLE options_trades ADD COLUMN budget_usdt REAL",
+        # 分批平仓累计:全平前写在最早一笔 open 上,全平后并入 premium_received
+        "ALTER TABLE options_trades ADD COLUMN close_recv_accum REAL DEFAULT 0",
+        "ALTER TABLE options_trades ADD COLUMN close_sheets_accum INTEGER DEFAULT 0",
+        "ALTER TABLE options_trades ADD COLUMN close_notional_accum REAL DEFAULT 0",
    ):
        try:
            conn.execute(ddl)
@@ -154,3 +158,86 @@ def sum_open_sheets(conn: sqlite3.Connection, inst_id: str) -> int | None:
    if not row or int(row["n"] or 0) < 1:
        return None
    return int(row["total"] or 0)
+
+
+def _earliest_open_trade_id(conn: sqlite3.Connection, inst_id: str) -> int | None:
+    inst = (inst_id or "").strip()
+    if not inst:
+        return None
+    row = conn.execute(
+        """
+        SELECT id FROM options_trades
+        WHERE inst_id = ? AND status = 'open'
+        ORDER BY id ASC
+        LIMIT 1
+        """,
+        (inst,),
+    ).fetchone()
+    return int(row["id"]) if row else None
+
+
+def get_open_close_accum(conn: sqlite3.Connection, inst_id: str) -> dict[str, float | int | str | None]:
+    """读取分批平仓累计(挂在最早一笔 open 上)."""
+    tid = _earliest_open_trade_id(conn, inst_id)
+    if tid is None:
+        return {"recv": 0.0, "sheets": 0, "notional": 0.0, "close_ord_id": None}
+    row = conn.execute(
+        """
+        SELECT close_recv_accum, close_sheets_accum, close_notional_accum, close_ord_id
+        FROM options_trades WHERE id = ?
+        """,
+        (tid,),
+    ).fetchone()
+    if not row:
+        return {"recv": 0.0, "sheets": 0, "notional": 0.0, "close_ord_id": None}
+    return {
+        "recv": float(row["close_recv_accum"] or 0),
+        "sheets": int(row["close_sheets_accum"] or 0),
+        "notional": float(row["close_notional_accum"] or 0),
+        "close_ord_id": str(row["close_ord_id"] or "").strip() or None,
+    }
+
+
+def add_open_close_accum(
+    conn: sqlite3.Connection,
+    inst_id: str,
+    *,
+    recv: float,
+    sheets: int,
+    px: float,
+    ord_id: str | None = None,
+) -> bool:
+    """分批平仓成交后累加收回权利金/张数/名义(供全平 VWAP 与总盈亏)."""
+    tid = _earliest_open_trade_id(conn, inst_id)
+    if tid is None:
+        return False
+    sheets_i = max(0, int(sheets or 0))
+    recv_f = float(recv or 0)
+    px_f = float(px or 0)
+    if sheets_i < 1 and abs(recv_f) < 1e-15:
+        return False
+    notional_add = px_f * sheets_i if sheets_i > 0 and px_f > 0 else 0.0
+    oid = str(ord_id or "").strip()
+    prev = conn.execute(
+        "SELECT close_ord_id FROM options_trades WHERE id = ?", (tid,)
+    ).fetchone()
+    prev_oids = str((prev["close_ord_id"] if prev else None) or "").strip()
+    if oid:
+        parts = [p for p in prev_oids.split(",") if p] if prev_oids else []
+        if oid not in parts:
+            parts.append(oid)
+        merged_oid = ",".join(parts)
+    else:
+        merged_oid = prev_oids or None
+    conn.execute(
+        """
+        UPDATE options_trades
+        SET close_recv_accum = COALESCE(close_recv_accum, 0) + ?,
+            close_sheets_accum = COALESCE(close_sheets_accum, 0) + ?,
+            close_notional_accum = COALESCE(close_notional_accum, 0) + ?,
+            close_ord_id = ?
+        WHERE id = ?
+        """,
+        (recv_f, sheets_i, notional_add, merged_oid, tid),
+    )
+    return True
@@ -0,0 +1,160 @@
+"""分批平仓:累计各批收回权利金后再算盈亏."""
+from __future__ import annotations
+
+import sqlite3
+import unittest
+from unittest.mock import patch
+
+from lib.options.options_close_exec_lib import close_option_by_bid1
+from lib.options.options_db import (
+    add_open_close_accum,
+    get_open_close_accum,
+    init_options_tables,
+)
+
+
+def _mem_db() -> sqlite3.Connection:
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    init_options_tables(conn)
+    return conn
+
+
+class CloseAccumDbTests(unittest.TestCase):
+    def test_accum_on_earliest_open(self):
+        conn = _mem_db()
+        conn.execute(
+            """
+            INSERT INTO options_trades
+            (inst_id, underlying, opt_type, sheets, eth_amount, premium_paid, status)
+            VALUES ('ETH-USD-260927-2700-P', 'ETH', 'P', 10, 0.1, 0.02, 'open')
+            """
+        )
+        conn.commit()
+        self.assertTrue(
+            add_open_close_accum(
+                conn, "ETH-USD-260927-2700-P", recv=0.005, sheets=5, px=0.05, ord_id="a1"
+            )
+        )
+        self.assertTrue(
+            add_open_close_accum(
+                conn, "ETH-USD-260927-2700-P", recv=0.004, sheets=5, px=0.04, ord_id="a2"
+            )
+        )
+        acc = get_open_close_accum(conn, "ETH-USD-260927-2700-P")
+        self.assertAlmostEqual(float(acc["recv"]), 0.009, places=6)
+        self.assertEqual(int(acc["sheets"]), 10)
+        self.assertAlmostEqual(float(acc["notional"]), 0.05 * 5 + 0.04 * 5, places=6)
+        self.assertEqual(acc["close_ord_id"], "a1,a2")
+
+
+class TwoStepCloseExecTests(unittest.TestCase):
+    def test_two_partial_closes_sum_premium_received(self):
+        conn = _mem_db()
+        inst = "ETH-USD-260927-2700-P"
+        conn.execute(
+            """
+            INSERT INTO options_trades
+            (inst_id, underlying, opt_type, strike, sheets, eth_amount,
+             open_quote, premium_paid, status, premium_ccy, margin_mode)
+            VALUES (?, 'ETH', 'P', 2700, 10, 0.1, 0.2, 0.02, 'open', 'ETH', 'coin')
+            """,
+            (inst,),
+        )
+        conn.commit()
+
+        # 持仓:第一次 10→5,第二次 5→0
+        pos_seq = [
+            [{"instId": inst, "availPos": "10", "pos": "10", "mgnMode": "isolated", "idxPx": "2500"}],
+            [{"instId": inst, "availPos": "5", "pos": "5", "mgnMode": "isolated", "idxPx": "2500"}],
+            [{"instId": inst, "availPos": "5", "pos": "5", "mgnMode": "isolated", "idxPx": "2500"}],
+            [],  # 第二次下单后已平
+        ]
+        pos_i = {"n": 0}
+
+        def fetch_positions(_ex):
+            i = pos_i["n"]
+            pos_i["n"] = min(i + 1, len(pos_seq) - 1)
+            return pos_seq[i]
+
+        order_px = {"n": 0}
+
+        def place_limit(_ex, **kwargs):
+            order_px["n"] += 1
+            px = 0.05 if order_px["n"] == 1 else 0.04
+            return {"ok": True, "px": px, "data": {"ordId": f"ord{order_px['n']}"}}
+
+        def quote(_ex, _inst):
+            return {"ok": True, "tick_sz": "0.0001", "ct_mult": 0.01, "bid": 0.05, "index_px": 2500}
+
+        def book(_ex, _inst, _n):
+            # 每次只够吃一半深度,强制分两批
+            return {"bids": [{"px": 0.05, "sz": 5}]} if order_px["n"] < 1 else {"bids": [{"px": 0.04, "sz": 5}]}
+
+        cfg = {
+            "get_db": lambda: _reuse_conn(conn),
+            "quote_option_contract": quote,
+            "fetch_option_positions": fetch_positions,
+            "fetch_option_book_depth": book,
+            "place_option_limit_order": place_limit,
+            "td_mode": "isolated",
+        }
+
+        with patch("lib.options.options_close_exec_lib.time.sleep", return_value=None), patch(
+            "lib.exchange.okx_options_lib.invalidate_option_positions_cache"
+        ), patch(
+            "lib.options.options_pricing_lib.fetch_option_mark_px", return_value=0.05
+        ), patch(
+            "lib.options.options_close_exec_lib._pos_close_refs", return_value=(0.05, 0.0)
+        ):
+            r1 = close_option_by_bid1(cfg, ex=object(), inst_id=inst, sheets=10)
+            self.assertTrue(r1.get("ok"), r1)
+            self.assertFalse(r1.get("fully_closed"), r1)
+            open_row = conn.execute(
+                "SELECT close_recv_accum, status, realized_pnl FROM options_trades WHERE inst_id=?",
+                (inst,),
+            ).fetchone()
+            self.assertEqual(open_row["status"], "open")
+            self.assertGreater(float(open_row["close_recv_accum"] or 0), 0)
+            self.assertIsNone(open_row["realized_pnl"])
+
+            # 第二次买一深度
+            def book2(_ex, _inst, _n):
+                return {"bids": [{"px": 0.04, "sz": 5}]}
+
+            cfg["fetch_option_book_depth"] = book2
+            r2 = close_option_by_bid1(cfg, ex=object(), inst_id=inst, sheets=5)
+            self.assertTrue(r2.get("ok"), r2)
+            self.assertTrue(r2.get("fully_closed"), r2)
+
+        closed = conn.execute(
+            """
+            SELECT premium_paid, premium_received, realized_pnl, status, close_recv_accum
+            FROM options_trades WHERE inst_id=?
+            """,
+            (inst,),
+        ).fetchone()
+        self.assertEqual(closed["status"], "closed")
+        # ct_mult=0.01: 5张×0.05 + 5张×0.04 = 0.0025 + 0.002 = 0.0045
+        self.assertAlmostEqual(float(closed["premium_received"]), 0.0045, places=6)
+        self.assertAlmostEqual(float(closed["realized_pnl"]), 0.0045 - 0.02, places=6)
+        self.assertEqual(float(closed["close_recv_accum"] or 0), 0.0)
+        # 若只记第二批,recv 会是 0.002,pnl=-0.018;合计应为 -0.0155
+        self.assertNotAlmostEqual(float(closed["premium_received"]), 0.002, places=6)
+
+
+def _reuse_conn(conn: sqlite3.Connection) -> sqlite3.Connection:
+    """get_db 每次返回同一连接;close 时 no-op,避免 :memory: 被关掉."""
+
+    class _Proxy:
+        def __getattr__(self, name):
+            return getattr(conn, name)
+
+        def close(self):
+            return None
+
+    return _Proxy()  # type: ignore[return-value]
+
+
+if __name__ == "__main__":
+    unittest.main()