修复币本位平仓/到期后未自动卖回,并消除模拟盘卖币时数据库锁定。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-07 10:47:25 +08:00
parent 474e0742f7
commit 265317672c
6 changed files with 156 additions and 4 deletions
+5 -1
View File
@@ -1607,8 +1607,12 @@ def _purge_key_monitors_if_full_margin():
def get_db(): def get_db():
conn = sqlite3.connect(DB_PATH) conn = sqlite3.connect(DB_PATH, timeout=30)
conn.row_factory = sqlite3.Row conn.row_factory = sqlite3.Row
try:
conn.execute("PRAGMA busy_timeout=30000")
except Exception:
pass
return conn return conn
+62
View File
@@ -475,3 +475,65 @@ def maybe_sell_spot_after_close(
conn.close() conn.close()
except Exception: except Exception:
pass pass
def sweep_spot_after_options_flat(
cfg: dict[str, Any],
ex: Any | None = None,
) -> dict[str, Any]:
"""币本位:某标的已无期权持仓时,把交易户该标的残留币卖回 USDT.
覆盖到期结算、限价平仓后同步关仓等未走买一平仓卖回路径的情况。
"""
if not is_coin_margin_mode():
return {"ok": True, "skipped": True, "msg": "非币本位"}
exchange = ex if ex is not None else cfg.get("exchange_options")
if exchange is None:
return {"ok": False, "msg": "exchange 未就绪"}
live_uly: set[str] = set()
try:
rows = cfg["fetch_option_positions"](exchange) or []
for p in rows:
try:
if abs(float(p.get("pos") or 0)) <= 1e-12:
continue
except (TypeError, ValueError):
continue
inst = str(p.get("instId") or p.get("inst_id") or "").strip()
if not inst:
continue
live_uly.add(inst.split("-")[0].upper())
except Exception as e:
return {"ok": False, "msg": f"持仓查询失败: {e}"}
from lib.options.options_db import init_options_tables
sold: list[dict[str, Any]] = []
for uly in ("BTC", "ETH"):
if uly in live_uly:
continue
try:
avail = _spot_bridge.fetch_trading_coin_available(exchange, uly)
if avail is None or float(avail) <= 1e-8:
continue
except Exception:
continue
conn = cfg["get_db"]()
try:
init_options_tables(conn)
out = _spot_bridge.sell_residual_after_option_flat(
conn, exchange, underlying=uly, inst_id=None
)
sold.append({"underlying": uly, **(out or {})})
try:
conn.commit()
except Exception:
pass
except Exception as e:
sold.append({"underlying": uly, "ok": False, "msg": str(e)})
finally:
try:
conn.close()
except Exception:
pass
return {"ok": True, "results": sold, "skipped": not sold}
+34
View File
@@ -285,6 +285,7 @@ def sync_open_options_trades(
""" """
).fetchall() ).fetchall()
updated = 0 updated = 0
closed_inst_ids: list[str] = []
now_ms = int(time.time() * 1000) now_ms = int(time.time() * 1000)
for row in rows: for row in rows:
inst_id = str(row["inst_id"] or "") inst_id = str(row["inst_id"] or "")
@@ -353,6 +354,7 @@ def sync_open_options_trades(
), ),
) )
updated += 1 updated += 1
closed_inst_ids.append(inst_id)
if notify_cfg is not None: if notify_cfg is not None:
try: try:
from lib.options.options_notify_lib import notify_options_close from lib.options.options_notify_lib import notify_options_close
@@ -373,9 +375,41 @@ def sync_open_options_trades(
pass pass
if notify_cfg is not None: if notify_cfg is not None:
_notify_recent_unsent_expiry_closes(conn, notify_cfg) _notify_recent_unsent_expiry_closes(conn, notify_cfg)
_sell_spot_after_synced_closes(notify_cfg, closed_inst_ids)
return updated return updated
def _sell_spot_after_synced_closes(
cfg: dict[str, Any],
closed_inst_ids: list[str],
) -> None:
"""同步关仓(含到期)后卖回币本位残留;空仓时再扫一遍交易户."""
if not isinstance(cfg, dict):
return
ex = cfg.get("exchange_options")
if ex is None:
return
try:
from lib.options.options_coin_open_lib import (
maybe_sell_spot_after_close,
sweep_spot_after_options_flat,
)
seen: set[str] = set()
for inst_id in closed_inst_ids or []:
iid = str(inst_id or "").strip()
if not iid or iid in seen:
continue
seen.add(iid)
try:
maybe_sell_spot_after_close(cfg, ex, inst_id=iid)
except Exception:
pass
sweep_spot_after_options_flat(cfg, ex)
except Exception:
pass
def _notify_recent_unsent_expiry_closes( def _notify_recent_unsent_expiry_closes(
conn: sqlite3.Connection, conn: sqlite3.Connection,
notify_cfg: dict[str, Any], notify_cfg: dict[str, Any],
+26 -3
View File
@@ -458,6 +458,13 @@ def _sync_options_trades(
conn.commit() conn.commit()
finally: finally:
conn.close() conn.close()
try:
from lib.options.options_coin_open_lib import sweep_spot_after_options_flat
sweep_spot_after_options_flat(cfg, ex)
_mark_balances_stale(cfg)
except Exception:
pass
def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None: def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
@@ -1694,12 +1701,20 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
data = request.get_json(silent=True) or {} data = request.get_json(silent=True) or {}
underlying = (data.get("underlying") or cfg.get("default_underly") or "ETH").strip().upper() underlying = (data.get("underlying") or cfg.get("default_underly") or "ETH").strip().upper()
inst_id = (data.get("inst_id") or "").strip() or None inst_id = (data.get("inst_id") or "").strip() or None
from lib.options.options_spot_bridge_lib import sell_residual_after_option_flat
conn = cfg["get_db"]() conn = cfg["get_db"]()
try: try:
init_options_tables(conn) init_options_tables(conn)
from lib.options.options_spot_bridge_lib import sell_residual_after_option_flat try:
conn.commit()
except Exception:
pass
out = sell_residual_after_option_flat(conn, ex, underlying=underlying, inst_id=inst_id) out = sell_residual_after_option_flat(conn, ex, underlying=underlying, inst_id=inst_id)
try:
conn.commit()
except Exception:
pass
if out.get("ok"): if out.get("ok"):
_mark_balances_stale(cfg) _mark_balances_stale(cfg)
return jsonify(out) return jsonify(out)
@@ -1976,12 +1991,20 @@ def _start_monitor_thread(app: Flask, cfg: dict[str, Any]) -> None:
return 0 return 0
live_ids = {str(p.get("instId") or "") for p in raw if str(p.get("instId") or "")} live_ids = {str(p.get("instId") or "") for p in raw if str(p.get("instId") or "")}
reconcile_live_open_trades(conn, live_inst_ids=live_ids) reconcile_live_open_trades(conn, live_inst_ids=live_ids)
return sync_open_options_trades( n = sync_open_options_trades(
conn, conn,
live_inst_ids=live_ids, live_inst_ids=live_ids,
fetch_history_fn=lambda inst_id: fetch_option_position_history(ex, inst_id), fetch_history_fn=lambda inst_id: fetch_option_position_history(ex, inst_id),
notify_cfg=cfg, notify_cfg=cfg,
) )
try:
from lib.options.options_coin_open_lib import sweep_spot_after_options_flat
sweep_spot_after_options_flat(cfg, ex)
_mark_balances_stale(cfg)
except Exception:
pass
return n
def _target_close(inst_id: str) -> dict[str, Any]: def _target_close(inst_id: str) -> dict[str, Any]:
from lib.options.options_target_lib import close_option_by_bid_depth from lib.options.options_target_lib import close_option_by_bid_depth
+10
View File
@@ -276,6 +276,11 @@ def rollback_bought_coin_to_usdt(
) -> dict[str, Any]: ) -> dict[str, Any]:
"""买币后开期权失败:卖回 USDT 并关闭桥.卖光交易户可用标的币.""" """买币后开期权失败:卖回 USDT 并关闭桥.卖光交易户可用标的币."""
_ = coin_amount # 兼容旧签名;定量以交易户可用为准 _ = coin_amount # 兼容旧签名;定量以交易户可用为准
# 卖币(模拟盘)会另开连接写钱包,须先释放本连接事务锁
try:
conn.commit()
except Exception:
pass
sell = spot_market_sell_coin_to_usdt(ex, underlying=underlying) sell = spot_market_sell_coin_to_usdt(ex, underlying=underlying)
if not sell.get("ok"): if not sell.get("ok"):
update_bridge( update_bridge(
@@ -313,6 +318,11 @@ def sell_residual_after_option_flat(
if not underlying or str(b.get("underlying") or "").upper() == underlying.upper(): if not underlying or str(b.get("underlying") or "").upper() == underlying.upper():
target = b target = b
break break
# 卖币(模拟盘)会另开连接写钱包,须先释放本连接事务锁,否则 database is locked
try:
conn.commit()
except Exception:
pass
sell = spot_market_sell_coin_to_usdt(ex, underlying=underlying) sell = spot_market_sell_coin_to_usdt(ex, underlying=underlying)
if target is None: if target is None:
if not sell.get("ok"): if not sell.get("ok"):
+19
View File
@@ -829,6 +829,25 @@ class SimBroker:
forget_close_gate_for_inst(inst_id) forget_close_gate_for_inst(inst_id)
except Exception: except Exception:
pass pass
try:
from lib.options.options_margin_mode_lib import margin_mode_from_inst_id
from lib.options.options_spot_bridge_lib import sell_residual_after_option_flat
if margin_mode_from_inst_id(inst_id) == "coin":
uly = inst_id.split("-")[0].upper() if "-" in inst_id else "ETH"
conn_s = self.get_db()
try:
sell_residual_after_option_flat(
conn_s,
exchange,
underlying=uly,
inst_id=inst_id,
)
conn_s.commit()
finally:
conn_s.close()
except Exception:
pass
settled.append( settled.append(
{ {
"inst_id": inst_id, "inst_id": inst_id,