feat(options): push chain asks/bids via OKX WS + SSE

Replace soft REST polling with OKX public tickers WS ingest and browser SSE patches so list quotes stay live while watching an expiry.

Co-authored-by: Cursor <cursoragent@cursor.com>

lib/options/options_quote_live_lib.py

@@ -0,0 +1,458 @@
+"""期权链实时报价:OKX 公共 WS tickers → 内存缓存 → SSE 推前端."""
+from __future__ import annotations
+
+import json
+import logging
+import os
+import queue
+import threading
+import time
+from collections.abc import Iterator
+from typing import Any, Callable
+
+from lib.exchange.okx_public_ws_lib import OkxPublicWs
+from lib.options.options_pricing_lib import (
+    expiry_breakeven_from_ask,
+    idx_distance_to_be,
+)
+
+logger = logging.getLogger(__name__)
+
+OPTIONS_QUOTE_SSE_HEARTBEAT_SEC = float(os.getenv("OKX_OPTIONS_QUOTE_SSE_HEARTBEAT_SEC", "20"))
+OPTIONS_QUOTE_FLUSH_MS = float(os.getenv("OKX_OPTIONS_QUOTE_FLUSH_MS", "120"))
+# OKX 单连接约 240 频道;当前到期日合约 + 指数通常够用
+OPTIONS_QUOTE_MAX_INST = int(os.getenv("OKX_OPTIONS_QUOTE_MAX_INST", "220"))
+
+
+def _safe_float(v: Any) -> float | None:
+    try:
+        if v is None or v == "":
+            return None
+        return float(v)
+    except (TypeError, ValueError):
+        return None
+
+
+class OptionsQuoteLive:
+    def __init__(self) -> None:
+        self._lock = threading.RLock()
+        self._watchers: dict[str, dict[str, Any]] = {}
+        self._meta: dict[str, dict[str, Any]] = {}
+        self._tickers: dict[str, dict[str, Any]] = {}
+        self._index_by_uly: dict[str, float] = {}
+        self._index_insts: set[str] = set()
+        self._dirty_inst: set[str] = set()
+        self._dirty_index: set[str] = set()
+        self._version = 0
+        self._subscribers: list[queue.Queue[str | None]] = []
+        self._stop = threading.Event()
+        self._flush_thread: threading.Thread | None = None
+        ws_url = (os.getenv("OKX_PUBLIC_WS_URL") or "").strip() or None
+        self._ws = OkxPublicWs(
+            on_data=self._on_ws_data,
+            name="okx-options-quote-ws",
+            **({"url": ws_url} if ws_url else {}),
+        )
+        self._started = False
+
+    def start(self) -> None:
+        if self._started:
+            return
+        self._started = True
+        self._stop.clear()
+        self._ws.start()
+        self._flush_thread = threading.Thread(
+            target=self._flush_loop, name="options-quote-flush", daemon=True
+        )
+        self._flush_thread.start()
+
+    def stop(self) -> None:
+        self._stop.set()
+        self._ws.stop()
+        self._broadcast(close=True)
+        self._started = False
+
+    def status(self) -> dict[str, Any]:
+        with self._lock:
+            uly = ""
+            exp = ""
+            index_inst = ""
+            if self._watchers:
+                last = next(reversed(list(self._watchers.values())))
+                uly = str(last.get("underlying") or "")
+                exp = str(last.get("exp_time") or "")
+                index_inst = str(last.get("index_inst") or "")
+            return {
+                "ok": True,
+                "started": self._started,
+                "ws_ok": self._ws.connected,
+                "underlying": uly,
+                "index_inst": index_inst,
+                "index_px": self._index_by_uly.get(uly),
+                "watch_exp": exp,
+                "watch_count": len(self._meta),
+                "watcher_count": len(self._watchers),
+                "version": self._version,
+                "last_msg_at": self._ws.last_msg_at,
+            }
+
+    def watch(
+        self,
+        *,
+        underlying: str,
+        exp_time: str | int | None,
+        contracts: list[dict[str, Any]],
+        index_inst_id: str | None = None,
+        watcher_id: str | None = None,
+    ) -> dict[str, Any]:
+        u = (underlying or "ETH").upper()
+        index_id = (index_inst_id or f"{u}-USD").strip()
+        wid = (watcher_id or "default").strip() or "default"
+        meta: dict[str, dict[str, Any]] = {}
+        for c in contracts or []:
+            if not isinstance(c, dict):
+                continue
+            inst_id = str(c.get("inst_id") or c.get("instId") or "").strip()
+            if not inst_id:
+                continue
+            meta[inst_id] = {
+                "inst_id": inst_id,
+                "opt_type": str(c.get("opt_type") or c.get("optType") or "").upper(),
+                "strike": _safe_float(c.get("strike")),
+                "tick_sz": c.get("tick_sz") or c.get("tickSz"),
+                "underlying": u,
+            }
+            if len(meta) >= max(1, OPTIONS_QUOTE_MAX_INST):
+                break
+        with self._lock:
+            self._watchers[wid] = {
+                "underlying": u,
+                "exp_time": str(exp_time or ""),
+                "index_inst": index_id,
+                "meta": meta,
+            }
+            self._rebuild_subscriptions_locked()
+        if not self._started:
+            self.start()
+        return self.status()
+
+    def _rebuild_subscriptions_locked(self) -> None:
+        merged: dict[str, dict[str, Any]] = {}
+        index_insts: set[str] = set()
+        for w in self._watchers.values():
+            index_insts.add(str(w.get("index_inst") or ""))
+            for inst_id, m in (w.get("meta") or {}).items():
+                if inst_id not in merged:
+                    merged[inst_id] = dict(m)
+                if len(merged) >= max(1, OPTIONS_QUOTE_MAX_INST):
+                    break
+            if len(merged) >= max(1, OPTIONS_QUOTE_MAX_INST):
+                break
+        index_insts = {x for x in index_insts if x}
+        self._meta = merged
+        self._index_insts = index_insts
+        keep = set(merged.keys())
+        for k in list(self._tickers.keys()):
+            if k not in keep:
+                self._tickers.pop(k, None)
+        args = [{"channel": "tickers", "instId": iid} for iid in merged]
+        for iid in sorted(index_insts):
+            args.append({"channel": "index-tickers", "instId": iid})
+        self._ws.set_subscriptions(args)
+
+    def seed_from_chain(
+        self,
+        chain: dict[str, Any],
+        *,
+        exp_time: str | int | None = None,
+        watcher_id: str | None = None,
+    ) -> None:
+        """REST 拉链后预填报价,并默认监视指定/最近到期."""
+        if not isinstance(chain, dict):
+            return
+        u = str(chain.get("underlying") or "ETH").upper()
+        index_px = _safe_float(chain.get("index_px"))
+        expiries = chain.get("expiries") or []
+        target = None
+        if exp_time is not None and str(exp_time):
+            for e in expiries:
+                if str(e.get("exp_time")) == str(exp_time):
+                    target = e
+                    break
+        if target is None and expiries:
+            target = expiries[0]
+        contracts = list((target or {}).get("contracts") or [])
+        if index_px is not None:
+            with self._lock:
+                self._index_by_uly[u] = index_px
+                self._dirty_index.add(u)
+        for c in contracts:
+            inst_id = str(c.get("inst_id") or "").strip()
+            if not inst_id:
+                continue
+            patch = {
+                "inst_id": inst_id,
+                "ask": c.get("ask"),
+                "bid": c.get("bid"),
+                "ask_sz": c.get("ask_sz"),
+                "bid_sz": c.get("bid_sz"),
+                "mark_px": c.get("mark_px"),
+                "ask_estimated": bool(c.get("ask_estimated")),
+                "expiry_be_px": c.get("expiry_be_px"),
+                "dist_expiry_be": c.get("dist_expiry_be"),
+                "underlying": u,
+            }
+            with self._lock:
+                self._tickers[inst_id] = patch
+                self._dirty_inst.add(inst_id)
+        self.watch(
+            underlying=u,
+            exp_time=(target or {}).get("exp_time"),
+            contracts=contracts,
+            index_inst_id=f"{u}-USD",
+            watcher_id=watcher_id or f"seed:{u}",
+        )
+
+    def _on_ws_data(self, payload: dict[str, Any]) -> None:
+        arg = payload.get("arg") or {}
+        channel = str(arg.get("channel") or "")
+        rows = payload.get("data") or []
+        if not isinstance(rows, list) or not rows:
+            return
+        if channel == "index-tickers":
+            row = rows[0] if isinstance(rows[0], dict) else {}
+            px = _safe_float(row.get("idxPx"))
+            inst = str(row.get("instId") or arg.get("instId") or "")
+            uly = inst.split("-")[0].upper() if inst else ""
+            if px is None or not uly:
+                return
+            with self._lock:
+                if self._index_by_uly.get(uly) == px:
+                    return
+                self._index_by_uly[uly] = px
+                self._dirty_index.add(uly)
+            return
+        if channel != "tickers":
+            return
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            inst_id = str(row.get("instId") or arg.get("instId") or "").strip()
+            if not inst_id:
+                continue
+            patch = self._ticker_to_patch(inst_id, row)
+            with self._lock:
+                prev = self._tickers.get(inst_id) or {}
+                if (
+                    prev.get("ask") == patch.get("ask")
+                    and prev.get("bid") == patch.get("bid")
+                    and prev.get("ask_sz") == patch.get("ask_sz")
+                    and prev.get("bid_sz") == patch.get("bid_sz")
+                    and prev.get("mark_px") == patch.get("mark_px")
+                ):
+                    continue
+                self._tickers[inst_id] = patch
+                self._dirty_inst.add(inst_id)
+
+    def _ticker_to_patch(self, inst_id: str, row: dict[str, Any]) -> dict[str, Any]:
+        ask = _safe_float(row.get("askPx"))
+        bid = _safe_float(row.get("bidPx"))
+        ask_sz = _safe_float(row.get("askSz"))
+        bid_sz = _safe_float(row.get("bidSz"))
+        mark = _safe_float(row.get("markPx"))
+        ask_estimated = False
+        with self._lock:
+            meta = dict(self._meta.get(inst_id) or {})
+            uly = str(meta.get("underlying") or inst_id.split("-")[0] or "").upper()
+            index_px = self._index_by_uly.get(uly)
+        if ask is None and mark is not None and mark > 0:
+            ask = mark
+            ask_estimated = True
+            ask_sz = None
+        if bid is None and mark is not None and mark > 0:
+            bid = mark
+        be = expiry_breakeven_from_ask(
+            opt_type=str(meta.get("opt_type") or ""),
+            strike=meta.get("strike"),
+            ask_px=None if ask_estimated else ask,
+            mark_px=mark,
+        )
+        dist = idx_distance_to_be(index_px, be)
+        return {
+            "inst_id": inst_id,
+            "underlying": uly,
+            "ask": ask,
+            "bid": bid,
+            "ask_sz": ask_sz,
+            "bid_sz": bid_sz,
+            "mark_px": mark,
+            "ask_estimated": ask_estimated,
+            "expiry_be_px": be,
+            "dist_expiry_be": dist,
+        }
+
+    def _flush_loop(self) -> None:
+        interval = max(0.05, OPTIONS_QUOTE_FLUSH_MS / 1000.0)
+        while not self._stop.is_set():
+            if self._stop.wait(interval):
+                break
+            event = self._build_flush_event()
+            if event is None:
+                continue
+            self._broadcast(event)
+
+    def _build_flush_event(self) -> str | None:
+        with self._lock:
+            if not self._dirty_inst and not self._dirty_index:
+                return None
+            dirty_uly = set(self._dirty_index)
+            self._dirty_index.clear()
+            quotes: list[dict[str, Any]] = []
+            for inst_id in list(self._dirty_inst):
+                q = self._tickers.get(inst_id)
+                if q:
+                    quotes.append(dict(q))
+            self._dirty_inst.clear()
+            for uly in dirty_uly:
+                index_px = self._index_by_uly.get(uly)
+                if index_px is None:
+                    continue
+                for inst_id, q in list(self._tickers.items()):
+                    if str(q.get("underlying") or "").upper() != uly:
+                        continue
+                    be = q.get("expiry_be_px")
+                    dist = idx_distance_to_be(index_px, be if be is not None else None)
+                    if q.get("dist_expiry_be") != dist:
+                        q2 = dict(q)
+                        q2["dist_expiry_be"] = dist
+                        self._tickers[inst_id] = q2
+                        quotes.append(q2)
+            self._version += 1
+            # 多标的时 index_px 取「最近一次 watch」的标的,前端仍以 payload.underlying 过滤
+            uly = ""
+            exp = ""
+            if self._watchers:
+                last = next(reversed(list(self._watchers.values())))
+                uly = str(last.get("underlying") or "")
+                exp = str(last.get("exp_time") or "")
+            # 若本批只有单一 underlying 的 quotes/index,优先用它
+            quote_ulys = {str(q.get("underlying") or "").upper() for q in quotes if q.get("underlying")}
+            if len(dirty_uly) == 1:
+                uly = next(iter(dirty_uly))
+            elif len(quote_ulys) == 1:
+                uly = next(iter(quote_ulys))
+            payload = {
+                "ok": True,
+                "live": True,
+                "ws_ok": self._ws.connected,
+                "version": self._version,
+                "underlying": uly,
+                "watch_exp": exp,
+                "index_px": self._index_by_uly.get(uly),
+                "indexes": dict(self._index_by_uly),
+                "quotes": quotes,
+                "ts": int(time.time() * 1000),
+            }
+        return json.dumps(payload, ensure_ascii=False)
+
+    def _broadcast(self, event: str | None = None, *, close: bool = False) -> None:
+        with self._lock:
+            subs = list(self._subscribers)
+        dead: list[queue.Queue[str | None]] = []
+        for q in subs:
+            try:
+                q.put_nowait(None if close else event)
+            except Exception:
+                dead.append(q)
+        if dead:
+            with self._lock:
+                for q in dead:
+                    if q in self._subscribers:
+                        self._subscribers.remove(q)
+
+    def _subscribe(self) -> queue.Queue[str | None]:
+        q: queue.Queue[str | None] = queue.Queue(maxsize=64)
+        with self._lock:
+            self._subscribers.append(q)
+        return q
+
+    def _unsubscribe(self, q: queue.Queue[str | None]) -> None:
+        with self._lock:
+            if q in self._subscribers:
+                self._subscribers.remove(q)
+
+    def iter_sse(self) -> Iterator[str]:
+        q = self._subscribe()
+        try:
+            yield self._format_event(
+                {
+                    "ok": True,
+                    "reason": "connect",
+                    **self.status(),
+                    "quotes": [],
+                    "ts": int(time.time() * 1000),
+                }
+            )
+            while True:
+                try:
+                    raw = q.get(timeout=OPTIONS_QUOTE_SSE_HEARTBEAT_SEC)
+                except queue.Empty:
+                    yield ": heartbeat\n\n"
+                    continue
+                if raw is None:
+                    break
+                yield f"event: quotes\ndata: {raw}\n\n"
+        finally:
+            self._unsubscribe(q)
+
+    @staticmethod
+    def _format_event(data: dict[str, Any]) -> str:
+        return "event: quotes\ndata: " + json.dumps(data, ensure_ascii=False) + "\n\n"
+
+
+options_quote_live = OptionsQuoteLive()
+
+
+def start_options_quote_live() -> OptionsQuoteLive:
+    options_quote_live.start()
+    return options_quote_live
+
+
+def register_options_quote_live_routes(app: Any, login_required: Callable) -> None:
+    from flask import Response, jsonify, request, stream_with_context
+
+    start_options_quote_live()
+
+    @app.route("/api/options/quotes/stream")
+    @login_required
+    def api_options_quotes_stream():
+        return Response(
+            stream_with_context(options_quote_live.iter_sse()),
+            mimetype="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+            },
+        )
+
+    @app.route("/api/options/quotes/watch", methods=["POST"])
+    @login_required
+    def api_options_quotes_watch():
+        data = request.get_json(silent=True) or {}
+        contracts = data.get("contracts") or []
+        if not contracts and data.get("inst_ids"):
+            contracts = [{"inst_id": x} for x in (data.get("inst_ids") or [])]
+        st = options_quote_live.watch(
+            underlying=str(data.get("underlying") or "ETH"),
+            exp_time=data.get("exp_time"),
+            contracts=contracts,
+            index_inst_id=data.get("index_inst_id"),
+            watcher_id=str(data.get("watcher_id") or "default"),
+        )
+        return jsonify({"ok": True, **st})
+
+    @app.route("/api/options/quotes/status")
+    @login_required
+    def api_options_quotes_status():
+        return jsonify(options_quote_live.status())

lib/options/options_register.py

@@ -61,6 +61,14 @@ def install_options_trading(app: Flask, repo_root: str, app_module: Any) -> None
     register_options_routes(app, cfg)
     _register_options_hub_bridge(app, cfg)
     if enabled:
+        try:
+            from lib.options.options_quote_live_lib import register_options_quote_live_routes
+
+            register_options_quote_live_routes(app, cfg["login_required"])
+        except Exception as e:
+            import logging
+
+            logging.getLogger(__name__).exception("options quote live init failed: %s", e)
         _start_monitor_thread(app, cfg)
 
 
@@ -381,6 +389,14 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
         # 热更新:每次读 env,保存配置后刷新链即可生效
         ask_liq_filter = _env_bool("OKX_OPTIONS_CHAIN_ASK_LIQ_FILTER_ENABLED", True)
         budget_buffer = _env_float("OKX_OPTIONS_BUDGET_BUFFER", 0.95)
+        if expiries:
+            try:
+                from lib.options.options_quote_live_lib import options_quote_live
+
+                watch_exp = (request.args.get("exp_time") or "").strip() or None
+                options_quote_live.seed_from_chain(chain, exp_time=watch_exp)
+            except Exception:
+                pass
         if not expiries:
             return jsonify(
                 {
@@ -401,6 +417,7 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
                 "ask_liq_filter_enabled": ask_liq_filter,
                 "budget_buffer": budget_buffer,
                 "trade_budget": cfg["trade_budget"],
+                "quote_live": True,
             }
         )
 

lib/options/templates/options_panel.html

@@ -322,4 +322,4 @@
     </div>
 </div>
 <script src="/static/options_expiry_countdown.js?v=1"></script>
-<script src="/static/options_panel.js?v=57"></script>
+<script src="/static/options_panel.js?v=58"></script>