diff --git a/lib/exchange/okx_options_lib.py b/lib/exchange/okx_options_lib.py index 7e9714e..d8c5c41 100644 --- a/lib/exchange/okx_options_lib.py +++ b/lib/exchange/okx_options_lib.py @@ -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)) diff --git a/lib/options/options_close_exec_lib.py b/lib/options/options_close_exec_lib.py index add5082..26ceb4b 100644 --- a/lib/options/options_close_exec_lib.py +++ b/lib/options/options_close_exec_lib.py @@ -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} 张" diff --git a/lib/options/options_db.py b/lib/options/options_db.py index d7fef4f..2b69371 100644 --- a/lib/options/options_db.py +++ b/lib/options/options_db.py @@ -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 diff --git a/tests/test_options_partial_close_accum.py b/tests/test_options_partial_close_accum.py new file mode 100644 index 0000000..3338961 --- /dev/null +++ b/tests/test_options_partial_close_accum.py @@ -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()