dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix options PnL backfill matching when the same contract is traded twice.

Browse Source 
Match exchange history by sheets and open time so an earlier close is not overwritten with the later trade's PnL.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-08-06 09:04:30 +08:00
parent 25ed46e3f2
commit f19500bcd9
4 changed files with 189 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/exchange/okx_options_lib.py
+53 -9
View File
@@ -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,
}
lib/hedge_plan/hedge_plan_settle_lib.py
+5 -1
View File
@@ -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"
lib/options/options_monitor_lib.py
+33 -9
View File
@@ -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
tests/test_options_sync.py
+98 -3
View File
@@ -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