Align options review PnL with OKX positions-history realizedPnl.

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

lib/options/options_monitor_lib.py

@@ -135,6 +135,98 @@ def _created_at_ms(created_at: Any) -> int | None:
     return None
 
 
+def _group_key_for_closed_trade(row: Any) -> str:
+    inst = str(row["inst_id"] or "").strip()
+    ord_id = str(row["close_ord_id"] or "").strip() if "close_ord_id" in row.keys() else ""
+    if ord_id:
+        return f"{inst}|ord:{ord_id}"
+    closed = str(row["closed_at"] or "").strip()
+    return f"{inst}|close:{(closed[:16] if closed else '')}"
+
+
+def backfill_closed_options_realized_pnl_from_history(
+    conn: sqlite3.Connection,
+    hist_rows: list[dict[str, Any]],
+    *,
+    trade_limit: int = 200,
+) -> int:
+    """
+    用 OKX positions-history 的 realizedPnl 覆盖本地已平记录.
+    同一次平仓多笔本地 open(加仓)按权利金占比分摊交易所总盈亏.
+    """
+    by_inst: dict[str, list[dict[str, Any]]] = {}
+    for raw in hist_rows or []:
+        if not isinstance(raw, dict):
+            continue
+        inst = str(raw.get("instId") or "").strip()
+        if not inst:
+            continue
+        by_inst.setdefault(inst, []).append(raw)
+
+    rows = conn.execute(
+        """
+        SELECT id, inst_id, sheets, premium_paid, realized_pnl, created_at, closed_at, close_ord_id
+        FROM options_trades
+        WHERE status = 'closed'
+        ORDER BY id DESC
+        LIMIT ?
+        """,
+        (int(trade_limit),),
+    ).fetchall()
+    if not rows:
+        return 0
+
+    groups: dict[str, list[Any]] = {}
+    for row in rows:
+        inst = str(row["inst_id"] or "").strip()
+        if not inst or inst not in by_inst:
+            continue
+        groups.setdefault(_group_key_for_closed_trade(row), []).append(row)
+
+    updated = 0
+    for group in groups.values():
+        inst = str(group[0]["inst_id"] or "").strip()
+        open_candidates = [_created_at_ms(r["created_at"]) for r in group]
+        open_ms = min((x for x in open_candidates if x is not None), default=None)
+        close_info = resolve_option_close_from_history(by_inst.get(inst) or [], open_ms=open_ms)
+        if not close_info:
+            continue
+        ex_pnl = _safe_float(close_info.get("realized_pnl"))
+        if ex_pnl is None:
+            continue
+        close_quote = _safe_float(close_info.get("close_quote"))
+        total_paid = 0.0
+        for r in group:
+            total_paid += float(_safe_float(r["premium_paid"]) or 0.0)
+        allocated = 0.0
+        for i, r in enumerate(group):
+            paid = float(_safe_float(r["premium_paid"]) or 0.0)
+            if i == len(group) - 1:
+                share = round(float(ex_pnl) - allocated, 4)
+            elif total_paid > 0:
+                share = round(float(ex_pnl) * (paid / total_paid), 4)
+                allocated += share
+            else:
+                share = round(float(ex_pnl) / len(group), 4)
+                allocated += share
+            local = _safe_float(r["realized_pnl"])
+            if local is not None and abs(local - share) < 1e-6:
+                continue
+            prem_recv = round(paid + share, 4)
+            conn.execute(
+                """
+                UPDATE options_trades
+                SET realized_pnl = ?,
+                    premium_received = ?,
+                    close_quote = COALESCE(?, close_quote)
+                WHERE id = ?
+                """,
+                (share, prem_recv, close_quote, int(r["id"])),
+            )
+            updated += 1
+    return updated
+
+
 def sync_open_options_trades(
     conn: sqlite3.Connection,
     *,

lib/options/options_register.py

@@ -263,8 +263,12 @@ def _sync_options_trades(
         if not force and now - _OPTIONS_SYNC_LAST_AT < _OPTIONS_SYNC_INTERVAL_SEC:
             return
         _OPTIONS_SYNC_LAST_AT = now
-    from lib.exchange.okx_options_lib import fetch_option_position_history
-    from lib.options.options_monitor_lib import reconcile_live_open_trades, sync_open_options_trades
+    from lib.exchange.okx_options_lib import fetch_all_option_positions_history, fetch_option_position_history
+    from lib.options.options_monitor_lib import (
+        backfill_closed_options_realized_pnl_from_history,
+        reconcile_live_open_trades,
+        sync_open_options_trades,
+    )
 
     if raw_positions is None:
         raw = cfg["fetch_option_positions"](ex)
@@ -282,6 +286,11 @@ def _sync_options_trades(
         init_options_tables(conn)
         reconcile_live_open_trades(conn, live_inst_ids=live_ids)
         sync_open_options_trades(conn, live_inst_ids=live_ids, fetch_history_fn=_hist)
+        try:
+            hist_all = fetch_all_option_positions_history(ex, limit=200)
+            backfill_closed_options_realized_pnl_from_history(conn, hist_all)
+        except Exception:
+            pass
         conn.commit()
     finally:
         conn.close()

lib/options/options_review_lib.py

@@ -556,8 +556,24 @@ def sync_all_review_sources(
     return out
 
 
-def ensure_local_review_synced(conn: sqlite3.Connection) -> dict[str, Any]:
-    """列表/统计前轻量刷新本地源."""
+def ensure_local_review_synced(
+    conn: sqlite3.Connection,
+    *,
+    ex: Any | None = None,
+    backfill_exchange_pnl: bool = True,
+) -> dict[str, Any]:
+    """列表/统计前轻量刷新本地源;有交易所时先用历史仓位盈亏覆盖本地再导入复盘."""
+    if backfill_exchange_pnl and ex is not None:
+        try:
+            from lib.exchange.okx_options_lib import fetch_all_option_positions_history
+            from lib.options.options_monitor_lib import (
+                backfill_closed_options_realized_pnl_from_history,
+            )
+
+            hist = fetch_all_option_positions_history(ex, limit=200)
+            backfill_closed_options_realized_pnl_from_history(conn, hist)
+        except Exception:
+            pass
     return sync_all_review_sources(conn, from_exchange=False)
 
 

lib/options/options_review_register.py

@@ -26,7 +26,6 @@ from lib.options.options_review_lib import (
     hide_review_trade,
     list_review_trades,
     save_review_entry,
-    sync_all_review_sources,
 )
 
 
@@ -112,11 +111,12 @@ def register_options_review_routes(app: Flask, cfg: dict[str, Any], repo_root: s
     @app.route("/api/options/review/sync", methods=["POST"])
     @lr
     def api_options_review_sync():
-        """刷新本地 options_trades + 已结束对冲计划(不访问交易所)."""
+        """刷新本地 options_trades + 已结束对冲计划;尽量用交易所历史盈亏覆盖本地估算."""
         conn = cfg["get_db"]()
         try:
             init_options_review_tables(conn)
-            result = sync_all_review_sources(conn, from_exchange=False)
+            ex, _err = _require_ex(cfg)
+            result = ensure_local_review_synced(conn, ex=ex if ex is not None else None)
             conn.commit()
             return jsonify(result)
         finally:
@@ -134,7 +134,8 @@ def register_options_review_routes(app: Flask, cfg: dict[str, Any], repo_root: s
                 "no",
             )
             if do_sync:
-                ensure_local_review_synced(conn)
+                ex, _err = _require_ex(cfg)
+                ensure_local_review_synced(conn, ex=ex if ex is not None else None)
                 conn.commit()
             filt = dict(
                 source_type=(request.args.get("source_type") or "").strip() or None,
@@ -266,7 +267,8 @@ def register_options_review_routes(app: Flask, cfg: dict[str, Any], repo_root: s
     def api_options_review_stats():
         conn = cfg["get_db"]()
         try:
-            ensure_local_review_synced(conn)
+            ex, _err = _require_ex(cfg)
+            ensure_local_review_synced(conn, ex=ex if ex is not None else None)
             conn.commit()
             stats = compute_review_stats(
                 conn,

tests/test_options_sync.py

@@ -173,3 +173,43 @@ def test_reconcile_live_open_trades_reopens_sync_artifact():
     assert row["realized_pnl"] == -0.5
     assert row["premium_received"] == 0.74
     assert row["close_ord_id"] == "pos-1"
+
+
+def test_backfill_closed_options_realized_pnl_from_history():
+    from lib.options.options_monitor_lib import backfill_closed_options_realized_pnl_from_history
+
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    init_options_tables(conn)
+    conn.execute(
+        """
+        INSERT INTO options_trades
+        (inst_id, underlying, opt_type, strike, exp_time, sheets, eth_amount,
+         open_quote, close_quote, premium_paid, premium_received, realized_pnl,
+         status, created_at, closed_at, close_ord_id)
+        VALUES (?, 'ETH', 'C', 1860, '', 43, 0.43, 22.4, 50.6, 9.632, 21.758, 12.126,
+                'closed', '2026-07-20 08:02:14', '2026-07-21 01:38:43', 'ord-1')
+        """,
+        ("ETH-USD_UM-260721-1860-C",),
+    )
+    conn.commit()
+    hist = [
+        {
+            "instId": "ETH-USD_UM-260721-1860-C",
+            "uTime": "1784564323000",
+            "realizedPnl": "11.64",
+            "closeAvgPx": "48.5",
+            "closeTotalPos": "43",
+            "posId": "pos-x",
+        }
+    ]
+    n = backfill_closed_options_realized_pnl_from_history(conn, hist)
+    assert n == 1
+    row = conn.execute(
+        "SELECT realized_pnl, premium_received, close_quote FROM options_trades WHERE id=1"
+    ).fetchone()
+    assert row["realized_pnl"] == 11.64
+    assert abs(float(row["premium_received"]) - (9.632 + 11.64)) < 1e-6
+    assert float(row["close_quote"]) == 48.5
+    # idempotent
+    assert backfill_closed_options_realized_pnl_from_history(conn, hist) == 0