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

修复模拟盘卖币握锁导致 database is locked 与中控超时离线。

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-07 11:04:06 +08:00
parent 066c90fd96
commit 80ef5ceccb
7 changed files with 173 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/options/options_coin_open_lib.py
+10 -25
View File
@@ -464,17 +464,12 @@ def maybe_sell_spot_after_close(
pass
parts = inst_id.split("-")
underlying = (parts[0] if parts else "ETH").upper()
conn = cfg["get_db"]()
try:
from lib.options.options_db import init_options_tables
init_options_tables(conn)
return _spot_bridge.sell_residual_after_option_flat(conn, ex, underlying=underlying, inst_id=inst_id)
finally:
try:
conn.close()
except Exception:
pass
return _spot_bridge.sell_residual_after_option_flat(
cfg["get_db"],
ex,
underlying=underlying,
inst_id=inst_id,
)
def sweep_spot_after_options_flat(
@@ -506,8 +501,6 @@ def sweep_spot_after_options_flat(
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:
@@ -518,22 +511,14 @@ def sweep_spot_after_options_flat(
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
cfg["get_db"],
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}
lib/options/options_monitor_lib.py
+6 -7
View File
@@ -573,16 +573,15 @@ def options_monitor_loop(
cfg=pe_cfg,
ex=pe_cfg.get("exchange_options"),
)
if sync_trades_fn is not None:
try:
conn.commit()
except Exception:
pass
sync_trades_fn(conn)
conn.commit()
finally:
conn.close()
# 币本位扫尾卖回:必须在监控连接关闭之后
# 同步/卖币必须在监控连接关闭之后
if sync_trades_fn is not None:
try:
sync_trades_fn(None)
except Exception:
pass
try:
pe_sell = dict(profit_exit_cfg or {})
pe_sell.setdefault("send_wechat", send_wechat)
lib/options/options_notify_lib.py
+6
View File
@@ -336,6 +336,12 @@ def notify_options_close(
trade_id=head.get("id") if len(rows) == 1 else None,
premium_ccy=ccy,
)
# 推送企业微信前先提交,避免握锁等待 webhook(可达数秒)导致卖币/快照 database is locked
if conn is not None:
try:
conn.commit()
except Exception:
pass
ok = notify_options_send(cfg, msg)
if ok and conn is not None:
for r in pending or rows:
lib/options/options_register.py
+58 -41
View File
@@ -437,8 +437,36 @@ def _sync_options_trades(
raw = raw_positions
live_ids = {str(p.get("instId") or "") for p in raw if str(p.get("instId") or "")}
def _hist(inst_id: str):
return fetch_option_position_history(ex, inst_id)
# 网络拉取历史必须在 DB 事务外,否则握锁期间卖币/中控快照会 database is locked
pending_insts: list[str] = []
conn = cfg["get_db"]()
try:
init_options_tables(conn)
rows = conn.execute(
"SELECT inst_id FROM options_trades WHERE status = 'open'"
).fetchall()
for r in rows:
iid = str(r["inst_id"] or "").strip()
if iid and iid not in live_ids:
pending_insts.append(iid)
try:
conn.commit()
except Exception:
pass
finally:
conn.close()
hist_by_inst: dict[str, list] = {}
for iid in pending_insts:
try:
hist_by_inst[iid] = fetch_option_position_history(ex, iid) or []
except Exception:
hist_by_inst[iid] = []
hist_all = None
try:
hist_all = fetch_all_option_positions_history(ex, limit=200)
except Exception:
hist_all = None
conn = cfg["get_db"]()
try:
@@ -447,17 +475,25 @@ def _sync_options_trades(
sync_open_options_trades(
conn,
live_inst_ids=live_ids,
fetch_history_fn=_hist,
fetch_history_fn=lambda inst_id: hist_by_inst.get(str(inst_id) or [], []),
notify_cfg=cfg,
)
if hist_all is not None:
try:
backfill_closed_options_realized_pnl_from_history(conn, hist_all)
except Exception:
pass
conn.commit()
except Exception:
try:
hist_all = fetch_all_option_positions_history(ex, limit=200)
backfill_closed_options_realized_pnl_from_history(conn, hist_all)
conn.rollback()
except Exception:
pass
conn.commit()
finally:
conn.close()
try:
conn.close()
except Exception:
pass
try:
from lib.options.options_monitor_lib import sell_spot_after_options_sync
@@ -1703,23 +1739,15 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> 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"]()
try:
init_options_tables(conn)
try:
conn.commit()
except Exception:
pass
out = sell_residual_after_option_flat(conn, ex, underlying=underlying, inst_id=inst_id)
try:
conn.commit()
except Exception:
pass
if out.get("ok"):
_mark_balances_stale(cfg)
return jsonify(out)
finally:
conn.close()
out = sell_residual_after_option_flat(
cfg["get_db"],
ex,
underlying=underlying,
inst_id=inst_id,
)
if out.get("ok"):
_mark_balances_stale(cfg)
return jsonify(out)
@app.route("/api/options/convert/quote", methods=["POST"])
@lr
@@ -1979,24 +2007,13 @@ def _start_monitor_thread(app: Flask, cfg: dict[str, Any]) -> None:
return []
return [cfg["format_position_row"](p) for p in raw]
def _sync(conn):
from lib.exchange.okx_options_lib import fetch_option_position_history
from lib.options.options_monitor_lib import reconcile_live_open_trades, sync_open_options_trades
ex = cfg.get("exchange_options")
if ex is None:
def _sync(_conn=None):
# 不用监控线程传入的连接;force=False 走 15s 节流,卖币由监控循环在关闭连接后另扫
try:
_sync_options_trades(cfg, force=False)
return 1
except Exception:
return 0
raw = cfg["fetch_option_positions"](ex)
if raw is None:
return 0
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)
return sync_open_options_trades(
conn,
live_inst_ids=live_ids,
fetch_history_fn=lambda inst_id: fetch_option_position_history(ex, inst_id),
notify_cfg=cfg,
)
def _target_close(inst_id: str) -> dict[str, Any]:
from lib.options.options_target_lib import close_option_by_bid_depth
lib/options/options_spot_bridge_lib.py
+77 -38
View File
@@ -303,60 +303,99 @@ def rollback_bought_coin_to_usdt(
def sell_residual_after_option_flat(
conn: sqlite3.Connection,
conn_or_get_db: Any,
ex: Any,
*,
underlying: str,
inst_id: str | None = None,
) -> dict[str, Any]:
"""期权已平:卖掉本桥残留标的币;优先关闭 matching holding/pending 桥."""
ensure_bridge_table(conn)
bridges = list_open_bridges(conn)
target = None
for b in bridges:
if str(b.get("status")) in (BRIDGE_HOLDING, BRIDGE_PENDING_SELL, BRIDGE_BOUGHT):
if not underlying or str(b.get("underlying") or "").upper() == underlying.upper():
target = b
break
# 卖币(模拟盘)会另开连接写钱包,须先释放本连接事务锁,否则 database is locked
try:
conn.commit()
except Exception:
pass
"""期权已平:卖掉本桥残留标的币;优先关闭 matching holding/pending 桥.
可传 get_db 可调用对象(推荐)或已打开的 Connection。
卖币(尤其模拟盘写钱包)期间绝不握着业务连接,否则会 database is locked。
"""
get_db = conn_or_get_db if callable(conn_or_get_db) else None
bid: int | None = None
if get_db is not None:
conn = get_db()
try:
ensure_bridge_table(conn)
bridges = list_open_bridges(conn)
for b in bridges:
if str(b.get("status")) in (BRIDGE_HOLDING, BRIDGE_PENDING_SELL, BRIDGE_BOUGHT):
if not underlying or str(b.get("underlying") or "").upper() == underlying.upper():
bid = int(b["id"])
break
try:
conn.commit()
except Exception:
pass
finally:
try:
conn.close()
except Exception:
pass
else:
conn = conn_or_get_db
ensure_bridge_table(conn)
bridges = list_open_bridges(conn)
for b in bridges:
if str(b.get("status")) in (BRIDGE_HOLDING, BRIDGE_PENDING_SELL, BRIDGE_BOUGHT):
if not underlying or str(b.get("underlying") or "").upper() == underlying.upper():
bid = int(b["id"])
break
try:
conn.commit()
except Exception:
pass
# 此处无业务连接:模拟盘 convert 可安全另开写钱包
sell = spot_market_sell_coin_to_usdt(ex, underlying=underlying)
if target is None:
if bid is None:
if not sell.get("ok"):
msg = str(sell.get("msg") or "")
if "无可用" in msg or "过小" in msg:
return {"ok": True, "msg": "无残留币需卖回", "skipped": True}
return {"ok": False, "msg": msg, "bridge_status": BRIDGE_PENDING_SELL}
return {"ok": True, "sell": sell, "bridge_status": None}
bid = int(target["id"])
if not sell.get("ok"):
def _apply_bridge(c: sqlite3.Connection) -> dict[str, Any]:
if not sell.get("ok"):
update_bridge(
c,
bid,
status=BRIDGE_PENDING_SELL,
inst_id=inst_id,
message=str(sell.get("msg") or "卖回 USDT 失败"),
)
return {
"ok": False,
"msg": sell.get("msg") or "卖回 USDT 失败",
"bridge_id": bid,
"bridge_status": BRIDGE_PENDING_SELL,
}
update_bridge(
conn,
c,
bid,
status=BRIDGE_PENDING_SELL,
status=BRIDGE_CLOSED,
sell_ord_id=str(sell.get("ord_id") or ""),
coin_sold=_safe_float(sell.get("coin_sold")),
inst_id=inst_id,
message=str(sell.get("msg") or "卖回 USDT 失败"),
message="期权已平,币已卖回 USDT",
close=True,
)
return {
"ok": False,
"msg": sell.get("msg") or "卖回 USDT 失败",
"bridge_id": bid,
"bridge_status": BRIDGE_PENDING_SELL,
}
update_bridge(
conn,
bid,
status=BRIDGE_CLOSED,
sell_ord_id=str(sell.get("ord_id") or ""),
coin_sold=_safe_float(sell.get("coin_sold")),
inst_id=inst_id,
message="期权已平,币已卖回 USDT",
close=True,
)
return {"ok": True, "sell": sell, "bridge_id": bid, "bridge_status": BRIDGE_CLOSED}
return {"ok": True, "sell": sell, "bridge_id": bid, "bridge_status": BRIDGE_CLOSED}
if get_db is not None:
conn2 = get_db()
try:
return _apply_bridge(conn2)
finally:
try:
conn2.close()
except Exception:
pass
return _apply_bridge(conn_or_get_db)
def bridge_blocks_new_open_msg(conn: sqlite3.Connection) -> str | None:
lib/sim/broker_lib.py
+6 -11
View File
@@ -835,17 +835,12 @@ class SimBroker:
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()
sell_residual_after_option_flat(
self.get_db,
exchange,
underlying=uly,
inst_id=inst_id,
)
except Exception:
pass
settled.append(