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

Stop repeat WeChat alerts on option target closes.

Browse Source 
Commit monitor status before notify, and use a closing state so unfilled limits retry silently instead of re-alerting.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-15 16:45:43 +08:00
parent ec30a3999c
commit 142b6de0f6
4 changed files with 318 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/options/options_hub_lib.py
+2 -2
View File
@@ -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 ""))
lib/options/options_register.py
+2 -2
View File
@@ -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()
lib/options/options_target_lib.py
+198 -46
View File
@@ -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
tests/test_options_target_lib.py
+116 -1
View File
@@ -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()