crypto_monitor/lib/options/options_target_lib.py

"""期权目标位委托:指数达价后限价平仓(无止损,到期由结算收口)."""
from __future__ import annotations

import sqlite3
import time
from typing import Any, Callable

from lib.options.options_db import init_options_tables
from lib.options.options_pricing_lib import estimate_close_by_bids, total_premium


def _safe_float(v: Any) -> float | None:
    if v is None or v == "":
        return None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def ensure_target_tables(conn: sqlite3.Connection) -> None:
    init_options_tables(conn)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS options_target_monitors (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            inst_id TEXT NOT NULL,
            underlying TEXT,
            opt_type TEXT,
            target_index REAL NOT NULL,
            trade_id INTEGER,
            sheets INTEGER,
            status TEXT DEFAULT 'active',
            trigger_idx REAL,
            close_ord_id TEXT,
            message TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            triggered_at TIMESTAMP
        )
        """
    )
    conn.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_options_target_monitors_status
        ON options_target_monitors(status)
        """
    )


def target_hit(*, opt_type: str | None, index_px: float, target_index: float) -> bool:
    """Call:指数涨到/超过目标平仓;Put:指数跌到/低于目标平仓."""
    ot = (opt_type or "").strip().upper()
    if ot == "P":
        return index_px <= target_index
    return index_px >= target_index


def upsert_target_monitor(
    conn: sqlite3.Connection,
    *,
    inst_id: str,
    target_index: float,
    underlying: str | None = None,
    opt_type: str | None = None,
    trade_id: int | None = None,
    sheets: int | None = None,
) -> dict[str, Any]:
    ensure_target_tables(conn)
    inst_id = (inst_id or "").strip()
    if not inst_id:
        return {"ok": False, "msg": "缺少 inst_id"}
    if target_index is None or float(target_index) <= 0:
        return {"ok": False, "msg": "目标位无效"}
    target_index = float(target_index)
    row = conn.execute(
        """
        SELECT id FROM options_target_monitors
        WHERE inst_id = ? AND status = 'active'
        ORDER BY id DESC LIMIT 1
        """,
        (inst_id,),
    ).fetchone()
    if row:
        conn.execute(
            """
            UPDATE options_target_monitors
            SET target_index = ?,
                underlying = COALESCE(?, underlying),
                opt_type = COALESCE(?, opt_type),
                trade_id = COALESCE(?, trade_id),
                sheets = COALESCE(?, sheets),
                message = NULL
            WHERE id = ?
            """,
            (target_index, underlying, opt_type, trade_id, sheets, int(row["id"])),
        )
        mon_id = int(row["id"])
    else:
        cur = conn.execute(
            """
            INSERT INTO options_target_monitors
            (inst_id, underlying, opt_type, target_index, trade_id, sheets, status)
            VALUES (?, ?, ?, ?, ?, ?, 'active')
            """,
            (inst_id, underlying, opt_type, target_index, trade_id, sheets),
        )
        mon_id = int(cur.lastrowid)
    return {"ok": True, "id": mon_id, "inst_id": inst_id, "target_index": target_index}


def cancel_target_monitor(conn: sqlite3.Connection, *, inst_id: str | None = None, monitor_id: int | None = None) -> int:
    ensure_target_tables(conn)
    if monitor_id is not None:
        cur = conn.execute(
            """
            UPDATE options_target_monitors
            SET status = 'cancelled', message = '手动取消'
            WHERE id = ? AND status = 'active'
            """,
            (int(monitor_id),),
        )
        return int(cur.rowcount or 0)
    if inst_id:
        cur = conn.execute(
            """
            UPDATE options_target_monitors
            SET status = 'cancelled', message = '手动取消'
            WHERE inst_id = ? AND status = 'active'
            """,
            (inst_id.strip(),),
        )
        return int(cur.rowcount or 0)
    return 0


def list_active_targets(conn: sqlite3.Connection) -> list[dict[str, Any]]:
    ensure_target_tables(conn)
    rows = conn.execute(
        """
        SELECT id, inst_id, underlying, opt_type, target_index, trade_id, sheets,
               status, message, created_at
        FROM options_target_monitors
        WHERE status = 'active'
        ORDER BY id DESC
        """
    ).fetchall()
    out: list[dict[str, Any]] = []
    for r in rows:
        out.append(
            {
                "id": int(r["id"]),
                "inst_id": r["inst_id"],
                "underlying": r["underlying"],
                "opt_type": r["opt_type"],
                "target_index": _safe_float(r["target_index"]),
                "trade_id": r["trade_id"],
                "sheets": r["sheets"],
                "status": r["status"],
                "message": r["message"],
                "created_at": r["created_at"],
            }
        )
    return out


def targets_by_inst(conn: sqlite3.Connection) -> dict[str, dict[str, Any]]:
    return {str(t["inst_id"]): t for t in list_active_targets(conn) if t.get("inst_id")}


def mark_monitor(
    conn: sqlite3.Connection,
    monitor_id: int,
    *,
    status: str,
    trigger_idx: float | None = None,
    close_ord_id: str | None = None,
    message: str | None = None,
) -> None:
    conn.execute(
        """
        UPDATE options_target_monitors
        SET status = ?,
            trigger_idx = COALESCE(?, trigger_idx),
            close_ord_id = COALESCE(?, close_ord_id),
            message = COALESCE(?, message),
            triggered_at = CASE WHEN ? IN ('triggered', 'expired') THEN CURRENT_TIMESTAMP ELSE triggered_at END
        WHERE id = ?
        """,
        (status, trigger_idx, close_ord_id, message, status, int(monitor_id)),
    )


def cancel_orphans_without_position(
    conn: sqlite3.Connection,
    *,
    live_inst_ids: set[str],
) -> int:
    """持仓已消失的目标委托标记为 expired(到期/已平),不挂止损."""
    ensure_target_tables(conn)
    rows = list_active_targets(conn)
    n = 0
    for t in rows:
        inst = str(t.get("inst_id") or "")
        if inst and inst not in live_inst_ids:
            mark_monitor(conn, int(t["id"]), status="expired", message="持仓已平/到期,委托结束")
            n += 1
    return n


def close_option_by_bid_depth(
    cfg: dict[str, Any],
    ex: Any,
    inst_id: str,
    *,
    sheets: int | None = None,
) -> dict[str, Any]:
    """按买盘拆分限价卖出(最多5档),供目标位自动平仓复用."""
    from lib.exchange.okx_options_lib import (
        _pos_side_from_position,
        invalidate_option_positions_cache,
    )

    q = cfg["quote_option_contract"](ex, inst_id)
    if not q.get("ok"):
        return {"ok": False, "msg": q.get("msg") or "报价失败"}
    tick_sz = q.get("tick_sz")
    ct_mult = float(q.get("ct_mult") or 0.01)
    raw_positions = cfg["fetch_option_positions"](ex)
    if raw_positions is None:
        return {"ok": False, "msg": "获取期权持仓失败"}
    pos = next((p for p in raw_positions if str(p.get("instId")) == inst_id), None)
    if not pos:
        return {"ok": False, "msg": "未找到持仓", "already_flat": True}

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

    avail = _avail(pos)
    close_sheets = int(sheets) if sheets else avail
    close_sheets = min(close_sheets, avail)
    if close_sheets < 1:
        return {"ok": False, "msg": "可平张数不足", "already_flat": True}
    td_mode = str(pos.get("mgnMode") or cfg.get("td_mode") or "isolated")
    pos_side = _pos_side_from_position(pos) or "net"

    remaining = close_sheets
    submitted_sheets = 0
    filled_or_reduced_sheets = 0
    total_received = 0.0
    orders: list[dict[str, Any]] = []
    stopped_reason = None
    ord_ids: list[str] = []

    for _ in range(5):
        if remaining <= 0:
            break
        invalidate_option_positions_cache()
        raw = cfg["fetch_option_positions"](ex)
        if raw is None:
            stopped_reason = "refresh_position_failed"
            break
        cur_pos = next((p for p in raw if str(p.get("instId")) == inst_id), None)
        current_avail = _avail(cur_pos) if cur_pos else 0
        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)
        levels = preview.get("levels") or []
        if not levels:
            # 无买盘时回退:报价买一 → 报价标记价 → 持仓标记价
            q2 = cfg["quote_option_contract"](ex, inst_id)
            cur_mark = _safe_float(cur_pos.get("markPx")) if cur_pos else None
            pos_mark = _safe_float(pos.get("markPx"))
            fallback_px = (
                _safe_float(q2.get("bid"))
                or _safe_float(q2.get("mark_px"))
                or _safe_float(q.get("bid"))
                or _safe_float(q.get("mark_px"))
                or cur_mark
                or pos_mark
            )
            if fallback_px is None or fallback_px <= 0:
                stopped_reason = "no_bid_depth"
                break
            levels = [{"sheets": remaining, "px": fallback_px, "fallback": True}]
        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})
        oid = str((order.get("data") or {}).get("ordId") or "")
        if oid:
            ord_ids.append(oid)
        submitted_sheets += level_sheets
        total_received += total_premium(px, level_sheets * ct_mult)
        time.sleep(0.6)
        invalidate_option_positions_cache()
        raw2 = cfg["fetch_option_positions"](ex)
        if raw2 is None:
            stopped_reason = "refresh_position_failed"
            break
        after_pos = next((p for p in raw2 if str(p.get("instId")) == inst_id), None)
        after_avail = _avail(after_pos) if after_pos else 0
        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:
        # 最后兜底:允许市价平仓时用市价
        if cfg.get("allow_market_close"):
            mkt = 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 mkt.get("ok"):
                oid = str((mkt.get("data") or {}).get("ordId") or "")
                return {
                    "ok": True,
                    "mode": "market",
                    "orders": [{"order": mkt, "sheets": close_sheets}],
                    "submitted_sheets": close_sheets,
                    "filled_or_reduced_sheets": close_sheets,
                    "remaining_sheets": 0,
                    "premium_received": None,
                    "close_ord_id": oid or None,
                    "fully_closed": True,
                }
            return {"ok": False, "msg": mkt.get("msg") or "市价平仓失败", "stopped_reason": stopped_reason}
        return {"ok": False, "msg": "暂无可用买盘深度,无法限价平仓", "stopped_reason": stopped_reason}

    avg_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
    close_ord_id = ",".join(ord_ids) if ord_ids else None

    conn = cfg["get_db"]()
    try:
        ensure_target_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,
                    signal_note = CASE
                        WHEN signal_note IS NULL OR TRIM(signal_note) = '' THEN '目标位平仓'
                        ELSE signal_note
                    END
                WHERE id = ?
                """,
                (avg_bid, prem_recv, pnl, close_ord_id, int(row["id"])),
            )
            conn.commit()
    finally:
        conn.close()

    invalidate_option_positions_cache()
    return {
        "ok": True,
        "mode": "depth_split",
        "orders": orders,
        "bid": avg_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,
        "close_ord_id": close_ord_id,
        "fully_closed": fully_submitted and remaining == 0,
    }


def run_options_target_closes(
    conn: sqlite3.Connection,
    positions: list[dict[str, Any]],
    *,
    close_fn: Callable[[str], dict[str, Any]],
    index_fn: Callable[[dict[str, Any]], float | None] | None = None,
    send_wechat: Callable[[str], None] | None = None,
    account_label: str = "OKX期权",
) -> int:
    """
    扫描 active 目标委托;指数到位后限价平仓.
    返回触发条数.
    """
    ensure_target_tables(conn)
    pos_by_inst = {str(p.get("inst_id") or p.get("instId") or ""): p for p in positions}
    live_ids = {k for k in pos_by_inst if k}
    cancel_orphans_without_position(conn, live_inst_ids=live_ids)

    triggered = 0
    for mon in list_active_targets(conn):
        inst_id = str(mon.get("inst_id") or "")
        target = _safe_float(mon.get("target_index"))
        if not inst_id or target is None:
            continue
        pos = pos_by_inst.get(inst_id)
        if not pos:
            continue
        if index_fn is not None:
            idx = index_fn(pos)
        else:
            idx = _safe_float(pos.get("idx_px") or pos.get("idxPx"))
        if idx is None:
            continue
        opt_type = mon.get("opt_type") or pos.get("opt_type") or pos.get("optType")
        if not target_hit(opt_type=str(opt_type) if opt_type else None, index_px=idx, target_index=target):
            continue

        result = close_fn(inst_id)
        if result.get("already_flat"):
            mark_monitor(conn, int(mon["id"]), status="expired", trigger_idx=idx, message="持仓已平")
            continue
        if not result.get("ok"):
            mark_monitor(
                conn,
                int(mon["id"]),
                status="active",
                trigger_idx=idx,
                message=str(result.get("msg") or result.get("stopped_reason") or "平仓未完成,将重试"),
            )
            continue

        mark_monitor(
            conn,
            int(mon["id"]),
            status="triggered",
            trigger_idx=idx,
            close_ord_id=result.get("close_ord_id"),
            message="目标位触发限价平仓",
        )
        triggered += 1
        if send_wechat:
            try:
                send_wechat(
                    "\n".join(
                        [
                            "【OKX期权·目标位平仓】",
                            f"账户:{account_label}",
                            f"合约:{inst_id}",
                            f"目标指数:{target:g}",
                            f"触发指数:{idx:g}",
                            f"提交张数:{result.get('submitted_sheets') or '—'}",
                            f"预估收回:{result.get('premium_received') if result.get('premium_received') is not None else '—'} USDC",
                        ]
                    )
                )
            except Exception:
                pass
    return triggered