crypto_monitor/lib/options/options_register.py

"""OKX 期权模块:Flask 路由注册."""
from __future__ import annotations

import os
import threading
import time
from typing import Any

from flask import Flask, jsonify, redirect, request, url_for
from jinja2 import ChoiceLoader, FileSystemLoader

from lib.options.options_db import init_options_tables
from lib.options.options_monitor_lib import options_monitor_loop
from lib.options.options_close_gate_lib import clear_close_gate, update_close_gate
from lib.options.options_pricing_lib import (
    calc_order_size,
    close_ref_prices,
    ct_mult_from_meta,
    estimate_close_by_bids,
    fetch_option_mark_px,
    filter_bids_for_close,
    is_stub_bid_px,
    min_sz_from_meta,
    premium_per_sheet,
    total_premium,
)
from lib.exchange.okx_options_lib import _pos_side_from_position, _safe_float, td_mode_for_option_buy


def _env_bool(key: str, default: bool = False) -> bool:
    raw = (os.getenv(key) or "").strip().lower()
    if not raw:
        return default
    return raw in ("1", "true", "yes", "on")


def _env_float(key: str, default: float) -> float:
    try:
        return float(os.getenv(key, str(default)))
    except (TypeError, ValueError):
        return default


def attach_options_templates(app: Flask, repo_root: str) -> None:
    tpl_dir = os.path.join(repo_root, "lib", "options", "templates")
    if not os.path.isdir(tpl_dir):
        return
    existing = app.jinja_loader
    loaders = [FileSystemLoader(tpl_dir)]
    if existing is not None:
        if isinstance(existing, ChoiceLoader):
            loaders = list(existing.loaders) + loaders
        else:
            loaders.insert(0, existing)
    app.jinja_loader = ChoiceLoader(loaders)


def install_options_trading(app: Flask, repo_root: str, app_module: Any) -> None:
    enabled = _env_bool("OKX_OPTIONS_ENABLED", False)
    attach_options_templates(app, repo_root)
    cfg = _build_cfg(app_module)
    app.extensions["options_cfg"] = cfg
    register_options_routes(app, cfg)
    _register_options_hub_bridge(app, cfg)
    if enabled:
        _start_monitor_thread(app, cfg)


def _register_options_hub_bridge(app: Flask, cfg: dict[str, Any]) -> None:
    from lib.options.options_hub_lib import build_options_hub_snapshot

    def snapshot_fn():
        return build_options_hub_snapshot(cfg)

    hub_ctx = dict(app.config.get("HUB_CTX") or {})
    hub_ctx["options_snapshot_fn"] = snapshot_fn
    app.config["HUB_CTX"] = hub_ctx


def _build_cfg(app_module: Any) -> dict[str, Any]:
    from lib.exchange.okx_options_lib import (
        build_option_chain,
        estimate_usdt_to_usdc,
        execute_convert,
        fetch_option_book_depth,
        fetch_option_positions,
        fetch_options_balances,
        format_position_row,
        options_api_ready,
        cancel_option_order,
        fetch_option_pending_orders,
        place_option_limit_order,
        place_option_market_order,
        quote_option_contract,
        spot_market_swap_usdt_usdc,
        transfer_ccy,
        transfer_main_sub_account,
    )

    return {
        "enabled": _env_bool("OKX_OPTIONS_ENABLED", False),
        "sub_account_name": (os.getenv("OKX_SUB_ACCOUNT_NAME") or "").strip(),
        "get_db": app_module.get_db,
        "login_required": app_module.login_required,
        "exchange_options": getattr(app_module, "exchange_options", None),
        "send_wechat": app_module.send_wechat_msg,
        "render_main_page": app_module.render_main_page,
        "trade_budget": _env_float("OKX_OPTIONS_TRADE_BUDGET_USDC", 10.0),
        "budget_buffer": _env_float("OKX_OPTIONS_BUDGET_BUFFER", 0.95),
        "default_underly": (os.getenv("OKX_OPTIONS_DEFAULT_UNDERLY") or "ETH").strip().upper(),
        "max_dte_days": _env_float("OKX_OPTIONS_MAX_DTE_DAYS", 2.0),
        "chain_max_dte_days": _env_float("OKX_OPTIONS_CHAIN_MAX_DTE_DAYS", 14.0),
        "itm_max_dist": _env_float("OKX_OPTIONS_ITM_MAX_DIST_USD", 30.0),
        "td_mode": (os.getenv("OKX_OPTIONS_TD_MODE") or "isolated").strip(),
        "allow_market_close": _env_bool("OKX_OPTIONS_ALLOW_MARKET_CLOSE", False),
        "profit_ratio": _env_float("OKX_OPTIONS_PROFIT_ALERT_RATIO", 1.0),
        "poll_seconds": _env_float("OKX_OPTIONS_POLL_SECONDS", 15.0),
        "account_label": (os.getenv("OKX_OPTIONS_ACCOUNT_LABEL") or "OKX期权").strip(),
        "build_option_chain": build_option_chain,
        "quote_option_contract": quote_option_contract,
        "fetch_option_book_depth": fetch_option_book_depth,
        "place_option_limit_order": place_option_limit_order,
        "place_option_market_order": place_option_market_order,
        "fetch_option_pending_orders": fetch_option_pending_orders,
        "cancel_option_order": cancel_option_order,
        "fetch_option_positions": fetch_option_positions,
        "fetch_options_balances": fetch_options_balances,
        "format_position_row": format_position_row,
        "estimate_usdt_to_usdc": estimate_usdt_to_usdc,
        "execute_convert": execute_convert,
        "transfer_ccy": transfer_ccy,
        "spot_market_swap_usdt_usdc": spot_market_swap_usdt_usdc,
        "transfer_main_sub_account": transfer_main_sub_account,
        "options_api_ready": options_api_ready,
        "app_module": app_module,
    }


def _mark_balances_stale(cfg: dict[str, Any]) -> None:
    from lib.exchange.okx_options_lib import invalidate_options_balance_cache
    from lib.instance.instance_live_push_lib import notify_instance_balance_changed

    invalidate_options_balance_cache()
    app_mod = cfg.get("app_module")
    if app_mod is not None and hasattr(app_mod, "invalidate_account_balance_cache"):
        app_mod.invalidate_account_balance_cache()
    try:
        notify_instance_balance_changed()
    except Exception:
        pass


def _require_options_ex(cfg: dict[str, Any]):
    if not cfg.get("enabled"):
        return None, "期权模块未启用,请在 .env 设置 OKX_OPTIONS_ENABLED=true 并重启 PM2"
    ex = cfg.get("exchange_options")
    ok, reason = cfg["options_api_ready"](ex)
    if not ok:
        return None, reason or "期权 API 未配置"
    return ex, ""


def _budget_full_usdc(cfg: dict[str, Any], ex: Any) -> tuple[float | None, str]:
    """交易账户 USDC 可用余额(由 calc_order_size 再乘 budget_buffer 留余量)."""
    from lib.exchange.okx_options_lib import fetch_options_trading_usdc

    raw = fetch_options_trading_usdc(ex)
    if raw is None or float(raw) <= 0:
        return None, "交易账户 USDC 可用余额不足"
    return float(raw), ""


def _open_premium_paid(cfg: dict[str, Any], inst_id: str) -> float | None:
    conn = cfg["get_db"]()
    try:
        init_options_tables(conn)
        rec = conn.execute(
            """
            SELECT premium_paid FROM options_trades
            WHERE inst_id = ? AND status = 'open'
            ORDER BY id DESC LIMIT 1
            """,
            (inst_id,),
        ).fetchone()
        if rec and rec["premium_paid"] is not None:
            return round(float(rec["premium_paid"]), 4)
    finally:
        conn.close()
    return None


def _position_avail_sheets(pos: dict[str, Any]) -> int:
    avail = _safe_float(pos.get("availPos"))
    if avail is None or avail <= 0:
        avail = abs(_safe_float(pos.get("pos")) or 0)
    return max(0, int(avail or 0))


def _find_position(rows: list[dict[str, Any]] | None, inst_id: str) -> dict[str, Any] | None:
    return next((p for p in rows or [] if str(p.get("instId")) == inst_id), None)


def _refresh_position_avail(cfg: dict[str, Any], ex: Any, inst_id: str) -> int | None:
    from lib.exchange.okx_options_lib import invalidate_option_positions_cache

    invalidate_option_positions_cache()
    raw = cfg["fetch_option_positions"](ex)
    if raw is None:
        return None
    pos = _find_position(raw, inst_id)
    if not pos:
        return 0
    return _position_avail_sheets(pos)


def _enrich_position_row_display(
    cfg: dict[str, Any],
    ex: Any,
    raw_pos: dict[str, Any],
    *,
    meta_cache: dict[str, dict[str, Any] | None] | None = None,
    premium_override: float | None = None,
) -> dict[str, Any]:
    from lib.options.options_history_lib import enrich_position_row_display

    return enrich_position_row_display(
        cfg,
        ex,
        raw_pos,
        meta_cache=meta_cache,
        premium_override=premium_override,
    )


def _attach_close_preview(
    cfg: dict[str, Any],
    ex: Any,
    row: dict[str, Any],
    *,
    sheets: int | None = None,
    premium_paid: float | None = None,
) -> dict[str, Any]:
    from lib.options.options_positions_lib import attach_close_preview

    return attach_close_preview(
        cfg,
        ex,
        row,
        sheets=sheets,
        premium_paid=premium_paid,
    )


_OPTIONS_SYNC_LOCK = threading.Lock()
_OPTIONS_SYNC_LAST_AT = 0.0
_OPTIONS_SYNC_INTERVAL_SEC = 15.0


def _sync_options_trades(
    cfg: dict[str, Any],
    *,
    raw_positions: list[dict[str, Any]] | None = None,
    force: bool = False,
) -> None:
    global _OPTIONS_SYNC_LAST_AT
    ex = cfg.get("exchange_options")
    if ex is None:
        return
    now = time.time()
    with _OPTIONS_SYNC_LOCK:
        if not force and now - _OPTIONS_SYNC_LAST_AT < _OPTIONS_SYNC_INTERVAL_SEC:
            return
        _OPTIONS_SYNC_LAST_AT = now
    from lib.exchange.okx_options_lib import fetch_option_position_history
    from lib.options.options_monitor_lib import reconcile_live_open_trades, sync_open_options_trades

    if raw_positions is None:
        raw = cfg["fetch_option_positions"](ex)
        if raw is None:
            return
    else:
        raw = raw_positions
    live_ids = {str(p.get("instId") or "") for p in raw if str(p.get("instId") or "")}

    def _hist(inst_id: str):
        return fetch_option_position_history(ex, inst_id)

    conn = cfg["get_db"]()
    try:
        init_options_tables(conn)
        reconcile_live_open_trades(conn, live_inst_ids=live_ids)
        sync_open_options_trades(conn, live_inst_ids=live_ids, fetch_history_fn=_hist)
        conn.commit()
    finally:
        conn.close()


def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
    lr = cfg["login_required"]

    @app.route("/api/options/balances")
    @lr
    def api_options_balances():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        force = (request.args.get("force") or "").strip().lower() in ("1", "true", "yes")
        scope = (request.args.get("scope") or "main").strip().lower()
        bal = cfg["fetch_options_balances"](
            ex,
            force=force,
            scope=scope,
            sub_acct=cfg.get("sub_account_name") or "",
        )
        return jsonify({"ok": True, **bal, "trade_budget": cfg["trade_budget"]})

    @app.route("/api/options/chain")
    @lr
    def api_options_chain():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        u = (request.args.get("underlying") or cfg["default_underly"]).upper()
        chain = cfg["build_option_chain"](
            ex,
            u,
            max_dte_days=cfg["chain_max_dte_days"],
            itm_only=False,
            itm_max_dist_usd=cfg["itm_max_dist"],
        )
        return jsonify({"ok": True, **chain, "chain_max_dte_days": cfg["chain_max_dte_days"]})

    @app.route("/api/options/quote")
    @lr
    def api_options_quote():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        inst_id = (request.args.get("inst_id") or "").strip()
        if not inst_id:
            return jsonify({"ok": False, "msg": "缺少 inst_id"})
        q = cfg["quote_option_contract"](ex, inst_id)
        if not q.get("ok"):
            return jsonify(q)
        ask = q.get("ask")
        ct_mult = q.get("ct_mult") or 0.01
        min_sz = q.get("min_sz") or 1
        mode = (request.args.get("mode") or "budget_full").strip()
        sheet_count = None
        try:
            if request.args.get("sheets"):
                sheet_count = int(request.args.get("sheets"))
        except (TypeError, ValueError):
            pass
        if mode == "close_preview":
            paid = _open_premium_paid(cfg, inst_id)
            target = sheet_count if sheet_count is not None else 0
            return jsonify(_attach_close_preview(cfg, ex, {**q, "pos": target, "premium_paid": paid}, sheets=target, premium_paid=paid))
        budget = cfg["trade_budget"]
        budget_cap = cfg["trade_budget"]
        available_usdc = None
        if mode == "budget_full":
            budget, budget_err = _budget_full_usdc(cfg, ex)
            if budget is None:
                return jsonify({"ok": False, "msg": budget_err})
            budget_cap = budget
            from lib.exchange.okx_options_lib import fetch_options_trading_usdc

            available_usdc = fetch_options_trading_usdc(ex)
        eth_amount = None
        try:
            if request.args.get("eth_amount"):
                eth_amount = float(request.args.get("eth_amount"))
        except (TypeError, ValueError):
            pass
        if ask is None or ask <= 0:
            return jsonify({**q, "ok": False, "msg": "暂无卖一价"})
        sizing = calc_order_size(
            quote_per_unit=float(ask),
            ct_mult=float(ct_mult),
            min_sz=int(min_sz),
            budget_usdc=budget if mode == "budget_full" else None,
            budget_buffer=cfg["budget_buffer"],
            eth_amount=eth_amount if mode == "eth_amount" else None,
            sheets=sheet_count if mode == "sheets" else None,
            budget_cap=budget_cap if mode in ("budget_full", "sheets", "eth_amount") else None,
        )
        q = _attach_close_preview(
            cfg,
            ex,
            q,
            sheets=int(sizing.get("sheets") or sheet_count or 0),
            premium_paid=_open_premium_paid(cfg, inst_id),
        )
        return jsonify(
            {
                **q,
                "quote_per_unit": ask,
                "premium_per_sheet": premium_per_sheet(float(ask), float(ct_mult)),
                "sizing": sizing,
                "available_usdc": available_usdc,
                "budget_full_usdc": budget if mode == "budget_full" else None,
            }
        )

    @app.route("/api/options/open", methods=["POST"])
    @lr
    def api_options_open():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        data = request.get_json(silent=True) or {}
        inst_id = (data.get("inst_id") or "").strip()
        mode = (data.get("mode") or "budget_full").strip()
        signal_note = (data.get("signal_note") or "").strip()
        target_index = None
        raw_target = data.get("target_index")
        if raw_target is not None and str(raw_target).strip() != "":
            try:
                target_index = float(raw_target)
            except (TypeError, ValueError):
                return jsonify({"ok": False, "msg": "目标位无效"})
            if target_index <= 0:
                return jsonify({"ok": False, "msg": "目标位无效"})
        if not inst_id:
            return jsonify({"ok": False, "msg": "缺少 inst_id"})
        q = cfg["quote_option_contract"](ex, inst_id)
        if not q.get("ok"):
            return jsonify(q)
        ask = q.get("ask")
        if ask is None or ask <= 0:
            return jsonify({"ok": False, "msg": "暂无卖一价,无法买入"})
        ct_mult = float(q.get("ct_mult") or 0.01)
        min_sz = int(q.get("min_sz") or 1)
        eth_amount = None
        sheet_count = None
        if mode == "eth_amount":
            try:
                eth_amount = float(data.get("eth_amount"))
            except (TypeError, ValueError):
                return jsonify({"ok": False, "msg": "ETH 数量无效"})
        elif mode == "sheets":
            try:
                sheet_count = int(data.get("sheets"))
            except (TypeError, ValueError):
                return jsonify({"ok": False, "msg": "张数无效"})
        budget = cfg["trade_budget"]
        budget_cap = cfg["trade_budget"]
        if mode == "budget_full":
            budget, budget_err = _budget_full_usdc(cfg, ex)
            if budget is None:
                return jsonify({"ok": False, "msg": budget_err})
            budget_cap = budget
        sizing = calc_order_size(
            quote_per_unit=float(ask),
            ct_mult=ct_mult,
            min_sz=min_sz,
            budget_usdc=budget if mode == "budget_full" else None,
            budget_buffer=cfg["budget_buffer"],
            eth_amount=eth_amount,
            sheets=sheet_count,
            budget_cap=budget_cap if mode in ("budget_full", "sheets", "eth_amount") else None,
        )
        if not sizing.get("ok"):
            return jsonify({"ok": False, "msg": sizing.get("msg") or "张数计算失败", "sizing": sizing})
        sheets = int(sizing["sheets"])
        tick_sz = q.get("tick_sz")
        order = cfg["place_option_limit_order"](
            ex,
            inst_id=inst_id,
            side="buy",
            sheets=sheets,
            price=float(ask),
            td_mode=td_mode_for_option_buy(cfg["td_mode"]),
            tick_sz=tick_sz,
        )
        if not order.get("ok"):
            return jsonify(order)
        conn = cfg["get_db"]()
        trade_id = None
        target_mon = None
        try:
            init_options_tables(conn)
            meta = q.get("meta") or {}
            u = str(meta.get("uly") or inst_id).split("-")[0]
            opt_type = meta.get("optType")
            cur = conn.execute(
                """
                INSERT INTO options_trades
                (inst_id, underlying, opt_type, strike, exp_time, sheets, eth_amount,
                 open_quote, premium_paid, status, signal_note, exchange_ord_id)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?)
                """,
                (
                    inst_id,
                    u,
                    opt_type,
                    q.get("strike"),
                    str(q.get("exp_time") or ""),
                    sheets,
                    sizing["eth_amount"],
                    float(ask),
                    sizing["total_premium"],
                    signal_note,
                    (order.get("data") or {}).get("ordId"),
                ),
            )
            trade_id = int(cur.lastrowid)
            if target_index is not None:
                from lib.options.options_target_lib import upsert_target_monitor

                target_mon = upsert_target_monitor(
                    conn,
                    inst_id=inst_id,
                    target_index=target_index,
                    underlying=u,
                    opt_type=str(opt_type) if opt_type else None,
                    trade_id=trade_id,
                    sheets=sheets,
                )
            conn.commit()
        finally:
            conn.close()
        from lib.exchange.okx_options_lib import invalidate_option_positions_cache

        invalidate_option_positions_cache()
        _sync_options_trades(cfg, force=True)
        return jsonify(
            {
                "ok": True,
                "order": order,
                "sizing": sizing,
                "trade_id": trade_id,
                "target_monitor": target_mon,
            }
        )

    @app.route("/api/options/orders/pending")
    @lr
    def api_options_orders_pending():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        inst_id = (request.args.get("inst_id") or "").strip() or None
        try:
            orders = cfg["fetch_option_pending_orders"](ex, inst_id)
        except Exception as e:
            return jsonify({"ok": False, "msg": f"获取委托失败: {e}"})
        return jsonify({"ok": True, "orders": orders, "count": len(orders)})

    @app.route("/api/options/orders/cancel", methods=["POST"])
    @lr
    def api_options_orders_cancel():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        data = request.get_json(silent=True) or {}
        inst_id = (data.get("inst_id") or "").strip()
        ord_id = (data.get("ord_id") or "").strip()
        if not inst_id or not ord_id:
            return jsonify({"ok": False, "msg": "缺少 inst_id 或 ord_id"})
        out = cfg["cancel_option_order"](ex, inst_id=inst_id, ord_id=ord_id)
        if out.get("ok"):
            from lib.exchange.okx_options_lib import invalidate_option_positions_cache

            invalidate_option_positions_cache()
            # 本地未成交开仓记录标记取消,避免假 open
            try:
                conn = cfg["get_db"]()
                try:
                    init_options_tables(conn)
                    conn.execute(
                        """
                        UPDATE options_trades
                        SET status = 'cancelled',
                            signal_note = CASE
                                WHEN signal_note IS NULL OR TRIM(signal_note) = '' THEN '委托撤销'
                                ELSE signal_note
                            END,
                            closed_at = CURRENT_TIMESTAMP
                        WHERE inst_id = ? AND exchange_ord_id = ? AND status = 'open'
                        """,
                        (inst_id, ord_id),
                    )
                    conn.commit()
                finally:
                    conn.close()
            except Exception:
                pass
            _sync_options_trades(cfg, force=True)
        return jsonify(out), (200 if out.get("ok") else 400)

    @app.route("/api/options/positions")
    @lr
    def api_options_positions():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        raw = cfg["fetch_option_positions"](ex)
        if raw is None:
            return jsonify({"ok": False, "msg": "获取期权持仓失败"})
        _sync_options_trades(cfg, raw_positions=raw)
        meta_cache: dict[str, dict[str, Any] | None] = {}
        conn = cfg["get_db"]()
        try:
            from lib.options.options_target_lib import targets_by_inst

            tgt_map = targets_by_inst(conn)
            rows = []
            for p in raw:
                inst = str(p.get("instId") or "").strip()
                premium_override = None
                if inst:
                    rec = conn.execute(
                        """
                        SELECT premium_paid FROM options_trades
                        WHERE inst_id = ? AND status = 'open'
                        ORDER BY id DESC LIMIT 1
                        """,
                        (inst,),
                    ).fetchone()
                    if rec and rec["premium_paid"] is not None:
                        premium_override = float(rec["premium_paid"])
                row = _enrich_position_row_display(
                    cfg,
                    ex,
                    p,
                    meta_cache=meta_cache,
                    premium_override=premium_override,
                )
                _attach_close_preview(cfg, ex, row, premium_paid=_safe_float(row.get("premium_paid")))
                mon = tgt_map.get(inst)
                if mon:
                    row["target_index"] = mon.get("target_index")
                    row["target_monitor_id"] = mon.get("id")
                    row["target_monitor"] = mon
                rows.append(row)
        finally:
            conn.close()
        return jsonify({"ok": True, "positions": rows})

    @app.route("/api/options/targets")
    @lr
    def api_options_targets():
        conn = cfg["get_db"]()
        try:
            from lib.options.options_target_lib import list_active_targets, list_closing_targets

            return jsonify({"ok": True, "targets": list_active_targets(conn) + list_closing_targets(conn)})
        finally:
            conn.close()

    @app.route("/api/options/target", methods=["POST"])
    @lr
    def api_options_target_set():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        data = request.get_json(silent=True) or {}
        inst_id = (data.get("inst_id") or "").strip()
        if not inst_id:
            return jsonify({"ok": False, "msg": "缺少 inst_id"})
        try:
            target_index = float(data.get("target_index"))
        except (TypeError, ValueError):
            return jsonify({"ok": False, "msg": "目标位无效"})
        if target_index <= 0:
            return jsonify({"ok": False, "msg": "目标位无效"})
        raw = cfg["fetch_option_positions"](ex)
        if raw is None:
            return jsonify({"ok": False, "msg": "获取期权持仓失败"})
        pos = _find_position(raw, inst_id)
        if not pos:
            return jsonify({"ok": False, "msg": "未找到持仓"})
        from lib.options.options_target_lib import upsert_target_monitor

        fmt = cfg["format_position_row"](pos)
        conn = cfg["get_db"]()
        try:
            trade = conn.execute(
                """
                SELECT id, sheets, opt_type, underlying FROM options_trades
                WHERE inst_id = ? AND status = 'open'
                ORDER BY id DESC LIMIT 1
                """,
                (inst_id,),
            ).fetchone()
            trade_id = int(trade["id"]) if trade else None
            sheets = int(trade["sheets"]) if trade and trade["sheets"] is not None else int(fmt.get("pos") or 0)
            opt_type = (trade["opt_type"] if trade else None) or fmt.get("opt_type")
            underlying = (trade["underlying"] if trade else None) or fmt.get("underlying")
            out = upsert_target_monitor(
                conn,
                inst_id=inst_id,
                target_index=target_index,
                underlying=str(underlying) if underlying else None,
                opt_type=str(opt_type) if opt_type else None,
                trade_id=trade_id,
                sheets=sheets,
            )
            conn.commit()
            return jsonify(out)
        finally:
            conn.close()

    @app.route("/api/options/target/cancel", methods=["POST"])
    @lr
    def api_options_target_cancel():
        data = request.get_json(silent=True) or {}
        inst_id = (data.get("inst_id") or "").strip() or None
        monitor_id = data.get("id")
        try:
            mid = int(monitor_id) if monitor_id is not None and str(monitor_id).strip() != "" else None
        except (TypeError, ValueError):
            return jsonify({"ok": False, "msg": "监控 id 无效"})
        if not inst_id and mid is None:
            return jsonify({"ok": False, "msg": "缺少 inst_id 或 id"})
        from lib.options.options_target_lib import cancel_target_monitor

        conn = cfg["get_db"]()
        try:
            n = cancel_target_monitor(conn, inst_id=inst_id, monitor_id=mid)
            conn.commit()
            return jsonify({"ok": True, "cancelled": n})
        finally:
            conn.close()

    @app.route("/api/options/close", methods=["POST"])
    @lr
    def api_options_close():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        data = request.get_json(silent=True) or {}
        inst_id = (data.get("inst_id") or "").strip()
        use_market = bool(data.get("market")) and cfg["allow_market_close"]
        close_mode = (data.get("mode") or "").strip()
        depth_split = close_mode == "depth_split" and not use_market
        if not inst_id:
            return jsonify({"ok": False, "msg": "缺少 inst_id"})
        sheets = data.get("sheets")
        q = cfg["quote_option_contract"](ex, inst_id)
        bid = q.get("bid")
        if not use_market and not depth_split and (bid is None or bid <= 0):
            return jsonify({"ok": False, "msg": "暂无买一价,无法限价平仓"})
        raw_positions = cfg["fetch_option_positions"](ex)
        if raw_positions is None:
            return jsonify({"ok": False, "msg": "获取期权持仓失败"})
        pos = _find_position(raw_positions, inst_id)
        if not pos:
            return jsonify({"ok": False, "msg": "未找到持仓"})
        avail = _position_avail_sheets(pos)
        close_sheets = int(sheets) if sheets else int(avail)
        close_sheets = min(close_sheets, int(avail))
        if close_sheets < 1:
            return jsonify({"ok": False, "msg": "可平张数不足"})
        td_mode = str(pos.get("mgnMode") or cfg["td_mode"])
        pos_side = _pos_side_from_position(pos) or "net"
        tick_sz = q.get("tick_sz")
        if use_market:
            order = cfg["place_option_market_order"](
                ex,
                inst_id=inst_id,
                side="sell",
                sheets=close_sheets,
                td_mode=td_mode,
                reduce_only=True,
                pos_side=pos_side,
            )
            if not order.get("ok"):
                return jsonify(order)
        elif depth_split:
            ct_mult = float(q.get("ct_mult") or 0.01)
            from lib.exchange.okx_options_lib import option_fields_from_inst_id

            mark_px = _safe_float(pos.get("markPx")) or _safe_float(q.get("mark_px") or q.get("mark"))
            if mark_px is None:
                mark_px = fetch_option_mark_px(ex, inst_id)
            opt_type = pos.get("optType") or q.get("opt_type")
            strike = _safe_float(pos.get("stk")) or _safe_float(q.get("strike"))
            if not opt_type or strike is None:
                pt, ps = option_fields_from_inst_id(inst_id)
                opt_type = opt_type or pt
                if strike is None:
                    strike = ps
            idx_px = _safe_float(pos.get("idxPx")) or _safe_float(q.get("index_px"))
            mark_px, intrinsic_px = close_ref_prices(
                mark_px=mark_px, opt_type=str(opt_type or ""), strike=strike, index_px=idx_px
            )
            book0 = cfg["fetch_option_book_depth"](ex, inst_id, 5)
            usable0, stub_only0, stub_reason0 = filter_bids_for_close(
                book0.get("bids") or [], mark_px=mark_px, intrinsic_px=intrinsic_px
            )
            paid = _open_premium_paid(cfg, inst_id)
            if stub_only0 or not usable0:
                bid_chk = None
                if book0.get("bids"):
                    bid_chk = _safe_float((book0.get("bids") or [{}])[0].get("px"))
                bid_chk = bid_chk or _safe_float(bid)
                stub, stub_reason = is_stub_bid_px(bid_chk, mark_px=mark_px, intrinsic_px=intrinsic_px)
                if stub or stub_only0:
                    update_close_gate(inst_id, recycle_usdc=None, premium_paid=paid)
                    return jsonify(
                        {
                            "ok": False,
                            "msg": stub_reason0 or stub_reason or "暂无有效买盘,禁止按买盘自动平仓",
                            "stopped_reason": "stub_bid",
                            "auto_close_blocked": True,
                        }
                    )
            preview_gate = estimate_close_by_bids(
                book0.get("bids") or [],
                close_sheets,
                ct_mult=ct_mult,
                premium_paid=paid,
                mark_px=mark_px,
                intrinsic_px=intrinsic_px,
            )
            if preview_gate.get("bid_invalid"):
                update_close_gate(inst_id, recycle_usdc=None, premium_paid=paid)
                return jsonify(
                    {
                        "ok": False,
                        "msg": preview_gate.get("bid_invalid_reason") or "暂无有效买盘,禁止按买盘自动平仓",
                        "stopped_reason": "stub_bid",
                        "auto_close_blocked": True,
                    }
                )
            gate = update_close_gate(
                inst_id,
                recycle_usdc=_safe_float(preview_gate.get("total_received")),
                premium_paid=paid,
            )
            if not gate.get("ready"):
                return jsonify(
                    {
                        "ok": False,
                        "msg": gate.get("msg") or "平仓门控未就绪(需可回收≥2×权利金并持续2分钟)",
                        "stopped_reason": "close_gate",
                        "auto_close_blocked": True,
                        "close_gate": gate,
                    }
                )
            remaining = close_sheets
            submitted_sheets = 0
            filled_or_reduced_sheets = 0
            total_received = 0.0
            orders: list[dict[str, Any]] = []
            stopped_reason = None
            for _ in range(5):
                if remaining <= 0:
                    break
                current_avail = _refresh_position_avail(cfg, ex, inst_id)
                if current_avail is None:
                    stopped_reason = "refresh_position_failed"
                    break
                if current_avail <= 0:
                    filled_or_reduced_sheets = close_sheets
                    remaining = 0
                    break
                remaining = min(remaining, current_avail)
                book = cfg["fetch_option_book_depth"](ex, inst_id, 5)
                preview = estimate_close_by_bids(
                    book.get("bids") or [],
                    remaining,
                    ct_mult=ct_mult,
                    mark_px=mark_px,
                    intrinsic_px=intrinsic_px,
                )
                if preview.get("auto_close_blocked") or preview.get("bid_invalid"):
                    return jsonify(
                        {
                            "ok": False,
                            "msg": preview.get("bid_invalid_reason") or "暂无有效买盘,禁止按买盘自动平仓",
                            "stopped_reason": "stub_bid",
                            "auto_close_blocked": True,
                        }
                    )
                levels = preview.get("levels") or []
                if not levels:
                    stopped_reason = "no_bid_depth"
                    break
                level = levels[0]
                level_sheets = int(level.get("sheets") or 0)
                level_px = float(level.get("px") or 0)
                if level_sheets <= 0 or level_px <= 0:
                    stopped_reason = "invalid_bid_depth"
                    break
                before_avail = current_avail
                order = cfg["place_option_limit_order"](
                    ex,
                    inst_id=inst_id,
                    side="sell",
                    sheets=level_sheets,
                    price=level_px,
                    td_mode=td_mode,
                    tick_sz=tick_sz,
                    reduce_only=True,
                    pos_side=pos_side,
                )
                if not order.get("ok"):
                    stopped_reason = order.get("msg") or "order_failed"
                    break
                px = float(order.get("px", level_px))
                orders.append({"order": order, "px": px, "sheets": level_sheets})
                submitted_sheets += level_sheets
                total_received += total_premium(px, level_sheets * ct_mult)
                time.sleep(0.6)
                after_avail = _refresh_position_avail(cfg, ex, inst_id)
                if after_avail is None:
                    stopped_reason = "refresh_position_failed"
                    break
                reduced = max(0, before_avail - after_avail)
                if reduced <= 0:
                    stopped_reason = "order_not_filled"
                    break
                filled_or_reduced_sheets += min(reduced, level_sheets)
                remaining = max(0, close_sheets - filled_or_reduced_sheets)
            if not orders:
                return jsonify({"ok": False, "msg": "暂无可用买盘深度,无法拆分平仓", "stopped_reason": stopped_reason})
            bid = (total_received / (submitted_sheets * ct_mult)) if submitted_sheets > 0 and ct_mult > 0 else 0
            prem_recv = round(total_received, 4)
            fully_submitted = submitted_sheets >= close_sheets and stopped_reason is None
            clear_close_gate(inst_id)
            conn = cfg["get_db"]()
            try:
                init_options_tables(conn)
                row = conn.execute(
                    "SELECT id, premium_paid FROM options_trades WHERE inst_id = ? AND status = 'open' ORDER BY id DESC LIMIT 1",
                    (inst_id,),
                ).fetchone()
                if row and fully_submitted:
                    paid = float(row["premium_paid"] or 0)
                    pnl = prem_recv - paid
                    conn.execute(
                        """
                        UPDATE options_trades
                        SET status = 'closed', close_quote = ?, premium_received = ?,
                            realized_pnl = ?, close_ord_id = ?, closed_at = CURRENT_TIMESTAMP
                        WHERE id = ?
                        """,
                        (
                            bid,
                            prem_recv,
                            pnl,
                            ",".join(str((o.get("order", {}).get("data") or {}).get("ordId") or "") for o in orders),
                            int(row["id"]),
                        ),
                    )
                    conn.commit()
            finally:
                conn.close()
            from lib.exchange.okx_options_lib import invalidate_option_positions_cache

            invalidate_option_positions_cache()
            _sync_options_trades(cfg, force=True)
            try:
                from lib.options.options_target_lib import cancel_target_monitor

                conn2 = cfg["get_db"]()
                try:
                    cancel_target_monitor(conn2, inst_id=inst_id)
                    conn2.commit()
                finally:
                    conn2.close()
            except Exception:
                pass
            return jsonify(
                {
                    "ok": True,
                    "mode": "depth_split",
                    "orders": orders,
                    "bid": bid,
                    "submitted_sheets": submitted_sheets,
                    "filled_or_reduced_sheets": filled_or_reduced_sheets,
                    "remaining_sheets": max(0, close_sheets - filled_or_reduced_sheets),
                    "premium_received": prem_recv,
                    "stopped_reason": stopped_reason,
                }
            )
        else:
            from lib.exchange.okx_options_lib import option_fields_from_inst_id

            mark_px = _safe_float(pos.get("markPx")) or _safe_float(q.get("mark_px") or q.get("mark"))
            if mark_px is None:
                mark_px = fetch_option_mark_px(ex, inst_id)
            opt_type = pos.get("optType") or q.get("opt_type")
            strike = _safe_float(pos.get("stk")) or _safe_float(q.get("strike"))
            if not opt_type or strike is None:
                pt, ps = option_fields_from_inst_id(inst_id)
                opt_type = opt_type or pt
                if strike is None:
                    strike = ps
            idx_px = _safe_float(pos.get("idxPx")) or _safe_float(q.get("index_px"))
            mark_px, intrinsic_px = close_ref_prices(
                mark_px=mark_px, opt_type=str(opt_type or ""), strike=strike, index_px=idx_px
            )
            close_px = float(bid)
            stub, stub_reason = is_stub_bid_px(close_px, mark_px=mark_px, intrinsic_px=intrinsic_px)
            if stub:
                return jsonify(
                    {
                        "ok": False,
                        "msg": stub_reason or "暂无有效买盘,禁止按买盘自动平仓",
                        "stopped_reason": "stub_bid",
                        "auto_close_blocked": True,
                    }
                )
            order = cfg["place_option_limit_order"](
                ex,
                inst_id=inst_id,
                side="sell",
                sheets=close_sheets,
                price=close_px,
                td_mode=td_mode,
                tick_sz=tick_sz,
                reduce_only=True,
                pos_side=pos_side,
            )
            if not order.get("ok"):
                return jsonify(order)
            bid = order.get("px", close_px)
        prem_recv = total_premium(float(bid or 0), close_sheets * float(q.get("ct_mult") or 0.01))
        conn = cfg["get_db"]()
        try:
            init_options_tables(conn)
            row = conn.execute(
                "SELECT id, premium_paid FROM options_trades WHERE inst_id = ? AND status = 'open' ORDER BY id DESC LIMIT 1",
                (inst_id,),
            ).fetchone()
            if row:
                paid = float(row["premium_paid"] or 0)
                pnl = prem_recv - paid
                conn.execute(
                    """
                    UPDATE options_trades
                    SET status = 'closed', close_quote = ?, premium_received = ?,
                        realized_pnl = ?, close_ord_id = ?, closed_at = CURRENT_TIMESTAMP
                    WHERE id = ?
                    """,
                    (
                        bid,
                        prem_recv,
                        pnl,
                        (order.get("data") or {}).get("ordId"),
                        int(row["id"]),
                    ),
                )
                conn.commit()
        finally:
            conn.close()
        from lib.exchange.okx_options_lib import invalidate_option_positions_cache

        invalidate_option_positions_cache()
        _sync_options_trades(cfg, force=True)
        try:
            from lib.options.options_target_lib import cancel_target_monitor

            conn2 = cfg["get_db"]()
            try:
                cancel_target_monitor(conn2, inst_id=inst_id)
                conn2.commit()
            finally:
                conn2.close()
        except Exception:
            pass
        return jsonify({"ok": True, "order": order, "bid": bid, "sheets": close_sheets})

    @app.route("/api/options/convert/quote", methods=["POST"])
    @lr
    def api_options_convert_quote():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        data = request.get_json(silent=True) or {}
        try:
            amount = float(data.get("amount"))
        except (TypeError, ValueError):
            return jsonify({"ok": False, "msg": "数量无效"})
        return jsonify(cfg["estimate_usdt_to_usdc"](ex, amount))

    @app.route("/api/options/convert/execute", methods=["POST"])
    @lr
    def api_options_convert_execute():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        data = request.get_json(silent=True) or {}
        quote_id = (data.get("quote_id") or "").strip()
        result = cfg["execute_convert"](ex, quote_id)
        if result.get("ok"):
            conn = cfg["get_db"]()
            try:
                init_options_tables(conn)
                conn.execute(
                    """
                    INSERT INTO options_convert_log (from_ccy, to_ccy, rfq_sz, received_sz, quote_id, status, message)
                    VALUES ('USDT', 'USDC', ?, ?, ?, 'ok', '')
                    """,
                    (
                        data.get("rfq_sz"),
                        (result.get("data") or {}).get("baseSz"),
                        quote_id,
                    ),
                )
                conn.commit()
            finally:
                conn.close()
        return jsonify(result)

    @app.route("/api/options/transfer", methods=["POST"])
    @lr
    def api_options_transfer():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        data = request.get_json(silent=True) or {}
        ccy = (data.get("ccy") or "USDC").upper()
        from_acct = (data.get("from") or "funding").strip()
        to_acct = (data.get("to") or "trading").strip()
        try:
            amount = float(data.get("amount"))
        except (TypeError, ValueError):
            return jsonify({"ok": False, "msg": "数量无效"})
        result = cfg["transfer_ccy"](ex, ccy, amount, from_acct, to_acct)
        if result.get("ok"):
            conn = cfg["get_db"]()
            try:
                init_options_tables(conn)
                conn.execute(
                    """
                    INSERT INTO options_transfer_log (ccy, amount, from_account, to_account, status, message)
                    VALUES (?, ?, ?, ?, 'ok', '')
                    """,
                    (ccy, amount, from_acct, to_acct),
                )
                conn.commit()
            finally:
                conn.close()
            _mark_balances_stale(cfg)
        return jsonify(result)

    @app.route("/api/options/spot/swap", methods=["POST"])
    @lr
    def api_options_spot_swap():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        data = request.get_json(silent=True) or {}
        direction = (data.get("direction") or "usdt_to_usdc").strip()
        try:
            amount = float(data.get("amount"))
        except (TypeError, ValueError):
            return jsonify({"ok": False, "msg": "数量无效"})
        result = cfg["spot_market_swap_usdt_usdc"](ex, direction=direction, amount=amount)
        if result.get("ok"):
            _mark_balances_stale(cfg)
        return jsonify(result)

    @app.route("/api/options/cross-transfer", methods=["POST"])
    @lr
    def api_options_cross_transfer():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        data = request.get_json(silent=True) or {}
        ccy = (data.get("ccy") or "USDT").upper()
        direction = (data.get("direction") or "sub_to_main").strip()
        from_account = (data.get("from_account") or data.get("account") or "funding").strip()
        to_account = (data.get("to_account") or data.get("account") or "funding").strip()
        try:
            amount = float(data.get("amount"))
        except (TypeError, ValueError):
            return jsonify({"ok": False, "msg": "数量无效"})
        main_to_sub = direction == "main_to_sub"
        result = cfg["transfer_main_sub_account"](
            ex,
            ccy=ccy,
            amount=amount,
            sub_acct=cfg.get("sub_account_name") or "",
            main_to_sub=main_to_sub,
            from_account=from_account,
            to_account=to_account,
        )
        if result.get("ok"):
            conn = cfg["get_db"]()
            try:
                init_options_tables(conn)
                conn.execute(
                    """
                    INSERT INTO options_transfer_log (ccy, amount, from_account, to_account, status, message)
                    VALUES (?, ?, ?, ?, 'ok', ?)
                    """,
                    (
                        ccy,
                        amount,
                        ("main" if main_to_sub else "sub") + ":" + from_account,
                        ("sub" if main_to_sub else "main") + ":" + to_account,
                        "cross",
                    ),
                )
                conn.commit()
            finally:
                conn.close()
            _mark_balances_stale(cfg)
        return jsonify(result)

    @app.route("/api/options/history")
    @lr
    def api_options_history():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        from lib.options.options_history_lib import load_options_history

        raw_live = cfg["fetch_option_positions"](ex)
        if raw_live is None:
            return jsonify({"ok": False, "msg": "获取期权持仓失败"})
        history = load_options_history(ex, cfg)
        live_ids = {str(x.get("inst_id") or "") for x in history if x.get("status") == "open"}
        return jsonify({"ok": True, "history": history, "live_inst_ids": sorted(live_ids)})

    @app.route("/api/options/stats")
    @lr
    def api_options_stats():
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        from lib.options.options_history_lib import load_options_history
        from lib.options.options_stats_lib import compute_options_stats_from_history

        raw_live = cfg["fetch_option_positions"](ex)
        if raw_live is None:
            return jsonify({"ok": False, "msg": "获取期权持仓失败"})
        history = load_options_history(ex, cfg)
        return jsonify({"ok": True, **compute_options_stats_from_history(history)})

    @app.route("/api/options/history/<path:history_key>", methods=["DELETE"])
    @lr
    def api_options_history_delete(history_key: str):
        ex, err = _require_options_ex(cfg)
        if ex is None:
            return jsonify({"ok": False, "msg": err})
        key = (history_key or "").strip()
        if not key:
            return jsonify({"ok": False, "msg": "缺少 history_key"})
        conn = cfg["get_db"]()
        try:
            init_options_tables(conn)
            conn.execute(
                "INSERT OR IGNORE INTO options_history_hidden (history_key) VALUES (?)",
                (key,),
            )
            conn.commit()
        finally:
            conn.close()
        return jsonify({"ok": True})


def _start_monitor_thread(app: Flask, cfg: dict[str, Any]) -> None:
    if app.extensions.get("options_monitor_started"):
        return
    app.extensions["options_monitor_started"] = True

    def _bid(inst_id: str) -> float | None:
        ex = cfg.get("exchange_options")
        if ex is None:
            return None
        try:
            q = cfg["quote_option_contract"](ex, inst_id)
            return q.get("bid")
        except Exception:
            return None

    def _positions():
        ex = cfg.get("exchange_options")
        if ex is None:
            return []
        raw = cfg["fetch_option_positions"](ex)
        if raw is None:
            return []
        return [cfg["format_position_row"](p) for p in raw]

    def _sync(conn):
        from lib.exchange.okx_options_lib import fetch_option_position_history
        from lib.options.options_monitor_lib import reconcile_live_open_trades, sync_open_options_trades

        ex = cfg.get("exchange_options")
        if ex is None:
            return 0
        raw = cfg["fetch_option_positions"](ex)
        if raw is None:
            return 0
        live_ids = {str(p.get("instId") or "") for p in raw if str(p.get("instId") or "")}
        reconcile_live_open_trades(conn, live_inst_ids=live_ids)
        return sync_open_options_trades(
            conn,
            live_inst_ids=live_ids,
            fetch_history_fn=lambda inst_id: fetch_option_position_history(ex, inst_id),
        )

    def _target_close(inst_id: str) -> dict[str, Any]:
        from lib.options.options_target_lib import close_option_by_bid_depth

        ex = cfg.get("exchange_options")
        if ex is None:
            return {"ok": False, "msg": "期权 exchange 未就绪"}
        result = close_option_by_bid_depth(cfg, ex, inst_id)
        if result.get("ok"):
            try:
                _sync_options_trades(cfg, force=True)
            except Exception:
                pass
            try:
                _mark_balances_stale(cfg)
            except Exception:
                pass
        return result

    t = threading.Thread(
        target=options_monitor_loop,
        kwargs={
            "enabled": True,
            "poll_seconds": cfg["poll_seconds"],
            "get_db": cfg["get_db"],
            "fetch_positions": _positions,
            "ticker_bid_fn": _bid,
            "send_wechat": cfg["send_wechat"],
            "account_label": cfg["account_label"],
            "profit_ratio": cfg["profit_ratio"],
            "sync_trades_fn": _sync,
            "target_close_fn": _target_close,
        },
        daemon=True,
        name="options-monitor",
    )
    t.start()