From 469e7a258ab5391e238f9755ce4097eafb52b7c3 Mon Sep 17 00:00:00 2001 From: dekun Date: Sun, 2 Aug 2026 14:41:00 +0800 Subject: [PATCH] Harden residual mid-close: IOC partial fills, exchange reconcile, atomic book. Co-authored-by: Cursor --- backend/app/live/binance_executor.py | 186 +++++++- backend/app/live/binance_trade.py | 21 +- backend/app/live/executor.py | 197 +++++++- backend/app/live/okx_trade.py | 28 +- backend/app/sim/ledger.py | 20 +- backend/app/sim/matcher.py | 458 +++++++++++-------- backend/app/strategy/engine.py | 2 + backend/tests/test_residual_premium_close.py | 55 +++ 8 files changed, 729 insertions(+), 238 deletions(-) diff --git a/backend/app/live/binance_executor.py b/backend/app/live/binance_executor.py index 02fdfc7..689efbc 100644 --- a/backend/app/live/binance_executor.py +++ b/backend/app/live/binance_executor.py @@ -1149,12 +1149,72 @@ class BinanceLiveExecutor(Matcher): }, ) + def _sync_residual_contracts_with_exchange(self, row: dict) -> dict | None: + option_inst_id = str(row.get("option_inst_id") or "") + group_id = str(row.get("group_id") or "") + client = self._client() + ex_sz = exchange_option_abs_size(client, option_inst_id) + if ex_sz is None: + return row + ct = self._ct_mult(option_inst_id) + local_c = float(row.get("option_qty_contracts") or 0) + if local_c <= 0: + local_c = float( + contracts_for_eth(float(row.get("option_qty_eth") or 0), ct) or 0 + ) + if ex_sz <= 1e-8: + now_ms = int(time.time() * 1000) + booked = self._book_residual_market_close( + row, + fill_px=0.0, + fee=0.0, + notional=0.0, + slip=0.0, + now_ms=now_ms, + note="LIVE-BN residual already flat on exchange", + exec_mode="LIVE", + filled_contracts=0.0, + remaining_contracts=0.0, + close_reason="residual_premium_close", + ) + logger.warning( + "residual %s already flat on exchange; local settled=%s", + group_id, + booked is not None, + ) + return None + if local_c > ex_sz + 1e-8: + rem_eth = eth_from_contracts(float(ex_sz), ct) + init = float(row.get("initial_premium") or 0) + local_eth = float(row.get("option_qty_eth") or 0) + if local_eth > 1e-12: + init = init * (rem_eth / local_eth) + with self.db._lock: + self.db._conn.execute( + """UPDATE residual_options SET + option_qty_eth=?, option_qty_contracts=?, initial_premium=? + WHERE group_id=? AND status='pending'""", + (rem_eth, float(ex_sz), init, group_id), + ) + self.db._conn.commit() + row = { + **row, + "option_qty_eth": rem_eth, + "option_qty_contracts": float(ex_sz), + "initial_premium": init, + } + return row + def try_close_one_residual(self, row: dict) -> dict | None: """LIVE-BN：权利金达标后按最新买一 IOC 限价卖出归档期权（不扫市价）。""" err = self._guard_live() if err: logger.warning("residual premium close blocked: %s", err) return None + synced = self._sync_residual_contracts_with_exchange(row) + if synced is None: + return None + row = synced skip, close_bid, _oq = self._evaluate_residual_premium_close(row) if skip or close_bid is None: if skip: @@ -1176,10 +1236,15 @@ class BinanceLiveExecutor(Matcher): ) return None oq2 = self._quote_held_option(option_inst_id) - bid_px = float(oq2.bid) if oq2 is not None and oq2.bid is not None else float(close_bid) - if bid_px <= 0: + if oq2 is None or oq2.bid is None: + return None + bid_px = float(oq2.bid) + skip2 = self._residual_bid_gate(row, bid=bid_px, oq=oq2) + if skip2: logger.debug( - "residual premium close skip %s: bid vanished", row.get("group_id") + "residual premium close recheck skip %s: %s", + row.get("group_id"), + skip2, ) return None client = self._client() @@ -1200,26 +1265,125 @@ class BinanceLiveExecutor(Matcher): return None of_px = float(opt_live.avg_px) of_fee = float(opt_live.fee) - filled_c = ( - float(opt_live.sz) - if opt_live.sz and float(opt_live.sz) > 0 - else opt_contracts + filled_c = float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0 + if filled_c <= 1e-12: + return None + ex_left = exchange_option_abs_size(client, option_inst_id) + remaining = ( + max(0.0, float(ex_left)) + if ex_left is not None + else max(0.0, opt_contracts - filled_c) ) - opt_qty = eth_from_contracts(filled_c, self._ct_mult(option_inst_id)) - row = {**row, "option_qty_eth": opt_qty, "option_qty_contracts": filled_c} - of_notional = of_px * opt_qty + fill_eth = eth_from_contracts(filled_c, self._ct_mult(option_inst_id)) now_ms = int(time.time() * 1000) return self._book_residual_market_close( row, fill_px=of_px, fee=of_fee, - notional=of_notional, + notional=of_px * fill_eth, slip=0.0, now_ms=now_ms, note=f"LIVE-BN residual mid-close at bid IOC px={bid_px}", exec_mode="LIVE", + filled_contracts=filled_c, + remaining_contracts=remaining, + close_reason="residual_premium_close", ) + def _try_exchange_flatten_residual( + self, row: dict, *, force: bool = False + ) -> dict | None: + err = self._guard_live() + if err: + return None + option_inst_id = str(row.get("option_inst_id") or "") + client = self._client() + ex_sz = exchange_option_abs_size(client, option_inst_id) + if ex_sz is not None and ex_sz <= 1e-8: + return { + "fill_px": 0.0, + "fee": 0.0, + "notional": 0.0, + "slip": 0.0, + "filled_contracts": 0.0, + "remaining_contracts": 0.0, + "note": "LIVE-BN residual flat on exchange before settle", + "exec_mode": "LIVE", + "close_reason": "emergency" if force else "expiry", + } + opt_contracts = float(row.get("option_qty_contracts") or 0) + if ex_sz is not None and ex_sz > 0: + opt_contracts = float(ex_sz) + if opt_contracts <= 0: + opt_contracts = float( + contracts_for_eth( + float(row.get("option_qty_eth") or 0), + self._ct_mult(option_inst_id), + ) + or 0 + ) + if opt_contracts <= 0: + return None + oq = self._quote_held_option(option_inst_id) + bid_px = float(oq.bid) if oq is not None and oq.bid is not None else 0.0 + try: + if bid_px > 0 and not force: + opt_live = client.place_option_ioc( + symbol=option_inst_id, + side="SELL", + quantity=max(1.0, opt_contracts), + price=bid_px, + reduce_only=True, + ) + else: + opt_live = client.place_option_market( + symbol=option_inst_id, + side="SELL", + quantity=max(1.0, opt_contracts), + reduce_only=True, + ) + except Exception as e: + logger.warning( + "residual exchange flatten failed %s force=%s: %s", + row.get("group_id"), + force, + e, + ) + return None + filled_c = float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0 + if filled_c <= 1e-12 and force: + try: + opt_live = client.place_option_market( + symbol=option_inst_id, + side="SELL", + quantity=max(1.0, opt_contracts), + reduce_only=True, + ) + filled_c = ( + float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0 + ) + except Exception as e: + logger.warning("residual emergency market sell failed: %s", e) + return None + if filled_c <= 1e-12: + return None + of_px = float(opt_live.avg_px) + of_fee = float(opt_live.fee) + fill_eth = eth_from_contracts(filled_c, self._ct_mult(option_inst_id)) + ex_left = exchange_option_abs_size(client, option_inst_id) + remaining = max(0.0, float(ex_left)) if ex_left is not None else 0.0 + return { + "fill_px": of_px, + "fee": of_fee, + "notional": of_px * fill_eth, + "slip": 0.0, + "filled_contracts": filled_c, + "remaining_contracts": remaining, + "note": f"LIVE-BN residual exchange flatten force={force}", + "exec_mode": "LIVE", + "close_reason": "emergency" if force else "expiry", + } + def close_perp_abandon_option( self, *, reason: str = "target_perp_only", require_deep_otm: bool = True ) -> CloseResult: diff --git a/backend/app/live/binance_trade.py b/backend/app/live/binance_trade.py index 065292d..0297295 100644 --- a/backend/app/live/binance_trade.py +++ b/backend/app/live/binance_trade.py @@ -228,9 +228,11 @@ class BinanceTradeClient: if reduce_only: params["reduceOnly"] = "true" data = self._signed(self._eapi, "POST", "/eapi/v1/order", params) - return self._fill_from_eapi(symbol, data) + return self._fill_from_eapi(symbol, data, allow_partial=True) - def _fill_from_eapi(self, symbol: str, data: dict[str, Any]) -> LiveFill: + def _fill_from_eapi( + self, symbol: str, data: dict[str, Any], *, allow_partial: bool = False + ) -> LiveFill: ord_id = str(data.get("orderId") or data.get("id") or "") avg = safe_float(data.get("avgPrice")) or safe_float(data.get("price")) sz = safe_float(data.get("executedQty")) or safe_float(data.get("quantity")) @@ -251,16 +253,25 @@ class BinanceTradeClient: if avg and avg > 0 and st == "FILLED": break if st in ("CANCELED", "REJECTED", "EXPIRED"): + if allow_partial and sz and sz > 1e-12 and avg and avg > 0: + break raise RuntimeError(f"币安期权订单失败 status={st} {q}") if st == "PARTIALLY_FILLED": continue if not avg or avg <= 0: raise RuntimeError(f"币安期权无成交均价 orderId={ord_id} last={data}") st_final = str(data.get("status") or "").upper() + executed = safe_float(data.get("executedQty")) or float(sz or 0) if st_final and st_final != "FILLED": - raise RuntimeError( - f"币安期权未完全成交 status={st_final} orderId={ord_id} last={data}" - ) + if not ( + allow_partial + and executed > 1e-12 + and st_final in ("CANCELED", "EXPIRED", "PARTIALLY_FILLED") + ): + raise RuntimeError( + f"币安期权未完全成交 status={st_final} orderId={ord_id} last={data}" + ) + sz = executed from .money import abs_fee_usdt fee = abs(safe_float(data.get("fee")) or 0.0) diff --git a/backend/app/live/executor.py b/backend/app/live/executor.py index 1c0ee51..690cc41 100644 --- a/backend/app/live/executor.py +++ b/backend/app/live/executor.py @@ -1192,12 +1192,74 @@ class OkxLiveExecutor(Matcher): }, ) + def _sync_residual_contracts_with_exchange(self, row: dict) -> dict | None: + """按交易所持仓修正本地残留数量；已空仓则直接结清。返回待卖 row 或 None(已处理/跳过)。""" + option_inst_id = str(row.get("option_inst_id") or "") + group_id = str(row.get("group_id") or "") + client = self._client() + ex_sz = exchange_option_abs_size(client, option_inst_id) + if ex_sz is None: + return row + ct = self._ct_mult(option_inst_id) + local_c = float(row.get("option_qty_contracts") or 0) + if local_c <= 0: + local_c = float( + contracts_for_eth(float(row.get("option_qty_eth") or 0), ct) or 0 + ) + if ex_sz <= 1e-8: + now_ms = int(time.time() * 1000) + booked = self._book_residual_market_close( + row, + fill_px=0.0, + fee=0.0, + notional=0.0, + slip=0.0, + now_ms=now_ms, + note="LIVE residual already flat on exchange", + exec_mode="LIVE", + filled_contracts=0.0, + remaining_contracts=0.0, + close_reason="residual_premium_close", + ) + logger.warning( + "residual %s already flat on exchange; local settled=%s", + group_id, + booked is not None, + ) + return None + # 交易所更少：缩到交易所数量，避免超卖 + if local_c > ex_sz + 1e-8: + rem_eth = eth_from_contracts(float(ex_sz), ct) + init = float(row.get("initial_premium") or 0) + local_eth = float(row.get("option_qty_eth") or 0) + if local_eth > 1e-12: + init = init * (rem_eth / local_eth) + with self.db._lock: + self.db._conn.execute( + """UPDATE residual_options SET + option_qty_eth=?, option_qty_contracts=?, initial_premium=? + WHERE group_id=? AND status='pending'""", + (rem_eth, float(ex_sz), init, group_id), + ) + self.db._conn.commit() + row = { + **row, + "option_qty_eth": rem_eth, + "option_qty_contracts": float(ex_sz), + "initial_premium": init, + } + return row + def try_close_one_residual(self, row: dict) -> dict | None: """LIVE：权利金达标后按最新买一 IOC 限价卖出归档期权（不扫市价）。""" err = self._guard_live() if err: logger.warning("residual premium close blocked: %s", err) return None + synced = self._sync_residual_contracts_with_exchange(row) + if synced is None: + return None + row = synced skip, close_bid, _oq = self._evaluate_residual_premium_close(row) if skip or close_bid is None: if skip: @@ -1218,12 +1280,16 @@ class OkxLiveExecutor(Matcher): "residual premium close skip %s: bad contracts", row.get("group_id") ) return None - # 下单前再刷一次买一，按最新盘口挂 IOC oq2 = self._quote_held_option(option_inst_id) - bid_px = float(oq2.bid) if oq2 is not None and oq2.bid is not None else float(close_bid) - if bid_px <= 0: + if oq2 is None or oq2.bid is None: + return None + bid_px = float(oq2.bid) + skip2 = self._residual_bid_gate(row, bid=bid_px, oq=oq2) + if skip2: logger.debug( - "residual premium close skip %s: bid vanished", row.get("group_id") + "residual premium close recheck skip %s: %s", + row.get("group_id"), + skip2, ) return None client = self._client() @@ -1245,26 +1311,131 @@ class OkxLiveExecutor(Matcher): return None of_px = float(opt_live.avg_px) of_fee = float(opt_live.fee) - filled_c = ( - float(opt_live.sz) - if opt_live.sz and float(opt_live.sz) > 0 - else opt_contracts - ) - opt_qty = eth_from_contracts(filled_c, self._ct_mult(option_inst_id)) - row = {**row, "option_qty_eth": opt_qty, "option_qty_contracts": filled_c} - of_notional = of_px * opt_qty + filled_c = float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0 + if filled_c <= 1e-12: + return None + ex_left = exchange_option_abs_size(client, option_inst_id) + if ex_left is not None: + remaining = max(0.0, float(ex_left)) + else: + remaining = max(0.0, opt_contracts - filled_c) + fill_eth = eth_from_contracts(filled_c, self._ct_mult(option_inst_id)) now_ms = int(time.time() * 1000) return self._book_residual_market_close( row, fill_px=of_px, fee=of_fee, - notional=of_notional, + notional=of_px * fill_eth, slip=0.0, now_ms=now_ms, note=f"LIVE residual mid-close at bid IOC px={bid_px}", exec_mode="LIVE", + filled_contracts=filled_c, + remaining_contracts=remaining, + close_reason="residual_premium_close", ) + def _try_exchange_flatten_residual( + self, row: dict, *, force: bool = False + ) -> dict | None: + """到期/紧急：优先交易所卖掉残留；失败返回 None 走内在价值。""" + err = self._guard_live() + if err: + return None + option_inst_id = str(row.get("option_inst_id") or "") + client = self._client() + ex_sz = exchange_option_abs_size(client, option_inst_id) + if ex_sz is not None and ex_sz <= 1e-8: + return { + "fill_px": 0.0, + "fee": 0.0, + "notional": 0.0, + "slip": 0.0, + "filled_contracts": 0.0, + "remaining_contracts": 0.0, + "note": "LIVE residual flat on exchange before settle", + "exec_mode": "LIVE", + "close_reason": "emergency" if force else "expiry", + } + opt_contracts = float(row.get("option_qty_contracts") or 0) + if ex_sz is not None and ex_sz > 0: + opt_contracts = float(ex_sz) + if opt_contracts <= 0: + opt_contracts = float( + contracts_for_eth( + float(row.get("option_qty_eth") or 0), + self._ct_mult(option_inst_id), + ) + or 0 + ) + if opt_contracts <= 0: + return None + oq = self._quote_held_option(option_inst_id) + bid_px = float(oq.bid) if oq is not None and oq.bid is not None else 0.0 + try: + if bid_px > 0 and not force: + opt_live = client.place_ioc( + inst_id=option_inst_id, + side="sell", + sz=str(max(1, int(round(opt_contracts)))), + px=bid_px, + td_mode="cash", + reduce_only=True, + ) + else: + opt_live = client.place_market( + inst_id=option_inst_id, + side="sell", + sz=str(max(1, int(round(opt_contracts)))), + td_mode="cash", + reduce_only=True, + ) + except Exception as e: + logger.warning( + "residual exchange flatten failed %s force=%s: %s", + row.get("group_id"), + force, + e, + ) + return None + filled_c = float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0 + if filled_c <= 1e-12 and force: + # 紧急：再试市价 + try: + opt_live = client.place_market( + inst_id=option_inst_id, + side="sell", + sz=str(max(1, int(round(opt_contracts)))), + td_mode="cash", + reduce_only=True, + ) + filled_c = ( + float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0 + ) + except Exception as e: + logger.warning("residual emergency market sell failed: %s", e) + return None + if filled_c <= 1e-12: + return None + of_px = float(opt_live.avg_px) + of_fee = float(opt_live.fee) + fill_eth = eth_from_contracts(filled_c, self._ct_mult(option_inst_id)) + ex_left = exchange_option_abs_size(client, option_inst_id) + remaining = max(0.0, float(ex_left)) if ex_left is not None else 0.0 + # 到期/紧急要求尽量结清：若仍有剩余且 force，不在此硬结（返回 None 让内在价值兜底会重复） + # 有成交则先入账已成交部分；剩余留 pending 由下次处理，除非交易所已空 + return { + "fill_px": of_px, + "fee": of_fee, + "notional": of_px * fill_eth, + "slip": 0.0, + "filled_contracts": filled_c, + "remaining_contracts": remaining, + "note": f"LIVE residual exchange flatten force={force}", + "exec_mode": "LIVE", + "close_reason": "emergency" if force else "expiry", + } + def close_perp_abandon_option( self, *, reason: str = "target_perp_only", require_deep_otm: bool = True ) -> CloseResult: diff --git a/backend/app/live/okx_trade.py b/backend/app/live/okx_trade.py index 57664fe..f5624bc 100644 --- a/backend/app/live/okx_trade.py +++ b/backend/app/live/okx_trade.py @@ -185,7 +185,7 @@ class OkxTradeClient: if not rows: raise RuntimeError("OKX IOC 下单无返回") ord_id = str(rows[0].get("ordId") or "") - return self._wait_fill(inst_id, ord_id) + return self._wait_fill(inst_id, ord_id, allow_partial=True) def _fill_from_order_row(self, inst_id: str, ord_id: str, row: dict[str, Any]) -> LiveFill: avg = safe_float(row.get("avgPx")) or 0.0 @@ -206,7 +206,14 @@ class OkxTradeClient: raw=row, ) - def _wait_fill(self, inst_id: str, ord_id: str, *, tries: int = 40) -> LiveFill: + def _wait_fill( + self, + inst_id: str, + ord_id: str, + *, + tries: int = 40, + allow_partial: bool = False, + ) -> LiveFill: path = f"/api/v5/trade/order?instId={inst_id}&ordId={ord_id}" last: dict[str, Any] = {} for _ in range(tries): @@ -215,21 +222,38 @@ class OkxTradeClient: last = rows[0] state = str(last.get("state") or "") avg = safe_float(last.get("avgPx")) + acc = safe_float(last.get("accFillSz")) or 0.0 # 仅完全成交；部分成交继续等，避免账本张数与交易所不一致 if state == "filled" and avg and avg > 0: return self._fill_from_order_row(inst_id, ord_id, last) if state in ("canceled", "failed"): + # IOC：未成交部分取消；若已有成交量则按部分成交入账 + if ( + allow_partial + and acc > 1e-12 + and avg + and avg > 0 + ): + return self._fill_from_order_row(inst_id, ord_id, last) raise RuntimeError(f"OKX 订单失败 state={state} {last}") time.sleep(0.3) # 超时兜底：仅接受完全成交；部分成交不得当全成记账（会错张数/对冲） state = str(last.get("state") or "") avg = safe_float(last.get("avgPx")) + acc = safe_float(last.get("accFillSz")) or 0.0 if state == "filled" and avg and avg > 0: logger.warning( "OKX fill wait timeout but order filled ordId=%s", ord_id, ) return self._fill_from_order_row(inst_id, ord_id, last) + if allow_partial and acc > 1e-12 and avg and avg > 0: + logger.warning( + "OKX IOC partial fill on timeout ordId=%s acc=%s", + ord_id, + acc, + ) + return self._fill_from_order_row(inst_id, ord_id, last) raise RuntimeError(f"OKX 订单未完全成交 ordId={ord_id} last={last}") def sum_fill_fees(self, inst_id: str, ord_id: str) -> tuple[float, str]: diff --git a/backend/app/sim/ledger.py b/backend/app/sim/ledger.py index 0951e85..136a1b0 100644 --- a/backend/app/sim/ledger.py +++ b/backend/app/sim/ledger.py @@ -28,10 +28,12 @@ class Ledger: group_id: str | None = None, note: str = "", allow_negative: bool = False, + commit: bool = True, ) -> float: """amount>0 入账；amount<0 出账。返回余额。 LIVE 实盘成交后本地账本仅作镜像，须 allow_negative=True，避免「交易所已成交、本地拒记」导致卡仓。 + commit=False：由调用方持锁并统一提交（与持仓/残留状态同事务）。 """ now = int(time.time() * 1000) with self.db._lock: @@ -49,15 +51,17 @@ class Ledger: "INSERT INTO ledger_entries(group_id, kind, amount, balance_after, note, ts_ms) VALUES (?,?,?,?,?,?)", (group_id, kind, float(amount), equity, note, now), ) - self.db._conn.commit() - try: - from ..config import get_settings - from .funds_wallets import SimFundsWallets + if commit: + self.db._conn.commit() + if commit: + try: + from ..config import get_settings + from .funds_wallets import SimFundsWallets - if get_settings().is_sim: - SimFundsWallets(self.db).mirror_cash(float(amount), kind=kind) - except Exception: - pass + if get_settings().is_sim: + SimFundsWallets(self.db).mirror_cash(float(amount), kind=kind) + except Exception: + pass return equity def reset_equity(self, amount: float, *, note: str = "重置模拟资金") -> float: diff --git a/backend/app/sim/matcher.py b/backend/app/sim/matcher.py index 7f0b59b..194814d 100644 --- a/backend/app/sim/matcher.py +++ b/backend/app/sim/matcher.py @@ -12,7 +12,7 @@ from ..exchange import get_exchange from ..models.db import Database, get_db from ..strategy.session import get_session from .ledger import Ledger -from .liquidity import bid_covers_eth, bid_mark_ok, contracts_for_eth +from .liquidity import bid_covers_eth, bid_mark_ok, contracts_for_eth, eth_from_contracts from .pricing import ( is_deep_otm, option_expiry_settle, @@ -837,72 +837,64 @@ class Matcher: ) ) + def _residual_bid_gate( + self, row: dict[str, Any], *, bid: float, oq: Any + ) -> str | None: + """权利金比例 + 深度 + 买一/标记偏差。通过返回 None。""" + s = get_settings() + initial_premium = float(row.get("initial_premium") or 0) + opt_qty = float(row.get("option_qty_eth") or 0) + if initial_premium <= 0 or opt_qty <= 0: + return "invalid_initial_premium_or_qty" + if bid <= 0: + return "option_bid_unavailable" + current_premium = float(bid) * opt_qty + min_pct = self._residual_min_premium_pct() + threshold = initial_premium * (min_pct / 100.0) + if current_premium + 1e-12 < threshold: + return ( + f"premium_below_threshold curr={current_premium:.4f} " + f"need>={threshold:.4f} ({min_pct:g}%)" + ) + option_inst_id = str(row.get("option_inst_id") or "") + ct_mult = self._ct_mult(option_inst_id) + if not bid_covers_eth( + bid_sz_contracts=getattr(oq, "bid_sz", None), + ct_mult=ct_mult, + need_eth=opt_qty, + ): + return "option_bid_liquidity_insufficient" + max_dev = self.ledger.get_setting_float( + "close_bid_mark_max_pct", s.close_bid_mark_max_pct + ) + ok_dev, why = bid_mark_ok( + bid=float(bid), + mark=getattr(oq, "mark_px", None), + max_dev_pct=max_dev, + ) + if not ok_dev: + return why or "bid_mark_deviation" + return None + def _evaluate_residual_premium_close( self, row: dict[str, Any] ) -> tuple[str | None, float | None, Any]: """ 残留中途平前置：权利金比例 + 买一流动性。 返回 (skip_reason, close_bid, option_quote)；skip_reason 非空则本轮不卖。 + 成交价口径：最新买一（不再抬到内在价值）。 """ - s = get_settings() - initial_premium = float(row.get("initial_premium") or 0) - opt_qty = float(row.get("option_qty_eth") or 0) - if initial_premium <= 0 or opt_qty <= 0: - return ("invalid_initial_premium_or_qty", None, None) - option_inst_id = str(row.get("option_inst_id") or "") if not option_inst_id: return ("missing_option_inst", None, None) - oq = self._quote_held_option(option_inst_id) if oq is None or oq.bid is None: return ("option_bid_unavailable", None, None) - close_bid = float(oq.bid) - current_premium = close_bid * opt_qty - min_pct = self._residual_min_premium_pct() - threshold = initial_premium * (min_pct / 100.0) - if current_premium + 1e-12 < threshold: - return ( - f"premium_below_threshold curr={current_premium:.4f} " - f"need>={threshold:.4f} ({min_pct:g}%)", - None, - None, - ) - - ct_mult = self._ct_mult(option_inst_id) - if not bid_covers_eth( - bid_sz_contracts=oq.bid_sz, - ct_mult=ct_mult, - need_eth=opt_qty, - ): - return ("option_bid_liquidity_insufficient", None, None) - - max_dev = self.ledger.get_setting_float( - "close_bid_mark_max_pct", s.close_bid_mark_max_pct - ) - ok_dev, why = bid_mark_ok(bid=close_bid, mark=oq.mark_px, max_dev_pct=max_dev) - if not ok_dev: - return (why or "bid_mark_deviation", None, None) - - strike = row.get("strike") - spot = self._close_spot_px(get_session().snapshot()) - intrinsic: float | None = None - if strike is not None and spot is not None: - intrinsic = option_intrinsic( - option_side=str(row["option_side"]), - strike=float(strike), - spot=float(spot), - ) - resolved = resolve_option_close_bid( - bid=close_bid, - mark=oq.mark_px, - intrinsic=intrinsic, - bypass_liquidity=False, - ) - if resolved is None: - return ("option_close_px_unavailable", None, None) - return (None, float(resolved), oq) + skip = self._residual_bid_gate(row, bid=close_bid, oq=oq) + if skip: + return (skip, None, None) + return (None, close_bid, oq) def _book_residual_market_close( self, @@ -915,50 +907,108 @@ class Matcher: now_ms: int, note: str, exec_mode: str | None = None, - ) -> dict[str, Any]: - """买一卖出残留后的入账与结清（SIM/LIVE 共用）。""" + filled_contracts: float | None = None, + remaining_contracts: float | None = None, + close_reason: str = "residual_premium_close", + ) -> dict[str, Any] | None: + """买一卖出残留后的入账（与 pending 状态同事务）。支持部分成交扣减数量。""" group_id = str(row["group_id"]) - opt_qty = float(row["option_qty_eth"]) + option_inst_id = str(row["option_inst_id"]) + ct_mult = self._ct_mult(option_inst_id) + local_c = float(row.get("option_qty_contracts") or 0) + local_eth = float(row.get("option_qty_eth") or 0) + zero_fill_ok = ( + filled_contracts is not None + and float(filled_contracts) <= 1e-12 + and remaining_contracts is not None + and float(remaining_contracts) <= 1e-12 + ) + if filled_contracts is not None and float(filled_contracts) > 0: + fill_c = float(filled_contracts) + fill_eth = eth_from_contracts(fill_c, ct_mult) + elif zero_fill_ok: + fill_c = 0.0 + fill_eth = 0.0 + else: + fill_eth = local_eth + fill_c = local_c if local_c > 0 else contracts_for_eth(fill_eth, ct_mult) + if fill_eth <= 0 and not zero_fill_ok: + return None + fill_notional = ( + float(notional) + if float(notional) > 0 + else float(fill_px) * fill_eth + ) opt_entry = float(row["option_entry_px"]) - opt_pnl = (float(fill_px) - opt_entry) * opt_qty - opt_cash = float(notional) - float(fee) - self.ledger.apply_cash( - opt_cash, - kind="close_option", - group_id=group_id, - note=note, - allow_negative=not get_settings().is_sim, - ) + opt_pnl = (float(fill_px) - opt_entry) * fill_eth if fill_eth > 0 else 0.0 + opt_cash = fill_notional - float(fee) + allow_neg = not get_settings().is_sim - fill_cols = ( - "group_id, leg, action, side, inst_id, qty_eth, qty_contracts, " - "base_px, fill_px, fee, slip, notional, ts_ms" - ) - fill_vals: list[Any] = [ - group_id, - "option", - "close", - "flat", - str(row["option_inst_id"]), - opt_qty, - float(row["option_qty_contracts"] or 0), - float(fill_px), - float(fill_px), - float(fee), - float(slip), - float(notional), - now_ms, - ] - if exec_mode: - fill_cols += ", exec_mode" - fill_vals.append(exec_mode) + if remaining_contracts is not None: + rem_c = max(0.0, float(remaining_contracts)) + else: + rem_c = max(0.0, local_c - fill_c) if local_c > 0 else 0.0 + rem_eth = eth_from_contracts(rem_c, ct_mult) if rem_c > 0 else 0.0 + fully_done = rem_c <= 1e-8 with self.db._lock: - self.db._conn.execute( - f"""INSERT INTO fills({fill_cols}) - VALUES ({",".join("?" for _ in fill_vals)})""", - tuple(fill_vals), - ) + pending = self.db._conn.execute( + "SELECT * FROM residual_options WHERE group_id=? AND status='pending'", + (group_id,), + ).fetchone() + if pending is None: + logger.warning( + "residual book skip %s: not pending (already settled?)", group_id + ) + return None + + if abs(opt_cash) > 1e-12: + self.ledger.apply_cash( + opt_cash, + kind="close_option", + group_id=group_id, + note=note, + allow_negative=allow_neg, + commit=False, + ) + if get_settings().is_sim: + try: + from .funds_wallets import SimFundsWallets + + SimFundsWallets(self.db).mirror_cash( + float(opt_cash), kind="close_option" + ) + except Exception: + pass + + if fill_eth > 1e-12 or float(fee) > 1e-12: + fill_cols = ( + "group_id, leg, action, side, inst_id, qty_eth, qty_contracts, " + "base_px, fill_px, fee, slip, notional, ts_ms" + ) + fill_vals: list[Any] = [ + group_id, + "option", + "close", + "flat", + option_inst_id, + fill_eth, + fill_c, + float(fill_px), + float(fill_px), + float(fee), + float(slip), + fill_notional, + now_ms, + ] + if exec_mode: + fill_cols += ", exec_mode" + fill_vals.append(exec_mode) + self.db._conn.execute( + f"""INSERT INTO fills({fill_cols}) + VALUES ({",".join("?" for _ in fill_vals)})""", + tuple(fill_vals), + ) fills = self.db._conn.execute( "SELECT * FROM fills WHERE group_id=? ORDER BY id ASC", (group_id,) ).fetchall() @@ -973,32 +1023,62 @@ class Matcher: ).fetchone() fees = float(g["fees"] or 0) + float(fee) if g else float(fee) slip_total = float(g["slip_cost"] or 0) + float(slip) if g else float(slip) - self.db._conn.execute( - """UPDATE residual_options SET status=?, settled_at_ms=?, settle_px=?, settle_pnl=?, note=? - WHERE group_id=?""", - ( - "settled", - now_ms, - float(fill_px), - opt_pnl, - note, - group_id, - ), - ) - self.db._conn.execute( - """UPDATE groups SET status=?, close_at_ms=COALESCE(close_at_ms, ?), - close_reason=COALESCE(close_reason, ?), realized_pnl=?, fees=?, slip_cost=? - WHERE group_id=?""", - ( - "closed", - now_ms, - "residual_premium_close", - float(net), - fees, - slip_total, - group_id, - ), - ) + + if fully_done: + cur = self.db._conn.execute( + """UPDATE residual_options SET status=?, settled_at_ms=?, settle_px=?, settle_pnl=?, note=? + WHERE group_id=? AND status='pending'""", + ( + "settled", + now_ms, + float(fill_px), + opt_pnl, + note, + group_id, + ), + ) + if cur.rowcount != 1: + self.db._conn.rollback() + logger.warning("residual settle race %s", group_id) + return None + self.db._conn.execute( + """UPDATE groups SET status=?, close_at_ms=COALESCE(close_at_ms, ?), + close_reason=?, realized_pnl=?, fees=?, slip_cost=? + WHERE group_id=?""", + ( + "closed", + now_ms, + close_reason, + float(net), + fees, + slip_total, + group_id, + ), + ) + else: + # 按初始权利金比例缩减门槛基准，避免部分成交后永远达不到原 20% + init_prem = float(pending["initial_premium"] or 0) + if local_eth > 1e-12 and rem_eth > 0: + init_prem = init_prem * (rem_eth / local_eth) + cur = self.db._conn.execute( + """UPDATE residual_options SET + option_qty_eth=?, option_qty_contracts=?, initial_premium=?, note=? + WHERE group_id=? AND status='pending'""", + ( + rem_eth, + rem_c, + init_prem, + f"{note}; partial rem_c={rem_c}", + group_id, + ), + ) + if cur.rowcount != 1: + self.db._conn.rollback() + return None + self.db._conn.execute( + """UPDATE groups SET realized_pnl=?, fees=?, slip_cost=? WHERE group_id=?""", + (float(net), fees, slip_total, group_id), + ) self.db._conn.commit() return { @@ -1006,9 +1086,12 @@ class Matcher: "option_pnl": opt_pnl, "settle_px": float(fill_px), "net_pnl": float(net), - "reason": "residual_premium_close", - "current_premium": float(fill_px) * opt_qty, + "reason": close_reason, + "current_premium": float(fill_px) * fill_eth, "initial_premium": float(row.get("initial_premium") or 0), + "filled_contracts": fill_c, + "remaining_contracts": rem_c, + "fully_done": fully_done, } def try_close_one_residual(self, row: dict[str, Any]) -> dict[str, Any] | None: @@ -1022,10 +1105,22 @@ class Matcher: skip, ) return None + # 下单前再刷买一并重跑门槛 + option_inst_id = str(row.get("option_inst_id") or "") + oq2 = self._quote_held_option(option_inst_id) or oq + bid2 = float(oq2.bid) if oq2.bid is not None else float(close_bid) + skip2 = self._residual_bid_gate(row, bid=bid2, oq=oq2) + if skip2: + logger.debug( + "residual premium close recheck skip %s: %s", + row.get("group_id"), + skip2, + ) + return None of = option_fill( action="close", - bid=float(close_bid), - ask=float(oq.ask or close_bid), + bid=float(bid2), + ask=float(oq2.ask or bid2), qty_eth=float(row["option_qty_eth"]), fee_rate=self._fee_rate(), ) @@ -1037,7 +1132,9 @@ class Matcher: notional=of.notional, slip=of.slip, now_ms=now_ms, - note=f"residual mid-close at bid px={close_bid}", + note=f"residual mid-close at bid px={bid2}", + filled_contracts=float(row.get("option_qty_contracts") or 0) or None, + remaining_contracts=0.0, ) def try_close_pending_residuals(self) -> list[dict[str, Any]]: @@ -1095,16 +1192,44 @@ class Matcher: out.append(r) return out + def _try_exchange_flatten_residual( + self, row: dict[str, Any], *, force: bool = False + ) -> dict[str, Any] | None: + """LIVE 覆盖：尽量在交易所卖掉残留。成功返回 fill 字段字典。""" + return None + def _settle_one_residual( self, row: dict[str, Any], *, now_ms: int, force: bool = False ) -> dict[str, Any] | None: group_id = str(row["group_id"]) + # LIVE：优先交易所卖出再入账 + ex_fill = self._try_exchange_flatten_residual(row, force=force) + if ex_fill is not None: + booked = self._book_residual_market_close( + row, + fill_px=float(ex_fill["fill_px"]), + fee=float(ex_fill.get("fee") or 0), + notional=float(ex_fill["notional"]), + slip=float(ex_fill.get("slip") or 0), + now_ms=now_ms, + note=str(ex_fill.get("note") or "residual exchange settle"), + exec_mode=ex_fill.get("exec_mode"), + filled_contracts=ex_fill.get("filled_contracts"), + remaining_contracts=float(ex_fill.get("remaining_contracts") or 0), + close_reason=str( + ex_fill.get("close_reason") + or ("emergency" if force else "expiry") + ), + ) + if booked is not None: + booked["forced"] = force + return booked + sess = get_session() snap = sess.snapshot() spot = self._close_spot_px(snap) strike = row["strike"] if strike is None or spot is None: - logger = __import__("logging").getLogger(__name__) logger.warning("residual settle skip %s: no strike/spot", group_id) return None fee_rate = self._fee_rate() @@ -1118,84 +1243,19 @@ class Matcher: qty_eth=float(row["option_qty_eth"]), fee_rate=fee_rate, ) - opt_entry = float(row["option_entry_px"]) - opt_qty = float(row["option_qty_eth"]) - opt_pnl = (of.fill_px - opt_entry) * opt_qty - opt_cash = of.notional - of.fee - from ..config import get_settings - - self.ledger.apply_cash( - opt_cash, - kind="close_option", - group_id=group_id, + return self._book_residual_market_close( + row, + fill_px=of.fill_px, + fee=of.fee, + notional=of.notional, + slip=of.slip, + now_ms=now_ms, note=f"residual option expiry settle{' force' if force else ''}", - # LIVE 本地账本仅镜像；拒记会导致 residual 永久 pending - allow_negative=not get_settings().is_sim, + filled_contracts=float(row.get("option_qty_contracts") or 0) or None, + remaining_contracts=0.0, + close_reason="expiry" if not force else "emergency", ) - with self.db._lock: - self.db._conn.execute( - """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, - base_px, fill_px, fee, slip, notional, ts_ms) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "option", - "close", - "flat", - str(row["option_inst_id"]), - opt_qty, - float(row["option_qty_contracts"] or 0), - of.base_px, - of.fill_px, - of.fee, - of.slip, - of.notional, - now_ms, - ), - ) - fills = self.db._conn.execute( - "SELECT * FROM fills WHERE group_id=? ORDER BY id ASC", (group_id,) - ).fetchall() - from ..sim.pnl import summarize_fills_pnl - - summary = summarize_fills_pnl(list(fills)) - net = summary.get("net_pnl") - if net is None: - net = opt_pnl - of.fee - g = self.db._conn.execute( - "SELECT fees, slip_cost FROM groups WHERE group_id=?", (group_id,) - ).fetchone() - fees = float(g["fees"] or 0) + of.fee - slip = float(g["slip_cost"] or 0) + of.slip - self.db._conn.execute( - """UPDATE residual_options SET status=?, settled_at_ms=?, settle_px=?, settle_pnl=?, note=? - WHERE group_id=?""", - ( - "settled", - now_ms, - of.fill_px, - opt_pnl, - "settled at intrinsic", - group_id, - ), - ) - self.db._conn.execute( - """UPDATE groups SET status=?, close_at_ms=COALESCE(close_at_ms, ?), - realized_pnl=?, fees=?, slip_cost=? - WHERE group_id=?""", - ("closed", now_ms, float(net), fees, slip, group_id), - ) - self.db._conn.commit() - - return { - "group_id": group_id, - "option_pnl": opt_pnl, - "settle_px": of.fill_px, - "net_pnl": net, - "forced": force, - } - def _quote_held_option(self, option_inst_id: str): """只取持仓合约盘口；缺失时 REST 补一次，绝不借用 ATM 对。""" if not option_inst_id: diff --git a/backend/app/strategy/engine.py b/backend/app/strategy/engine.py index 903a2f3..49a73fe 100644 --- a/backend/app/strategy/engine.py +++ b/backend/app/strategy/engine.py @@ -570,6 +570,8 @@ class StrategyEngine: if not closed: return for item in closed: + if isinstance(item, dict) and item.get("fully_done") is False: + continue try: from ..notify import wecom diff --git a/backend/tests/test_residual_premium_close.py b/backend/tests/test_residual_premium_close.py index c802303..67c00d7 100644 --- a/backend/tests/test_residual_premium_close.py +++ b/backend/tests/test_residual_premium_close.py @@ -134,6 +134,61 @@ def test_residual_liquidity_fail_skips(tmp_path, monkeypatch) -> None: db.close() +def test_residual_recheck_bid_drop_skips(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("MODE", "SIM") + db = Database(tmp_path / "recheck.db") + db.set_setting("residual_min_premium_pct", "20") + _seed_residual(db, initial_premium=100.0, qty=2.0) + m = Matcher(db) + + good = SimpleNamespace(bid=15.0, ask=15.5, bid_sz=10_000.0, mark_px=15.0) + bad = SimpleNamespace(bid=5.0, ask=5.5, bid_sz=10_000.0, mark_px=5.0) + quotes = iter([good, bad]) + monkeypatch.setattr(m, "_quote_held_option", lambda _id: next(quotes)) + monkeypatch.setattr(m, "_close_spot_px", lambda _snap: 1900.0) + monkeypatch.setattr(m, "_ct_mult", lambda _id: 0.01) + + assert m.try_close_one_residual(m.list_residual_options()[0]) is None + row = db.fetchone( + "SELECT status FROM residual_options WHERE group_id=?", ("G-res",) + ) + assert row is not None and row["status"] == "pending" + db.close() + + +def test_residual_book_pending_guard(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("MODE", "SIM") + db = Database(tmp_path / "guard.db") + _seed_residual(db, initial_premium=100.0, qty=2.0) + m = Matcher(db) + row = m.list_residual_options()[0] + first = m._book_residual_market_close( + row, + fill_px=15.0, + fee=0.01, + notional=30.0, + slip=0.0, + now_ms=1_700_000_100_000, + note="first", + filled_contracts=200.0, + remaining_contracts=0.0, + ) + assert first is not None and first.get("fully_done") is True + second = m._book_residual_market_close( + row, + fill_px=15.0, + fee=0.01, + notional=30.0, + slip=0.0, + now_ms=1_700_000_200_000, + note="second", + filled_contracts=200.0, + remaining_contracts=0.0, + ) + assert second is None + db.close() + + def test_settings_exposes_residual_min_premium_pct(tmp_path, monkeypatch) -> None: monkeypatch.setenv("MODE", "SIM") from app.api import settings as settings_api