Add options index target monitors that auto limit-close on hit.

Position and order forms can arm a target; right-side and hub panels show active monitors; expiry remains the stop with no separate SL. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-15 15:27:46 +08:00
parent 8d2da921ee
commit d6823a2903
13 changed files with 1011 additions and 18 deletions
@@ -0,0 +1,449 @@
+"""期权目标位委托:指数达价后限价平仓(无止损,到期由结算收口)."""
+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:
+            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})
+        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:
+        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