diff --git a/lib/exchange/okx_options_lib.py b/lib/exchange/okx_options_lib.py index bc73ce0..d477ab3 100644 --- a/lib/exchange/okx_options_lib.py +++ b/lib/exchange/okx_options_lib.py @@ -158,7 +158,10 @@ def option_history_row_key( pos_id = (pos_id or "").strip() if source == "live": return f"live:{inst_id}:{pos_id or close_ms or '0'}" + # OKX 可能对同合约多次开平复用 posId,必须带上平仓时间区分 if pos_id: + if close_ms: + return f"ex:{pos_id}:{int(close_ms)}" return f"ex:{pos_id}" return f"ex:{inst_id}:{close_ms or 0}" @@ -1396,28 +1399,69 @@ def resolve_option_close_from_history( hist_rows: list[dict[str, Any]], *, open_ms: int | None = None, + close_ms: int | None = None, + sheets: float | int | None = None, ) -> dict[str, Any] | None: - """从 positions-history 中选取最近一条有效平仓/结算记录.""" - best: dict[str, Any] | None = None - best_utime = -1 + """从 positions-history 选取匹配的平仓记录. + + 同合约多次开平时,优先按开仓时间(cTime≈open_ms)对齐,再按平仓时间/张数; + 无锚点时取开仓后最晚一条(供刚平掉的持仓同步)。 + """ + candidates: list[tuple[int, dict[str, Any]]] = [] for row in hist_rows: u_ms = _safe_float(row.get("uTime")) if u_ms is None or u_ms <= 0: continue - if open_ms is not None and u_ms < int(open_ms) - 60_000: + u_i = int(u_ms) + # 本地时间偶发与交易所差整时区时,放宽到 12h,主要靠 cTime/张数精配 + if open_ms is not None and u_i < int(open_ms) - 12 * 3600_000: continue - if u_ms > best_utime: - best = row - best_utime = int(u_ms) - if not best: + candidates.append((u_i, row)) + if not candidates: return None + + has_ctime = any(_safe_float(row.get("cTime")) is not None for _, row in candidates) + want_sheets = _safe_float(sheets) + + def _score(item: tuple[int, dict[str, Any]]) -> tuple: + u_i, row = item + c_ms = _safe_float(row.get("cTime")) + parts: list[float] = [] + # 张数优先:同合约多笔时最稳,且不受本地/交易所时区偏差影响 + if want_sheets is not None: + hist_sheets = _safe_float(row.get("closeTotalPos")) + if hist_sheets is None: + hist_sheets = _safe_float(row.get("openMaxPos")) + parts.append( + abs(float(hist_sheets) - float(want_sheets)) + if hist_sheets is not None + else 1e12 + ) + if open_ms is not None and c_ms is not None: + 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: + parts.append(float(-u_i)) + # 同距时偏向更晚平仓 + parts.append(float(-u_i)) + return tuple(parts) + + if open_ms is None and close_ms is None and want_sheets is None: + u_i, best = max(candidates, key=lambda item: item[0]) + elif open_ms is not None and close_ms is None and want_sheets is None and not has_ctime: + # 兼容旧调用:只有 open_ms 时仍取最晚一条 + u_i, best = max(candidates, key=lambda item: item[0]) + else: + u_i, best = min(candidates, key=_score) + realized = _safe_float(best.get("realizedPnl")) if realized is None: realized = _safe_float(best.get("pnl")) return { "close_quote": _safe_float(best.get("closeAvgPx")), "realized_pnl": realized, - "close_ms": best_utime, + "close_ms": u_i, "pos_id": str(best.get("posId") or "").strip() or None, } diff --git a/lib/hedge_plan/hedge_plan_settle_lib.py b/lib/hedge_plan/hedge_plan_settle_lib.py index f835181..b1d6803 100644 --- a/lib/hedge_plan/hedge_plan_settle_lib.py +++ b/lib/hedge_plan/hedge_plan_settle_lib.py @@ -109,7 +109,11 @@ def resolve_option_leg_realized_pnl( except Exception: rows = None if rows: - info = resolve_option_close_from_history(rows, open_ms=open_ms) + close_ms = _parse_opened_ms(leg.get("closed_at")) + sheets = _sf(leg.get("size")) or _sf(leg.get("sheets")) + info = resolve_option_close_from_history( + rows, open_ms=open_ms, close_ms=close_ms, sheets=sheets + ) pnl = _sf((info or {}).get("realized_pnl")) if info else None if pnl is not None: return round(float(pnl), 4), "exchange" diff --git a/lib/options/options_monitor_lib.py b/lib/options/options_monitor_lib.py index 1c25a67..f77450d 100644 --- a/lib/options/options_monitor_lib.py +++ b/lib/options/options_monitor_lib.py @@ -142,11 +142,13 @@ def _created_at_ms(created_at: Any) -> int | 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 '')}" + close_prefix = closed[:16] if closed else "" + ord_id = str(row["close_ord_id"] or "").strip() if "close_ord_id" in row.keys() else "" + # 即使 close_ord_id/posId 相同,也要按平仓时间拆开(OKX 可能复用 posId) + if ord_id: + return f"{inst}|ord:{ord_id}|close:{close_prefix}" + return f"{inst}|close:{close_prefix}" def backfill_closed_options_realized_pnl_from_history( @@ -170,7 +172,8 @@ def backfill_closed_options_realized_pnl_from_history( rows = conn.execute( """ - SELECT id, inst_id, sheets, premium_paid, realized_pnl, created_at, closed_at, close_ord_id + SELECT id, inst_id, sheets, premium_paid, realized_pnl, close_quote, + created_at, closed_at, close_ord_id FROM options_trades WHERE status = 'closed' ORDER BY id DESC @@ -193,13 +196,26 @@ def backfill_closed_options_realized_pnl_from_history( 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) + close_candidates = [_created_at_ms(r["closed_at"]) for r in group] + close_ms = max((x for x in close_candidates if x is not None), default=None) + sheets_hint = None + try: + sheets_hint = sum(float(_safe_float(r["sheets"]) or 0.0) for r in group) or None + except (TypeError, ValueError): + sheets_hint = None + close_info = resolve_option_close_from_history( + by_inst.get(inst) or [], + open_ms=open_ms, + close_ms=close_ms, + sheets=sheets_hint, + ) 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")) + matched_pos = str(close_info.get("pos_id") or "").strip() or None total_paid = 0.0 for r in group: total_paid += float(_safe_float(r["premium_paid"]) or 0.0) @@ -215,7 +231,14 @@ def backfill_closed_options_realized_pnl_from_history( 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: + local_close = _safe_float(r["close_quote"]) + local_ord = str(r["close_ord_id"] or "").strip() + pnl_ok = local is not None and abs(local - share) < 1e-6 + quote_ok = close_quote is None or ( + local_close is not None and abs(local_close - float(close_quote)) < 1e-6 + ) + ord_ok = (not matched_pos) or (local_ord == matched_pos) + if pnl_ok and quote_ok and ord_ok: continue prem_recv = round(paid + share, 4) conn.execute( @@ -223,10 +246,11 @@ def backfill_closed_options_realized_pnl_from_history( UPDATE options_trades SET realized_pnl = ?, premium_received = ?, - close_quote = COALESCE(?, close_quote) + close_quote = COALESCE(?, close_quote), + close_ord_id = COALESCE(?, close_ord_id) WHERE id = ? """, - (share, prem_recv, close_quote, int(r["id"])), + (share, prem_recv, close_quote, matched_pos, int(r["id"])), ) updated += 1 return updated diff --git a/tests/test_options_sync.py b/tests/test_options_sync.py index afe0678..c6e8d5f 100644 --- a/tests/test_options_sync.py +++ b/tests/test_options_sync.py @@ -45,7 +45,7 @@ def test_format_option_history_row(): assert row["status_label"] == "已平" assert row["open_avg_px_fmt"] == "380" assert row["premium_paid_fmt"] == "3.80" - assert row["history_key"] == "ex:pos-btc" + assert row["history_key"] == "ex:pos-btc:1784088035000" def test_resolve_option_close_from_history_picks_latest(): @@ -59,6 +59,40 @@ def test_resolve_option_close_from_history_picks_latest(): assert got["pos_id"] == "9" +def test_resolve_option_close_from_history_matches_open_and_sheets(): + rows = [ + { + "instId": "ETH-USD_UM-260806-1875-C", + "cTime": "1785932775047", + "uTime": "1785933936733", + "realizedPnl": "-3.036", + "closeAvgPx": "12.4", + "closeTotalPos": "57", + "posId": "same-pos", + }, + { + "instId": "ETH-USD_UM-260806-1875-C", + "cTime": "1785938392445", + "uTime": "1785957764279", + "realizedPnl": "16.937", + "closeAvgPx": "41.0", + "closeTotalPos": "66", + "posId": "same-pos", + }, + ] + # 本地时间相对交易所偏 8h 时,仍应按 cTime/张数对齐到正确一笔 + early = resolve_option_close_from_history( + rows, open_ms=1785902775000, close_ms=1785903937000, sheets=57 + ) + late = resolve_option_close_from_history( + rows, open_ms=1785908392000, close_ms=1785927764000, sheets=66 + ) + assert early is not None and early["realized_pnl"] == -3.036 + assert early["close_quote"] == 12.4 + assert late is not None and late["realized_pnl"] == 16.937 + assert late["close_quote"] == 41.0 + + def test_sync_open_options_trades_marks_expired_closed(): conn = sqlite3.connect(":memory:") conn.row_factory = sqlite3.Row @@ -96,9 +130,9 @@ def test_sync_open_options_trades_skips_without_close_evidence(): INSERT INTO options_trades (inst_id, underlying, opt_type, strike, exp_time, sheets, eth_amount, open_quote, premium_paid, status, created_at) - VALUES (?, 'BTC', 'P', 62000, '', 1, 0.01, 380.0, 3.8, 'open', '2026-07-09 08:00:00') + VALUES (?, 'BTC', 'P', 62000, '', 1, 0.01, 380.0, 3.8, 'open', '2026-08-05 08:00:00') """, - ("BTC-USD_UM-260710-62000-P",), + ("BTC-USD_UM-261231-62000-P",), ) conn.commit() @@ -213,3 +247,64 @@ def test_backfill_closed_options_realized_pnl_from_history(): assert float(row["close_quote"]) == 48.5 # idempotent assert backfill_closed_options_realized_pnl_from_history(conn, hist) == 0 + + +def test_backfill_does_not_overwrite_earlier_close_with_later_pnl(): + 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, realized_pnl, status, created_at, closed_at) + VALUES (?, 'ETH', 'C', 1875, '', 57, 0.57, 16.6, 41.0, 9.462, 16.9368, + 'closed', '2026-08-05 12:26:15', '2026-08-05 12:45:37') + """, + ("ETH-USD_UM-260806-1875-C",), + ) + conn.execute( + """ + INSERT INTO options_trades + (inst_id, underlying, opt_type, strike, exp_time, sheets, eth_amount, + open_quote, close_quote, premium_paid, realized_pnl, status, created_at, closed_at) + VALUES (?, 'ETH', 'C', 1875, '', 66, 0.66, 14.2, 41.0, 9.372, 16.9368, + 'closed', '2026-08-05 13:59:52', '2026-08-05 19:22:45') + """, + ("ETH-USD_UM-260806-1875-C",), + ) + conn.commit() + hist = [ + { + "instId": "ETH-USD_UM-260806-1875-C", + "uTime": "1785933936733", + "cTime": "1785932775047", + "realizedPnl": "-3.03616314", + "closeAvgPx": "12.4", + "closeTotalPos": "57", + "posId": "3806091806281486337", + }, + { + "instId": "ETH-USD_UM-260806-1875-C", + "uTime": "1785957764279", + "cTime": "1785938392445", + "realizedPnl": "16.9368375", + "closeAvgPx": "41.0", + "closeTotalPos": "66", + "posId": "3806091806281486337", + }, + ] + n = backfill_closed_options_realized_pnl_from_history(conn, hist) + assert n >= 1 + rows = { + int(r["id"]): r + for r in conn.execute( + "SELECT id, realized_pnl, close_quote FROM options_trades ORDER BY id" + ).fetchall() + } + assert abs(float(rows[1]["realized_pnl"]) - (-3.0362)) < 1e-3 + assert float(rows[1]["close_quote"]) == 12.4 + assert abs(float(rows[2]["realized_pnl"]) - 16.9368) < 1e-3 + assert float(rows[2]["close_quote"]) == 41.0