crypto_monitor/lib/options/options_monitor_lib.py

"""期权持仓监控:浮盈翻倍微信提醒 + 平仓/到期状态同步."""
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,
    target_close_fn: Callable[[str], dict[str, Any]] | None = None,
    stale_pending_fn: Callable[[], dict[str, Any]] | 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 target_close_fn is not None:
                    from lib.options.options_target_lib import run_options_target_closes

                    run_options_target_closes(
                        conn,
                        positions,
                        close_fn=target_close_fn,
                        send_wechat=send_wechat,
                        account_label=account_label,
                    )
                if sync_trades_fn is not None:
                    sync_trades_fn(conn)
                conn.commit()
            finally:
                conn.close()
            # 平仓限价挂单超时撤单(独立于 DB 事务)
            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)))