"""本地模拟撮合:永续市价 + 期权只买开/卖平。""" from __future__ import annotations import logging import time from dataclasses import dataclass from typing import Any from ..config import get_settings 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, eth_from_contracts from .pricing import ( is_deep_otm, option_expiry_settle, option_fill, option_intrinsic, perp_fill, resolve_option_close_bid, ) logger = logging.getLogger(__name__) # 禁止新开仓的本地仓位状态(实盘防卡) BLOCKING_STATUSES = frozenset( {"open", "half_open", "option_closed_perp_pending", "opening"} ) @dataclass(slots=True) class OpenResult: ok: bool group_id: str | None = None detail: str = "" data: dict[str, Any] | None = None @dataclass(slots=True) class CloseResult: ok: bool detail: str = "" liquidity_wait: bool = False data: dict[str, Any] | None = None class Matcher: def __init__(self, db: Database | None = None) -> None: self.db = db or get_db() self.ledger = Ledger(self.db) def _fee_rate(self) -> float: return self.ledger.get_setting_float("fee_rate", get_settings().fee_rate) def _ct_mult(self, option_inst_id: str) -> float: try: from ..exchange.runtime import load_runtime_settings s = load_runtime_settings() except Exception: s = get_settings() try: return get_exchange().get_ct_mult( option_inst_id, s.option_inst_family, s.option_ct_mult_default ) except Exception: pass return float(s.option_ct_mult_default) def current_position(self) -> dict[str, Any]: row = self.db.fetchone("SELECT * FROM positions WHERE id=1") assert row is not None return dict(row) def has_open_position(self) -> bool: """是否禁止新开:含 open / half_open / option_closed_perp_pending / opening。""" pos = self.current_position() st = str(pos.get("status") or "") if st not in BLOCKING_STATUSES: return False if st == "opening": return True return bool(pos.get("group_id") or pos.get("option_inst_id")) def position_status(self) -> str: return str(self.current_position().get("status") or "flat") def _liquidity_wait(self, group_id: str, detail: str) -> CloseResult: note = f"liquidity_wait:{int(time.time())}:{detail[:80]}" self.db.execute( "UPDATE groups SET note=? WHERE group_id=? AND status='open'", (note, group_id), ) return CloseResult(ok=False, detail=detail, liquidity_wait=True) def _group_strike(self, group_id: str, option_inst_id: str) -> float | None: g = self.db.fetchone( "SELECT strike FROM groups WHERE group_id=?", (group_id,) ) if g is not None and g["strike"] is not None: try: return float(g["strike"]) except (TypeError, ValueError): pass try: from ..exchange.okx.parse import parse_option_inst_id _, stk, _ = parse_option_inst_id(option_inst_id) if stk is not None: return float(stk) except Exception: pass try: from ..exchange.binance.parse import parse_option_symbol _, stk, _ = parse_option_symbol(option_inst_id) if stk is not None: return float(stk) except Exception: pass return None def _close_spot_px(self, snap: Any) -> float | None: if getattr(snap, "index_px", None) is not None: try: px = float(snap.index_px) if px > 0: return px except (TypeError, ValueError): pass perp = getattr(snap, "perp", None) if not perp: return None for attr in ("mark_px", "last"): v = getattr(perp, attr, None) if v is not None: try: px = float(v) if px > 0: return px except (TypeError, ValueError): pass if perp.bid is not None and perp.ask is not None: return (float(perp.bid) + float(perp.ask)) / 2.0 if perp.bid is not None: return float(perp.bid) if perp.ask is not None: return float(perp.ask) return None def open_group( self, *, group_id: str, bias: str, option_side: str, # call|put perp_side: str, # long|short option_inst_id: str, entry_index_px: float, strike: float | None = None, expiry_ymd: str | None = None, ) -> OpenResult: s = get_settings() pos = self.current_position() st = str(pos.get("status") or "flat") if st in BLOCKING_STATUSES and ( st == "opening" or bool(pos.get("group_id") or pos.get("option_inst_id")) ): return OpenResult(ok=False, detail=f"已有持仓状态({st}),请先平仓") if pos.get("status") == "open" and pos.get("group_id"): return OpenResult(ok=False, detail="已有持仓组,请先平仓") sess = get_session() snap = sess.snapshot() if not snap.perp or snap.perp.bid is None or snap.perp.ask is None: return OpenResult(ok=False, detail="永续盘口不可用") oq = snap.call if option_side == "call" else snap.put # 若 ATM 对与持仓合约不一致,直接取持仓合约盘口 held = get_exchange().quote(option_inst_id) if held and held.ask is not None: oq = held if not oq or oq.ask is None: return OpenResult(ok=False, detail="期权卖一不可用") fee_rate = self._fee_rate() s = get_settings() perp_qty = self.ledger.get_setting_float("perp_qty_eth", s.perp_qty_eth) opt_qty = self.ledger.get_setting_float("option_qty_eth", s.option_qty_eth) ct_mult = self._ct_mult(option_inst_id) opt_contracts = contracts_for_eth(opt_qty, ct_mult) # 1) 先成交期权(吃卖一);失败则整组不开 of = option_fill( action="open", bid=float(oq.bid or 0), ask=float(oq.ask), qty_eth=opt_qty, fee_rate=fee_rate, ) initial_premium = of.fill_px * opt_qty # 锁定口径:成交价×名义,不含费 premium_cost = of.notional + of.fee try: self.ledger.apply_cash( -premium_cost, kind="open_option", group_id=group_id, note=f"open option {group_id}", ) except RuntimeError as e: return OpenResult(ok=False, detail=str(e)) # 2) 期权确认后再市价成交永续(重新取盘口) snap2 = sess.snapshot() if not snap2.perp or snap2.perp.bid is None or snap2.perp.ask is None: self.ledger.apply_cash( premium_cost, kind="open_option_rollback", group_id=group_id, note=f"rollback option {group_id}: perp book missing", ) return OpenResult( ok=False, detail="期权已成交但永续盘口不可用,已回滚期权", ) pf = perp_fill( side=perp_side, action="open", bid=float(snap2.perp.bid), ask=float(snap2.perp.ask), qty_eth=perp_qty, fee_rate=fee_rate, ) try: self.ledger.apply_cash( -pf.fee, kind="open_perp_fee", group_id=group_id, note=f"open perp {group_id}", ) except RuntimeError as e: self.ledger.apply_cash( premium_cost, kind="open_option_rollback", group_id=group_id, note=f"rollback option {group_id}: {e}", ) return OpenResult(ok=False, detail=f"期权已成交但永续扣费失败并已回滚: {e}") now = int(time.time() * 1000) with self.db._lock: self.db._conn.execute( """INSERT INTO groups( group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id, strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost, exec_mode ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, "open", bias, option_side, perp_side, option_inst_id, s.perp_inst_id, strike, expiry_ymd, entry_index_px, initial_premium, now, pf.fee + of.fee, pf.slip + of.slip, "SIM", ), ) # 成交顺序:期权先、永续后(时间戳差 1ms 便于审计) 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, exec_mode) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, "option", "open", "long", option_inst_id, opt_qty, opt_contracts, of.base_px, of.fill_px, of.fee, of.slip, of.notional, now, "SIM", ), ) 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, exec_mode) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, "perp", "open", perp_side, s.perp_inst_id, perp_qty, None, pf.base_px, pf.fill_px, pf.fee, pf.slip, pf.notional, now + 1, "SIM", ), ) self.db._conn.execute( """UPDATE positions SET group_id=?, perp_side=?, perp_qty_eth=?, perp_entry_px=?, option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?, option_entry_px=?, entry_index_px=?, initial_premium=?, status=? WHERE id=1""", ( group_id, perp_side, perp_qty, pf.fill_px, option_inst_id, option_side, opt_qty, opt_contracts, of.fill_px, entry_index_px, initial_premium, "open", ), ) self.db._conn.commit() try: from ..strategy.exits import lock_trade_exit_target lock_trade_exit_target( self.db, group_id=group_id, initial_premium=initial_premium ) except Exception: logger.exception("lock exit target failed group=%s", group_id) leverage = self.ledger.get_setting_float("leverage", s.leverage) perp_margin = ( abs(float(pf.fill_px) * float(perp_qty)) / float(leverage) if leverage and float(leverage) > 0 else None ) return OpenResult( ok=True, group_id=group_id, detail="opened", data={ "group_id": group_id, "bias": bias, "option_side": option_side, "perp_side": perp_side, "option_inst_id": option_inst_id, "strike": strike, "expiry_ymd": expiry_ymd, "perp": pf.to_dict(), "option": of.to_dict(), "perp_qty_eth": float(perp_qty), "option_qty_eth": float(opt_qty), "perp_entry_px": float(pf.fill_px), "option_entry_px": float(of.fill_px), "initial_premium": initial_premium, "perp_margin": perp_margin, "leverage": float(leverage) if leverage else None, "fees": pf.fee + of.fee, "open_sequence": ["option", "perp"], }, ) def open_oo_group( self, *, group_id: str, call_inst_id: str, put_inst_id: str, call_strike: float, put_strike: float, entry_index_px: float, expiry_ymd: str | None = None, ) -> OpenResult: """期期:买 Call 再买 Put,无永续。""" if not get_settings().is_sim: return OpenResult( ok=False, detail="LIVE 期期开仓须走 LiveExecutor.open_oo_group", ) pos = self.current_position() st = str(pos.get("status") or "flat") if st in BLOCKING_STATUSES and ( st == "opening" or bool(pos.get("group_id") or pos.get("option_inst_id")) ): return OpenResult(ok=False, detail=f"已有持仓状态({st}),请先平仓") if pos.get("status") == "open" and pos.get("group_id"): return OpenResult(ok=False, detail="已有持仓组,请先平仓") ex = get_exchange() cq = ex.quote(call_inst_id) pq = ex.quote(put_inst_id) if cq is None or cq.ask is None: _, asks, _ = ex.fetch_book(call_inst_id, depth=5) if asks: from types import SimpleNamespace cq = SimpleNamespace(ask=asks[0].px, bid=None) if pq is None or pq.ask is None: _, asks, _ = ex.fetch_book(put_inst_id, depth=5) if asks: from types import SimpleNamespace pq = SimpleNamespace(ask=asks[0].px, bid=None) if not cq or cq.ask is None or not pq or pq.ask is None: return OpenResult(ok=False, detail="期期 Call/Put 卖一不可用") fee_rate = self._fee_rate() opt_qty = self.ledger.get_setting_float("option_qty_eth", 0.1) if opt_qty < 0.1 - 1e-12: return OpenResult(ok=False, detail="期期名义 qty 无效") call_ct = self._ct_mult(call_inst_id) put_ct = self._ct_mult(put_inst_id) call_contracts = contracts_for_eth(opt_qty, call_ct) put_contracts = contracts_for_eth(opt_qty, put_ct) cf = option_fill( action="open", bid=float(getattr(cq, "bid", None) or 0), ask=float(cq.ask), qty_eth=opt_qty, fee_rate=fee_rate, ) call_prem = cf.fill_px * opt_qty call_cost = cf.notional + cf.fee try: self.ledger.apply_cash( -call_cost, kind="open_option", group_id=group_id, note=f"open oo call {group_id}", ) except RuntimeError as e: return OpenResult(ok=False, detail=str(e)) # 再买 Put;失败则尝试卖回 Call pq2 = ex.quote(put_inst_id) or pq ask2 = float(pq2.ask) if pq2 and pq2.ask else float(pq.ask) pf = option_fill( action="open", bid=float(getattr(pq2, "bid", None) or 0), ask=ask2, qty_eth=opt_qty, fee_rate=fee_rate, ) put_prem = pf.fill_px * opt_qty put_cost = pf.notional + pf.fee try: self.ledger.apply_cash( -put_cost, kind="open_option", group_id=group_id, note=f"open oo put {group_id}", ) except RuntimeError as e: # 回滚 Call:按买一卖出估算 bid = float(getattr(cq, "bid", None) or cf.fill_px) rb = option_fill( action="close", bid=bid, ask=float(cq.ask), qty_eth=opt_qty, fee_rate=fee_rate, ) self.ledger.apply_cash( rb.notional - rb.fee, kind="open_option_rollback", group_id=group_id, note=f"rollback oo call {group_id}: {e}", ) return OpenResult(ok=False, detail=f"Call 已成交但 Put 扣费失败并已回滚: {e}") now = int(time.time() * 1000) total_prem = call_prem + put_prem with self.db._lock: self.db._conn.execute( """INSERT INTO groups( group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id, strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost, exec_mode, hedge_mode, option2_inst_id, option2_side, strike2, initial_premium2 ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, "open", "option_option", "call", None, call_inst_id, None, float(call_strike), expiry_ymd, entry_index_px, call_prem, now, cf.fee + pf.fee, cf.slip + pf.slip, "SIM", "option_option", put_inst_id, "put", float(put_strike), put_prem, ), ) 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, exec_mode) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, "option", "open", "long", call_inst_id, opt_qty, call_contracts, cf.base_px, cf.fill_px, cf.fee, cf.slip, cf.notional, now, "SIM", ), ) 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, exec_mode) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, "option2", "open", "long", put_inst_id, opt_qty, put_contracts, pf.base_px, pf.fill_px, pf.fee, pf.slip, pf.notional, now + 1, "SIM", ), ) self.db._conn.execute( """UPDATE positions SET group_id=?, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?, option_entry_px=?, entry_index_px=?, initial_premium=?, status=?, hedge_mode=?, option2_inst_id=?, option2_side=?, option2_qty_eth=?, option2_qty_contracts=?, option2_entry_px=?, strike2=?, initial_premium2=? WHERE id=1""", ( group_id, call_inst_id, "call", opt_qty, call_contracts, cf.fill_px, entry_index_px, call_prem, "open", "option_option", put_inst_id, "put", opt_qty, put_contracts, pf.fill_px, float(put_strike), put_prem, ), ) self.db._conn.commit() try: from ..strategy.exits import lock_trade_exit_target lock_trade_exit_target( self.db, group_id=group_id, initial_premium=total_prem ) except Exception: logger.exception("lock exit target failed oo group=%s", group_id) return OpenResult( ok=True, group_id=group_id, detail="opened_oo", data={ "group_id": group_id, "hedge_mode": "option_option", "call_inst_id": call_inst_id, "put_inst_id": put_inst_id, "call_strike": float(call_strike), "put_strike": float(put_strike), "option_qty_eth": float(opt_qty), "initial_premium": total_prem, "fees": cf.fee + pf.fee, "open_sequence": ["call", "put"], }, ) def close_winning_oo_leave_residual( self, *, reason: str = "target_oo_win", skip_market: bool = False ) -> CloseResult: """期期达标:平盈利腿,亏损腿进 residual。skip_market=True 时假定已在交易所卖掉盈利腿。""" pos = self.current_position() st = str(pos.get("status") or "") if st not in ("open", "closing") or not pos.get("group_id"): return CloseResult(ok=False, detail="无期期持仓可平") if str(pos.get("hedge_mode") or "") != "option_option": # 兼容:有 option2 即视为期期 if not pos.get("option2_inst_id"): return CloseResult(ok=False, detail="非期期持仓") if st == "closing" and not skip_market: skip_market = True group_id = str(pos["group_id"]) call_id = str(pos.get("option_inst_id") or "") put_id = str(pos.get("option2_inst_id") or "") qty = float(pos.get("option_qty_eth") or 0) qty2 = float(pos.get("option2_qty_eth") or qty) if not call_id or not put_id or qty <= 0: return CloseResult(ok=False, detail="期期腿不完整") upl = self.unrealized() call_upl = float(upl.get("option_upl") or 0) put_upl = float(upl.get("option2_upl") or 0) # 盈利腿:UPL 更高且 > 0 if call_upl >= put_upl and call_upl > 0: win_leg, lose_leg = "option", "option2" win_id, lose_id = call_id, put_id win_side, lose_side = "call", "put" win_qty = qty lose_qty = qty2 win_entry = float(pos.get("option_entry_px") or 0) lose_entry = float(pos.get("option2_entry_px") or 0) lose_strike = float(pos.get("strike2") or 0) lose_prem = float(pos.get("initial_premium2") or 0) win_contracts = float(pos.get("option_qty_contracts") or 0) lose_contracts = float(pos.get("option2_qty_contracts") or 0) elif put_upl > call_upl and put_upl > 0: win_leg, lose_leg = "option2", "option" win_id, lose_id = put_id, call_id win_side, lose_side = "put", "call" win_qty = qty2 lose_qty = qty win_entry = float(pos.get("option2_entry_px") or 0) lose_entry = float(pos.get("option_entry_px") or 0) g = self.db.fetchone( "SELECT strike FROM groups WHERE group_id=?", (group_id,) ) lose_strike = float(g["strike"] or 0) if g else 0.0 lose_prem = float(pos.get("initial_premium") or 0) win_contracts = float(pos.get("option2_qty_contracts") or 0) lose_contracts = float(pos.get("option_qty_contracts") or 0) else: return CloseResult(ok=False, detail="无明确盈利腿,暂不平") fee_rate = self._fee_rate() if skip_market: oq = self._quote_held_option(win_id) fill_px = float(oq.bid) if oq and oq.bid else float(win_entry) of = option_fill( action="close", bid=fill_px, ask=fill_px, qty_eth=win_qty, fee_rate=fee_rate, ) else: oq = self._quote_held_option(win_id) if oq is None or oq.bid is None or float(oq.bid) <= 0: return CloseResult(ok=False, detail="盈利腿买一不可用") gate = self._residual_bid_gate( { "option_inst_id": win_id, "option_qty_eth": win_qty, "initial_premium": win_entry * win_qty, }, bid=float(oq.bid), oq=oq, require_premium_ratio=False, ) if gate: return CloseResult(ok=False, detail=f"盈利腿流动性不足: {gate}") of = option_fill( action="close", bid=float(oq.bid), ask=float(oq.ask or oq.bid), qty_eth=win_qty, fee_rate=fee_rate, ) cash = of.notional - of.fee if get_settings().is_sim or not skip_market: self.ledger.apply_cash( cash, kind="close_option", group_id=group_id, note=f"oo win {win_leg}" ) elif skip_market: # LIVE:交易所已成交,仍记本地账本现金(与其它 LIVE 平仓一致) try: self.ledger.apply_cash( cash, kind="close_option", group_id=group_id, note=f"oo win live {win_leg}", ) except Exception: logger.exception("oo win live ledger cash failed") now = int(time.time() * 1000) expiry_ymd = None expiry_ms = None g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,)) if g: expiry_ymd = g["expiry_ymd"] try: from ..exchange.expiry import expiry_ms_from_ymd if expiry_ymd: expiry_ms = int(expiry_ms_from_ymd(str(expiry_ymd))) except Exception: expiry_ms = None 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, exec_mode) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, win_leg, "close", "sell", win_id, win_qty, win_contracts, of.base_px, of.fill_px, of.fee, of.slip, of.notional, now, "SIM", ), ) self.db._conn.execute( """INSERT INTO residual_options( group_id, option_inst_id, option_side, option_qty_eth, option_qty_contracts, option_entry_px, strike, expiry_ymd, expiry_ms, entry_index_px, initial_premium, status, created_at_ms, note ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, lose_id, lose_side, lose_qty, lose_contracts, lose_entry, lose_strike, expiry_ymd, expiry_ms, float(pos.get("entry_index_px") or 0), lose_prem, "pending", now, f"oo losing leg after {reason}; win={win_leg}", ), ) # 组:记部分实现盈亏(赢腿),状态 residual win_pnl = (of.fill_px - win_entry) * win_qty - of.fee self.db._conn.execute( """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?, fees=COALESCE(fees,0)+?, note=? WHERE group_id=?""", ( "option_residual", now, reason, float(win_pnl), float(of.fee), f"oo win closed {win_leg}; lose {lose_leg} residual", group_id, ), ) self.db._conn.execute( """UPDATE positions SET group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, exit_target_usdt=NULL, status='flat', hedge_mode=NULL, option2_inst_id=NULL, option2_side=NULL, option2_qty_eth=0, option2_qty_contracts=0, option2_entry_px=NULL, strike2=NULL, initial_premium2=NULL WHERE id=1""" ) self.db._conn.commit() return CloseResult( ok=True, detail="oo_win_closed_lose_residual", data={ "group_id": group_id, "reason": reason, "win_leg": win_leg, "lose_leg": lose_leg, "win_pnl": win_pnl, }, ) def close_oo_full( self, *, reason: str = "expiry", bypass_liquidity: bool = False ) -> CloseResult: """期期全平两腿(到期/紧急);无永续。""" pos = self.current_position() if str(pos.get("status") or "") != "open" or not pos.get("group_id"): return CloseResult(ok=False, detail="无期期持仓可平") if not ( str(pos.get("hedge_mode") or "") == "option_option" or pos.get("option2_inst_id") ): return CloseResult(ok=False, detail="非期期持仓") group_id = str(pos["group_id"]) legs = [ ( "option", str(pos.get("option_inst_id") or ""), float(pos.get("option_qty_eth") or 0), float(pos.get("option_qty_contracts") or 0), float(pos.get("option_entry_px") or 0), float(pos.get("initial_premium") or 0), ), ( "option2", str(pos.get("option2_inst_id") or ""), float(pos.get("option2_qty_eth") or 0), float(pos.get("option2_qty_contracts") or 0), float(pos.get("option2_entry_px") or 0), float(pos.get("initial_premium2") or 0), ), ] fee_rate = self._fee_rate() now = int(time.time() * 1000) total_pnl = 0.0 total_fees = 0.0 for leg, inst, qty, contracts, entry, prem in legs: if not inst or qty <= 0: continue oq = self._quote_held_option(inst) if reason == "expiry": # 到期:尽量用买一,否则按 0 权利金结算 bid = float(oq.bid) if oq and oq.bid is not None else 0.0 ask = float(oq.ask) if oq and oq.ask is not None else bid else: if oq is None or oq.bid is None or float(oq.bid) <= 0: if not bypass_liquidity: return CloseResult( ok=False, detail=f"期期全平缺买一: {inst}" ) bid = float(entry) ask = bid else: if not bypass_liquidity: gate = self._residual_bid_gate( { "option_inst_id": inst, "option_qty_eth": qty, "initial_premium": prem or entry * qty, }, bid=float(oq.bid), oq=oq, require_premium_ratio=False, ) if gate: return CloseResult( ok=False, detail=f"期期全平流动性: {gate}" ) bid = float(oq.bid) ask = float(oq.ask or oq.bid) of = option_fill( action="close", bid=bid, ask=ask if ask > 0 else bid, qty_eth=qty, fee_rate=fee_rate, ) cash = of.notional - of.fee if get_settings().is_sim: self.ledger.apply_cash( cash, kind="close_option", group_id=group_id, note=f"oo full {leg}", ) else: try: self.ledger.apply_cash( cash, kind="close_option", group_id=group_id, note=f"oo full {leg}", ) except Exception: logger.exception("oo full ledger cash failed leg=%s", leg) total_pnl += (of.fill_px * qty - (prem or entry * qty)) - of.fee total_fees += of.fee 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, exec_mode) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, leg, "close", "sell", inst, qty, contracts, of.base_px, of.fill_px, of.fee, of.slip, of.notional, now, "SIM" if get_settings().is_sim else "LIVE", ), ) self.db._conn.commit() now += 1 with self.db._lock: self.db._conn.execute( """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?, fees=COALESCE(fees,0)+?, note=? WHERE group_id=?""", ( "closed", int(time.time() * 1000), reason, float(total_pnl), float(total_fees), f"oo full close {reason}", group_id, ), ) self.db._conn.execute( """UPDATE positions SET group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, exit_target_usdt=NULL, status='flat', hedge_mode=NULL, option2_inst_id=NULL, option2_side=NULL, option2_qty_eth=0, option2_qty_contracts=0, option2_entry_px=NULL, strike2=NULL, initial_premium2=NULL WHERE id=1""" ) self.db._conn.commit() return CloseResult( ok=True, detail="oo_full_closed", data={"group_id": group_id, "reason": reason, "net": total_pnl}, ) def close_group(self, *, reason: str, bypass_liquidity: bool = False) -> CloseResult: """ 全平一组。默认校验期权买一深度 + 买一/标记偏差(默认≤30%)。 reason=expiry:对齐实盘,期权按标的结算价的内在价值入账(不吃盘口)。 bypass_liquidity=True(紧急):绕过深度/偏差闸门,成交价仍按买一(对齐 OKX 市价卖,不用标记)。 成交顺序:先平期权 → 再瞬时平永续;永续盘口失败则回滚期权入账。 """ s = get_settings() pos = self.current_position() if pos.get("status") != "open" or not pos.get("group_id"): return CloseResult(ok=False, detail="无持仓可平") group_id = str(pos["group_id"]) sess = get_session() snap = sess.snapshot() if not snap.perp or snap.perp.bid is None or snap.perp.ask is None: return CloseResult(ok=False, detail="永续盘口不可用") option_inst_id = str(pos["option_inst_id"]) option_side = str(pos["option_side"]) # 严禁回退到 ATM 对:持仓行权价可能已偏离当前 ATM oq = self._quote_held_option(option_inst_id) if oq is None and reason != "expiry": return CloseResult( ok=False, detail=f"持仓期权盘口不可用: {option_inst_id}", ) ct_mult = self._ct_mult(option_inst_id) need_eth = float(pos["option_qty_eth"] or s.option_qty_eth) max_dev = self.ledger.get_setting_float( "close_bid_mark_max_pct", s.close_bid_mark_max_pct ) strike = self._group_strike(group_id, option_inst_id) spot = self._close_spot_px(snap) intrinsic: float | None = None if strike is not None and spot is not None: intrinsic = option_intrinsic( option_side=option_side, strike=strike, spot=spot ) fee_rate = self._fee_rate() is_expiry = reason == "expiry" if is_expiry: # 实盘到期:直接按内在价值结算,不依赖盘口 if intrinsic is None: return CloseResult( ok=False, detail="到期结算失败:缺少行权价或标的结算价", ) of = option_expiry_settle( intrinsic=float(intrinsic), qty_eth=float(pos["option_qty_eth"]), fee_rate=fee_rate, ) close_bid = float(intrinsic) else: if not oq: return CloseResult( ok=False, detail="期权盘口不可用", liquidity_wait=not bypass_liquidity, ) close_bid = oq.bid if not bypass_liquidity: if close_bid is None: return self._liquidity_wait(group_id, "期权买一不可用") if not bid_covers_eth( bid_sz_contracts=oq.bid_sz, ct_mult=ct_mult, need_eth=need_eth, ): return self._liquidity_wait(group_id, "期权买一流动性不足") ok_dev, why = bid_mark_ok( bid=close_bid, mark=oq.mark_px, max_dev_pct=max_dev ) if not ok_dev: return self._liquidity_wait(group_id, why) resolved = resolve_option_close_bid( bid=float(close_bid), mark=oq.mark_px, intrinsic=intrinsic, bypass_liquidity=False, ) if resolved is None: return self._liquidity_wait(group_id, "期权平仓价不可用") close_bid = resolved else: resolved = resolve_option_close_bid( bid=close_bid, mark=oq.mark_px, intrinsic=intrinsic, bypass_liquidity=True, ) if resolved is None: return CloseResult( ok=False, detail="紧急全平失败:无买一(对齐 OKX,不能用标记价平仓)", ) close_bid = resolved of = option_fill( action="close", bid=float(close_bid), ask=float(oq.ask or close_bid), qty_eth=float(pos["option_qty_eth"]), fee_rate=fee_rate, ) perp_side = str(pos["perp_side"]) perp_qty = float(pos["perp_qty_eth"]) opt_qty = float(pos["option_qty_eth"]) perp_entry = float(pos["perp_entry_px"]) opt_entry = float(pos["option_entry_px"]) # 1) 先平期权;永续对冲暂留 opt_pnl = (of.fill_px - opt_entry) * opt_qty opt_cash = of.notional - of.fee self.ledger.apply_cash( opt_cash, kind="close_option", group_id=group_id, note=f"close option {reason}", ) # 2) 期权确认后再瞬时平永续(重取盘口) snap2 = sess.snapshot() if not snap2.perp or snap2.perp.bid is None or snap2.perp.ask is None: self.ledger.apply_cash( -opt_cash, kind="close_option_rollback", group_id=group_id, note=f"rollback option close {group_id}: perp book missing", ) return CloseResult( ok=False, detail="期权已平但永续盘口不可用,已回滚期权入账", ) pf = perp_fill( side=perp_side, action="close", bid=float(snap2.perp.bid), ask=float(snap2.perp.ask), qty_eth=perp_qty, fee_rate=fee_rate, ) if perp_side == "long": perp_pnl = (pf.fill_px - perp_entry) * perp_qty else: perp_pnl = (perp_entry - pf.fill_px) * perp_qty self.ledger.apply_cash( perp_pnl - pf.fee, kind="close_perp", group_id=group_id, note=f"close perp {reason}", ) net = perp_pnl + opt_pnl - pf.fee - of.fee # 组已累计开仓手续费;实现净盈亏扣开+平全部手续费 open_fees = float( ( self.db.fetchone( "SELECT fees FROM groups WHERE group_id=?", (group_id,) ) or {"fees": 0} )["fees"] or 0 ) net_after_all_fees = perp_pnl + opt_pnl - open_fees - pf.fee - of.fee now = int(time.time() * 1000) 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", option_inst_id, opt_qty, float(pos["option_qty_contracts"] or 0), of.base_px, of.fill_px, of.fee, of.slip, of.notional, now, ), ) 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, "perp", "close", "flat", s.perp_inst_id, perp_qty, None, pf.base_px, pf.fill_px, pf.fee, pf.slip, pf.notional, now + 1, ), ) g = self.db._conn.execute( "SELECT fees, slip_cost FROM groups WHERE group_id=?", (group_id,) ).fetchone() fees = float(g["fees"] or 0) + pf.fee + of.fee slip = float(g["slip_cost"] or 0) + pf.slip + of.slip settle_index = float(spot) if spot is not None else None self.db._conn.execute( """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?, fees=?, slip_cost=?, note=NULL, settle_index_px=? WHERE group_id=?""", ( "closed", now, reason, net_after_all_fees, fees, slip, settle_index, group_id, ), ) self.db._conn.execute( """UPDATE positions SET group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, exit_target_usdt=NULL, status='flat' WHERE id=1""" ) self.db._conn.commit() return CloseResult( ok=True, detail="closed", data={ "group_id": group_id, "reason": reason, "perp_pnl": perp_pnl, "option_pnl": opt_pnl, "net": net_after_all_fees, "net_pnl": net_after_all_fees, "fees_open": open_fees, "fees_close": pf.fee + of.fee, "fees": float(open_fees) + float(pf.fee) + float(of.fee), "close_sequence": ["option", "perp"], "cash_delta": opt_cash + perp_pnl - pf.fee, "option_close_bid": float(close_bid), "option_intrinsic": intrinsic, "settle_spot": spot, }, ) def option_is_deep_otm(self) -> bool: """活跃组期权是否远虚(内在价值≈0)。""" pos = self.current_position() if pos.get("status") != "open" or not pos.get("group_id"): return False group_id = str(pos["group_id"]) option_inst_id = str(pos.get("option_inst_id") or "") option_side = str(pos.get("option_side") or "") strike = self._group_strike(group_id, option_inst_id) spot = self._close_spot_px(get_session().snapshot()) if strike is None or spot is None: return False return is_deep_otm( option_side=option_side, strike=float(strike), spot=float(spot) ) def close_perp_abandon_option( self, *, reason: str = "target_perp_only", require_deep_otm: bool = True ) -> CloseResult: """ 目标平仓 B:只平永续,期权归档为到期残留(不再盯盘、不挡新开)。 """ s = get_settings() pos = self.current_position() if pos.get("status") != "open" or not pos.get("group_id"): return CloseResult(ok=False, detail="无持仓可平") group_id = str(pos["group_id"]) sess = get_session() snap = sess.snapshot() if not snap.perp or snap.perp.bid is None or snap.perp.ask is None: return CloseResult(ok=False, detail="永续盘口不可用") option_inst_id = str(pos["option_inst_id"]) option_side = str(pos["option_side"]) strike = self._group_strike(group_id, option_inst_id) spot = self._close_spot_px(snap) if strike is None or spot is None: return CloseResult(ok=False, detail="无法判断远虚:缺行权价或标的价") if require_deep_otm and not is_deep_otm( option_side=option_side, strike=float(strike), spot=float(spot) ): return CloseResult(ok=False, detail="期权非远虚,应走双腿全平") fee_rate = self._fee_rate() perp_side = str(pos["perp_side"]) perp_qty = float(pos["perp_qty_eth"]) perp_entry = float(pos["perp_entry_px"]) pf = perp_fill( side=perp_side, action="close", bid=float(snap.perp.bid), ask=float(snap.perp.ask), qty_eth=perp_qty, fee_rate=fee_rate, ) if perp_side == "long": perp_pnl = (pf.fill_px - perp_entry) * perp_qty else: perp_pnl = (perp_entry - pf.fill_px) * perp_qty self.ledger.apply_cash( perp_pnl - pf.fee, kind="close_perp", group_id=group_id, note=f"close perp {reason} abandon option", ) g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,)) expiry_ymd = str(g["expiry_ymd"]) if g and g["expiry_ymd"] else None expiry_ms = None if expiry_ymd: try: from ..exchange.expiry import expiry_ms_from_ymd expiry_ms = int(expiry_ms_from_ymd(expiry_ymd)) except Exception: expiry_ms = None now = int(time.time() * 1000) open_fees = float((g["fees"] if g else 0) or 0) fees = open_fees + pf.fee slip = float((g["slip_cost"] if g else 0) or 0) + pf.slip # 暂记永续段实现盈亏;期权到期结算后再按全部成交重算 interim_net = perp_pnl - open_fees - pf.fee 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, "perp", "close", "flat", s.perp_inst_id, perp_qty, None, pf.base_px, pf.fill_px, pf.fee, pf.slip, pf.notional, now, ), ) self.db._conn.execute( """INSERT INTO residual_options( group_id, option_inst_id, option_side, option_qty_eth, option_qty_contracts, option_entry_px, strike, expiry_ymd, expiry_ms, entry_index_px, initial_premium, status, created_at_ms, note ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( group_id, option_inst_id, option_side, float(pos["option_qty_eth"]), float(pos["option_qty_contracts"] or 0), float(pos["option_entry_px"]), float(strike), expiry_ymd, expiry_ms, float(pos["entry_index_px"] or 0), float(pos["initial_premium"] or 0), "pending", now, f"abandoned after {reason}; deep_otm spot={spot:.4f} K={strike}", ), ) self.db._conn.execute( """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?, fees=?, slip_cost=?, note=? WHERE group_id=?""", ( "option_residual", now, reason, interim_net, fees, slip, f"perp_closed; option residual until expiry", group_id, ), ) self.db._conn.execute( """UPDATE positions SET group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, exit_target_usdt=NULL, status='flat' WHERE id=1""" ) self.db._conn.commit() return CloseResult( ok=True, detail="perp_closed_option_residual", data={ "group_id": group_id, "reason": reason, "mode": "target_perp_only", "perp_pnl": perp_pnl, "option_pnl": None, "interim_net": interim_net, "net": interim_net, "net_pnl": interim_net, "option_abandoned": True, "strike": float(strike), "spot": float(spot), }, ) def list_residual_options(self, *, pending_only: bool = True) -> list[dict[str, Any]]: if pending_only: rows = self.db.fetchall( "SELECT * FROM residual_options WHERE status='pending' ORDER BY created_at_ms ASC" ) else: rows = self.db.fetchall( "SELECT * FROM residual_options ORDER BY created_at_ms DESC LIMIT 100" ) return [dict(r) for r in rows] def _residual_min_premium_pct(self) -> float: s = get_settings() return float( self.ledger.get_setting_float( "residual_min_premium_pct", s.residual_min_premium_pct ) ) def _residual_bid_gate( self, row: dict[str, Any], *, bid: float, oq: Any, require_premium_ratio: bool = True, ) -> 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 opt_qty <= 0: return "invalid_qty" if bid <= 0: return "option_bid_unavailable" if require_premium_ratio: if initial_premium <= 0: return "invalid_initial_premium_or_qty" 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], *, require_premium_ratio: bool = True ) -> tuple[str | None, float | None, Any]: """ 残留平前置:买一流动性;自动路径另加权利金比例。 返回 (skip_reason, close_bid, option_quote);skip_reason 非空则本轮不卖。 """ 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) skip = self._residual_bid_gate( row, bid=close_bid, oq=oq, require_premium_ratio=require_premium_ratio, ) if skip: return (skip, None, None) return (None, close_bid, oq) def list_residual_options_enriched(self) -> list[dict[str, Any]]: """pending 残留 + 买一价/量、买一权利金、回收占比、流动性是否可手动平。""" out: list[dict[str, Any]] = [] for row in self.list_residual_options(pending_only=True): d = dict(row) option_inst_id = str(d.get("option_inst_id") or "") opt_qty = float(d.get("option_qty_eth") or 0) init = float(d.get("initial_premium") or 0) oq = self._quote_held_option(option_inst_id) if option_inst_id else None bid = float(oq.bid) if oq is not None and oq.bid is not None else None bid_sz = None bid_sz_eth = None if oq is not None and getattr(oq, "bid_sz", None) is not None: try: bid_sz = float(oq.bid_sz) except (TypeError, ValueError): bid_sz = None if bid_sz is not None and bid_sz >= 0: ct = self._ct_mult(option_inst_id) if option_inst_id else 0.01 bid_sz_eth = eth_from_contracts(bid_sz, ct) # 权利金 = 最新买一价 × 持仓数量;占比 = 买一权利金 / 开仓权利金 cur = float(bid) * opt_qty if bid is not None else None ratio = (cur / init * 100.0) if cur is not None and init > 1e-12 else None liq_detail: str | None if bid is None or oq is None: liq_detail = "option_bid_unavailable" else: liq_detail = self._residual_bid_gate( d, bid=bid, oq=oq, require_premium_ratio=False ) d.update( { "bid_px": bid, "bid_sz": bid_sz, "bid_sz_eth": bid_sz_eth, "current_premium": cur, "recovery_pct": ratio, "liquidity_ok": liq_detail is None, "liquidity_detail": liq_detail, } ) out.append(d) return out def close_residual_manual(self, group_id: str) -> CloseResult: """中控手动平单条残留:只验流动性,不验权利金比例。""" gid = str(group_id or "").strip() if not gid: return CloseResult(ok=False, detail="缺少 group_id") row = self.db.fetchone( "SELECT * FROM residual_options WHERE group_id=? AND status='pending'", (gid,), ) if row is None: return CloseResult(ok=False, detail="无该组 pending 残留") d = dict(row) skip, _bid, _oq = self._evaluate_residual_premium_close( d, require_premium_ratio=False ) if skip: return CloseResult(ok=False, detail=skip, liquidity_wait=True) booked = self.try_close_one_residual(d, skip_premium_ratio=True) if not booked: return CloseResult( ok=False, detail="平残留失败(流动性变化或下单未成交)", liquidity_wait=True, ) return CloseResult(ok=True, detail="residual_manual_closed", data=booked) def _book_residual_market_close( self, row: dict[str, Any], *, fill_px: float, fee: float, notional: float, slip: float, now_ms: int, note: str, exec_mode: str | None = None, 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"]) 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) * fill_eth if fill_eth > 0 else 0.0 opt_cash = fill_notional - float(fee) allow_neg = not get_settings().is_sim 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: 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() 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 - float(fee) g = self.db._conn.execute( "SELECT fees, slip_cost FROM groups WHERE group_id=?", (group_id,) ).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) 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 { "group_id": group_id, "option_pnl": opt_pnl, "settle_px": float(fill_px), "net_pnl": float(net), "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], *, skip_premium_ratio: bool = False ) -> dict[str, Any] | None: """SIM:流动性通过则吃买一平残留;自动路径另要求权利金比例。""" require_ratio = not skip_premium_ratio skip, close_bid, oq = self._evaluate_residual_premium_close( row, require_premium_ratio=require_ratio ) if skip or close_bid is None or oq is None: if skip: logger.debug( "residual close skip %s: %s", row.get("group_id"), 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, require_premium_ratio=require_ratio ) if skip2: logger.debug( "residual close recheck skip %s: %s", row.get("group_id"), skip2, ) return None of = option_fill( action="close", bid=float(bid2), ask=float(oq2.ask or bid2), qty_eth=float(row["option_qty_eth"]), fee_rate=self._fee_rate(), ) now_ms = int(time.time() * 1000) note = ( f"residual manual close at bid px={bid2}" if skip_premium_ratio else f"residual mid-close at bid px={bid2}" ) 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=note, filled_contracts=float(row.get("option_qty_contracts") or 0) or None, remaining_contracts=0.0, close_reason=( "residual_manual_close" if skip_premium_ratio else "residual_premium_close" ), ) def try_close_pending_residuals(self) -> list[dict[str, Any]]: """巡检全部 pending 残留,尝试权利金回收平仓。""" out: list[dict[str, Any]] = [] for row in self.list_residual_options(pending_only=True): try: r = self.try_close_one_residual(row) except Exception: logger.exception( "try_close_one_residual failed group=%s", row.get("group_id") ) continue if r: out.append(r) return out def settle_due_residuals(self, *, now_ms: int | None = None) -> list[dict[str, Any]]: """到期结算所有 pending 残留期权(不扫描进活跃组平仓)。""" now = int(now_ms if now_ms is not None else time.time() * 1000) pending = self.db.fetchall( "SELECT * FROM residual_options WHERE status='pending' ORDER BY id ASC" ) out: list[dict[str, Any]] = [] for row in pending: ems = row["expiry_ms"] if ems is None: ymd = row["expiry_ymd"] if ymd: try: from ..exchange.expiry import expiry_ms_from_ymd ems = int(expiry_ms_from_ymd(str(ymd))) except Exception: continue else: continue if now < int(ems): continue r = self._settle_one_residual(dict(row), now_ms=now) if r: out.append(r) return out def settle_all_residuals_now(self) -> list[dict[str, Any]]: """紧急:立即按内在价值结算全部残留(不等到期)。""" pending = self.db.fetchall( "SELECT * FROM residual_options WHERE status='pending' ORDER BY id ASC" ) now = int(time.time() * 1000) out: list[dict[str, Any]] = [] for row in pending: r = self._settle_one_residual(dict(row), now_ms=now, force=True) if r: 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.warning("residual settle skip %s: no strike/spot", group_id) return None fee_rate = self._fee_rate() intrinsic = option_intrinsic( option_side=str(row["option_side"]), strike=float(strike), spot=float(spot), ) of = option_expiry_settle( intrinsic=float(intrinsic), qty_eth=float(row["option_qty_eth"]), fee_rate=fee_rate, ) 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 ''}", filled_contracts=float(row.get("option_qty_contracts") or 0) or None, remaining_contracts=0.0, close_reason="expiry" if not force else "emergency", ) def _quote_held_option(self, option_inst_id: str): """只取持仓合约盘口;缺失时 REST 补一次,绝不借用 ATM 对。""" if not option_inst_id: return None ex = get_exchange() oq = ex.quote(option_inst_id) if oq is not None and (oq.bid is not None or oq.ask is not None or oq.mark_px is not None): return oq try: bids, asks, ts = ex.fetch_book(option_inst_id, depth=5) cache = getattr(ex, "cache", None) if cache is not None and (bids or asks): cache.upsert_book(option_inst_id, bids=bids, asks=asks, ts_ms=ts) try: mp = ex.fetch_mark(option_inst_id) if mp: cache.set_mark_px(option_inst_id, mp) except Exception: pass return ex.quote(option_inst_id) except Exception: return ex.quote(option_inst_id) def unrealized(self) -> dict[str, Any]: pos = self.current_position() if pos.get("status") != "open": return { "has_position": False, "perp_upl": 0.0, "option_upl": 0.0, "option2_upl": 0.0, "net_pnl": 0.0, "est_close_fees": 0.0, "index_px": None, "move_points": 0.0, "move_pct": 0.0, "premium_gap": None, } if str(pos.get("hedge_mode") or "") == "option_option" or pos.get( "option2_inst_id" ): return self._unrealized_oo(pos) return self._unrealized_perp(pos) def _unrealized_oo(self, pos: dict[str, Any]) -> dict[str, Any]: sess = get_session() snap = sess.snapshot() fee_rate = self._fee_rate() index_px = snap.index_px if index_px is None and snap.perp: index_px = snap.perp.mark_px qty = float(pos.get("option_qty_eth") or 0) qty2 = float(pos.get("option2_qty_eth") or qty) prem1 = float(pos.get("initial_premium") or 0) prem2 = float(pos.get("initial_premium2") or 0) call_id = str(pos.get("option_inst_id") or "") put_id = str(pos.get("option2_inst_id") or "") oq1 = self._quote_held_option(call_id) if call_id else None oq2 = self._quote_held_option(put_id) if put_id else None option_upl = 0.0 option2_upl = 0.0 fees = 0.0 if oq1 and oq1.bid is not None and qty > 0: bid = float(oq1.bid) of = option_fill( action="close", bid=bid, ask=float(oq1.ask or bid), qty_eth=qty, fee_rate=fee_rate, ) fees += of.fee option_upl = bid * qty - prem1 if oq2 and oq2.bid is not None and qty2 > 0: bid = float(oq2.bid) of = option_fill( action="close", bid=bid, ask=float(oq2.ask or bid), qty_eth=qty2, fee_rate=fee_rate, ) fees += of.fee option2_upl = bid * qty2 - prem2 g = None gid = pos.get("group_id") if gid: g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (gid,)) paid = float(g["fees"] or 0) if g else 0.0 net = option_upl + option2_upl - paid - fees entry_idx = float(pos.get("entry_index_px") or 0) move = abs(float(index_px) - entry_idx) if index_px is not None and entry_idx else 0.0 move_pct = (move / entry_idx * 100.0) if entry_idx > 0 else 0.0 return { "has_position": True, "hedge_mode": "option_option", "group_id": gid, "status": "open", "perp_upl": 0.0, "option_upl": option_upl, "option2_upl": option2_upl, "net_pnl": net, "fees_paid": paid, "est_close_fees": fees, "index_px": float(index_px) if index_px is not None else None, "entry_index_px": entry_idx, "move_points": move, "move_pct": move_pct, "initial_premium": prem1 + prem2, "option_inst_id": call_id, "option2_inst_id": put_id, "option_side": "call", "option2_side": "put", "option_qty_eth": qty, "option2_qty_eth": qty2, "option_entry_px": float(pos.get("option_entry_px") or 0), "option2_entry_px": float(pos.get("option2_entry_px") or 0), "strike": float(g["strike"]) if g and g["strike"] is not None else None, "strike2": float(pos.get("strike2") or 0) or None, "expiry_ymd": g["expiry_ymd"] if g else None, "open_at_ms": int(g["open_at_ms"]) if g and g["open_at_ms"] else None, "perp_side": None, "perp_qty_eth": 0.0, "premium_gap": None, } def _unrealized_perp(self, pos: dict[str, Any]) -> dict[str, Any]: sess = get_session() snap = sess.snapshot() s = get_settings() fee_rate = self._fee_rate() index_px = snap.index_px if index_px is None and snap.perp: index_px = snap.perp.mark_px perp_side = str(pos["perp_side"]) perp_entry = float(pos["perp_entry_px"]) perp_qty = float(pos["perp_qty_eth"]) opt_qty = float(pos["option_qty_eth"] or 0) opt_entry = float(pos["option_entry_px"] or 0) # 与平仓一致:用对手价估算可平盈亏 + 手续费 perp_upl = 0.0 est_perp_close_fee = 0.0 mark = None if snap.perp and snap.perp.bid is not None and snap.perp.ask is not None: pf = perp_fill( side=perp_side, action="close", bid=float(snap.perp.bid), ask=float(snap.perp.ask), qty_eth=perp_qty, fee_rate=fee_rate, ) if perp_side == "long": perp_upl = (pf.fill_px - perp_entry) * perp_qty else: perp_upl = (perp_entry - pf.fill_px) * perp_qty est_perp_close_fee = pf.fee mark = pf.fill_px elif snap.perp: if perp_side == "long": mark = snap.perp.bid or snap.perp.mark_px else: mark = snap.perp.ask or snap.perp.mark_px if mark is not None: if perp_side == "long": perp_upl = (float(mark) - perp_entry) * perp_qty else: perp_upl = (perp_entry - float(mark)) * perp_qty option_side = str(pos["option_side"]) opt_inst = str(pos.get("option_inst_id") or "") oq = self._quote_held_option(opt_inst) initial_premium = float(pos["initial_premium"] or 0) option_upl = 0.0 est_opt_close_fee = 0.0 opt_mark = None opt_bid_sz = None if oq and oq.bid is not None: bid = float(oq.bid) of = option_fill( action="close", bid=bid, ask=float(oq.ask or bid), qty_eth=opt_qty, fee_rate=fee_rate, ) est_opt_close_fee = of.fee opt_mark = bid opt_bid_sz = float(oq.bid_sz) if oq.bid_sz is not None else None # 浮盈亏:买一×数量 − 初始权利金(对齐可市价卖出) option_upl = bid * opt_qty - initial_premium elif oq: opt_mark = oq.bid or oq.mark_px opt_bid_sz = float(oq.bid_sz) if oq.bid_sz is not None else None if opt_mark is not None: option_upl = float(opt_mark) * opt_qty - initial_premium book_close_fees = est_perp_close_fee + est_opt_close_fee entry_idx = float(pos["entry_index_px"] or 0) move = abs(float(index_px) - entry_idx) if index_px is not None and entry_idx else 0.0 move_pct = (move / entry_idx * 100.0) if entry_idx > 0 else 0.0 premium_gap = initial_premium - perp_upl leverage = self.ledger.get_setting_float("leverage", s.leverage) notional = abs(perp_entry * perp_qty) margin = notional / leverage if leverage > 0 else None from ..strategy.selection import option_leverage as _opt_lev opt_lev = _opt_lev(entry_idx, opt_entry) if entry_idx > 0 and opt_entry > 0 else None if opt_lev is not None: opt_lev = round(float(opt_lev), 1) group_id = pos.get("group_id") g = ( self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,)) if group_id else None ) strike = float(g["strike"]) if g and g["strike"] is not None else None expiry_ymd = str(g["expiry_ymd"]) if g and g["expiry_ymd"] else None open_at_ms = int(g["open_at_ms"]) if g and g["open_at_ms"] else None open_fees = abs(float(g["fees"] or 0)) if g else 0.0 # 净盈利:浮盈 − 入场手续费×2(离场费按入场估算);无入场费时退回盘口估平仓费 if open_fees > 1e-12: est_close_fees = open_fees net_pnl = perp_upl + option_upl - open_fees * 2.0 else: est_close_fees = book_close_fees net_pnl = perp_upl + option_upl - est_close_fees expiry_ms = None if expiry_ymd and len(expiry_ymd) == 6: try: from ..exchange.expiry import expiry_ms_from_ymd expiry_ms = expiry_ms_from_ymd(expiry_ymd) except Exception: expiry_ms = None perp_inst_id = ( str(g["perp_inst_id"]) if g and g["perp_inst_id"] else s.perp_inst_id ) return { "has_position": True, "group_id": group_id, "open_at_ms": open_at_ms, "perp_side": perp_side, "option_side": option_side, "perp_inst_id": perp_inst_id, "perp_entry_px": perp_entry, "perp_qty_eth": perp_qty, "perp_mark_px": float(mark) if mark is not None else None, "perp_notional": notional, "perp_margin": margin, "leverage": leverage, "option_inst_id": pos.get("option_inst_id"), "option_entry_px": opt_entry, "option_qty_eth": opt_qty, "option_qty_contracts": float(pos["option_qty_contracts"] or 0), "option_mark_px": float(opt_mark) if opt_mark is not None else None, "option_bid_sz": float(opt_bid_sz) if opt_bid_sz is not None else None, "option_leverage": float(opt_lev) if opt_lev is not None else None, "strike": strike, "expiry_ymd": expiry_ymd, "expiry_ms": expiry_ms, "perp_upl": perp_upl, "option_upl": option_upl, "fees_paid": open_fees, "est_close_fees": est_close_fees, "net_pnl": net_pnl, "index_px": index_px, "entry_index_px": entry_idx, "move_points": move, "move_pct": move_pct, "initial_premium": initial_premium, "exit_target_usdt": ( float(pos["exit_target_usdt"]) if pos.get("exit_target_usdt") is not None else None ), "premium_gap": premium_gap, "status": pos.get("status"), }