Files
eth_hedge_sim/backend/app/sim/matcher.py
T
2026-07-31 15:44:52 +08:00

1128 lines
42 KiB
Python

"""本地模拟撮合:永续市价 + 期权只买开/卖平。"""
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
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)
return OpenResult(
ok=True,
group_id=group_id,
detail="opened",
data={
"group_id": group_id,
"perp": pf.to_dict(),
"option": of.to_dict(),
"initial_premium": initial_premium,
"fees": pf.fee + of.fee,
"open_sequence": ["option", "perp"],
},
)
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,
"fees_open": open_fees,
"fees_close": pf.fee + 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,
"interim_net": 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 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 _settle_one_residual(
self, row: dict[str, Any], *, now_ms: int, force: bool = False
) -> dict[str, Any] | None:
group_id = str(row["group_id"])
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()
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,
)
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,
note=f"residual option expiry settle{' force' if force else ''}",
# LIVE 本地账本仅镜像;拒记会导致 residual 永久 pending
allow_negative=not get_settings().is_sim,
)
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:
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,
"net_pnl": 0.0,
"est_close_fees": 0.0,
"index_px": None,
"move_points": 0.0,
"move_pct": 0.0,
"premium_gap": None,
}
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
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
# 浮盈亏:买一×数量 − 初始权利金(对齐可市价卖出)
option_upl = bid * opt_qty - initial_premium
elif oq:
opt_mark = oq.bid or oq.mark_px
if opt_mark is not None:
option_upl = float(opt_mark) * opt_qty - initial_premium
est_close_fees = est_perp_close_fee + est_opt_close_fee
# 净盈利:永续浮盈 + 期权浮盈 − 预估平仓手续费
net_pnl = perp_upl + option_upl - est_close_fees
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
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
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,
"strike": strike,
"expiry_ymd": expiry_ymd,
"expiry_ms": expiry_ms,
"perp_upl": perp_upl,
"option_upl": option_upl,
"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"),
}