dekun/eth_hedge_sim
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Recover stuck opening and harden LIVE open/close reconcile.

Browse Source 
Stamp open intent, recover opening from exchange option/perp state, skip resell/reopen when already flat, and persist Binance margin mode.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-29 20:32:45 +08:00
parent 44fd0371b9
commit e2a19a1614
8 changed files with 893 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/live/binance_executor.py
+227 -23
View File
@@ -15,8 +15,11 @@ from .binance_trade import BinanceTradeClient
from .reconcile import (
assert_safe_to_open_live,
claim_open_slot,
exchange_option_abs_size,
perp_close_qty_eth_binance,
recover_stuck_opening,
release_open_slot_if_opening,
stamp_opening_intent,
)
from .symbols import live_settings, resolve_perp_inst_id
@@ -39,6 +42,17 @@ class BinanceLiveExecutor(Matcher):
return reason
return None
def _perp_margin_mode(self) -> str:
from ..config import get_settings
s = get_settings()
raw = str(
self.ledger.get_setting_str("perp_margin_mode", s.perp_margin_mode)
or s.perp_margin_mode
or "cross"
).strip().lower()
return "isolated" if raw == "isolated" else "cross"
def unrealized(self) -> dict:
base = super().unrealized()
if not base.get("has_position"):
@@ -106,6 +120,17 @@ class BinanceLiveExecutor(Matcher):
ct_mult = self._ct_mult(option_inst_id)
opt_contracts = contracts_for_eth(opt_qty, ct_mult)
stamp_opening_intent(
self.db,
group_id=group_id,
option_inst_id=option_inst_id,
option_side=option_side,
perp_side=perp_side,
option_qty_eth=opt_qty,
option_qty_contracts=float(opt_contracts),
entry_index_px=entry_index_px,
)
try:
opt_fill = client.place_option_market(
symbol=option_inst_id,
@@ -129,14 +154,30 @@ class BinanceLiveExecutor(Matcher):
)
opt_contracts = filled_opt_contracts
opt_qty = eth_from_contracts(opt_contracts, ct_mult)
stamp_opening_intent(
self.db,
group_id=group_id,
option_inst_id=option_inst_id,
option_side=option_side,
perp_side=perp_side,
option_qty_eth=opt_qty,
option_qty_contracts=float(opt_contracts),
entry_index_px=entry_index_px,
option_entry_px=float(opt_fill.avg_px),
)
# 永续市价失败（多为保证金不足）→ 必须回滚期权
mgn = self._perp_margin_mode()
try:
if perp_side == "long":
side, pos_side = "BUY", "LONG"
else:
side, pos_side = "SELL", "SHORT"
leverage = self.ledger.get_setting_float("leverage", s.leverage)
try:
client.set_margin_type(perp_inst, mgn)
except Exception as e_mgn:
logger.warning("binance set_margin_type failed: %s", e_mgn)
try:
client.set_leverage(perp_inst, leverage)
except Exception as e_lev:
@@ -149,6 +190,21 @@ class BinanceLiveExecutor(Matcher):
)
except Exception as e:
logger.exception("binance live open perp failed (likely margin); rollback option")
try:
live_perp = client.get_perp_pos_sz(
perp_inst,
position_side=("LONG" if perp_side == "long" else "SHORT"),
)
except Exception:
live_perp = None
if live_perp is not None and live_perp > 1e-8:
return OpenResult(
ok=False,
detail=(
f"永续可能已成交但未确认成交明细(保留 opening): {e}; "
f"ex_perp={live_perp}"
),
)
try:
client.place_option_market(
symbol=option_inst_id,
@@ -220,8 +276,8 @@ class BinanceLiveExecutor(Matcher):
"""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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
exec_mode, perp_margin_mode
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
group_id,
"open",
@@ -238,6 +294,7 @@ class BinanceLiveExecutor(Matcher):
of_fee + pf_fee,
0.0,
"LIVE",
mgn,
),
)
self.db._conn.execute(
@@ -505,6 +562,101 @@ class BinanceLiveExecutor(Matcher):
data={"group_id": group_id, "exec_mode": "LIVE", "exchange": "binance"},
)
def recover_opening(self) -> CloseResult:
err = self._guard_live()
if err:
return CloseResult(ok=False, detail=err)
r = recover_stuck_opening(self)
if r is None:
return CloseResult(ok=False, detail="非 opening 状态")
return r
def _promote_opening_to_open(
self,
*,
pos: dict,
perp_inst: str,
opt_sz: float,
perp_total: float,
) -> CloseResult:
s = live_settings()
group_id = str(pos.get("group_id") or f"RCV-{int(time.time())}")
option_inst_id = str(pos.get("option_inst_id") or "")
option_side = str(pos.get("option_side") or "call")
perp_side = str(pos.get("perp_side") or "long")
of_px = float(pos.get("option_entry_px") or 0) or 0.0
opt_qty = float(pos.get("option_qty_eth") or 0)
opt_contracts = float(pos.get("option_qty_contracts") or opt_sz or 0)
if opt_qty <= 0 and opt_contracts > 0:
opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id))
perp_qty = float(pos.get("perp_qty_eth") or 0) or float(
self.ledger.get_setting_float("perp_qty_eth", s.perp_qty_eth)
)
if perp_total > 0:
perp_qty = float(perp_total)
entry_index = float(pos.get("entry_index_px") or 0) or 0.0
pf_px = entry_index if entry_index > 0 else of_px
initial_premium = of_px * opt_qty
mgn = self._perp_margin_mode()
now = int(time.time() * 1000)
with self.db._lock:
existing = self.db._conn.execute(
"SELECT group_id FROM groups WHERE group_id=?", (group_id,)
).fetchone()
if existing is None:
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, perp_margin_mode, note
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
group_id,
"open",
"recover",
option_side,
perp_side,
option_inst_id,
perp_inst,
None,
None,
entry_index,
initial_premium,
now,
0.0,
0.0,
"LIVE",
mgn,
"recover_opening both legs",
),
)
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='open'
WHERE id=1""",
(
group_id,
perp_side,
perp_qty,
pf_px,
option_inst_id,
option_side,
opt_qty,
opt_contracts,
of_px,
entry_index,
initial_premium,
),
)
self.db._conn.commit()
return CloseResult(
ok=True,
detail="recover_opening: 已提升为 open",
data={"group_id": group_id, "exec_mode": "LIVE"},
)
def close_group(self, *, reason: str, bypass_liquidity: bool = False) -> CloseResult:
err = self._guard_live()
if err:
@@ -513,6 +665,8 @@ class BinanceLiveExecutor(Matcher):
s = live_settings()
pos = self.current_position()
st = str(pos.get("status") or "")
if st == "opening":
return self.recover_opening()
if st == "half_open":
return self.repair_half_open()
if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"):
@@ -579,7 +733,34 @@ class BinanceLiveExecutor(Matcher):
opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id))
of_notional = of_px * opt_qty
except Exception as e:
if is_expiry and intrinsic is not None:
ex_opt = exchange_option_abs_size(client, option_inst_id)
if ex_opt is not None and ex_opt <= 1e-8:
prev = self.db.fetchone(
"""SELECT fill_px, fee, notional, slip FROM fills
WHERE group_id=? AND leg='option' AND action='close'
ORDER BY id DESC LIMIT 1""",
(group_id,),
)
if prev is not None:
of_px = float(prev["fill_px"])
of_fee = float(prev["fee"] or 0)
of_notional = float(prev["notional"] or (of_px * opt_qty))
of_slip = float(prev["slip"] or 0)
elif is_expiry and intrinsic is not None:
of = option_expiry_settle(
intrinsic=float(intrinsic), qty_eth=opt_qty, fee_rate=fee_rate
)
of_px, of_fee, of_notional = of.fill_px, of.fee, of.notional
of_slip = 0.0
else:
of_px = float(pos.get("option_entry_px") or 0) or 0.0
of_fee = 0.0
of_notional = of_px * opt_qty
of_slip = 0.0
logger.warning(
"binance option already flat on exchange; skip resell: %s", e
)
elif is_expiry and intrinsic is not None:
of = option_expiry_settle(
intrinsic=float(intrinsic), qty_eth=opt_qty, fee_rate=fee_rate
)
@@ -601,18 +782,32 @@ class BinanceLiveExecutor(Matcher):
else:
return CloseResult(ok=False, detail=f"币安平期权失败: {e}")
# 期权已平（或到期本地结算）：立刻落 pending，避免永续失败后重试再卖期权
self._mark_option_closed_perp_pending(
group_id=group_id,
option_inst_id=option_inst_id,
opt_qty=opt_qty,
opt_contracts=opt_contracts,
of_px=of_px,
of_fee=of_fee,
of_notional=of_notional,
of_slip=of_slip,
reason=reason,
st_now = str(self.current_position().get("status") or "")
prev_close = self.db.fetchone(
"""SELECT id FROM fills
WHERE group_id=? AND leg='option' AND action='close'
ORDER BY id DESC LIMIT 1""",
(group_id,),
)
if st_now == "option_closed_perp_pending" or prev_close is not None:
if st_now != "option_closed_perp_pending":
with self.db._lock:
self.db._conn.execute(
"UPDATE positions SET status='option_closed_perp_pending' WHERE id=1"
)
self.db._conn.commit()
else:
self._mark_option_closed_perp_pending(
group_id=group_id,
option_inst_id=option_inst_id,
opt_qty=opt_qty,
opt_contracts=opt_contracts,
of_px=of_px,
of_fee=of_fee,
of_notional=of_notional,
of_slip=of_slip,
reason=reason,
)
pending_perp_only = True
try:
@@ -625,16 +820,25 @@ class BinanceLiveExecutor(Matcher):
perp_inst=perp_inst,
perp_side=perp_side,
perp_qty_eth=perp_qty,
allow_db_fallback=not pending_perp_only,
)
perp_live = client.place_perp_market(
symbol=perp_inst,
side=side,
qty_eth=perp_qty_close,
position_side=pos_side,
reduce_only=True,
)
pf_px = float(perp_live.avg_px)
pf_fee = float(perp_live.fee)
if perp_qty_close <= 0:
pf_px = float(pos.get("perp_entry_px") or 0) or 0.0
pf_fee = 0.0
logger.warning(
"binance perp already flat; finalize without order group=%s",
group_id,
)
else:
perp_live = client.place_perp_market(
symbol=perp_inst,
side=side,
qty_eth=perp_qty_close,
position_side=pos_side,
reduce_only=True,
)
pf_px = float(perp_live.avg_px)
pf_fee = float(perp_live.fee)
except Exception as e:
return CloseResult(
ok=False,
backend/app/live/binance_trade.py
+63 -3
View File
@@ -162,7 +162,7 @@ class BinanceTradeClient:
sz = safe_float(q.get("executedQty")) or sz
data = q
if not avg or avg <= 0:
raise RuntimeError(f"币安永续无成交均价: {data}")
raise RuntimeError(f"币安永续无成交均价 orderId={ord_id} last={data}")
from .money import abs_fee_usdt
fee = abs(safe_float(data.get("cumCommission")) or 0.0)
@@ -226,10 +226,12 @@ class BinanceTradeClient:
if st == "PARTIALLY_FILLED":
continue
if not avg or avg <= 0:
raise RuntimeError(f"币安期权无成交均价: {data}")
raise RuntimeError(f"币安期权无成交均价 orderId={ord_id} last={data}")
st_final = str(data.get("status") or "").upper()
if st_final and st_final != "FILLED":
raise RuntimeError(f"币安期权未完全成交 status={st_final} {data}")
raise RuntimeError(
f"币安期权未完全成交 status={st_final} orderId={ord_id} last={data}"
)
from .money import abs_fee_usdt
fee = abs(safe_float(data.get("fee")) or 0.0)
@@ -336,6 +338,64 @@ class BinanceTradeClient:
return abs(float(amt))
return 0.0
def get_option_pos_sz(self, symbol: str) -> float | None:
"""期权持仓绝对张数；查不到接口时返回 None。"""
try:
rows = self._signed(self._eapi, "GET", "/eapi/v1/position", {"symbol": symbol})
except Exception as e:
logger.warning("binance get_option_pos_sz failed: %s", e)
return None
if isinstance(rows, dict):
rows = [rows]
total = 0.0
hit = False
for row in rows:
if not isinstance(row, dict):
continue
if str(row.get("symbol") or "") and str(row.get("symbol")) != symbol:
continue
qty = safe_float(row.get("quantity")) or safe_float(row.get("positionAmt")) or 0.0
hit = True
total += abs(float(qty))
return total if hit else 0.0
def any_option_pos_abs(self) -> float | None:
"""账户任意期权绝对持仓合计（ETH 期权）。"""
try:
rows = self._signed(self._eapi, "GET", "/eapi/v1/position", {})
except Exception as e:
logger.warning("binance any_option_pos_abs failed: %s", e)
return None
if isinstance(rows, dict):
rows = [rows]
total = 0.0
for row in rows:
if not isinstance(row, dict):
continue
sym = str(row.get("symbol") or "")
if sym and not sym.upper().startswith("ETH"):
continue
qty = safe_float(row.get("quantity")) or safe_float(row.get("positionAmt")) or 0.0
total += abs(float(qty))
return total
def set_margin_type(self, symbol: str, margin_type: str) -> None:
"""ISOLATED | CROSSED。"""
mt = "ISOLATED" if str(margin_type).lower() == "isolated" else "CROSSED"
try:
self._signed(
self._fapi,
"POST",
"/fapi/v1/marginType",
{"symbol": symbol, "marginType": mt},
)
except Exception as e:
# 已是目标模式时币安常报错，忽略
msg = str(e).lower()
if "no need to change" in msg or "-4046" in msg:
return
raise
def set_leverage(self, symbol: str, leverage: int | float) -> None:
lev = int(round(float(leverage)))
if lev < 1:
backend/app/live/executor.py
+225 -24
View File
@@ -17,8 +17,11 @@ from .okx_trade import OkxTradeClient
from .reconcile import (
assert_safe_to_open_live,
claim_open_slot,
exchange_option_abs_size,
perp_close_contracts_okx,
recover_stuck_opening,
release_open_slot_if_opening,
stamp_opening_intent,
)
from .symbols import live_settings, resolve_perp_inst_id
@@ -132,6 +135,18 @@ class OkxLiveExecutor(Matcher):
ct_mult = self._ct_mult(option_inst_id)
opt_contracts = contracts_for_eth(opt_qty, ct_mult)
# 意图先落库：崩溃后仍可 recover（含 option_inst_id）
stamp_opening_intent(
self.db,
group_id=group_id,
option_inst_id=option_inst_id,
option_side=option_side,
perp_side=perp_side,
option_qty_eth=opt_qty,
option_qty_contracts=float(opt_contracts),
entry_index_px=entry_index_px,
)
# 期权：买入，张数 = contracts
try:
opt_fill = client.place_market(
@@ -158,6 +173,17 @@ class OkxLiveExecutor(Matcher):
)
opt_contracts = filled_opt_contracts
opt_qty = eth_from_contracts(opt_contracts, ct_mult)
stamp_opening_intent(
self.db,
group_id=group_id,
option_inst_id=option_inst_id,
option_side=option_side,
perp_side=perp_side,
option_qty_eth=opt_qty,
option_qty_contracts=float(opt_contracts),
entry_index_px=entry_index_px,
option_entry_px=float(opt_fill.avg_px),
)
# 永续市价：按产品假设，失败原因实质为保证金不足 → 必须回滚期权
mgn = self._perp_margin_mode()
@@ -190,6 +216,21 @@ class OkxLiveExecutor(Matcher):
)
except Exception as e:
logger.exception("live open perp failed (likely margin); rollback option")
# 永续可能已成交：先查仓，有仓则不得回滚期权
try:
live_perp = client.get_perp_pos_sz(
perp_inst, pos_side=("long" if perp_side == "long" else "short")
)
except Exception:
live_perp = None
if live_perp is not None and live_perp > 1e-8:
return OpenResult(
ok=False,
detail=(
f"永续可能已成交但未确认成交明细(保留 opening): {e}; "
f"ex_perp={live_perp}"
),
)
try:
client.place_market(
inst_id=option_inst_id,
@@ -544,6 +585,109 @@ class OkxLiveExecutor(Matcher):
data={"group_id": group_id, "exec_mode": "LIVE"},
)
def recover_opening(self) -> CloseResult:
"""恢复 stuck opening（交易所对账后 half_open/open/清槽）。"""
err = self._guard_live()
if err:
return CloseResult(ok=False, detail=err)
r = recover_stuck_opening(self)
if r is None:
return CloseResult(ok=False, detail="非 opening 状态")
return r
def _promote_opening_to_open(
self,
*,
pos: dict,
perp_inst: str,
opt_sz: float,
perp_total: float,
) -> CloseResult:
"""opening + 交易所双边有仓 → 落本地 open（用 stamp/设置数量）。"""
s = live_settings()
group_id = str(pos.get("group_id") or f"RCV-{int(time.time())}")
option_inst_id = str(pos.get("option_inst_id") or "")
option_side = str(pos.get("option_side") or "call")
perp_side = str(pos.get("perp_side") or "long")
of_px = float(pos.get("option_entry_px") or 0) or 0.0
opt_qty = float(pos.get("option_qty_eth") or 0)
opt_contracts = float(pos.get("option_qty_contracts") or opt_sz or 0)
if opt_qty <= 0 and opt_contracts > 0:
opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id))
perp_qty = float(pos.get("perp_qty_eth") or 0) or float(
self.ledger.get_setting_float("perp_qty_eth", s.perp_qty_eth)
)
if perp_total > 0:
try:
ct_val = self._client().get_ct_val(perp_inst, inst_type="SWAP")
if ct_val > 0:
perp_qty = float(perp_total) * float(ct_val)
except Exception:
pass
entry_index = float(pos.get("entry_index_px") or 0) or 0.0
# 永续入场价未知时用指数近似（仅恢复镜像）
pf_px = entry_index if entry_index > 0 else of_px
initial_premium = of_px * opt_qty
mgn = self._perp_margin_mode()
now = int(time.time() * 1000)
with self.db._lock:
existing = self.db._conn.execute(
"SELECT group_id FROM groups WHERE group_id=?", (group_id,)
).fetchone()
if existing is None:
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, perp_margin_mode, note
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
group_id,
"open",
"recover",
option_side,
perp_side,
option_inst_id,
perp_inst,
None,
None,
entry_index,
initial_premium,
now,
0.0,
0.0,
"LIVE",
mgn,
"recover_opening both legs",
),
)
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='open'
WHERE id=1""",
(
group_id,
perp_side,
perp_qty,
pf_px,
option_inst_id,
option_side,
opt_qty,
opt_contracts,
of_px,
entry_index,
initial_premium,
),
)
self.db._conn.commit()
return CloseResult(
ok=True,
detail="recover_opening: 已提升为 open",
data={"group_id": group_id, "exec_mode": "LIVE"},
)
def close_group(self, *, reason: str, bypass_liquidity: bool = False) -> CloseResult:
err = self._guard_live()
if err:
@@ -552,6 +696,8 @@ class OkxLiveExecutor(Matcher):
s = live_settings()
pos = self.current_position()
st = str(pos.get("status") or "")
if st == "opening":
return self.recover_opening()
if st == "half_open":
return self.repair_half_open()
if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"):
@@ -619,7 +765,35 @@ class OkxLiveExecutor(Matcher):
opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id))
of_notional = of_px * opt_qty
except Exception as e:
if is_expiry and intrinsic is not None:
# 交易所期权可能已空（上次卖出成功但未 mark）：跳过再卖，直接 pending
ex_opt = exchange_option_abs_size(client, option_inst_id)
if ex_opt is not None and ex_opt <= 1e-8:
prev = self.db.fetchone(
"""SELECT fill_px, fee, notional, slip FROM fills
WHERE group_id=? AND leg='option' AND action='close'
ORDER BY id DESC LIMIT 1""",
(group_id,),
)
if prev is not None:
of_px = float(prev["fill_px"])
of_fee = float(prev["fee"] or 0)
of_notional = float(prev["notional"] or (of_px * opt_qty))
of_slip = float(prev["slip"] or 0)
elif is_expiry and intrinsic is not None:
of = option_expiry_settle(
intrinsic=float(intrinsic), qty_eth=opt_qty, fee_rate=fee_rate
)
of_px, of_fee, of_notional = of.fill_px, of.fee, of.notional
of_slip = 0.0
else:
of_px = float(pos.get("option_entry_px") or 0) or 0.0
of_fee = 0.0
of_notional = of_px * opt_qty
of_slip = 0.0
logger.warning(
"option already flat on exchange; skip resell: %s", e
)
elif is_expiry and intrinsic is not None:
# 到期后交易所可能已不能交易：用本地结算，仍进入 pending 再平永续
of = option_expiry_settle(
intrinsic=float(intrinsic), qty_eth=opt_qty, fee_rate=fee_rate
@@ -643,42 +817,69 @@ class OkxLiveExecutor(Matcher):
return CloseResult(ok=False, detail=f"实盘平期权失败: {e}")
# 期权已平（或到期本地结算）：立刻落 pending，避免永续失败后重试再卖期权
self._mark_option_closed_perp_pending(
group_id=group_id,
option_inst_id=option_inst_id,
opt_qty=opt_qty,
opt_contracts=opt_contracts,
of_px=of_px,
of_fee=of_fee,
of_notional=of_notional,
of_slip=of_slip,
reason=reason,
st_now = str(self.current_position().get("status") or "")
prev_close = self.db.fetchone(
"""SELECT id FROM fills
WHERE group_id=? AND leg='option' AND action='close'
ORDER BY id DESC LIMIT 1""",
(group_id,),
)
if st_now == "option_closed_perp_pending" or prev_close is not None:
# 已入账过平期权：只保证 pending，禁止二次现金
if st_now != "option_closed_perp_pending":
with self.db._lock:
self.db._conn.execute(
"UPDATE positions SET status='option_closed_perp_pending' WHERE id=1"
)
self.db._conn.commit()
else:
self._mark_option_closed_perp_pending(
group_id=group_id,
option_inst_id=option_inst_id,
opt_qty=opt_qty,
opt_contracts=opt_contracts,
of_px=of_px,
of_fee=of_fee,
of_notional=of_notional,
of_slip=of_slip,
reason=reason,
)
pending_perp_only = True
try:
ct_val = client.get_ct_val(perp_inst, inst_type="SWAP")
# pending 路径：交易所已空则禁止用 DB 数量再下单
perp_sz = perp_close_contracts_okx(
client,
perp_inst=perp_inst,
perp_side=perp_side,
perp_qty_eth=perp_qty,
ct_val=ct_val,
allow_db_fallback=not pending_perp_only,
)
if perp_side == "long":
side, pos_side = "sell", "long"
if perp_sz <= 0:
# 永续已在交易所平掉：用入场价近似 finalize（净盈亏由对账校正）
pf_px = float(pos.get("perp_entry_px") or 0) or 0.0
pf_fee = 0.0
logger.warning(
"perp already flat on exchange; finalize without order group=%s",
group_id,
)
else:
side, pos_side = "buy", "short"
perp_live = client.place_market(
inst_id=perp_inst,
side=side,
sz=str(perp_sz),
td_mode=self._perp_margin_mode_for_group(group_id),
pos_side=pos_side,
reduce_only=True,
)
pf_px = float(perp_live.avg_px)
pf_fee = float(perp_live.fee)
if perp_side == "long":
side, pos_side = "sell", "long"
else:
side, pos_side = "buy", "short"
perp_live = client.place_market(
inst_id=perp_inst,
side=side,
sz=str(perp_sz),
td_mode=self._perp_margin_mode_for_group(group_id),
pos_side=pos_side,
reduce_only=True,
)
pf_px = float(perp_live.avg_px)
pf_fee = float(perp_live.fee)
except Exception as e:
return CloseResult(
ok=False,
backend/app/live/okx_trade.py
+29
View File
@@ -265,6 +265,35 @@ class OkxTradeClient:
return abs(float(pos))
return 0.0
def get_option_pos_sz(self, inst_id: str) -> float | None:
"""期权绝对持仓张数。"""
try:
rows = self._request(
"GET",
f"/api/v5/account/positions?instType=OPTION&instId={inst_id}",
)
except Exception as e:
logger.warning("okx get_option_pos_sz failed: %s", e)
return None
total = 0.0
for row in rows:
pos = safe_float(row.get("pos")) or 0.0
total += abs(float(pos))
return total
def any_option_pos_abs(self) -> float | None:
"""账户任意期权绝对持仓张数合计。"""
try:
rows = self._request("GET", "/api/v5/account/positions?instType=OPTION")
except Exception as e:
logger.warning("okx any_option_pos_abs failed: %s", e)
return None
total = 0.0
for row in rows:
pos = safe_float(row.get("pos")) or 0.0
total += abs(float(pos))
return total
def set_leverage(
self,
inst_id: str,
backend/app/live/reconcile.py
+228 -9
View File
@@ -1,4 +1,4 @@
"""LIVE 开仓对账：占槽、交易所持仓核对、平永续数量解析。"""
"""LIVE 开仓对账：占槽、交易所持仓核对、平永续数量解析、stuck opening 恢复。"""
from __future__ import annotations
@@ -7,12 +7,13 @@ from typing import Any
from ..config import get_settings
from ..exchange.runtime import load_runtime_settings
from ..sim.matcher import BLOCKING_STATUSES
from ..sim.matcher import BLOCKING_STATUSES, CloseResult
from .symbols import resolve_perp_inst_id
logger = logging.getLogger(__name__)
_PERP_EPS = 1e-8
_OPT_EPS = 1e-8
def claim_open_slot(db) -> tuple[bool, str]:
@@ -33,11 +34,59 @@ def claim_open_slot(db) -> tuple[bool, str]:
return True, "ok"
def stamp_opening_intent(
db,
*,
group_id: str,
option_inst_id: str,
option_side: str,
perp_side: str,
option_qty_eth: float,
option_qty_contracts: float,
entry_index_px: float | None = None,
option_entry_px: float | None = None,
) -> None:
"""开仓意图落库：崩溃后仍可按 option_inst_id 恢复，禁止「opening 无元数据」。"""
with db._lock:
self_row = db._conn.execute(
"SELECT status FROM positions WHERE id=1"
).fetchone()
st = str(self_row["status"] or "") if self_row else ""
if st != "opening":
return
db._conn.execute(
"""UPDATE positions SET
group_id=?, option_inst_id=?, option_side=?, perp_side=?,
option_qty_eth=?, option_qty_contracts=?,
entry_index_px=COALESCE(?, entry_index_px),
option_entry_px=COALESCE(?, option_entry_px),
status='opening'
WHERE id=1 AND status='opening'""",
(
group_id,
option_inst_id,
option_side,
perp_side,
float(option_qty_eth),
float(option_qty_contracts),
entry_index_px,
option_entry_px,
),
)
db._conn.commit()
def release_open_slot_if_opening(db) -> None:
"""开仓失败且未落 half_open/open 时，释放 opening 占槽。"""
with db._lock:
db._conn.execute(
"UPDATE positions SET status='flat' WHERE id=1 AND status='opening'"
"""UPDATE positions SET
status='flat', group_id=NULL, option_inst_id=NULL,
option_side=NULL, perp_side=NULL,
option_qty_eth=0, option_qty_contracts=0,
option_entry_px=NULL, perp_qty_eth=0, perp_entry_px=NULL,
entry_index_px=NULL, initial_premium=0
WHERE id=1 AND status='opening'"""
)
db._conn.commit()
@@ -51,7 +100,7 @@ def exchange_perp_abs_size(
"""查询交易所永续绝对持仓：OKX 张数，Binance ETH。"""
ex = (exchange or "").strip().lower()
try:
if ex == "binance":
if ex in ("binance", "bn"):
ps = "LONG" if perp_side == "long" else "SHORT"
return client.get_perp_pos_sz(perp_inst_id, position_side=ps)
ps = "long" if perp_side == "long" else "short"
@@ -61,8 +110,26 @@ def exchange_perp_abs_size(
return None
def exchange_option_abs_size(
client: Any, option_inst_id: str
) -> float | None:
try:
return client.get_option_pos_sz(option_inst_id)
except Exception as e:
logger.warning("exchange_option_abs_size failed: %s", e)
return None
def exchange_any_option_abs(client: Any) -> float | None:
try:
return client.any_option_pos_abs()
except Exception as e:
logger.warning("exchange_any_option_abs failed: %s", e)
return None
def assert_safe_to_open_live(executor) -> tuple[bool, str]:
"""LIVE 开仓前：本地无 blocking 仓，且交易所无残留永续（flat/opening 时）。"""
"""LIVE 开仓前：本地无 blocking 仓，且交易所无残留永续/期权（flat/opening 时）。"""
if get_settings().is_sim:
return True, "ok"
@@ -76,7 +143,6 @@ def assert_safe_to_open_live(executor) -> tuple[bool, str]:
return False, "无法核对交易所持仓"
perp_inst = resolve_perp_inst_id(executor.db)
# flat/opening：两侧都查，避免只查默认 long 漏掉 short 残留
if st in ("flat", "", "opening"):
sides = ("long", "short")
else:
@@ -93,6 +159,27 @@ def assert_safe_to_open_live(executor) -> tuple[bool, str]:
False,
"交易所有永续仓但本地无持仓，禁止新开，请人工核对",
)
# 期权：有具体合约则查该合约；flat 时查账户任意期权残留
opt_inst = str(pos.get("option_inst_id") or "")
if opt_inst:
opt_sz = exchange_option_abs_size(client, opt_inst)
if opt_sz is None:
return False, "无法核对交易所期权持仓"
if opt_sz > _OPT_EPS and st in ("flat", "", "opening"):
return (
False,
f"交易所有期权仓({opt_inst})但本地未确认持仓，禁止新开，请人工核对",
)
elif st in ("flat", ""):
any_opt = exchange_any_option_abs(client)
if any_opt is None:
return False, "无法核对交易所期权持仓"
if any_opt > _OPT_EPS:
return (
False,
"交易所有期权残留仓但本地无持仓，禁止新开，请人工核对",
)
return True, "ok"
@@ -102,9 +189,16 @@ def log_exchange_db_mismatch(executor) -> None:
return
ok, msg = assert_safe_to_open_live(executor)
if ok:
logger.info("LIVE startup reconcile: exchange/DB perp OK")
logger.info("LIVE startup reconcile: exchange/DB OK")
else:
logger.warning("LIVE startup reconcile mismatch: %s", msg)
# 启动时尝试恢复 stuck opening
try:
r = recover_stuck_opening(executor)
if r is not None:
logger.info("LIVE startup recover_opening: ok=%s detail=%s", r.ok, r.detail)
except Exception:
logger.exception("LIVE startup recover_opening failed")
def perp_close_contracts_okx(
@@ -114,12 +208,21 @@ def perp_close_contracts_okx(
perp_side: str,
perp_qty_eth: float,
ct_val: float,
allow_db_fallback: bool = True,
) -> int:
"""平永续张数：优先交易所持仓，否则 DB qty/ct_val。"""
"""平永续张数：优先交易所持仓。
allow_db_fallback=False 且交易所已空仓时返回 0（勿用 DB 数量再下单，防反向开仓）。
"""
ps = "long" if perp_side == "long" else "short"
ex_sz = client.get_perp_pos_sz(perp_inst, pos_side=ps)
if ex_sz is not None and ex_sz > _PERP_EPS:
return max(1, int(round(ex_sz)))
if ex_sz is not None and ex_sz <= _PERP_EPS:
if not allow_db_fallback:
return 0
if not allow_db_fallback:
return 0
return max(1, int(round(perp_qty_eth / ct_val)))
@@ -129,15 +232,131 @@ def perp_close_qty_eth_binance(
perp_inst: str,
perp_side: str,
perp_qty_eth: float,
allow_db_fallback: bool = True,
) -> float:
"""平永续 ETH 数量：优先交易所持仓，否则 DB qty。"""
"""平永续 ETH 数量：优先交易所持仓。交易所空且不允许 fallback → 0。"""
ps = "LONG" if perp_side == "long" else "SHORT"
ex_sz = client.get_perp_pos_sz(perp_inst, position_side=ps)
if ex_sz is not None and ex_sz > _PERP_EPS:
return float(ex_sz)
if ex_sz is not None and ex_sz <= _PERP_EPS:
if not allow_db_fallback:
return 0.0
if not allow_db_fallback:
return 0.0
return float(perp_qty_eth)
def recover_stuck_opening(executor) -> CloseResult | None:
"""恢复本地 status=opening：
- 交易所期权+永续皆空 → 清槽
- 仅期权 → half_open 并尝试 repair
- 期权+永续 → 提升为 open（用本地已 stamp 的数量/均价）
- 无元数据且交易所仍有仓 → 保持 opening，返回失败详情
"""
if get_settings().is_sim:
return None
pos = executor.current_position()
st = str(pos.get("status") or "")
if st != "opening":
return None
client, ex_name = _executor_client_and_exchange(executor)
if client is None:
return CloseResult(ok=False, detail="recover_opening: 无交易客户端")
option_inst = str(pos.get("option_inst_id") or "")
perp_side = str(pos.get("perp_side") or "long")
group_id = str(pos.get("group_id") or "")
perp_inst = resolve_perp_inst_id(executor.db, group_id=group_id or None)
perp_total = 0.0
for side in ("long", "short"):
sz = exchange_perp_abs_size(client, ex_name or "", perp_inst, side)
if sz is None:
return CloseResult(ok=False, detail="recover_opening: 无法查永续")
perp_total += float(sz)
opt_sz = 0.0
if option_inst:
raw = exchange_option_abs_size(client, option_inst)
if raw is None:
return CloseResult(ok=False, detail="recover_opening: 无法查期权")
opt_sz = float(raw)
else:
any_opt = exchange_any_option_abs(client)
if any_opt is None:
return CloseResult(ok=False, detail="recover_opening: 无法查期权")
if any_opt > _OPT_EPS:
return CloseResult(
ok=False,
detail=(
"recover_opening: opening 无 option_inst_id 但交易所有期权仓，"
"禁止自动清槽，请人工核对"
),
)
opt_sz = 0.0
# 两边皆空 → 清槽
if opt_sz <= _OPT_EPS and perp_total <= _PERP_EPS:
release_open_slot_if_opening(executor.db)
return CloseResult(ok=True, detail="recover_opening: 交易所空仓，已释放 opening")
# 无元数据但有仓 → 不自动处理
if not option_inst:
return CloseResult(
ok=False,
detail="recover_opening: 交易所有仓但本地 opening 缺 option_inst_id",
)
# 仅期权 → half_open + repair
if opt_sz > _OPT_EPS and perp_total <= _PERP_EPS:
persist = getattr(executor, "_persist_half_open", None)
if not callable(persist):
return CloseResult(ok=False, detail="recover_opening: 无 half_open 落库")
of_px = float(pos.get("option_entry_px") or 0) or 0.0
opt_qty = float(pos.get("option_qty_eth") or 0)
opt_contracts = float(pos.get("option_qty_contracts") or 0)
if opt_contracts <= 0 and hasattr(executor, "_ct_mult"):
from ..sim.liquidity import contracts_for_eth
ct = executor._ct_mult(option_inst)
opt_contracts = float(contracts_for_eth(opt_qty or opt_sz, ct)) if opt_qty else float(opt_sz)
if opt_qty <= 0 and hasattr(executor, "_ct_mult"):
from ..sim.liquidity import eth_from_contracts
opt_qty = eth_from_contracts(opt_contracts or opt_sz, executor._ct_mult(option_inst))
persist(
group_id=group_id or f"RCV-{option_inst[-12:]}",
bias="recover",
option_side=str(pos.get("option_side") or "call"),
perp_side=perp_side,
option_inst_id=option_inst,
entry_index_px=float(pos.get("entry_index_px") or 0) or 0.0,
strike=None,
expiry_ymd=None,
opt_qty=opt_qty,
opt_contracts=opt_contracts or opt_sz,
of_px=of_px,
of_fee=0.0,
detail="recover_opening option-only → half_open",
)
repair = getattr(executor, "repair_half_open", None)
if callable(repair):
return repair()
return CloseResult(ok=True, detail="recover_opening: 已落 half_open")
# 期权+永续 → 提升为 open
promote = getattr(executor, "_promote_opening_to_open", None)
if callable(promote):
return promote(pos=pos, perp_inst=perp_inst, opt_sz=opt_sz, perp_total=perp_total)
return CloseResult(
ok=False,
detail="recover_opening: 双边有仓但执行器无 promote，请人工核对",
)
def _executor_client_and_exchange(executor) -> tuple[Any | None, str | None]:
if get_settings().is_sim:
return None, None
backend/app/strategy/engine.py
+71 -17
View File
@@ -242,23 +242,32 @@ class StrategyEngine:
if r.ok:
self.enter_rest_after_close()
elif st == "opening":
# 开仓未确认：不自动清槽（可能期权已成交）；紧急全平仅告警，防重复开
ok = False
detail = (
"stuck opening：请核对交易所期权/永续后人工处理"
"（已占槽防重复开，紧急全平不会自动释放）"
)
close_data = {"status": "opening"}
try:
from ..notify import wecom
wecom.notify_fault(
title="紧急全平遇到 stuck opening",
detail=detail,
dedupe_key="emergency:opening",
recover = getattr(self.matcher, "recover_opening", None)
if callable(recover):
r = recover()
ok = r.ok
detail = r.detail
close_data = r.data
st2 = str(self.matcher.current_position().get("status") or "")
if r.ok and st2 in ("flat",):
self.enter_rest_after_close()
else:
ok = False
detail = (
"stuck opening：请核对交易所期权/永续后人工处理"
"（已占槽防重复开）"
)
except Exception:
pass
close_data = {"status": "opening"}
try:
from ..notify import wecom
wecom.notify_fault(
title="紧急全平遇到 stuck opening",
detail=detail,
dedupe_key="emergency:opening",
)
except Exception:
pass
elif st in ("open", "option_closed_perp_pending"):
# A：双腿（或续平永续）
r = self.matcher.close_group(reason="emergency", bypass_liquidity=True)
@@ -534,6 +543,46 @@ class StrategyEngine:
pos = self.matcher.current_position()
st_pos = str(pos.get("status") or "flat")
if st_pos == "opening":
allowed, left = self._retry_allowed("opening")
if not allowed:
self._set_state(
phase="opening_stuck",
last_error=f"opening 恢复退避中，{left:.0f}s 后再试",
)
return
recover = getattr(self.matcher, "recover_opening", None)
if callable(recover):
r = await asyncio.to_thread(recover)
self._note_retry_result("opening", ok=r.ok, detail=r.detail)
if r.ok:
self._set_state(phase="idle", last_error=None)
if "half_open" in (r.detail or "") or "提升为 open" in (r.detail or ""):
pass # 继续本 tick 后续逻辑由下一轮处理
try:
from ..notify import wecom
wecom.notify_fault(
title="opening 已自动恢复",
detail=r.detail,
dedupe_key=f"opening_ok:{r.detail[:60]}",
)
except Exception:
pass
else:
self._set_state(phase="opening_stuck", last_error=r.detail)
try:
from ..notify import wecom
wecom.notify_fault(
title="opening 恢复失败",
detail=r.detail,
dedupe_key=f"opening_fail:{r.detail[:60]}",
)
except Exception:
pass
return
if st_pos == "half_open":
allowed, left = self._retry_allowed("half_open")
if not allowed:
@@ -733,7 +782,12 @@ class StrategyEngine:
except Exception:
logger.exception("wecom notify_open failed")
else:
self._set_state(phase="idle", last_error=r.detail)
# 保留 opening 时勿伪装 idle
pos_after = self.matcher.current_position()
if str(pos_after.get("status") or "") == "opening":
self._set_state(phase="opening_stuck", last_error=r.detail)
else:
self._set_state(phase="idle", last_error=r.detail)
try:
from ..notify import wecom