From 142b6de0f6890a0e752993ba2e8c706a29179028 Mon Sep 17 00:00:00 2001 From: dekun Date: Wed, 15 Jul 2026 16:45:43 +0800 Subject: [PATCH] Stop repeat WeChat alerts on option target closes. Commit monitor status before notify, and use a closing state so unfilled limits retry silently instead of re-alerting. Co-authored-by: Cursor --- lib/options/options_hub_lib.py | 4 +- lib/options/options_register.py | 4 +- lib/options/options_target_lib.py | 244 ++++++++++++++++++++++++------ tests/test_options_target_lib.py | 117 +++++++++++++- 4 files changed, 318 insertions(+), 51 deletions(-) diff --git a/lib/options/options_hub_lib.py b/lib/options/options_hub_lib.py index 3c1bfdf..c859af9 100644 --- a/lib/options/options_hub_lib.py +++ b/lib/options/options_hub_lib.py @@ -34,9 +34,9 @@ def build_options_hub_snapshot(cfg: dict[str, Any]) -> dict[str, Any]: try: conn = cfg["get_db"]() try: - from lib.options.options_target_lib import list_active_targets, targets_by_inst + from lib.options.options_target_lib import list_active_targets, list_closing_targets, targets_by_inst - target_monitors = list_active_targets(conn) + target_monitors = list_active_targets(conn) + list_closing_targets(conn) tgt_map = targets_by_inst(conn) for p in positions: mon = tgt_map.get(str(p.get("inst_id") or "")) diff --git a/lib/options/options_register.py b/lib/options/options_register.py index 244bf22..a4e8daf 100644 --- a/lib/options/options_register.py +++ b/lib/options/options_register.py @@ -580,9 +580,9 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None: def api_options_targets(): conn = cfg["get_db"]() try: - from lib.options.options_target_lib import list_active_targets + from lib.options.options_target_lib import list_active_targets, list_closing_targets - return jsonify({"ok": True, "targets": list_active_targets(conn)}) + return jsonify({"ok": True, "targets": list_active_targets(conn) + list_closing_targets(conn)}) finally: conn.close() diff --git a/lib/options/options_target_lib.py b/lib/options/options_target_lib.py index ae2853b..1347e64 100644 --- a/lib/options/options_target_lib.py +++ b/lib/options/options_target_lib.py @@ -75,8 +75,9 @@ def upsert_target_monitor( row = conn.execute( """ SELECT id FROM options_target_monitors - WHERE inst_id = ? AND status = 'active' - ORDER BY id DESC LIMIT 1 + WHERE inst_id = ? AND status IN ('active', 'closing') + ORDER BY CASE status WHEN 'active' THEN 0 WHEN 'closing' THEN 1 ELSE 2 END, id DESC + LIMIT 1 """, (inst_id,), ).fetchone() @@ -89,12 +90,25 @@ def upsert_target_monitor( opt_type = COALESCE(?, opt_type), trade_id = COALESCE(?, trade_id), sheets = COALESCE(?, sheets), - message = NULL + status = 'active', + trigger_idx = NULL, + close_ord_id = NULL, + message = NULL, + triggered_at = NULL WHERE id = ? """, (target_index, underlying, opt_type, trade_id, sheets, int(row["id"])), ) mon_id = int(row["id"]) + # 同一合约其他进行中的委托取消,避免双轨触发重复推送 + conn.execute( + """ + UPDATE options_target_monitors + SET status = 'cancelled', message = '被新目标位覆盖' + WHERE inst_id = ? AND id != ? AND status IN ('active', 'closing') + """, + (inst_id, mon_id), + ) else: cur = conn.execute( """ @@ -115,7 +129,7 @@ def cancel_target_monitor(conn: sqlite3.Connection, *, inst_id: str | None = Non """ UPDATE options_target_monitors SET status = 'cancelled', message = '手动取消' - WHERE id = ? AND status = 'active' + WHERE id = ? AND status IN ('active', 'closing') """, (int(monitor_id),), ) @@ -125,7 +139,7 @@ def cancel_target_monitor(conn: sqlite3.Connection, *, inst_id: str | None = Non """ UPDATE options_target_monitors SET status = 'cancelled', message = '手动取消' - WHERE inst_id = ? AND status = 'active' + WHERE inst_id = ? AND status IN ('active', 'closing') """, (inst_id.strip(),), ) @@ -133,6 +147,21 @@ def cancel_target_monitor(conn: sqlite3.Connection, *, inst_id: str | None = Non return 0 +def _row_to_target(r: sqlite3.Row) -> dict[str, Any]: + return { + "id": int(r["id"]), + "inst_id": r["inst_id"], + "underlying": r["underlying"], + "opt_type": r["opt_type"], + "target_index": _safe_float(r["target_index"]), + "trade_id": r["trade_id"], + "sheets": r["sheets"], + "status": r["status"], + "message": r["message"], + "created_at": r["created_at"], + } + + def list_active_targets(conn: sqlite3.Connection) -> list[dict[str, Any]]: ensure_target_tables(conn) rows = conn.execute( @@ -144,27 +173,32 @@ def list_active_targets(conn: sqlite3.Connection) -> list[dict[str, Any]]: ORDER BY id DESC """ ).fetchall() - out: list[dict[str, Any]] = [] - for r in rows: - out.append( - { - "id": int(r["id"]), - "inst_id": r["inst_id"], - "underlying": r["underlying"], - "opt_type": r["opt_type"], - "target_index": _safe_float(r["target_index"]), - "trade_id": r["trade_id"], - "sheets": r["sheets"], - "status": r["status"], - "message": r["message"], - "created_at": r["created_at"], - } - ) - return out + return [_row_to_target(r) for r in rows] + + +def list_closing_targets(conn: sqlite3.Connection) -> list[dict[str, Any]]: + """已挂出平仓单、等待成交的目标(不再重复推送微信).""" + ensure_target_tables(conn) + rows = conn.execute( + """ + SELECT id, inst_id, underlying, opt_type, target_index, trade_id, sheets, + status, message, created_at + FROM options_target_monitors + WHERE status = 'closing' + ORDER BY id DESC + """ + ).fetchall() + return [_row_to_target(r) for r in rows] def targets_by_inst(conn: sqlite3.Connection) -> dict[str, dict[str, Any]]: - return {str(t["inst_id"]): t for t in list_active_targets(conn) if t.get("inst_id")} + """UI/持仓挂载:active 与 closing 都算进行中.""" + out: dict[str, dict[str, Any]] = {} + for t in list_closing_targets(conn) + list_active_targets(conn): + inst = str(t.get("inst_id") or "") + if inst and inst not in out: + out[inst] = t + return out def mark_monitor( @@ -183,7 +217,10 @@ def mark_monitor( trigger_idx = COALESCE(?, trigger_idx), close_ord_id = COALESCE(?, close_ord_id), message = COALESCE(?, message), - triggered_at = CASE WHEN ? IN ('triggered', 'expired') THEN CURRENT_TIMESTAMP ELSE triggered_at END + triggered_at = CASE + WHEN ? IN ('triggered', 'expired', 'closing') THEN COALESCE(triggered_at, CURRENT_TIMESTAMP) + ELSE triggered_at + END WHERE id = ? """, (status, trigger_idx, close_ord_id, message, status, int(monitor_id)), @@ -197,7 +234,7 @@ def cancel_orphans_without_position( ) -> int: """持仓已消失的目标委托标记为 expired(到期/已平),不挂止损.""" ensure_target_tables(conn) - rows = list_active_targets(conn) + rows = list_active_targets(conn) + list_closing_targets(conn) n = 0 for t in rows: inst = str(t.get("inst_id") or "") @@ -207,6 +244,14 @@ def cancel_orphans_without_position( return n +def _commit_monitor(conn: sqlite3.Connection) -> None: + """状态变更立刻落库,避免后续 sync 异常回滚后重复触发/推送.""" + try: + conn.commit() + except Exception: + pass + + def close_option_by_bid_depth( cfg: dict[str, Any], ex: Any, @@ -246,7 +291,42 @@ def close_option_by_bid_depth( td_mode = str(pos.get("mgnMode") or cfg.get("td_mode") or "isolated") pos_side = _pos_side_from_position(pos) or "net" - # 先撤本合约未完成卖单,避免重复挂单被当成开空占用保证金 + # 已有未成交卖平单时先等成交,避免每轮撤单重挂反复推送/吃档 + try: + pending = ex.private_get_trade_orders_pending({"instType": "OPTION", "instId": inst_id}) or {} + sell_pending = [ + o + for o in (pending.get("data") or []) + if str(o.get("side") or "").lower() == "sell" and o.get("ordId") + ] + if sell_pending: + time.sleep(0.5) + invalidate_option_positions_cache() + raw_positions = cfg["fetch_option_positions"](ex) + if raw_positions is None: + return {"ok": False, "msg": "获取期权持仓失败"} + pos = next((p for p in raw_positions if str(p.get("instId")) == inst_id), None) + if not pos or _avail(pos) < 1: + return { + "ok": True, + "already_flat": True, + "msg": "已有限价卖单成交", + "close_ord_id": ",".join(str(o.get("ordId")) for o in sell_pending), + "fully_closed": True, + "submitted_sheets": close_sheets, + "remaining_sheets": 0, + } + # 仍持仓且卖单挂着:本轮不撤不重挂,交给下一轮 + return { + "ok": False, + "msg": "等待已有买一限价卖单成交", + "stopped_reason": "pending_close_order", + "close_ord_id": ",".join(str(o.get("ordId")) for o in sell_pending), + } + except Exception: + pass + + # 无挂单时再清理残留卖单(兼容旧路径)并按买一重新挂出 try: pending = ex.private_get_trade_orders_pending({"instType": "OPTION", "instId": inst_id}) or {} for o in pending.get("data") or []: @@ -431,6 +511,46 @@ def close_option_by_bid_depth( } +def _notify_target_close( + send_wechat: Callable[[str], None] | None, + *, + account_label: str, + inst_id: str, + target: float, + idx: float, + result: dict[str, Any], +) -> None: + if not send_wechat: + return + try: + send_wechat( + "\n".join( + [ + "【OKX期权·目标位平仓】", + f"账户:{account_label}", + f"合约:{inst_id}", + f"目标指数:{target:g}", + f"触发指数:{idx:g}", + f"提交张数:{result.get('submitted_sheets') or '—'}", + f"预估收回:{result.get('premium_received') if result.get('premium_received') is not None else '—'} USDC", + ] + ) + ) + except Exception: + pass + + +def _result_fully_done(result: dict[str, Any]) -> bool: + if result.get("already_flat"): + return True + if result.get("fully_closed"): + return True + remaining = result.get("remaining_sheets") + if remaining is not None and int(remaining) <= 0 and result.get("ok"): + return True + return False + + def run_options_target_closes( conn: sqlite3.Connection, positions: list[dict[str, Any]], @@ -442,12 +562,47 @@ def run_options_target_closes( ) -> int: """ 扫描 active 目标委托;指数到位后限价平仓. - 返回触发条数. + 状态先 commit 再推微信,避免 sync 失败回滚导致同一笔反复推送. + 未完全成交进入 closing,仅重试平仓不再推送. + 返回本次新触发(并推送)的条数. """ ensure_target_tables(conn) pos_by_inst = {str(p.get("inst_id") or p.get("instId") or ""): p for p in positions} live_ids = {k for k in pos_by_inst if k} cancel_orphans_without_position(conn, live_inst_ids=live_ids) + _commit_monitor(conn) + + # 先处理已挂单等待成交的,绝不再发微信 + for mon in list_closing_targets(conn): + inst_id = str(mon.get("inst_id") or "") + if not inst_id: + continue + if inst_id not in pos_by_inst: + mark_monitor(conn, int(mon["id"]), status="expired", message="持仓已平") + _commit_monitor(conn) + continue + 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): + mark_monitor( + conn, + int(mon["id"]), + status="triggered", + trigger_idx=idx, + close_ord_id=result.get("close_ord_id"), + message="目标位限价平仓完成", + ) + _commit_monitor(conn) + continue + mark_monitor( + conn, + int(mon["id"]), + status="closing", + trigger_idx=idx, + close_ord_id=result.get("close_ord_id"), + message=str(result.get("msg") or result.get("stopped_reason") or "等待买一成交"), + ) + _commit_monitor(conn) triggered = 0 for mon in list_active_targets(conn): @@ -471,6 +626,7 @@ def run_options_target_closes( result = close_fn(inst_id) if result.get("already_flat"): mark_monitor(conn, int(mon["id"]), status="expired", trigger_idx=idx, message="持仓已平") + _commit_monitor(conn) continue if not result.get("ok"): mark_monitor( @@ -480,32 +636,28 @@ def run_options_target_closes( trigger_idx=idx, message=str(result.get("msg") or result.get("stopped_reason") or "平仓未完成,将重试"), ) + _commit_monitor(conn) continue + done = _result_fully_done(result) + status = "triggered" if done else "closing" mark_monitor( conn, int(mon["id"]), - status="triggered", + status=status, trigger_idx=idx, close_ord_id=result.get("close_ord_id"), - message="目标位触发限价平仓", + message="目标位触发限价平仓" if done else "目标位已挂买一限价,等待成交", ) + # 关键:先落库,再推送——否则后续 sync 异常回滚会让同一笔反复推微信 + _commit_monitor(conn) triggered += 1 - if send_wechat: - try: - send_wechat( - "\n".join( - [ - "【OKX期权·目标位平仓】", - f"账户:{account_label}", - f"合约:{inst_id}", - f"目标指数:{target:g}", - f"触发指数:{idx:g}", - f"提交张数:{result.get('submitted_sheets') or '—'}", - f"预估收回:{result.get('premium_received') if result.get('premium_received') is not None else '—'} USDC", - ] - ) - ) - except Exception: - pass + _notify_target_close( + send_wechat, + account_label=account_label, + inst_id=inst_id, + target=target, + idx=idx, + result=result, + ) return triggered diff --git a/tests/test_options_target_lib.py b/tests/test_options_target_lib.py index 69547dc..2962f1d 100644 --- a/tests/test_options_target_lib.py +++ b/tests/test_options_target_lib.py @@ -7,6 +7,7 @@ import unittest from lib.options.options_target_lib import ( ensure_target_tables, list_active_targets, + list_closing_targets, run_options_target_closes, target_hit, upsert_target_monitor, @@ -38,7 +39,14 @@ class OptionsTargetLibTests(unittest.TestCase): def close_fn(inst_id: str): closed.append(inst_id) - return {"ok": True, "submitted_sheets": 1, "premium_received": 1.2, "close_ord_id": "oid1"} + return { + "ok": True, + "submitted_sheets": 1, + "premium_received": 1.2, + "close_ord_id": "oid1", + "fully_closed": True, + "remaining_sheets": 0, + } n = run_options_target_closes( conn, @@ -49,6 +57,113 @@ class OptionsTargetLibTests(unittest.TestCase): self.assertEqual(closed, ["ETH-USD_UM-260717-1900-C"]) self.assertEqual(len(list_active_targets(conn)), 0) + def test_partial_fill_notifies_once_then_closing_retry_silent(self): + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + ensure_target_tables(conn) + upsert_target_monitor( + conn, + inst_id="ETH-USD_UM-260715-1870-P", + target_index=1872, + opt_type="P", + sheets=1, + ) + conn.commit() + notices: list[str] = [] + calls = {"n": 0} + + def close_fn(inst_id: str): + calls["n"] += 1 + if calls["n"] == 1: + return { + "ok": True, + "submitted_sheets": 1, + "premium_received": 0.032, + "close_ord_id": "oid-a", + "fully_closed": False, + "remaining_sheets": 1, + "stopped_reason": "order_not_filled", + } + return { + "ok": True, + "submitted_sheets": 1, + "premium_received": 0.032, + "close_ord_id": "oid-b", + "fully_closed": True, + "remaining_sheets": 0, + "already_flat": True, + } + + pos = [{"inst_id": "ETH-USD_UM-260715-1870-P", "idx_px": 1867.5, "opt_type": "P"}] + n1 = run_options_target_closes( + conn, + pos, + close_fn=close_fn, + send_wechat=notices.append, + account_label="主账户·期权", + ) + self.assertEqual(n1, 1) + self.assertEqual(len(notices), 1) + self.assertEqual(len(list_active_targets(conn)), 0) + self.assertEqual(len(list_closing_targets(conn)), 1) + + # 模拟后续 sync 异常也不会再推:closing 重试静默 + n2 = run_options_target_closes( + conn, + pos, + close_fn=close_fn, + send_wechat=notices.append, + account_label="主账户·期权", + ) + self.assertEqual(n2, 0) + self.assertEqual(len(notices), 1) + self.assertEqual(len(list_closing_targets(conn)), 0) + + def test_commit_before_wechat_survives_later_rollback(self): + """状态在推送前已 commit,外层异常回滚不应让委托回到 active.""" + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + ensure_target_tables(conn) + upsert_target_monitor( + conn, + inst_id="ETH-USD_UM-260715-1870-P", + target_index=1872, + opt_type="P", + ) + conn.commit() + notices: list[str] = [] + + def close_fn(inst_id: str): + return { + "ok": True, + "submitted_sheets": 1, + "premium_received": 0.03, + "close_ord_id": "oid1", + "fully_closed": True, + "remaining_sheets": 0, + } + + run_options_target_closes( + conn, + [{"inst_id": "ETH-USD_UM-260715-1870-P", "idx_px": 1860, "opt_type": "P"}], + close_fn=close_fn, + send_wechat=notices.append, + ) + # 模拟 loop 后续 sync 抛错后 close 未再 commit —— 但 status 已提前 commit + conn.rollback() + self.assertEqual(len(notices), 1) + self.assertEqual(len(list_active_targets(conn)), 0) + + # 下一轮不应再次触发推送 + n2 = run_options_target_closes( + conn, + [{"inst_id": "ETH-USD_UM-260715-1870-P", "idx_px": 1860, "opt_type": "P"}], + close_fn=close_fn, + send_wechat=notices.append, + ) + self.assertEqual(n2, 0) + self.assertEqual(len(notices), 1) + if __name__ == "__main__": unittest.main()