crypto_monitor/lib/options/options_monitor_lib.py

"""期权持仓监控:浮盈翻倍微信提醒."""
from __future__ import annotations

import sqlite3
import time
from typing import Any, Callable


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 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,
    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,
                )
                conn.commit()
            finally:
                conn.close()
        except Exception:
            pass
        time.sleep(max(5.0, float(poll_seconds)))