diff --git a/lib/hedge_plan/hedge_plan_monitor_lib.py b/lib/hedge_plan/hedge_plan_monitor_lib.py index 717cd19..4255473 100644 --- a/lib/hedge_plan/hedge_plan_monitor_lib.py +++ b/lib/hedge_plan/hedge_plan_monitor_lib.py @@ -1,12 +1,14 @@ -"""对冲计划监控:永期 TP/SL 与期期目标价/到期收口.""" +"""对冲计划监控:永期 TP/SL、期期目标价、到期结算与微信收口推送.""" from __future__ import annotations import os from datetime import datetime, timezone from typing import Any, Optional -from lib.hedge_plan.hedge_plan_db import get_plan_legs, list_plans, update_plan +from lib.hedge_plan.hedge_plan_db import get_plan, get_plan_legs, list_plans, update_plan +from lib.hedge_plan.hedge_plan_notify_lib import notify_hedge, notify_plan_end, build_hedge_alert_message from lib.hedge_plan.hedge_plan_orders_lib import _sell_option +from lib.hedge_plan.hedge_plan_settle_lib import leg_is_expired, settle_option_leg_at_spot def _now() -> str: @@ -47,11 +49,21 @@ def _index_px(cfg: dict[str, Any], underlying: str) -> Optional[float]: return fn(ex, underlying) except Exception: return None + # 无期权账户时回退永续 ticker + ex_perp = cfg.get("exchange") + if ex_perp is not None: + try: + base = (underlying or "ETH").upper() + sym = f"{base}/USDT:USDT" + t = ex_perp.fetch_ticker(sym) + return _sf((t.get("info") or {}).get("markPx")) or _sf(t.get("last")) + except Exception: + return None return None def tick_active_plans(cfg: dict[str, Any]) -> dict[str, Any]: - """扫描 active 计划并按规则收口.返回处理摘要.""" + """扫描 active 计划 + 止盈后遗留期权到期收口.返回处理摘要.""" get_db = cfg.get("get_db") if not callable(get_db): return {"ok": False, "msg": "get_db missing"} @@ -61,31 +73,44 @@ def tick_active_plans(cfg: dict[str, Any]) -> dict[str, Any]: from lib.hedge_plan.hedge_plan_db import init_hedge_plan_tables init_hedge_plan_tables(conn) - plans = list_plans(conn, status="active", limit=20) + plans = list_plans(conn, status="active", limit=40) for plan in plans: r = _tick_one(cfg, conn, plan) if r: acted.append(r) + orphaned = _settle_orphaned_after_tp(cfg, conn) + acted.extend(orphaned) conn.commit() finally: conn.close() return {"ok": True, "acted": acted} +def _notify_end_reload(cfg: dict[str, Any], conn: Any, plan_id: int) -> None: + plan = get_plan(conn, int(plan_id)) + if plan: + notify_plan_end(cfg, conn, plan) + + def _tick_one(cfg: dict[str, Any], conn: Any, plan: dict[str, Any]) -> Optional[dict[str, Any]]: pt = plan.get("plan_type") legs = get_plan_legs(conn, int(plan["id"])) if pt == "perp_options": - return _tick_po(cfg, conn, plan, legs) + # 先判断期权是否已过期且永续仍在(罕见);主路径仍是永续平仓侦测 + r = _tick_po(cfg, conn, plan, legs) + return r if pt == "options_options": - return _tick_oo(cfg, conn, plan, legs) + r = _tick_oo_expiry(cfg, conn, plan, legs) + if r: + return r + return _tick_oo_target(cfg, conn, plan, legs) return None def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]]) -> Optional[dict[str, Any]]: perp = next((x for x in legs if x.get("leg_role") == "perp"), None) opt = next((x for x in legs if x.get("leg_role") == "option_hedge"), None) - if not perp: + if not perp or perp.get("status") != "open": return None symbol = perp.get("symbol") or "" direction = (plan.get("direction") or "long").lower() @@ -117,7 +142,6 @@ def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di elif direction == "short" and mark <= tp: reason = "perp_tp" premium = float(plan.get("premium_total") or 0) - # 粗算永续已实现 cs = float(cfg.get("default_contract_size") or 0.01) get_cs = cfg.get("get_contract_size") if callable(get_cs) and symbol: @@ -140,13 +164,20 @@ def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di inst_id=str(opt.get("inst_id") or ""), sheets=float(opt.get("size") or 1), ) - # 简化:平仓失败仍结束计划并记 −权利金 + if not close_r.get("ok"): + notify_hedge( + cfg, + build_hedge_alert_message( + title="永续止损后期权强制平仓失败", + plan_id=plan.get("id"), + detail=str(close_r.get("msg") or close_r), + ), + ) if close_r.get("ok"): - # 无法精确拿到卖出价差时仍用 −premium 作为下限;有 bid 则近似 bid = _sf(close_r.get("bid")) ask_open = _sf(opt.get("avg_open")) if bid is not None and ask_open is not None: - ct = 0.01 + ct = float(opt.get("ct_mult") or 0.01) opt_pnl = (bid - ask_open) * float(opt.get("size") or 1) * ct else: opt_pnl = -premium @@ -155,9 +186,17 @@ def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di ("closed", reason, _now(), opt_pnl, opt["id"]), ) elif reason == "perp_tp" and opt: - # 止盈默认不平期权 if _env_bool("HEDGE_PLAN_ON_PERP_TP_CLOSE_OPTIONS", False): - _sell_option(cfg, inst_id=str(opt.get("inst_id") or ""), sheets=float(opt.get("size") or 1)) + close_r = _sell_option(cfg, inst_id=str(opt.get("inst_id") or ""), sheets=float(opt.get("size") or 1)) + if not close_r.get("ok"): + notify_hedge( + cfg, + build_hedge_alert_message( + title="永续止盈后期权平仓失败", + plan_id=plan.get("id"), + detail=str(close_r.get("msg") or close_r), + ), + ) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=? WHERE id=?", ("closed", reason, _now(), opt["id"]), @@ -170,9 +209,9 @@ def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di opt_pnl = -premium if reason == "perp_tp": - total = perp_pnl + opt_pnl # = 止盈盈利 − 权利金 + total = perp_pnl + opt_pnl else: - total = opt_pnl + perp_pnl # 期权盈亏 + 永续盈亏 + total = opt_pnl + perp_pnl conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", @@ -189,33 +228,32 @@ def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di stats_bucket="tp" if reason == "perp_tp" else "sl", closed_at=_now(), ) + _notify_end_reload(cfg, conn, int(plan["id"])) return {"plan_id": plan["id"], "close_reason": reason, "total": total} -def _tick_oo(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]]) -> Optional[dict[str, Any]]: +def _tick_oo_target( + cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]] +) -> Optional[dict[str, Any]]: target = _sf(plan.get("target_price")) idx = _index_px(cfg, str(plan.get("underlying") or "ETH")) if target is None or idx is None: return None - # 简化:接近目标价(相对 0.15%)时平盈利腿 - if abs(idx - target) / max(target, 1) > 0.0015 and not (idx >= target or idx <= target): - pass near = abs(idx - target) / max(abs(target), 1.0) <= 0.002 if not near: return None if not _env_bool("HEDGE_PLAN_OO_CLOSE_WINNER_ONLY", True): return None - open_legs = [x for x in legs if x.get("status") == "open" and x.get("leg_role", "").startswith("option")] + open_legs = [x for x in legs if x.get("status") == "open" and str(x.get("leg_role") or "").startswith("option")] if len(open_legs) < 2: return None - # 用内在价值粗判盈利腿 winners = [] for leg in open_legs: strike = _sf(leg.get("strike")) or 0 o = (leg.get("opt_type") or "").upper() intrinsic = max(0.0, idx - strike) if o == "C" else max(0.0, strike - idx) premium = float(leg.get("premium") or 0) - pnl = intrinsic * float(leg.get("size") or 1) * 0.01 - premium + pnl = intrinsic * float(leg.get("size") or 1) * float(leg.get("ct_mult") or 0.01) - premium winners.append((pnl, leg)) winners.sort(key=lambda x: x[0], reverse=True) best_pnl, best = winners[0] @@ -223,11 +261,130 @@ def _tick_oo(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di return None close_r = _sell_option(cfg, inst_id=str(best.get("inst_id") or ""), sheets=float(best.get("size") or 1)) if not close_r.get("ok"): + notify_hedge( + cfg, + build_hedge_alert_message( + title="期期平盈利腿失败", + plan_id=plan.get("id"), + detail=str(close_r.get("msg") or close_r), + ), + ) return {"plan_id": plan["id"], "msg": "平盈利腿失败", "close": close_r} conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", "target_win_leg", _now(), best_pnl, best["id"]), ) - # 计划暂不 closed,等另一腿到期;先标 note update_plan(conn, int(plan["id"]), close_reason="target_win_leg") + mid = dict(plan) + mid["close_reason"] = "target_win_leg" + mid["status"] = "active" + notify_plan_end(cfg, conn, mid) return {"plan_id": plan["id"], "close_reason": "target_win_leg", "closed_leg": best.get("id")} + + +def _tick_oo_expiry( + cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]] +) -> Optional[dict[str, Any]]: + """期期:剩余期权腿全部到期 → 结算合计并结束计划.""" + pending = [ + x + for x in legs + if str(x.get("leg_role") or "").startswith("option") + and str(x.get("status") or "") in ("open", "hold_to_expiry") + ] + if not pending: + # 若腿已全部 closed 但计划仍 active(异常残留)则用腿合计收口 + closed_opts = [ + x for x in legs if str(x.get("leg_role") or "").startswith("option") and x.get("status") == "closed" + ] + if len(closed_opts) < 1: + return None + total_opts = sum(float(x.get("realized_pnl") or 0) for x in closed_opts) + reason = "oo_expiry_loss" if total_opts <= 0 else "oo_expiry_win" + update_plan( + conn, + int(plan["id"]), + status="closed", + close_reason=reason, + realized_pnl_options=round(total_opts, 4), + realized_pnl_total=round(total_opts, 4), + stats_bucket=reason if reason == "oo_expiry_loss" else "oo_target", + closed_at=_now(), + ) + _notify_end_reload(cfg, conn, int(plan["id"])) + return {"plan_id": plan["id"], "close_reason": reason, "total": total_opts} + + if not all(leg_is_expired(x) for x in pending): + return None + + spot = _index_px(cfg, str(plan.get("underlying") or "ETH")) + if spot is None: + return None + + settled_sum = 0.0 + for leg in pending: + pnl = settle_option_leg_at_spot(leg, float(spot)) + settled_sum += pnl + conn.execute( + "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", + ("closed", "expiry", _now(), round(pnl, 4), leg["id"]), + ) + + already = sum( + float(x.get("realized_pnl") or 0) + for x in legs + if str(x.get("leg_role") or "").startswith("option") and x.get("status") == "closed" + ) + total = already + settled_sum + reason = "oo_expiry_loss" if total <= 0 else "oo_expiry_win" + bucket = "oo_expiry_loss" if reason == "oo_expiry_loss" else "oo_target" + update_plan( + conn, + int(plan["id"]), + status="closed", + close_reason=reason, + realized_pnl_options=round(total, 4), + realized_pnl_total=round(total, 4), + stats_bucket=bucket, + closed_at=_now(), + ) + _notify_end_reload(cfg, conn, int(plan["id"])) + return {"plan_id": plan["id"], "close_reason": reason, "total": total, "spot": spot} + + +def _settle_orphaned_after_tp(cfg: dict[str, Any], conn: Any) -> list[dict[str, Any]]: + """永期止盈后 hold_to_expiry 期权到期:只更新腿,不回写计划合计.""" + rows = conn.execute( + """ + SELECT l.id AS leg_id, l.plan_id, l.inst_id, l.opt_type, l.strike, l.size, l.premium, l.status, + p.underlying, p.status AS plan_status + FROM hedge_plan_legs l + JOIN hedge_plans p ON p.id = l.plan_id + WHERE l.status = 'hold_to_expiry' AND l.close_reason = 'orphaned_after_tp' + LIMIT 40 + """ + ).fetchall() + acted: list[dict[str, Any]] = [] + for row in rows: + leg = dict(row) + if not leg_is_expired(leg): + continue + spot = _index_px(cfg, str(leg.get("underlying") or "ETH")) + if spot is None: + continue + pnl = settle_option_leg_at_spot(leg, float(spot)) + conn.execute( + "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", + ("closed", "expiry", _now(), round(pnl, 4), leg["leg_id"]), + ) + # 故意不 UPDATE hedge_plans.realized_pnl_* + acted.append( + { + "plan_id": leg["plan_id"], + "close_reason": "orphaned_option_expiry", + "leg_id": leg["leg_id"], + "leg_pnl": round(pnl, 4), + "note": "不回写计划合计", + } + ) + return acted diff --git a/lib/hedge_plan/hedge_plan_notify_lib.py b/lib/hedge_plan/hedge_plan_notify_lib.py new file mode 100644 index 0000000..a5f8736 --- /dev/null +++ b/lib/hedge_plan/hedge_plan_notify_lib.py @@ -0,0 +1,173 @@ +"""对冲计划企业微信推送(起止必发,幂等落库标记).""" +from __future__ import annotations + +from typing import Any, Callable, Optional + +from lib.hedge_plan.hedge_plan_db import update_plan + + +def _fmt(v: Any, d: int = 2) -> str: + try: + if v is None or v == "": + return "—" + return f"{float(v):.{d}f}" + except (TypeError, ValueError): + return str(v) + + +def _type_label(plan_type: str) -> str: + return "永期对冲" if (plan_type or "") == "perp_options" else "期期对冲" + + +def _dir_label(direction: str) -> str: + d = (direction or "").lower() + if d == "long": + return "做多" + if d == "short": + return "做空" + return "—" + + +def build_hedge_start_message(plan: dict[str, Any], *, legs: Optional[list[dict[str, Any]]] = None) -> str: + pt = plan.get("plan_type") or "" + lines = [ + f"🟢 对冲计划启动 #{plan.get('id')}", + f"📌 类型:{_type_label(pt)}", + f"🪙 标的:{plan.get('underlying') or '—'}", + ] + if pt == "perp_options": + lines.extend( + [ + f"📈 方向:{_dir_label(plan.get('direction') or '')}", + f"💵 开仓参考:{_fmt(plan.get('entry_mark'))}", + f"🎯 止盈:{_fmt(plan.get('tp'))}｜止损:{_fmt(plan.get('sl'))}", + f"📦 永续张数:{_fmt(plan.get('perp_size'), 4)}｜杠杆:{_fmt(plan.get('leverage'), 0)}x", + f"💎 期权保费合计:{_fmt(plan.get('premium_total'), 4)} USDC", + ] + ) + else: + lines.extend( + [ + f"🎯 目标价 S*:{_fmt(plan.get('target_price'))}", + f"💎 期权保费合计:{_fmt(plan.get('premium_total'), 4)} USDC", + ] + ) + if legs: + for leg in legs: + role = leg.get("leg_role") or "" + if role == "perp": + lines.append(f"· 永续腿 {leg.get('symbol') or ''} ×{_fmt(leg.get('size'), 4)}") + else: + lines.append( + f"· {role} {(leg.get('opt_type') or '')} K{_fmt(leg.get('strike'), 0)} " + f"×{_fmt(leg.get('size'), 0)}张 {leg.get('inst_id') or ''}" + ) + lines.append("📎 独立模块推送,不进普通交易复盘") + return "\n".join(lines) + + +def build_hedge_end_message(plan: dict[str, Any]) -> str: + reason = plan.get("close_reason") or "—" + total = plan.get("realized_pnl_total") + try: + tv = float(total) if total is not None else None + except (TypeError, ValueError): + tv = None + head = "🔴" if (tv is not None and tv < 0) else "🟢" + reason_map = { + "perp_tp": "永续止盈(期权默认不平)", + "perp_sl": "永续止损(期权强制平)", + "target_win_leg": "期期已平盈利腿(中间态)", + "oo_expiry_loss": "期期到期无盈利·总亏损", + "oo_expiry_win": "期期到期仍盈利", + "expiry": "到期收口", + "manual": "人工结束", + "partial_fail": "半腿失败收尾", + "cancelled": "已取消", + } + lines = [ + f"{head} 对冲计划结束 #{plan.get('id')}", + f"📌 类型:{_type_label(plan.get('plan_type') or '')}", + f"🪙 标的:{plan.get('underlying') or '—'}", + f"📎 原因:{reason_map.get(reason, reason)}", + f"💰 合计≈U:{_fmt(total)}", + f"· 永续分项:{_fmt(plan.get('realized_pnl_perp'))} USDT", + f"· 期权分项:{_fmt(plan.get('realized_pnl_options'))} USDC(≈U 1:1)", + f"⏱ 开仓:{plan.get('opened_at') or '—'}｜结束:{plan.get('closed_at') or '—'}", + ] + return "\n".join(lines) + + +def build_hedge_alert_message( + *, + title: str, + plan_id: Any = None, + detail: str = "", +) -> str: + lines = [f"⚠️ 对冲计划告警{(' #' + str(plan_id)) if plan_id else ''}", f"📌 {title}"] + if detail: + lines.append(str(detail)[:800]) + return "\n".join(lines) + + +def notify_hedge( + cfg: dict[str, Any], + content: str, +) -> bool: + send: Optional[Callable[[str], Any]] = cfg.get("send_wechat") + if not callable(send): + return False + try: + send(content) + return True + except Exception: + return False + + +def notify_plan_start( + cfg: dict[str, Any], + conn: Any, + plan: dict[str, Any], + legs: Optional[list[dict[str, Any]]] = None, +) -> bool: + if int(plan.get("wechat_start_sent") or 0): + return False + ok = notify_hedge(cfg, build_hedge_start_message(plan, legs=legs)) + if ok and plan.get("id") is not None: + update_plan(conn, int(plan["id"]), wechat_start_sent=1) + plan["wechat_start_sent"] = 1 + return ok + + +def notify_plan_end(cfg: dict[str, Any], conn: Any, plan: dict[str, Any]) -> bool: + if int(plan.get("wechat_end_sent") or 0): + return False + # 中间态 target_win_leg 不算正式结束推送(用告警) + if (plan.get("close_reason") or "") == "target_win_leg" and (plan.get("status") or "") != "closed": + notify_hedge( + cfg, + build_hedge_alert_message( + title="期期已平盈利腿,亏损腿继续持有至到期", + plan_id=plan.get("id"), + detail=f"目标价 {_fmt(plan.get('target_price'))}", + ), + ) + return True + ok = notify_hedge(cfg, build_hedge_end_message(plan)) + if ok and plan.get("id") is not None: + update_plan(conn, int(plan["id"]), wechat_end_sent=1) + plan["wechat_end_sent"] = 1 + return ok + + +def notify_partial_fail(cfg: dict[str, Any], *, plan_type: str, msg: str, results: Any = None) -> bool: + detail = msg + if results: + try: + detail = f"{msg}\n路径结果:{results}"[:800] + except Exception: + pass + return notify_hedge( + cfg, + build_hedge_alert_message(title=f"{_type_label(plan_type)}半腿失败", detail=detail), + ) diff --git a/lib/hedge_plan/hedge_plan_orders_lib.py b/lib/hedge_plan/hedge_plan_orders_lib.py index 18b6077..9a335ea 100644 --- a/lib/hedge_plan/hedge_plan_orders_lib.py +++ b/lib/hedge_plan/hedge_plan_orders_lib.py @@ -278,9 +278,19 @@ def execute_perp_options_start( sheets=float(opt_res.get("sheets") or body.get("sheets") or 1), ) results.append({"step": "options_auto_close_on_perp_fail", **close_r}) + msg = perp_res.get("msg") or "永续开仓失败" + if not dry_run: + try: + from lib.hedge_plan.hedge_plan_notify_lib import notify_partial_fail + + notify_partial_fail( + cfg, plan_type="perp_options", msg=msg, results=results + ) + except Exception: + pass return { "ok": False, - "msg": perp_res.get("msg") or "永续开仓失败", + "msg": msg, "path": path, "results": results, "partial": True, @@ -322,9 +332,17 @@ def execute_options_options_start( if not dry_run and _env_bool("HEDGE_PLAN_PARTIAL_AUTO_CLOSE_OPTION", True): close_r = _sell_option(cfg, inst_id=str(a_res.get("inst_id") or ""), sheets=float(a_res.get("sheets") or 1)) results.append({"step": "options_auto_close_leg_a", **close_r}) + msg = b_res.get("msg") or "腿B开仓失败" + if not dry_run: + try: + from lib.hedge_plan.hedge_plan_notify_lib import notify_partial_fail + + notify_partial_fail(cfg, plan_type="options_options", msg=msg, results=results) + except Exception: + pass return { "ok": False, - "msg": b_res.get("msg") or "腿B开仓失败", + "msg": msg, "path": path, "results": results, "partial": True, diff --git a/lib/hedge_plan/hedge_plan_register.py b/lib/hedge_plan/hedge_plan_register.py index e863b29..aab3c67 100644 --- a/lib/hedge_plan/hedge_plan_register.py +++ b/lib/hedge_plan/hedge_plan_register.py @@ -98,6 +98,7 @@ def _build_cfg(app_module: Any) -> dict[str, Any]: "perp_account_label": (os.getenv("OKX_ACCOUNT_LABEL") or "合约账户").strip(), "options_account_label": (os.getenv("OKX_OPTIONS_ACCOUNT_LABEL") or "期权账户").strip(), "live_trading": _env_bool("LIVE_TRADING_ENABLED", False), + "send_wechat": getattr(app_module, "send_wechat_msg", None), } @@ -171,7 +172,14 @@ def _maybe_start_monitor(cfg: dict[str, Any]) -> None: def _persist_po(cfg: dict[str, Any], result: dict[str, Any], body: dict[str, Any]) -> int: - from lib.hedge_plan.hedge_plan_db import init_hedge_plan_tables, insert_leg, insert_plan + from lib.hedge_plan.hedge_plan_db import ( + get_plan, + get_plan_legs, + init_hedge_plan_tables, + insert_leg, + insert_plan, + ) + from lib.hedge_plan.hedge_plan_notify_lib import notify_plan_start conn = cfg["get_db"]() try: @@ -229,13 +237,25 @@ def _persist_po(cfg: dict[str, Any], result: dict[str, Any], body: dict[str, Any }, ) conn.commit() + plan = get_plan(conn, plan_id) + legs = get_plan_legs(conn, plan_id) + if plan: + notify_plan_start(cfg, conn, plan, legs) + conn.commit() return plan_id finally: conn.close() def _persist_oo(cfg: dict[str, Any], result: dict[str, Any], body: dict[str, Any]) -> int: - from lib.hedge_plan.hedge_plan_db import init_hedge_plan_tables, insert_leg, insert_plan + from lib.hedge_plan.hedge_plan_db import ( + get_plan, + get_plan_legs, + init_hedge_plan_tables, + insert_leg, + insert_plan, + ) + from lib.hedge_plan.hedge_plan_notify_lib import notify_plan_start conn = cfg["get_db"]() try: @@ -274,6 +294,11 @@ def _persist_oo(cfg: dict[str, Any], result: dict[str, Any], body: dict[str, Any }, ) conn.commit() + plan = get_plan(conn, plan_id) + legs = get_plan_legs(conn, plan_id) + if plan: + notify_plan_start(cfg, conn, plan, legs) + conn.commit() return plan_id finally: conn.close() diff --git a/lib/hedge_plan/hedge_plan_settle_lib.py b/lib/hedge_plan/hedge_plan_settle_lib.py new file mode 100644 index 0000000..a864e98 --- /dev/null +++ b/lib/hedge_plan/hedge_plan_settle_lib.py @@ -0,0 +1,62 @@ +"""对冲计划结算辅助:到期内在价值与期权腿收口.""" +from __future__ import annotations + +import time +from typing import Any, Optional + +from lib.exchange.okx_options_lib import normalize_option_exp_ms +from lib.hedge_plan.hedge_plan_calc_lib import option_expiry_pnl + + +def _sf(v: Any) -> Optional[float]: + try: + if v is None or v == "": + return None + return float(v) + except (TypeError, ValueError): + return None + + +def leg_exp_ms(leg: dict[str, Any]) -> Optional[int]: + return normalize_option_exp_ms(leg.get("exp_time"), str(leg.get("inst_id") or "")) + + +def leg_is_expired(leg: dict[str, Any], *, now_ms: Optional[int] = None) -> bool: + exp = leg_exp_ms(leg) + if exp is None: + return False + now = int(now_ms if now_ms is not None else time.time() * 1000) + return now >= int(exp) + + +def settle_option_leg_at_spot(leg: dict[str, Any], spot: float) -> float: + """按到期结算口径估算腿盈亏(USDC).""" + premium = float(leg.get("premium") or 0) + strike = _sf(leg.get("strike")) + if strike is None: + return -premium + sheets = float(leg.get("size") or 1) + # ct_mult 未入库时默认 0.01 + ct = float(leg.get("ct_mult") or 0.01) + return float( + option_expiry_pnl( + opt_type=str(leg.get("opt_type") or "P"), + strike=float(strike), + spot=float(spot), + sheets=sheets, + ct_mult=ct, + premium_paid=premium, + ) + ) + + +def all_option_legs_expired(legs: list[dict[str, Any]], *, now_ms: Optional[int] = None) -> bool: + opts = [ + x + for x in legs + if str(x.get("leg_role") or "").startswith("option") + and str(x.get("status") or "") in ("open", "hold_to_expiry") + ] + if not opts: + return False + return all(leg_is_expired(x, now_ms=now_ms) for x in opts) diff --git a/tests/test_hedge_plan_notify_settle.py b/tests/test_hedge_plan_notify_settle.py new file mode 100644 index 0000000..a6e8bbb --- /dev/null +++ b/tests/test_hedge_plan_notify_settle.py @@ -0,0 +1,220 @@ +"""对冲计划微信文案与到期结算.""" +import sqlite3 +import time +import unittest +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock + +from lib.exchange.okx_options_lib import expiry_ms_from_inst_id +from lib.hedge_plan.hedge_plan_db import get_plan, init_hedge_plan_tables, insert_leg, insert_plan +from lib.hedge_plan.hedge_plan_monitor_lib import _tick_oo_expiry, _settle_orphaned_after_tp +from lib.hedge_plan.hedge_plan_notify_lib import ( + build_hedge_end_message, + build_hedge_start_message, + notify_plan_end, + notify_plan_start, +) +from lib.hedge_plan.hedge_plan_settle_lib import leg_is_expired, settle_option_leg_at_spot + + +def _mem_db(): + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + init_hedge_plan_tables(conn) + return conn + + +class TestHedgeNotify(unittest.TestCase): + def test_start_end_copy(self): + plan = { + "id": 7, + "plan_type": "perp_options", + "underlying": "ETH", + "direction": "long", + "entry_mark": 1800, + "tp": 1900, + "sl": 1700, + "perp_size": 2, + "leverage": 10, + "premium_total": 1.5, + "close_reason": "perp_tp", + "realized_pnl_total": 12.3, + "realized_pnl_perp": 15, + "realized_pnl_options": -1.5, + "opened_at": "2026-07-01 10:00:00", + "closed_at": "2026-07-01 12:00:00", + } + s = build_hedge_start_message(plan) + self.assertIn("启动 #7", s) + self.assertIn("永期", s) + e = build_hedge_end_message(plan) + self.assertIn("结束 #7", e) + self.assertIn("止盈", e) + + def test_idempotent_flags(self): + conn = _mem_db() + sent = [] + cfg = {"send_wechat": lambda c: sent.append(c)} + pid = insert_plan( + conn, + { + "plan_type": "options_options", + "status": "active", + "underlying": "ETH", + "target_price": 2000, + "premium_total": 2.0, + "opened_at": "t0", + }, + ) + plan = get_plan(conn, pid) + self.assertTrue(notify_plan_start(cfg, conn, plan, [])) + plan = get_plan(conn, pid) + self.assertEqual(int(plan["wechat_start_sent"]), 1) + self.assertFalse(notify_plan_start(cfg, conn, plan, [])) + self.assertEqual(len(sent), 1) + + plan["status"] = "closed" + plan["close_reason"] = "oo_expiry_loss" + plan["realized_pnl_total"] = -2 + plan["closed_at"] = "t1" + self.assertTrue(notify_plan_end(cfg, conn, plan)) + plan = get_plan(conn, pid) + self.assertEqual(int(plan["wechat_end_sent"]), 1) + self.assertFalse(notify_plan_end(cfg, conn, plan)) + self.assertEqual(len(sent), 2) + + +class TestHedgeSettle(unittest.TestCase): + def test_put_expiry_otm(self): + pnl = settle_option_leg_at_spot( + {"opt_type": "P", "strike": 1700, "size": 2, "premium": 1.2, "ct_mult": 0.01}, + spot=1800, + ) + self.assertAlmostEqual(pnl, -1.2) + + def test_call_expiry_itm(self): + # intrinsic (1900-1800)*2*0.01 - 0.5 = 2 - 0.5 + pnl = settle_option_leg_at_spot( + {"opt_type": "C", "strike": 1800, "size": 2, "premium": 0.5, "ct_mult": 0.01}, + spot=1900, + ) + self.assertAlmostEqual(pnl, 1.5) + + def test_leg_expired_from_inst(self): + # past date in inst_id + past = datetime.now(timezone.utc) - timedelta(days=3) + yy = past.year % 100 + tag = f"ETH-USD-{yy:02d}{past.month:02d}{past.day:02d}-1800-P" + self.assertTrue(leg_is_expired({"inst_id": tag})) + future = datetime.now(timezone.utc) + timedelta(days=10) + tag2 = f"ETH-USD-{future.year % 100:02d}{future.month:02d}{future.day:02d}-1800-P" + self.assertFalse(leg_is_expired({"inst_id": tag2})) + self.assertIsNotNone(expiry_ms_from_inst_id(tag)) + + +class TestHedgeMonitorExpiry(unittest.TestCase): + def test_oo_expiry_loss_closes_plan(self): + conn = _mem_db() + past = datetime.now(timezone.utc) - timedelta(days=1) + tag = f"ETH-USD-{past.year % 100:02d}{past.month:02d}{past.day:02d}-1800-P" + pid = insert_plan( + conn, + { + "plan_type": "options_options", + "status": "active", + "underlying": "ETH", + "target_price": 2000, + "premium_total": 2.0, + }, + ) + insert_leg( + conn, + { + "plan_id": pid, + "leg_role": "option_a", + "inst_id": tag, + "opt_type": "P", + "strike": 1800, + "size": 1, + "premium": 1.0, + "status": "open", + }, + ) + insert_leg( + conn, + { + "plan_id": pid, + "leg_role": "option_b", + "inst_id": tag.replace("-P", "-C").replace("1800", "1900"), + "opt_type": "C", + "strike": 1900, + "size": 1, + "premium": 1.0, + "status": "open", + }, + ) + sent = [] + cfg = { + "send_wechat": lambda c: sent.append(c), + "fetch_index_price": lambda ex, u: 1850.0, + "exchange_options": object(), + } + plan = get_plan(conn, pid) + legs = [ + dict(r) + for r in conn.execute("SELECT * FROM hedge_plan_legs WHERE plan_id=?", (pid,)).fetchall() + ] + r = _tick_oo_expiry(cfg, conn, plan, legs) + self.assertIsNotNone(r) + self.assertEqual(r["close_reason"], "oo_expiry_loss") + plan2 = get_plan(conn, pid) + self.assertEqual(plan2["status"], "closed") + self.assertLessEqual(float(plan2["realized_pnl_total"]), 0) + self.assertTrue(any("结束" in x for x in sent)) + + def test_orphaned_option_does_not_rewrite_plan_total(self): + conn = _mem_db() + past = datetime.now(timezone.utc) - timedelta(days=1) + tag = f"ETH-USD-{past.year % 100:02d}{past.month:02d}{past.day:02d}-1700-P" + pid = insert_plan( + conn, + { + "plan_type": "perp_options", + "status": "closed", + "underlying": "ETH", + "direction": "long", + "close_reason": "perp_tp", + "realized_pnl_total": 10.0, + "realized_pnl_options": -1.0, + "stats_bucket": "tp", + }, + ) + insert_leg( + conn, + { + "plan_id": pid, + "leg_role": "option_hedge", + "inst_id": tag, + "opt_type": "P", + "strike": 1700, + "size": 1, + "premium": 1.0, + "status": "hold_to_expiry", + "close_reason": "orphaned_after_tp", + }, + ) + cfg = { + "fetch_index_price": lambda ex, u: 1800.0, + "exchange_options": object(), + } + acted = _settle_orphaned_after_tp(cfg, conn) + self.assertEqual(len(acted), 1) + plan = get_plan(conn, pid) + self.assertAlmostEqual(float(plan["realized_pnl_total"]), 10.0) + leg = conn.execute("SELECT * FROM hedge_plan_legs WHERE plan_id=?", (pid,)).fetchone() + self.assertEqual(leg["status"], "closed") + self.assertEqual(leg["close_reason"], "expiry") + + +if __name__ == "__main__": + unittest.main()