From ea3e029eb9ac66a407fd45dd6f27f3b83df0810d Mon Sep 17 00:00:00 2001 From: dekun Date: Wed, 7 Oct 2026 11:28:29 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E7=B2=89=E5=B0=98=E6=A1=A5?= =?UTF-8?q?=E6=8B=A6=E5=BC=80=E4=BB=93=EF=BC=8C=E5=B9=B6=E7=BB=A7=E7=BB=AD?= =?UTF-8?q?=E6=B6=88=E9=99=A4=E6=A8=A1=E6=8B=9F=E7=9B=98=E5=B5=8C=E5=A5=97?= =?UTF-8?q?=E6=8F=A1=E9=94=81=E5=AF=BC=E8=87=B4=E7=9A=84=20database=20is?= =?UTF-8?q?=20locked=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- lib/options/options_coin_open_lib.py | 21 +++++ lib/options/options_monitor_lib.py | 85 ++++++++++++++------- lib/options/options_notify_lib.py | 6 ++ lib/options/options_profit_exit_lib.py | 3 + lib/options/options_spot_bridge_lib.py | 101 ++++++++++++++++++++++++- lib/options/options_target_lib.py | 3 + lib/sim/broker_lib.py | 10 ++- 7 files changed, 196 insertions(+), 33 deletions(-) diff --git a/lib/options/options_coin_open_lib.py b/lib/options/options_coin_open_lib.py index d540c9a..dea96a4 100644 --- a/lib/options/options_coin_open_lib.py +++ b/lib/options/options_coin_open_lib.py @@ -69,6 +69,12 @@ def open_coin_option_buy_full( parts = inst_id.split("-") underlying = (parts[0] if parts else "ETH").upper() + # 先清粉尘 pending 桥(自管短连接),再开业务连接,避免模拟盘读余额嵌套锁库 + try: + _spot_bridge.clear_dust_pending_sell_bridges(cfg["get_db"], ex) + except Exception: + pass + conn = cfg["get_db"]() try: init_options_tables(conn) @@ -142,6 +148,11 @@ def open_coin_option_buy_full( buy_usdt = float(plan["buy_usdt"]) sheets = int(plan["sheets"]) + # 买币/读余额(模拟盘写钱包)前必须提交释放锁 + try: + conn.commit() + except Exception: + pass # 1) 仅买「权利金×现货缓冲」所需 USDT,不全额兑换预算 coin_before = _spot_bridge.fetch_trading_coin_available(ex, underlying) or 0.0 buy = _spot_bridge.spot_market_buy_coin_with_usdt(ex, underlying=underlying, usdt_amount=buy_usdt) @@ -161,6 +172,10 @@ def open_coin_option_buy_full( inst_id=inst_id, message="已买币,待开期权", ) + try: + conn.commit() + except Exception: + pass # 等余额落账 time.sleep(1.5) try: @@ -501,6 +516,12 @@ def sweep_spot_after_options_flat( except Exception as e: return {"ok": False, "msg": f"持仓查询失败: {e}"} + # 先关掉只剩粉尘的 pending 桥,避免拦开仓又看不到重试按钮 + try: + _spot_bridge.clear_dust_pending_sell_bridges(cfg["get_db"], exchange) + except Exception: + pass + sold: list[dict[str, Any]] = [] for uly in ("BTC", "ETH"): if uly in live_uly: diff --git a/lib/options/options_monitor_lib.py b/lib/options/options_monitor_lib.py index b30e188..ecda50c 100644 --- a/lib/options/options_monitor_lib.py +++ b/lib/options/options_monitor_lib.py @@ -522,14 +522,32 @@ def options_monitor_loop( if stop_event is not None and getattr(stop_event, "is_set", lambda: False)(): break try: - conn = get_db() + # 持仓拉取不握 DB,避免与模拟盘钱包写互相锁 try: - if cond_order_tick_fn is not None: + positions = fetch_positions() + except Exception: + positions = [] + + pe_cfg = dict(profit_exit_cfg or {}) + pe_cfg.setdefault("send_wechat", send_wechat) + pe_cfg.setdefault("account_label", account_label) + + # 条件单可能嵌套开仓/卖币:单独短连接 + if cond_order_tick_fn is not None: + conn = get_db() + try: + cond_order_tick_fn(conn) + conn.commit() + except Exception: + pass + finally: try: - cond_order_tick_fn(conn) + conn.close() except Exception: pass - positions = fetch_positions() + + conn = get_db() + try: run_options_profit_alerts( conn, positions, @@ -538,17 +556,22 @@ def options_monitor_loop( account_label=account_label, ticker_bid_fn=ticker_bid_fn, ) - pe_cfg = dict(profit_exit_cfg or {}) - pe_cfg.setdefault("send_wechat", send_wechat) - pe_cfg.setdefault("account_label", account_label) - # 平仓/卖币会另开 DB 连接;先提交释放写锁,避免中控快照被锁超时→全部离线 - if target_close_fn is not None: - from lib.options.options_target_lib import run_options_target_closes + conn.commit() + except Exception: + pass + finally: + try: + conn.close() + except Exception: + pass - try: - conn.commit() - except Exception: - pass + # 目标/翻倍出场:短连接内扫描;close_fn 在连接关闭后由内部仍可能嵌套— + # 此处每段自管连接,平仓前必 close + if target_close_fn is not None: + from lib.options.options_target_lib import run_options_target_closes + + conn = get_db() + try: run_options_target_closes( conn, positions, @@ -557,13 +580,20 @@ def options_monitor_loop( account_label=account_label, cfg=pe_cfg, ) - if profit_exit_close_fn is not None: - from lib.options.options_profit_exit_lib import run_options_profit_exits - + conn.commit() + except Exception: + pass + finally: try: - conn.commit() + conn.close() except Exception: pass + + if profit_exit_close_fn is not None: + from lib.options.options_profit_exit_lib import run_options_profit_exits + + conn = get_db() + try: run_options_profit_exits( conn, positions, @@ -573,23 +603,24 @@ def options_monitor_loop( cfg=pe_cfg, ex=pe_cfg.get("exchange_options"), ) - conn.commit() - finally: - conn.close() - # 同步/卖币必须在监控连接关闭之后 + conn.commit() + except Exception: + pass + finally: + try: + conn.close() + except Exception: + pass + 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) - pe_sell.setdefault("account_label", account_label) - sell_spot_after_options_sync(pe_sell) + sell_spot_after_options_sync(pe_cfg) except Exception: pass - # 平仓限价挂单超时撤单(独立于 DB 事务) if stale_pending_fn is not None: try: stale_pending_fn() diff --git a/lib/options/options_notify_lib.py b/lib/options/options_notify_lib.py index bc80856..4854d9c 100644 --- a/lib/options/options_notify_lib.py +++ b/lib/options/options_notify_lib.py @@ -232,6 +232,12 @@ def notify_options_open( trade_id=trade_id, 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 and trade_id is not None: conn.execute( diff --git a/lib/options/options_profit_exit_lib.py b/lib/options/options_profit_exit_lib.py index b6909be..e006999 100644 --- a/lib/options/options_profit_exit_lib.py +++ b/lib/options/options_profit_exit_lib.py @@ -335,6 +335,8 @@ def run_options_profit_exits( continue if state == "closing": + # 平仓/卖币会另开连接写库,先提交释放锁(模拟盘尤其敏感) + _commit(conn) result = close_fn(inst_id) if result.get("already_flat") or _result_fully_done(result): _mark_state(conn, inst_id, "done") @@ -369,6 +371,7 @@ def run_options_profit_exits( if not profit_exit_hit(premium_paid=prem, recycle_usdc=recycle, mult=mult): continue + _commit(conn) result = close_fn(inst_id) if result.get("already_flat"): _mark_state(conn, inst_id, "done") diff --git a/lib/options/options_spot_bridge_lib.py b/lib/options/options_spot_bridge_lib.py index 721c02b..c157fe4 100644 --- a/lib/options/options_spot_bridge_lib.py +++ b/lib/options/options_spot_bridge_lib.py @@ -514,7 +514,94 @@ def sell_residual_after_option_flat( return _apply_bridge(conn_or_get_db) -def bridge_blocks_new_open_msg(conn: sqlite3.Connection) -> str | None: +def clear_dust_pending_sell_bridges( + conn_or_get_db: Any, + ex: Any | None, +) -> int: + """待卖回桥若交易户只剩粉尘(或已空),关闭桥并返回关闭条数. + + 推荐传 get_db:余额查询(模拟盘会另开连接)绝不能套在未关闭的业务连接上。 + """ + if ex is None: + return 0 + get_db = conn_or_get_db if callable(conn_or_get_db) else None + + def _pending(c: sqlite3.Connection) -> list[dict[str, Any]]: + return [ + b + for b in list_open_bridges(c) + if str(b.get("status") or "") == BRIDGE_PENDING_SELL + ] + + if get_db is not None: + conn = get_db() + try: + pending = _pending(conn) + try: + conn.commit() + except Exception: + pass + finally: + try: + conn.close() + except Exception: + pass + else: + conn = conn_or_get_db + pending = _pending(conn) + try: + conn.commit() + except Exception: + pass + + to_close: list[dict[str, Any]] = [] + for b in pending: + uly = str(b.get("underlying") or "ETH").upper() or "ETH" + try: + avail = fetch_trading_coin_available(ex, uly) + except Exception: + continue + if avail is None: + continue + if is_sellable_spot_coin(ex, uly, avail): + continue + to_close.append(b) + if not to_close: + return 0 + + def _close_all(c: sqlite3.Connection) -> int: + n = 0 + for b in to_close: + try: + update_bridge( + c, + int(b["id"]), + status=BRIDGE_CLOSED, + message="粉尘低于现货最小可卖,已忽略", + close=True, + ) + n += 1 + except Exception: + pass + return n + + if get_db is not None: + conn2 = get_db() + try: + return _close_all(conn2) + finally: + try: + conn2.close() + except Exception: + pass + return _close_all(conn_or_get_db) + + +def bridge_blocks_new_open_msg( + conn: sqlite3.Connection, + ex: Any | None = None, +) -> str | None: + _ = ex # 清粉尘请调用方先 clear_dust_pending_sell_bridges(get_db, ex) bridges = list_open_bridges(conn) if not bridges: return None @@ -528,8 +615,18 @@ def bridge_blocks_new_open_msg(conn: sqlite3.Connection) -> str | None: return "存在未完成的币本位资金桥,暂不可开仓" -def mode_switch_block_msg(conn: sqlite3.Connection, ex: Any | None = None) -> str | None: +def mode_switch_block_msg( + conn: sqlite3.Connection, + ex: Any | None = None, + *, + get_db: Any | None = None, +) -> str | None: """有单笔期权仓或未完成桥时禁止切换本位.""" + try: + if get_db is not None: + clear_dust_pending_sell_bridges(get_db, ex) + except Exception: + pass if has_unfinished_bridge(conn): return "存在未完成的币本位资金桥,禁止切换期权本位模式" if ex is not None: diff --git a/lib/options/options_target_lib.py b/lib/options/options_target_lib.py index a404f06..0f9cb9b 100644 --- a/lib/options/options_target_lib.py +++ b/lib/options/options_target_lib.py @@ -390,6 +390,8 @@ def run_options_target_closes( mark_monitor(conn, int(mon["id"]), status="expired", message="持仓已平") _commit_monitor(conn) continue + # 平仓/卖币另开连接,先提交释放锁 + _commit_monitor(conn) result = close_fn(inst_id) idx = _safe_float(pos_by_inst[inst_id].get("idx_px") or pos_by_inst[inst_id].get("idxPx")) if result.get("already_flat") or _result_fully_done(result): @@ -432,6 +434,7 @@ def run_options_target_closes( if not target_hit(opt_type=str(opt_type) if opt_type else None, index_px=idx, target_index=target): continue + _commit_monitor(conn) result = close_fn(inst_id) if result.get("already_flat"): mark_monitor(conn, int(mon["id"]), status="expired", trigger_idx=idx, message="持仓已平") diff --git a/lib/sim/broker_lib.py b/lib/sim/broker_lib.py index 64a4853..cccf78d 100644 --- a/lib/sim/broker_lib.py +++ b/lib/sim/broker_lib.py @@ -273,6 +273,8 @@ class SimBroker: side = (direction or "").lower().strip() if side not in ("long", "short"): raise ValueError("direction 须为 long 或 short") + # 行情先拉,再开库,避免握锁等网络 + bid, ask = _ticker_bid_ask(exchange, symbol) conn = self.get_db() try: row = conn.execute( @@ -289,7 +291,6 @@ class SimBroker: if close_c <= 0: raise ValueError("平仓张数无效") fr = sim_fee_rate(fee_rate) - bid, ask = _ticker_bid_ask(exchange, symbol) qty = close_c * ct_sz pr = perp_fill(side=side, action="close", bid=bid, ask=ask, qty=qty, fee_rate=fr) if side == "long": @@ -492,6 +493,10 @@ class SimBroker: n = int(sheets) if n < 1: return {"ok": False, "msg": "张数至少为 1"} + # 行情先拉,再开库,避免握锁等网络/嵌套写钱包 + bid, ask, _ = _option_bid_ask(exchange_options, inst_id) + if price is not None and float(price) > 0: + bid = max(float(bid), float(price)) if float(price) > 0 else float(bid) conn = self.get_db() try: row = conn.execute( @@ -506,9 +511,6 @@ class SimBroker: if close_n <= 0: return {"ok": False, "msg": "平仓张数无效"} fr = sim_fee_rate(fee_rate) - bid, ask, _ = _option_bid_ask(exchange_options, inst_id) - if price is not None and float(price) > 0: - bid = max(float(bid), float(price)) if float(price) > 0 else float(bid) qty = close_n * ct_mult pr = option_fill(action="close", bid=bid, ask=ask, qty=qty, fee_rate=fr) credit = pr.notional - pr.fee