"""期权持仓监控:浮盈翻倍微信提醒 + 平仓/到期状态同步.""" from __future__ import annotations import sqlite3 import time from datetime import datetime, timezone from typing import Any, Callable from lib.exchange.okx_options_lib import normalize_option_exp_ms, resolve_option_close_from_history 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, ) -> str: 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} USDC", f"未实现盈亏:{upl:+.4f} USDC({pct})", f"当前买一:{bid_txt}(可考虑限价平仓锁利)", ] ) 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} rows = conn.execute( """ SELECT id, inst_id, premium_paid, profit_alert_sent FROM options_trades WHERE status = 'open' """ ).fetchall() for row in rows: if int(row["profit_alert_sent"] or 0): continue inst_id = str(row["inst_id"] or "") prem = _safe_float(row["premium_paid"]) if not inst_id or prem is None or prem <= 0: continue pos = pos_by_inst.get(inst_id) if not pos: continue 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, ) try: send_wechat(msg) conn.execute( "UPDATE options_trades SET profit_alert_sent = 1 WHERE id = ?", (int(row["id"]),), ) sent += 1 except Exception: pass return sent def _created_at_ms(created_at: Any) -> int | None: if not created_at: return None raw = str(created_at).strip() if not raw: return None for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%f"): try: dt = datetime.strptime(raw[:26], fmt).replace(tzinfo=timezone.utc) return int(dt.timestamp() * 1000) except ValueError: continue return None def sync_open_options_trades( conn: sqlite3.Connection, *, live_inst_ids: set[str], fetch_history_fn: Callable[[str], list[dict[str, Any]]], ) -> int: """ 交易所已无持仓时,将本地 open 记录同步为 closed. 优先用 positions-history 回填盈亏;否则到期后按归零处理. """ rows = conn.execute( """ SELECT id, inst_id, premium_paid, exp_time, created_at FROM options_trades WHERE status = 'open' """ ).fetchall() updated = 0 now_ms = int(time.time() * 1000) 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") if realized_pnl is not None: prem_recv = round(paid + float(realized_pnl), 4) close_ms = close_info.get("close_ms") if close_ms: closed_at = datetime.fromtimestamp(int(close_ms) / 1000, tz=timezone.utc).strftime( "%Y-%m-%d %H:%M:%S" ) 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 = datetime.fromtimestamp(int(exp_ms) / 1000, tz=timezone.utc).strftime( "%Y-%m-%d %H:%M:%S" ) else: continue 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, CURRENT_TIMESTAMP), 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_reason, int(row["id"]), ), ) updated += 1 return updated 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, 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: conn = get_db() try: positions = fetch_positions() run_options_profit_alerts( conn, positions, profit_ratio=profit_ratio, send_wechat=send_wechat, account_label=account_label, ticker_bid_fn=ticker_bid_fn, ) if sync_trades_fn is not None: sync_trades_fn(conn) conn.commit() finally: conn.close() except Exception: pass time.sleep(max(5.0, float(poll_seconds)))