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 IN ('active', 'closing')
        ORDER BY CASE status WHEN 'active' THEN 0 WHEN 'closing' THEN 1 ELSE 2 END, 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),
                status = 'active',
                trigger_idx = NULL,
                close_ord_id = NULL,
                message = NULL,
                triggered_at = NULL
            WHERE id = ?
            """,
            (target_index, underlying, opt_type, trade_id, sheets, int(row["id"])),
        )
        mon_id = int(row["id"])
        # 同一合约其他进行中的委托取消,避免双轨触发重复推送
        conn.execute(
            """
            UPDATE options_target_monitors
            SET status = 'cancelled', message = '被新目标位覆盖'
            WHERE inst_id = ? AND id != ? AND status IN ('active', 'closing')
            """,
            (inst_id, mon_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 IN ('active', 'closing')
            """,
            (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 IN ('active', 'closing')
            """,
            (inst_id.strip(),),
        )
        return int(cur.rowcount or 0)
    return 0


def _row_to_target(r: sqlite3.Row) -> dict[str, Any]:
    return {
        "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"],
    }


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()
    return [_row_to_target(r) for r in rows]


def list_closing_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 = 'closing'
        ORDER BY id DESC
        """
    ).fetchall()
    return [_row_to_target(r) for r in rows]


def targets_by_inst(conn: sqlite3.Connection) -> dict[str, dict[str, Any]]:
    """UI/持仓挂载:active 与 closing 都算进行中."""
    out: dict[str, dict[str, Any]] = {}
    for t in list_closing_targets(conn) + list_active_targets(conn):
        inst = str(t.get("inst_id") or "")
        if inst and inst not in out:
            out[inst] = t
    return out


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', 'closing') THEN COALESCE(triggered_at, 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) + list_closing_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 _commit_monitor(conn: sqlite3.Connection) -> None:
    """状态变更立刻落库,避免后续 sync 异常回滚后重复触发/推送."""
    try:
        conn.commit()
    except Exception:
        pass


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"

    # 已有未成交卖平单时先等成交,避免每轮撤单重挂反复推送/吃档
    try:
        pending = ex.private_get_trade_orders_pending({"instType": "OPTION", "instId": inst_id}) or {}
        sell_pending = [
            o
            for o in (pending.get("data") or [])
            if str(o.get("side") or "").lower() == "sell" and o.get("ordId")
        ]
        if sell_pending:
            time.sleep(0.5)
            invalidate_option_positions_cache()
            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 or _avail(pos) < 1:
                return {
                    "ok": True,
                    "already_flat": True,
                    "msg": "已有限价卖单成交",
                    "close_ord_id": ",".join(str(o.get("ordId")) for o in sell_pending),
                    "fully_closed": True,
                    "submitted_sheets": close_sheets,
                    "remaining_sheets": 0,
                }
            # 仍持仓且卖单挂着:本轮不撤不重挂,交给下一轮
            return {
                "ok": False,
                "msg": "等待已有买一限价卖单成交",
                "stopped_reason": "pending_close_order",
                "close_ord_id": ",".join(str(o.get("ordId")) for o in sell_pending),
            }
    except Exception:
        pass

    # 无挂单时再清理残留卖单(兼容旧路径)并按买一重新挂出
    try:
        pending = ex.private_get_trade_orders_pending({"instType": "OPTION", "instId": inst_id}) or {}
        for o in pending.get("data") or []:
            if str(o.get("side") or "").lower() != "sell":
                continue
            oid = o.get("ordId")
            if not oid:
                continue
            try:
                ex.private_post_trade_cancel_order({"instId": inst_id, "ordId": oid})
            except Exception:
                pass
        time.sleep(0.3)
        invalidate_option_positions_cache()
        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}
        avail = _avail(pos)
        close_sheets = min(close_sheets, avail)
        if close_sheets < 1:
            return {"ok": False, "msg": "可平张数不足", "already_flat": True}
    except Exception:
        pass

    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)
            bid_px = _safe_float(q2.get("bid")) or _safe_float(q.get("bid"))
            if bid_px is None or bid_px <= 0:
                stopped_reason = "no_bid"
                break
            levels = [{"sheets": remaining, "px": bid_px}]
        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 or "no_bid",
        }

    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 _notify_target_close(
    send_wechat: Callable[[str], None] | None,
    *,
    account_label: str,
    inst_id: str,
    target: float,
    idx: float,
    result: dict[str, Any],
) -> None:
    if not send_wechat:
        return
    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


def _result_fully_done(result: dict[str, Any]) -> bool:
    if result.get("already_flat"):
        return True
    if result.get("fully_closed"):
        return True
    remaining = result.get("remaining_sheets")
    if remaining is not None and int(remaining) <= 0 and result.get("ok"):
        return True
    return False


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 目标委托;指数到位后限价平仓.
    状态先 commit 再推微信,避免 sync 失败回滚导致同一笔反复推送.
    未完全成交进入 closing,仅重试平仓不再推送.
    返回本次新触发(并推送)的条数.
    """
    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)
    _commit_monitor(conn)

    # 先处理已挂单等待成交的,绝不再发微信
    for mon in list_closing_targets(conn):
        inst_id = str(mon.get("inst_id") or "")
        if not inst_id:
            continue
        if inst_id not in pos_by_inst:
            mark_monitor(conn, int(mon["id"]), status="expired", message="持仓已平")
            _commit_monitor(conn)
            continue
        result = close_fn(inst_id)
        idx = _safe_float(pos_by_inst[inst_id].get("idx_px") or pos_by_inst[inst_id].get("idxPx"))
        if result.get("already_flat") or _result_fully_done(result):
            mark_monitor(
                conn,
                int(mon["id"]),
                status="triggered",
                trigger_idx=idx,
                close_ord_id=result.get("close_ord_id"),
                message="目标位限价平仓完成",
            )
            _commit_monitor(conn)
            continue
        mark_monitor(
            conn,
            int(mon["id"]),
            status="closing",
            trigger_idx=idx,
            close_ord_id=result.get("close_ord_id"),
            message=str(result.get("msg") or result.get("stopped_reason") or "等待买一成交"),
        )
        _commit_monitor(conn)

    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="持仓已平")
            _commit_monitor(conn)
            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 "平仓未完成,将重试"),
            )
            _commit_monitor(conn)
            continue

        done = _result_fully_done(result)
        status = "triggered" if done else "closing"
        mark_monitor(
            conn,
            int(mon["id"]),
            status=status,
            trigger_idx=idx,
            close_ord_id=result.get("close_ord_id"),
            message="目标位触发限价平仓" if done else "目标位已挂买一限价,等待成交",
        )
        # 关键:先落库,再推送——否则后续 sync 异常回滚会让同一笔反复推微信
        _commit_monitor(conn)
        triggered += 1
        _notify_target_close(
            send_wechat,
            account_label=account_label,
            inst_id=inst_id,
            target=target,
            idx=idx,
            result=result,
        )
    return triggered