From 80ef5ceccb6328948c4d4551c6f14e506896fafa Mon Sep 17 00:00:00 2001 From: dekun Date: Wed, 7 Oct 2026 11:04:06 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=A8=A1=E6=8B=9F=E7=9B=98?= =?UTF-8?q?=E5=8D=96=E5=B8=81=E6=8F=A1=E9=94=81=E5=AF=BC=E8=87=B4=20databa?= =?UTF-8?q?se=20is=20locked=20=E4=B8=8E=E4=B8=AD=E6=8E=A7=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E7=A6=BB=E7=BA=BF=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- app.py | 14 ++- lib/options/options_coin_open_lib.py | 35 +++----- lib/options/options_monitor_lib.py | 13 ++- lib/options/options_notify_lib.py | 6 ++ lib/options/options_register.py | 99 ++++++++++++--------- lib/options/options_spot_bridge_lib.py | 115 +++++++++++++++++-------- lib/sim/broker_lib.py | 17 ++-- 7 files changed, 173 insertions(+), 126 deletions(-) diff --git a/app.py b/app.py index 019a576..eca3f0c 100644 --- a/app.py +++ b/app.py @@ -1611,6 +1611,9 @@ def get_db(): conn.row_factory = sqlite3.Row try: conn.execute("PRAGMA busy_timeout=30000") + # WAL:读(中控快照)与写(卖币/同步)可并行,避免 database is locked→全部超时离线 + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") except Exception: pass return conn @@ -5119,11 +5122,14 @@ def background_task(): while True: try: auto_transfer_once_per_day() - conn = get_db() + # force_close 自带连接,勿套在另一未关闭连接上 force_close_before_reset() - reconcile_external_closes(conn) - conn.commit() - conn.close() + conn = get_db() + try: + reconcile_external_closes(conn) + conn.commit() + finally: + conn.close() check_key_monitors() check_order_monitors() except Exception as e: diff --git a/lib/options/options_coin_open_lib.py b/lib/options/options_coin_open_lib.py index 2c9b66a..19b111a 100644 --- a/lib/options/options_coin_open_lib.py +++ b/lib/options/options_coin_open_lib.py @@ -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} diff --git a/lib/options/options_monitor_lib.py b/lib/options/options_monitor_lib.py index 8b34c93..b30e188 100644 --- a/lib/options/options_monitor_lib.py +++ b/lib/options/options_monitor_lib.py @@ -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) diff --git a/lib/options/options_notify_lib.py b/lib/options/options_notify_lib.py index 9babdea..bc80856 100644 --- a/lib/options/options_notify_lib.py +++ b/lib/options/options_notify_lib.py @@ -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: diff --git a/lib/options/options_register.py b/lib/options/options_register.py index d1d9007..5491f64 100644 --- a/lib/options/options_register.py +++ b/lib/options/options_register.py @@ -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 diff --git a/lib/options/options_spot_bridge_lib.py b/lib/options/options_spot_bridge_lib.py index 4cec591..2bdb606 100644 --- a/lib/options/options_spot_bridge_lib.py +++ b/lib/options/options_spot_bridge_lib.py @@ -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: diff --git a/lib/sim/broker_lib.py b/lib/sim/broker_lib.py index 3c36f71..64a4853 100644 --- a/lib/sim/broker_lib.py +++ b/lib/sim/broker_lib.py @@ -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(