"""期权持仓监控:浮盈翻倍微信提醒 + 平仓/到期状态同步.""" from __future__ import annotations import sqlite3 import time from typing import Any, Callable from lib.exchange.okx_options_lib import normalize_option_exp_ms, resolve_option_close_from_history from lib.options.options_time_lib import ms_to_options_app_str, options_app_now_str, options_wall_clock_to_ms def _safe_float(v: Any) -> float | None: if v is None: return None try: return float(v) except (TypeError, ValueError): return None def build_profit_alert_message( *, account_label: str, inst_id: str, premium_paid: float, upl: float, upl_ratio: float | None, bid: float | None, premium_ccy: str = "USDC", ) -> str: from lib.options.options_notify_lib import resolve_premium_ccy ccy = resolve_premium_ccy(premium_ccy, inst_id=inst_id) pct = f"{upl_ratio * 100:.1f}%" if upl_ratio is not None else "—" bid_txt = f"{bid:.4f}" if bid is not None else "—" return "\n".join( [ "【OKX期权·翻倍提醒】", f"账户:{account_label}", f"合约:{inst_id}", f"已付权利金:{premium_paid:.4f} {ccy}", f"未实现盈亏:{upl:+.4f} {ccy}({pct})", f"当前买一:{bid_txt} {ccy}(可考虑限价平仓锁利)", ] ) def run_options_profit_alerts( conn: sqlite3.Connection, positions: list[dict[str, Any]], *, profit_ratio: float, send_wechat: Callable[[str], None], account_label: str, ticker_bid_fn: Callable[[str], float | None], ) -> int: """ 对比 DB 中 open 记录与交易所持仓;达到阈值发微信. 返回发送条数. """ sent = 0 pos_by_inst = {str(p.get("inst_id") or p.get("instId") or ""): p for p in positions} try: rows = conn.execute( """ SELECT id, inst_id, premium_paid, profit_alert_sent, premium_ccy FROM options_trades WHERE status = 'open' ORDER BY id ASC """ ).fetchall() except Exception: rows = conn.execute( """ SELECT id, inst_id, premium_paid, profit_alert_sent FROM options_trades WHERE status = 'open' ORDER BY id ASC """ ).fetchall() # 同合约多腿加仓:按合约汇总权利金,整仓只告警一次 by_inst: dict[str, dict[str, Any]] = {} for row in rows: inst_id = str(row["inst_id"] or "") if not inst_id: continue bucket = by_inst.setdefault( inst_id, {"ids": [], "premium": 0.0, "all_sent": True, "has_prem": False, "premium_ccy": None}, ) bucket["ids"].append(int(row["id"])) prem = _safe_float(row["premium_paid"]) if prem is not None: bucket["premium"] += float(prem) bucket["has_prem"] = True if not bucket.get("premium_ccy"): try: bucket["premium_ccy"] = row["premium_ccy"] except (KeyError, IndexError, TypeError): bucket["premium_ccy"] = None if not int(row["profit_alert_sent"] or 0): bucket["all_sent"] = False for inst_id, bucket in by_inst.items(): if bucket["all_sent"] or not bucket["has_prem"] or bucket["premium"] <= 0: continue pos = pos_by_inst.get(inst_id) if not pos: continue prem = float(bucket["premium"]) upl = _safe_float(pos.get("upl")) upl_ratio = _safe_float(pos.get("upl_ratio_pct")) if upl_ratio is not None: ratio = upl_ratio / 100.0 elif upl is not None: ratio = upl / prem else: continue if ratio < float(profit_ratio): continue bid = ticker_bid_fn(inst_id) msg = build_profit_alert_message( account_label=account_label, inst_id=inst_id, premium_paid=prem, upl=upl or 0.0, upl_ratio=ratio, bid=bid, premium_ccy=str(bucket.get("premium_ccy") or pos.get("premium_ccy") or ""), ) try: # 推送前先提交,避免握锁等 webhook try: conn.commit() except Exception: pass send_wechat(msg) conn.execute( f"UPDATE options_trades SET profit_alert_sent = 1 WHERE id IN ({','.join('?' * len(bucket['ids']))})", tuple(bucket["ids"]), ) try: conn.commit() except Exception: pass sent += 1 except Exception: pass return sent def _created_at_ms(created_at: Any) -> int | None: """墙钟 created_at → UTC ms.库内时间为业务时区(默认 Asia/Shanghai),不可当 UTC.""" return options_wall_clock_to_ms(created_at) def _group_key_for_closed_trade(row: Any) -> str: inst = str(row["inst_id"] or "").strip() closed = str(row["closed_at"] or "").strip() close_prefix = closed[:16] if closed else "" ord_id = str(row["close_ord_id"] or "").strip() if "close_ord_id" in row.keys() else "" # 即使 close_ord_id/posId 相同,也要按平仓时间拆开(OKX 可能复用 posId) if ord_id: return f"{inst}|ord:{ord_id}|close:{close_prefix}" return f"{inst}|close:{close_prefix}" def backfill_closed_options_realized_pnl_from_history( conn: sqlite3.Connection, hist_rows: list[dict[str, Any]], *, trade_limit: int = 200, ) -> int: """ 用 OKX positions-history 的 realizedPnl 覆盖本地已平记录. 同一次平仓多笔本地 open(加仓)按权利金占比分摊交易所总盈亏. """ by_inst: dict[str, list[dict[str, Any]]] = {} for raw in hist_rows or []: if not isinstance(raw, dict): continue inst = str(raw.get("instId") or "").strip() if not inst: continue by_inst.setdefault(inst, []).append(raw) rows = conn.execute( """ SELECT id, inst_id, sheets, premium_paid, realized_pnl, close_quote, created_at, closed_at, close_ord_id, realized_pnl_usdt, open_fee, close_fee, premium_ccy, underlying FROM options_trades WHERE status = 'closed' ORDER BY id DESC LIMIT ? """, (int(trade_limit),), ).fetchall() if not rows: return 0 from lib.options.options_pnl_settle_lib import ( resolve_close_index_px, settle_exchange_realized_pnl, ) groups: dict[str, list[Any]] = {} for row in rows: # 已冻结 U 的本地结算不再被交易所历史覆盖 if _safe_float(row["realized_pnl_usdt"]) is not None: continue inst = str(row["inst_id"] or "").strip() if not inst or inst not in by_inst: continue groups.setdefault(_group_key_for_closed_trade(row), []).append(row) updated = 0 for group in groups.values(): inst = str(group[0]["inst_id"] or "").strip() open_candidates = [_created_at_ms(r["created_at"]) for r in group] open_ms = min((x for x in open_candidates if x is not None), default=None) close_candidates = [_created_at_ms(r["closed_at"]) for r in group] close_ms = max((x for x in close_candidates if x is not None), default=None) sheets_hint = None try: sheets_hint = sum(float(_safe_float(r["sheets"]) or 0.0) for r in group) or None except (TypeError, ValueError): sheets_hint = None close_info = resolve_option_close_from_history( by_inst.get(inst) or [], open_ms=open_ms, close_ms=close_ms, sheets=sheets_hint, ) if not close_info: continue ex_pnl = _safe_float(close_info.get("realized_pnl")) if ex_pnl is None: continue close_quote = _safe_float(close_info.get("close_quote")) matched_pos = str(close_info.get("pos_id") or "").strip() or None total_paid = 0.0 for r in group: total_paid += float(_safe_float(r["premium_paid"]) or 0.0) underly = str(group[0]["underlying"] or "").strip() close_idx = resolve_close_index_px(None, inst_id=inst, underlying=underly) allocated = 0.0 for i, r in enumerate(group): paid = float(_safe_float(r["premium_paid"]) or 0.0) if i == len(group) - 1: share = round(float(ex_pnl) - allocated, 4) elif total_paid > 0: share = round(float(ex_pnl) * (paid / total_paid), 4) allocated += share else: share = round(float(ex_pnl) / len(group), 4) allocated += share local = _safe_float(r["realized_pnl"]) local_close = _safe_float(r["close_quote"]) local_ord = str(r["close_ord_id"] or "").strip() pnl_ok = local is not None and abs(local - share) < 1e-6 quote_ok = close_quote is None or ( local_close is not None and abs(local_close - float(close_quote)) < 1e-6 ) ord_ok = (not matched_pos) or (local_ord == matched_pos) if pnl_ok and quote_ok and ord_ok: continue # OKX realizedPnl 已含手续费,只冻结折 U,禁止再减 open_fee/close_fee prem_ccy = str(r["premium_ccy"] or "").strip() or "ETH" settled = settle_exchange_realized_pnl( realized_pnl=share, premium_ccy=prem_ccy, index_px=close_idx, ) # 展示用收回≈已付+净盈亏+已记手续费(毛额近似,不参与再扣费) ofee = float(_safe_float(r["open_fee"]) or 0.0) cfee = float(_safe_float(r["close_fee"]) or 0.0) prem_recv = round(paid + share + ofee + cfee, 4) conn.execute( """ UPDATE options_trades SET realized_pnl = ?, premium_received = ?, close_quote = COALESCE(?, close_quote), close_ord_id = COALESCE(?, close_ord_id), close_index_px = COALESCE(?, close_index_px), realized_pnl_usdt = COALESCE(?, realized_pnl_usdt) WHERE id = ? AND realized_pnl_usdt IS NULL """, ( settled.get("realized_pnl"), prem_recv, close_quote, matched_pos, settled.get("close_index_px"), settled.get("realized_pnl_usdt"), int(r["id"]), ), ) updated += 1 return updated def sync_open_options_trades( conn: sqlite3.Connection, *, live_inst_ids: set[str], fetch_history_fn: Callable[[str], list[dict[str, Any]]], notify_cfg: dict[str, Any] | None = None, ) -> int: """ 交易所已无持仓时,将本地 open 记录同步为 closed. 优先用 positions-history 回填盈亏;否则到期后按归零处理. """ rows = conn.execute( """ SELECT id, inst_id, premium_paid, exp_time, created_at, open_fee, premium_ccy, underlying, exchange_ord_id, budget_usdt, open_index_px, fee_ccy FROM options_trades WHERE status = 'open' """ ).fetchall() updated = 0 closed_inst_ids: list[str] = [] now_ms = int(time.time() * 1000) from lib.options.options_pnl_settle_lib import ( resolve_close_index_px, settle_closed_pnl, ) for row in rows: inst_id = str(row["inst_id"] or "") if not inst_id or inst_id in live_inst_ids: continue paid = _safe_float(row["premium_paid"]) or 0.0 open_ms = _created_at_ms(row["created_at"]) exp_ms = normalize_option_exp_ms(row["exp_time"], inst_id) close_quote: float | None = None prem_recv: float | None = None realized_pnl: float | None = None close_ord_id: str | None = None closed_at: str | None = None close_reason = "exchange" close_info = resolve_option_close_from_history( fetch_history_fn(inst_id), open_ms=open_ms, ) if close_info: close_quote = close_info.get("close_quote") realized_pnl = close_info.get("realized_pnl") close_ord_id = close_info.get("pos_id") close_ms = close_info.get("close_ms") if close_ms: closed_at = ms_to_options_app_str(close_ms) elif exp_ms is not None and now_ms >= int(exp_ms): close_reason = "expired" close_quote = 0.0 prem_recv = 0.0 realized_pnl = round(-paid, 4) if exp_ms: closed_at = ms_to_options_app_str(exp_ms) else: continue if not closed_at: closed_at = options_app_now_str() open_fee = float(_safe_float(row["open_fee"]) or 0.0) prem_ccy = str(row["premium_ccy"] or "").strip() or "ETH" underly = str(row["underlying"] or "").strip() close_idx = resolve_close_index_px(None, inst_id=inst_id, underlying=underly) budget = _safe_float(row["budget_usdt"]) open_idx = _safe_float(row["open_index_px"]) fee_ccy = str(row["fee_ccy"] or prem_ccy or "") if close_reason == "exchange" and realized_pnl is not None: # OKX realizedPnl 已含费 → 反推毛收回,再按开仓成本折 U(不再二次扣费) prem_recv = round(paid + float(realized_pnl) + open_fee, 4) settled = settle_closed_pnl( premium_paid=paid, premium_received=prem_recv, open_fee=open_fee, close_fee=0, premium_ccy=prem_ccy, index_px=close_idx, open_index_px=open_idx, budget_usdt=budget, fee_ccy=fee_ccy, ) else: if prem_recv is None: prem_recv = 0.0 settled = settle_closed_pnl( premium_paid=paid, premium_received=prem_recv, open_fee=open_fee, close_fee=0, premium_ccy=prem_ccy, index_px=close_idx, open_index_px=open_idx, budget_usdt=budget, fee_ccy=fee_ccy, ) realized_pnl = settled.get("realized_pnl") # BTC/ETH:真盈亏U与平仓价等卖回VWAP,禁止用指数占位 pc = str(prem_ccy or "").strip().upper() if pc in ("ETH", "BTC"): pnl_usdt = None close_px_store = None else: pnl_usdt = settled.get("realized_pnl_usdt") close_px_store = settled.get("close_index_px") conn.execute( """ UPDATE options_trades SET status = 'closed', close_quote = ?, premium_received = ?, realized_pnl = ?, close_ord_id = COALESCE(?, close_ord_id), closed_at = COALESCE(?, closed_at), close_index_px = COALESCE(?, close_index_px), open_index_px = COALESCE(open_index_px, ?), realized_pnl_usdt = COALESCE(?, realized_pnl_usdt), fee_ccy = COALESCE(fee_ccy, ?), signal_note = CASE WHEN ? = 'expired' AND (signal_note IS NULL OR TRIM(signal_note) = '') THEN '到期结算' ELSE signal_note END WHERE id = ? """, ( close_quote, prem_recv, realized_pnl, close_ord_id, closed_at, close_px_store, settled.get("open_index_px"), pnl_usdt, prem_ccy, close_reason, int(row["id"]), ), ) updated += 1 closed_inst_ids.append(inst_id) if notify_cfg is not None: try: from lib.options.options_notify_lib import notify_options_close reason = "到期结算" if close_reason == "expired" else "交易所平仓" notify_options_close( notify_cfg, conn, inst_id=inst_id, reason=reason, trade_id=int(row["id"]), premium_paid=paid, premium_received=prem_recv, realized_pnl=realized_pnl, close_quote=close_quote, ) except Exception: pass if notify_cfg is not None: _notify_recent_unsent_expiry_closes(conn, notify_cfg) # 卖币不在此处做:调用方须先 commit/close 本连接,再调 sell_spot_after_options_sync, # 否则监控线程握锁时中控快照会超时,表现为全部离线。 return updated def sell_spot_after_options_sync( cfg: dict[str, Any] | None, closed_inst_ids: list[str] | None = None, ) -> None: """在释放业务 DB 连接后调用:同步关仓/到期后的币本位卖回.""" if not isinstance(cfg, dict): return ex = cfg.get("exchange_options") if ex is None: return try: from lib.options.options_coin_open_lib import ( maybe_sell_spot_after_close, sweep_spot_after_options_flat, ) seen: set[str] = set() for inst_id in closed_inst_ids or []: iid = str(inst_id or "").strip() if not iid or iid in seen: continue seen.add(iid) try: maybe_sell_spot_after_close(cfg, ex, inst_id=iid) except Exception: pass sweep_spot_after_options_flat(cfg, ex) except Exception: pass def _notify_recent_unsent_expiry_closes( conn: sqlite3.Connection, notify_cfg: dict[str, Any], *, within_hours: float = 48.0, ) -> None: """页面同步曾漏推时,补发近期到期结算.""" try: from lib.options.options_notify_lib import notify_options_close_trade_ids from lib.options.options_time_lib import options_wall_clock_to_ms now_ms = int(time.time() * 1000) by_inst: dict[str, list[int]] = {} rows = conn.execute( """ SELECT id, inst_id, closed_at FROM options_trades WHERE status = 'closed' AND COALESCE(wechat_close_sent, 0) = 0 AND COALESCE(signal_note, '') = '到期结算' ORDER BY id DESC LIMIT 20 """ ).fetchall() for r in rows: closed_ms = options_wall_clock_to_ms(r["closed_at"]) if closed_ms is None or now_ms - int(closed_ms) > float(within_hours) * 3600 * 1000: continue inst = str(r["inst_id"] or "") by_inst.setdefault(inst, []).append(int(r["id"])) for ids in by_inst.values(): notify_options_close_trade_ids(notify_cfg, conn, ids, reason="到期结算") except Exception: pass def reconcile_live_open_trades( conn: sqlite3.Connection, *, live_inst_ids: set[str], ) -> int: """交易所有持仓但本地误标 closed 时恢复为 open.""" fixed = 0 for inst_id in live_inst_ids: if not inst_id: continue open_row = conn.execute( "SELECT id FROM options_trades WHERE inst_id = ? AND status = 'open' LIMIT 1", (inst_id,), ).fetchone() if open_row: continue row = conn.execute( """ SELECT id, close_ord_id, realized_pnl FROM options_trades WHERE inst_id = ? AND status = 'closed' ORDER BY id DESC LIMIT 1 """, (inst_id,), ).fetchone() if not row: continue if row["close_ord_id"]: continue if row["realized_pnl"] is not None: continue conn.execute( """ UPDATE options_trades SET status = 'open', close_quote = NULL, premium_received = NULL, realized_pnl = NULL, closed_at = NULL, signal_note = CASE WHEN signal_note = '到期结算' THEN NULL ELSE signal_note END WHERE id = ? """, (int(row["id"]),), ) fixed += 1 return fixed def options_monitor_loop( *, enabled: bool, poll_seconds: float, get_db: Callable[[], sqlite3.Connection], fetch_positions: Callable[[], list[dict[str, Any]]], ticker_bid_fn: Callable[[str], float | None], send_wechat: Callable[[str], None], account_label: str, profit_ratio: float, sync_trades_fn: Callable[[sqlite3.Connection], int] | None = None, target_close_fn: Callable[[str], dict[str, Any]] | None = None, profit_exit_close_fn: Callable[[str], dict[str, Any]] | None = None, profit_exit_cfg: dict[str, Any] | None = None, stale_pending_fn: Callable[[], dict[str, Any]] | None = None, cond_order_tick_fn: Callable[[sqlite3.Connection], None] | None = None, stop_event: Any = None, ) -> None: if not enabled: return while True: if stop_event is not None and getattr(stop_event, "is_set", lambda: False)(): break try: # 持仓拉取不握 DB,避免与模拟盘钱包写互相锁 try: 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: try: cond_order_tick_fn(None) except Exception: pass conn = get_db() try: run_options_profit_alerts( conn, positions, profit_ratio=profit_ratio, send_wechat=send_wechat, account_label=account_label, ticker_bid_fn=ticker_bid_fn, ) conn.commit() except Exception: pass finally: try: conn.close() except Exception: pass # 目标/翻倍出场:传 get_db,平仓时不握业务连接 if target_close_fn is not None: from lib.options.options_target_lib import run_options_target_closes try: run_options_target_closes( get_db, positions, close_fn=target_close_fn, send_wechat=send_wechat, account_label=account_label, cfg=pe_cfg, ) except Exception: pass if profit_exit_close_fn is not None: from lib.options.options_profit_exit_lib import run_options_profit_exits try: run_options_profit_exits( get_db, positions, close_fn=profit_exit_close_fn, send_wechat=send_wechat, account_label=account_label, cfg=pe_cfg, ex=pe_cfg.get("exchange_options"), ) except Exception: pass if sync_trades_fn is not None: try: sync_trades_fn(None) except Exception: pass try: sell_spot_after_options_sync(pe_cfg) except Exception: pass if stale_pending_fn is not None: try: stale_pending_fn() except Exception: pass except Exception: pass time.sleep(max(5.0, float(poll_seconds)))