Align hedge plan option leg PnL with OKX exchange history.

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

lib/hedge_plan/hedge_plan_monitor_lib.py

@@ -8,7 +8,11 @@ from typing import Any, Optional
 from lib.hedge_plan.hedge_plan_db import get_plan, get_plan_legs, list_plans, update_plan
 from lib.hedge_plan.hedge_plan_notify_lib import notify_hedge, notify_plan_end, build_hedge_alert_message
 from lib.hedge_plan.hedge_plan_orders_lib import _sell_option
-from lib.hedge_plan.hedge_plan_settle_lib import leg_is_expired, settle_option_leg_at_spot
+from lib.hedge_plan.hedge_plan_settle_lib import (
+    leg_is_expired,
+    resolve_option_leg_realized_pnl,
+    settle_option_leg_at_spot,
+)
 
 
 def _now() -> str:
@@ -69,6 +73,7 @@ def tick_active_plans(cfg: dict[str, Any]) -> dict[str, Any]:
         return {"ok": False, "msg": "get_db missing"}
     conn = get_db()
     acted: list[dict[str, Any]] = []
+    backfill_stats: dict[str, int] = {}
     try:
         from lib.hedge_plan.hedge_plan_db import init_hedge_plan_tables
 
@@ -80,10 +85,22 @@ def tick_active_plans(cfg: dict[str, Any]) -> dict[str, Any]:
                 acted.append(r)
         orphaned = _settle_orphaned_after_tp(cfg, conn)
         acted.extend(orphaned)
+        try:
+            ex = cfg.get("exchange_options")
+            if ex is not None:
+                from lib.exchange.okx_options_lib import fetch_all_option_positions_history
+                from lib.hedge_plan.hedge_plan_settle_lib import (
+                    backfill_hedge_option_legs_realized_pnl,
+                )
+
+                hist = fetch_all_option_positions_history(ex, limit=200)
+                backfill_stats = backfill_hedge_option_legs_realized_pnl(conn, hist)
+        except Exception:
+            pass
         conn.commit()
     finally:
         conn.close()
-    return {"ok": True, "acted": acted}
+    return {"ok": True, "acted": acted, "pnl_backfill": backfill_stats}
 
 
 def _notify_end_reload(cfg: dict[str, Any], conn: Any, plan_id: int) -> None:
@@ -230,9 +247,10 @@ def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di
             ask_open = _sf(opt.get("avg_open"))
             if bid is not None and ask_open is not None:
                 ct = float(opt.get("ct_mult") or 0.01)
-                opt_pnl = (bid - ask_open) * float(opt.get("size") or 1) * ct
+                est = (bid - ask_open) * float(opt.get("size") or 1) * ct
             else:
-                opt_pnl = -premium
+                est = -premium
+            opt_pnl = _option_leg_pnl_after_close(cfg, opt, fallback=est)
         conn.execute(
             "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?",
             ("closed", reason, _now(), opt_pnl, opt["id"]),
@@ -284,6 +302,18 @@ def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di
     return {"plan_id": plan["id"], "close_reason": reason, "total": total}
 
 
+def _option_leg_pnl_after_close(
+    cfg: dict[str, Any],
+    leg: dict[str, Any],
+    *,
+    fallback: float,
+) -> float:
+    """平仓后写腿盈亏:优先交易所历史,否则用估算."""
+    ex = cfg.get("exchange_options")
+    pnl, _src = resolve_option_leg_realized_pnl(ex=ex, leg=leg, fallback=fallback)
+    return float(pnl if pnl is not None else fallback)
+
+
 def _estimate_leg_close_pnl(leg: dict[str, Any], idx: Optional[float], bid: Optional[float]) -> float:
     """残腿平仓盈亏估算:优先买一回收 − 权利金;无买一则用内在价值."""
     premium = float(leg.get("premium") or 0)
@@ -341,7 +371,8 @@ def _tick_oo_close_rest(
             update_plan(conn, int(plan["id"]), close_reason="oo_rest_closing")
             return {"plan_id": plan["id"], "msg": "残腿平仓失败", "close": close_r, "retry": True}
         bid = _sf(close_r.get("bid"))
-        pnl = _estimate_leg_close_pnl(leg, idx, bid)
+        est = _estimate_leg_close_pnl(leg, idx, bid)
+        pnl = _option_leg_pnl_after_close(cfg, leg, fallback=est)
         conn.execute(
             "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?",
             ("closed", "oo_rest_close", _now(), round(pnl, 4), leg["id"]),
@@ -418,9 +449,11 @@ def _tick_oo_target(
         )
         return {"plan_id": plan["id"], "msg": "平盈利腿失败", "close": close_r}
     reason = "target_up_win_leg" if hit_side == "up" else "target_down_win_leg"
+    # 选腿用内在估算;落库优先交易所已实现盈亏
+    closed_pnl = _option_leg_pnl_after_close(cfg, best, fallback=float(best_pnl))
     conn.execute(
         "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?",
-        ("closed", reason, _now(), best_pnl, best["id"]),
+        ("closed", reason, _now(), closed_pnl, best["id"]),
     )
     rest_mode = resolve_oo_rest_close_mode(plan)
     update_plan(conn, int(plan["id"]), close_reason=reason)
@@ -505,7 +538,8 @@ def _tick_oo_expiry(
 
     settled_sum = 0.0
     for leg in pending:
-        pnl = settle_option_leg_at_spot(leg, float(spot))
+        est = settle_option_leg_at_spot(leg, float(spot))
+        pnl = _option_leg_pnl_after_close(cfg, leg, fallback=est)
         settled_sum += pnl
         conn.execute(
             "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?",
@@ -554,7 +588,11 @@ def _settle_orphaned_after_tp(cfg: dict[str, Any], conn: Any) -> list[dict[str,
         spot = _index_px(cfg, str(leg.get("underlying") or "ETH"))
         if spot is None:
             continue
-        pnl = settle_option_leg_at_spot(leg, float(spot))
+        pnl_est = settle_option_leg_at_spot(leg, float(spot))
+        # orphan row uses leg_id; map to id for resolver
+        leg_for_pnl = dict(leg)
+        leg_for_pnl["id"] = leg.get("leg_id")
+        pnl = _option_leg_pnl_after_close(cfg, leg_for_pnl, fallback=pnl_est)
         conn.execute(
             "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?",
             ("closed", "expiry", _now(), round(pnl, 4), leg["leg_id"]),

lib/hedge_plan/hedge_plan_settle_lib.py

@@ -2,9 +2,10 @@
 from __future__ import annotations
 
 import time
-from typing import Any, Optional
+from datetime import datetime, timezone
+from typing import Any, Callable, Optional
 
-from lib.exchange.okx_options_lib import normalize_option_exp_ms
+from lib.exchange.okx_options_lib import normalize_option_exp_ms, resolve_option_close_from_history
 from lib.hedge_plan.hedge_plan_calc_lib import option_expiry_pnl
 
 
@@ -60,3 +61,148 @@ def all_option_legs_expired(legs: list[dict[str, Any]], *, now_ms: Optional[int]
     if not opts:
         return False
     return all(leg_is_expired(x, now_ms=now_ms) for x in opts)
+
+
+def _parse_opened_ms(raw: Any) -> Optional[int]:
+    if raw is None or raw == "":
+        return None
+    s = str(raw).strip()
+    if not s:
+        return None
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%f"):
+        try:
+            dt = datetime.strptime(s[:26], fmt).replace(tzinfo=timezone.utc)
+            return int(dt.timestamp() * 1000)
+        except ValueError:
+            continue
+    return None
+
+
+def resolve_option_leg_realized_pnl(
+    *,
+    ex: Any = None,
+    leg: dict[str, Any],
+    fallback: Optional[float] = None,
+    fetch_history_fn: Optional[Callable[[str], list[dict[str, Any]]]] = None,
+    hist_rows: Optional[list[dict[str, Any]]] = None,
+) -> tuple[Optional[float], str]:
+    """
+    期权腿已实现盈亏:优先 OKX positions-history realizedPnl.
+    返回 (pnl, source) source=exchange|fallback|none.
+    """
+    inst_id = str(leg.get("inst_id") or "").strip()
+    open_ms = _parse_opened_ms(leg.get("opened_at"))
+    rows = hist_rows
+    if rows is None and inst_id:
+        try:
+            if callable(fetch_history_fn):
+                rows = fetch_history_fn(inst_id)
+            elif ex is not None:
+                from lib.exchange.okx_options_lib import fetch_option_position_history
+
+                rows = fetch_option_position_history(ex, inst_id)
+        except Exception:
+            rows = None
+    if rows:
+        info = resolve_option_close_from_history(rows, open_ms=open_ms)
+        pnl = _sf((info or {}).get("realized_pnl")) if info else None
+        if pnl is not None:
+            return round(float(pnl), 4), "exchange"
+    if fallback is not None:
+        return round(float(fallback), 4), "fallback"
+    return None, "none"
+
+
+def backfill_hedge_option_legs_realized_pnl(
+    conn: Any,
+    hist_rows: list[dict[str, Any]],
+    *,
+    update_plan_fn: Optional[Callable[..., Any]] = None,
+) -> dict[str, int]:
+    """用交易所历史覆盖已平期权腿盈亏,并重算已结束计划合计."""
+    from lib.hedge_plan.hedge_plan_db import get_plan, get_plan_legs, update_plan
+
+    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 inst:
+            by_inst.setdefault(inst, []).append(raw)
+
+    legs = conn.execute(
+        """
+        SELECT * FROM hedge_plan_legs
+        WHERE status = 'closed'
+          AND inst_id IS NOT NULL AND TRIM(inst_id) != ''
+          AND (leg_role LIKE 'option%' OR opt_type IS NOT NULL)
+        ORDER BY id DESC
+        LIMIT 400
+        """
+    ).fetchall()
+    updated_legs = 0
+    touched_plans: set[int] = set()
+    for row in legs:
+        leg = dict(row)
+        inst = str(leg.get("inst_id") or "").strip()
+        if not inst or inst not in by_inst:
+            continue
+        pnl, src = resolve_option_leg_realized_pnl(
+            leg=leg,
+            hist_rows=by_inst[inst],
+            fallback=None,
+        )
+        if src != "exchange" or pnl is None:
+            continue
+        local = _sf(leg.get("realized_pnl"))
+        if local is not None and abs(local - pnl) < 1e-6:
+            continue
+        conn.execute(
+            "UPDATE hedge_plan_legs SET realized_pnl=? WHERE id=?",
+            (pnl, int(leg["id"])),
+        )
+        updated_legs += 1
+        touched_plans.add(int(leg["plan_id"]))
+
+    updated_plans = 0
+    updater = update_plan_fn or update_plan
+    for pid in touched_plans:
+        plan = get_plan(conn, pid)
+        if not plan or str(plan.get("status") or "") != "closed":
+            continue
+        plan_legs = get_plan_legs(conn, pid)
+        opt_sum = 0.0
+        for lg in plan_legs:
+            role = str(lg.get("leg_role") or "")
+            if not (role.startswith("option") or lg.get("opt_type")):
+                continue
+            if str(lg.get("status") or "") != "closed":
+                continue
+            opt_sum += float(_sf(lg.get("realized_pnl")) or 0.0)
+        perp = float(_sf(plan.get("realized_pnl_perp")) or 0.0)
+        ptype = str(plan.get("plan_type") or "")
+        if ptype == "options_options":
+            total = opt_sum
+            kwargs: dict[str, Any] = {
+                "realized_pnl_options": round(opt_sum, 4),
+                "realized_pnl_total": round(total, 4),
+            }
+        else:
+            total = perp + opt_sum
+            kwargs = {
+                "realized_pnl_perp": round(perp, 4),
+                "realized_pnl_options": round(opt_sum, 4),
+                "realized_pnl_total": round(total, 4),
+            }
+        old_total = _sf(plan.get("realized_pnl_total"))
+        old_opts = _sf(plan.get("realized_pnl_options"))
+        if (
+            old_total is not None
+            and abs(old_total - total) < 1e-6
+            and old_opts is not None
+            and abs(old_opts - opt_sum) < 1e-6
+        ):
+            continue
+        updater(conn, pid, **kwargs)
+        updated_plans += 1
+    return {"legs": updated_legs, "plans": updated_plans}

lib/options/options_review_lib.py

@@ -566,12 +566,16 @@ def ensure_local_review_synced(
     if backfill_exchange_pnl and ex is not None:
         try:
             from lib.exchange.okx_options_lib import fetch_all_option_positions_history
+            from lib.hedge_plan.hedge_plan_settle_lib import (
+                backfill_hedge_option_legs_realized_pnl,
+            )
             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)
+            backfill_hedge_option_legs_realized_pnl(conn, hist)
         except Exception:
             pass
     return sync_all_review_sources(conn, from_exchange=False)