Files
eth_hedge_sim/backend/app/sim/matcher.py
T
2026-07-26 16:07:20 +08:00

700 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""本地模拟撮合:永续市价 + 期权只买开/卖平。"""
from __future__ import annotations
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 (
option_fill,
option_intrinsic,
perp_fill,
resolve_option_close_bid,
)
@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:
pos = self.current_position()
return pos.get("status") == "open" and bool(pos.get("group_id"))
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()
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
) 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,
),
)
# 成交顺序:期权先、永续后(时间戳差 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)
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,
),
)
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",
"open",
perp_side,
s.perp_inst_id,
perp_qty,
None,
pf.base_px,
pf.fill_px,
pf.fee,
pf.slip,
pf.notional,
now + 1,
),
)
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()
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%)。
bypass_liquidity=True:紧急/到期可绕过深度闸门;平仓价取 max(买一, 标记, 内在价值),
避免到期垃圾盘口把实值期权按近零价卖掉。
成交顺序:先平期权(薄)→ 再瞬时平永续(对冲先留着);永续盘口失败则回滚期权入账。
"""
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"])
oq = get_exchange().quote(option_inst_id) or (
snap.call if option_side == "call" else snap.put
)
if not oq:
return CloseResult(
ok=False,
detail="期权盘口不可用",
liquidity_wait=not bypass_liquidity,
)
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
)
close_bid = oq.bid
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
)
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:
# 到期/紧急:买一/标记可能枯死,用 max(买一, 标记, 内在价值)
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="紧急全平失败:无买一/标记/内在价值",
)
close_bid = resolved
fee_rate = self._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) 先平期权(买一流动性差);永续对冲暂留
of = option_fill(
action="close",
bid=float(close_bid),
ask=float(oq.ask or close_bid),
qty_eth=opt_qty,
fee_rate=fee_rate,
)
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
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
self.db._conn.execute(
"""UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?,
fees=?, slip_cost=?, note=NULL WHERE group_id=?""",
("closed", now, reason, net, fees, slip, 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, 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,
"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 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 = get_exchange().quote(opt_inst) if opt_inst else None
if oq is None:
oq = snap.call if option_side == "call" else snap.put
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
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,
"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,
"premium_gap": premium_gap,
"status": pos.get("status"),
}