fix(hub): restore pre-WS quote path and relieve Gate/account contention

Two audit rounds after WS rollback: lighten hub options snapshot, cache hub balances without extra fetch_balance, soft-poll single-flight, and document fixes in R1/R2 reports.

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

crypto_monitor_binance/app.py

@@ -10042,14 +10042,14 @@ def _hub_meta_bundle():
 
 
 def _hub_account_bundle():
-    funding_capital, trading_capital = get_exchange_capitals(force=True)
+    # 中控看板高频拉取:仅走余额缓存,避免额外 fetch_balance
+    funding_capital, trading_capital = get_exchange_capitals(force=False)
     funding_usdt = round(funding_capital, FUNDS_DECIMALS) if funding_capital is not None else None
     trading_usdt = round(trading_capital, FUNDS_DECIMALS) if trading_capital is not None else None
-    available = get_available_trading_usdt()
     return {
         "funding_usdt": funding_usdt,
         "trading_usdt": trading_usdt,
-        "available_trading_usdt": round(available, FUNDS_DECIMALS) if available is not None else None,
+        "available_trading_usdt": trading_usdt,
         "trading_day": get_trading_day(app_now()),
     }
 

crypto_monitor_gate/app.py

@@ -467,6 +467,8 @@ from lib.exchange.gate_ccxt_lib import gate_ccxt_class
 # Gate.io USDT 永续(swap)
 exchange = gate_ccxt_class()({
     "enableRateLimit": True,
+    # 避免关键位监控/账户拉取无限挂起拖垮中控
+    "timeout": int(os.getenv("GATE_CCXT_TIMEOUT_MS", "8000")),
     "options": {
         "defaultType": "swap",
         "defaultMarginMode": _GATE_DEFAULT_MARGIN_MODE,
@@ -4611,14 +4613,25 @@ def _finalize_key_monitor_one_shot(conn, row, last_msg, close_reason):
     conn.execute("DELETE FROM key_monitors WHERE id=?", (row["id"],))
 
 
+_RS_BAR_CACHE: dict[str, dict] = {}
+_RS_BAR_CACHE_TTL_SEC = float(os.getenv("GATE_RS_BAR_CACHE_SEC", "45"))
+
+
 def _fetch_last_closed_bar(symbol):
-    """最近一根闭合 K:[ts, o, h, l, c, v] 或 None."""
+    """最近一根闭合 K:[ts, o, h, l, c, v] 或 None.短缓存减轻关键位监控打爆 ccxt."""
     ex_sym = normalize_exchange_symbol(symbol)
+    now = time.time()
+    cached = _RS_BAR_CACHE.get(ex_sym)
+    if cached and now - float(cached.get("updated_at") or 0) < _RS_BAR_CACHE_TTL_SEC:
+        return cached.get("bar")
     bars = exchange.fetch_ohlcv(ex_sym, timeframe=KLINE_TIMEFRAME, limit=5) or []
     if len(bars) < 2:
+        _RS_BAR_CACHE[ex_sym] = {"updated_at": now, "bar": None}
         return None
     closed = bars[:-1]
-    return closed[-1] if closed else None
+    bar = closed[-1] if closed else None
+    _RS_BAR_CACHE[ex_sym] = {"updated_at": now, "bar": bar}
+    return bar
 
 
 def _key_rs_gate_preview(symbol, upper, lower):
@@ -9893,14 +9906,14 @@ def _hub_meta_bundle():
 
 
 def _hub_account_bundle():
-    funding_capital, trading_capital = get_exchange_capitals(force=True)
+    # 中控看板高频拉取:仅走余额缓存;不再额外 fetch_balance(会与关键位监控争用 ccxt)
+    funding_capital, trading_capital = get_exchange_capitals(force=False)
     funding_usdt = round(funding_capital, 2) if funding_capital is not None else None
     trading_usdt = round(trading_capital, 2) if trading_capital is not None else None
-    available = get_available_trading_usdt()
     return {
         "funding_usdt": funding_usdt,
         "trading_usdt": trading_usdt,
-        "available_trading_usdt": round(available, 2) if available is not None else None,
+        "available_trading_usdt": trading_usdt,
         "trading_day": get_trading_day(app_now()),
     }
 

crypto_monitor_okx/app.py

@@ -9665,14 +9665,14 @@ def _hub_meta_bundle():
 
 
 def _hub_account_bundle():
-    funding_capital, trading_capital = get_exchange_capitals(force=True)
+    # 中控看板高频拉取:仅走余额缓存,避免额外 fetch_balance
+    funding_capital, trading_capital = get_exchange_capitals(force=False)
     funding_usdt = round(funding_capital, FUNDS_DECIMALS) if funding_capital is not None else None
     trading_usdt = round(trading_capital, FUNDS_DECIMALS) if trading_capital is not None else None
-    available = get_available_trading_usdt()
     return {
         "funding_usdt": funding_usdt,
         "trading_usdt": trading_usdt,
-        "available_trading_usdt": round(available, FUNDS_DECIMALS) if available is not None else None,
+        "available_trading_usdt": trading_usdt,
         "trading_day": get_trading_day(app_now()),
     }
 

docs/审计修复报告-WS回滚与中控可用-2026-08-11-R1.md

@@ -0,0 +1,53 @@
+# 审计修复报告 · WS 回滚与中控可用性(2026-08-11 · 第 1 轮)
+
+## 背景
+
+期权链接入 OKX WS 推送后,生产中控出现「期权数据不可用 / 子代理不可用」。按要求**先回滚 WS 链路**,再全量审计并修复。
+
+## 回滚
+
+| 提交 | 说明 |
+|------|------|
+| `a488e2f` | Revert fast-path(依赖 WS 热缓存) |
+| `d592632` | Revert OKX WS + SSE 推送整栈 |
+
+恢复为 **REST 拉链 + 前端约 15s soft-poll**(commit `24bb853` 行为),删除:
+
+- `lib/exchange/okx_public_ws_lib.py`
+- `lib/options/options_quote_live_lib.py`
+- `tests/test_options_quote_live_lib.py`
+
+## 根因结论(非仅 WS)
+
+| 级别 | 问题 | 证据 |
+|------|------|------|
+| Critical | Gate 关键位 RS 监控在后台线程高频 `fetch_ohlcv`,与 `/api/hub/account` 争用同一 ccxt 客户端,账户/子代理超时 | 日志 `[key_rs_level_alert] BTC/USDT id=13`;本机 `5000/api/hub/account` 25s 超时 |
+| Critical | 中控期权快照对每仓拉 books 深度,易超 `HUB_FLASK_TIMEOUT=10` | `build_display_option_positions` → `attach_close_preview` → `fetch_option_book_depth` |
+| High | Soft 拉链 3 次重试 + 无单飞,易与 SSE tick 叠打 OKX | `options_panel.js` loadChain |
+| High | 中控账户接口每轮 `force=True` 绕过余额缓存 | Gate/OKX/Binance `_hub_account_bundle` |
+| Medium | Flask 超时错误只有 `error` 无 `msg`,前端易落默认文案 | `hub.py` `_fetch_flask_json` |
+| Medium | `options` 为 null 时前端当成「0 仓」而非不可用 | `app.js` renderOptionsSectionBody |
+
+WS 部署触发的**全进程重启**放大了 Gate 争用与快照超时,表现为「全不可用」;OKX 快照在轻负载下仍可 `ok:true`。
+
+## 本轮修复
+
+1. **Hub 期权快照**关闭逐仓 `close_preview`/books(`with_close_preview=False`)
+2. **Hub 账户**三所改为 `get_exchange_capitals(force=False)`
+3. **Gate ccxt** 增加 `timeout=8000ms`;RS K 线 **45s 缓存**
+4. **期权 tickers** 恢复 **10s** 短缓存(无 WS)
+5. **前端 soft 拉链**:单飞 + soft 仅 1 次尝试;已有链不先清空表格
+6. **Hub**:超时补 `msg`;期权快照与 account/monitor **并行 gather**
+7. **中控 UI**:capabilities 含 options 且 snapshot 缺失时显式「期权数据不可用」
+
+## 测试建议
+
+- 强刷中控监控区:OKX 期权资金/持仓应恢复;Gate 子代理 status 应在数秒内恢复
+- 期权页「刷新链」不应长时间白屏;指数行显示约 15s 静默刷新
+- Gate 关键位监控日志不应再每秒刷屏 `fetch_ohlcv` 失败
+
+## 残留风险(交第 2 轮)
+
+- Gate 仍与监控共用单一 ccxt 客户端(未加全局锁)
+- 中控 board 仍可能被最慢交易所拉长整轮等待
+- Soft-poll 仍是 REST,非真·实时

docs/审计修复报告-WS回滚与中控可用-2026-08-11-R2.md

@@ -0,0 +1,41 @@
+# 审计修复报告 · WS 回滚与中控可用性(2026-08-11 · 第 2 轮)
+
+## 范围
+
+复查第 1 轮修复是否引入回归,并扫清仍会导致「中控不可用」的残留高优先级问题。
+
+## 复查结论
+
+| 项 | 结论 |
+|----|------|
+| Hub 并行 options 索引进位 | 正确(day / options 组合无错位) |
+| Hub 关闭 close_preview | UI 降级为 upl/`—`,不崩 |
+| Gate RS 缓存 / timeout | timeout 已为 int;缓存可接受 |
+| board row capabilities | `_fetch_agent_status` 始终带上 |
+| `with_close_preview` 默认 | 实例路径仍为 True |
+
+## 本轮新发现问题与修复
+
+| 级别 | 问题 | 修复 |
+|------|------|------|
+| High | Hub 账户在 `force=False` 后仍调用 `get_available_trading_usdt()` 再打一枪 `fetch_balance`,Gate 争用依旧 | 三所 `_hub_account_bundle` 改为用缓存的 `trading_usdt` 作为 `available_trading_usdt` |
+| Medium | `loadChain` soft 门禁在 `seq++` 之后,叠刷可导致 `chainLoadInFlight` 永不清理 | soft 门禁移到 `seq++` 之前 |
+
+## 与第 1 轮一并交付的状态
+
+- WS 推送链路已回滚(REST + 15s soft-poll)
+- 中控期权快照轻量化 + 并行拉取
+- Gate RS K 线短缓存 + ccxt timeout
+- 期权 tickers 10s 缓存;soft 单飞/单次尝试
+
+## 已知残留(不阻塞本次部署)
+
+- Gate 监控与账户仍共用单一 ccxt 客户端(无全局锁)
+- 中控 board 仍可能被最慢交易所拉长整轮
+- Soft-poll 非真·实时报价
+
+## 部署后验收
+
+1. 中控强刷:OKX 期权区有资金数字,不再长期「期权数据不可用」
+2. Gate 卡:子代理恢复绿色/有资金;不再长时间「子代理不可用」
+3. 期权页刷新链不白屏;约 15s 静默更新时间戳

lib/common/static/options_panel.js

@@ -7,10 +7,6 @@
   root.setAttribute("data-options-booted", "1");
 
   const panelCache = (window.__optionsPanelCache = window.__optionsPanelCache || {});
-  if (!panelCache.quoteWatcherId) {
-    panelCache.quoteWatcherId =
-      "w" + Date.now().toString(36) + Math.random().toString(36).slice(2, 8);
-  }
 
   const state = {
     underlying: root.dataset.defaultUnderly || "ETH",
@@ -44,17 +40,12 @@
   let chainSoftTimer = null;
   let lastChainSoftAt = 0;
   let chainQuotedAt = 0;
-  let quoteLiveEs = null;
-  let quoteLiveReconnectTimer = null;
-  let quoteLiveOk = false;
-  let quoteLiveWsOk = false;
-  let lastOrderQuoteLiveAt = 0;
+  let chainLoadInFlight = false;
   let pendingTtlSeconds = 600;
   const POSITIONS_STALE_MS = 45000;
   const PENDING_POLL_MS = 8000;
-  /** SSE/WS 断开时的 REST 兜底;连上后停用 */
-  const CHAIN_SOFT_POLL_MS = 30000;
-  const ORDER_QUOTE_LIVE_MIN_MS = 800;
+  /** 链卖一/买一静默刷新节流:无推送,靠拉;过密会撞 OKX 50011 */
+  const CHAIN_SOFT_POLL_MS = 15000;
   const orderPanelHome = (function () {
     const host = document.getElementById("opt-order-panel-host");
     return host ? host.parentElement : null;
@@ -669,214 +660,17 @@
     const line = document.getElementById("opt-index-line");
     if (line) {
       const liqHint = askLiqFilterOn() ? "仅显示卖一深度≥1张" : "显示全部卖一(含估算~)";
-      let liveHint = "";
-      if (quoteLiveOk && quoteLiveWsOk) {
-        liveHint = chainQuotedAt
-          ? " · WS实时 " + fmtChainQuotedAt()
-          : " · WS实时";
-      } else if (quoteLiveOk) {
-        liveHint = " · 推送已连,等待 OKX WS…";
-      } else if (chainQuotedAt) {
-        liveHint = " · 链报价 " + fmtChainQuotedAt() + "（REST兜底）";
-      }
+      const ageHint = chainQuotedAt ? " · 链报价 " + fmtChainQuotedAt() + "（约每15s静默刷新）" : "";
       line.textContent =
         "指数 " + state.underlying + " ≈ " + fmt(idx, 2) +
-        " · 默认最近一期 · " + liqHint + " · 实值含平值 · 虚值=价外" + liveHint;
+        " · 默认最近一期 · " + liqHint + " · 实值含平值 · 虚值=价外" + ageHint;
     }
   }
 
-  function findChainContract(instId) {
-    if (!state.chain || !instId) return null;
-    const exps = state.chain.expiries || [];
-    for (let i = 0; i < exps.length; i++) {
-      const contracts = exps[i].contracts || [];
-      for (let j = 0; j < contracts.length; j++) {
-        if (String(contracts[j].inst_id) === String(instId)) return contracts[j];
-      }
-    }
-    return null;
-  }
-
-  function currentExpiryContracts() {
-    if (!state.chain) return [];
-    const expMs = (document.getElementById("opt-exp-select") || {}).value;
-    const exp = (state.chain.expiries || []).find(function (e) {
-      return String(e.exp_time) === String(expMs);
-    });
-    return (exp && exp.contracts) || [];
-  }
-
-  async function watchCurrentExpiryQuotes() {
-    if (!state.chain) return;
-    const expMs = (document.getElementById("opt-exp-select") || {}).value;
-    const contracts = currentExpiryContracts().map(function (c) {
-      return {
-        inst_id: c.inst_id,
-        opt_type: c.opt_type,
-        strike: c.strike,
-        tick_sz: c.tick_sz,
-      };
-    });
-    if (!contracts.length) return;
-    try {
-      await apiJson("/api/options/quotes/watch", {
-        method: "POST",
-        headers: { "Content-Type": "application/json" },
-        body: JSON.stringify({
-          underlying: state.underlying,
-          exp_time: expMs,
-          contracts: contracts,
-          index_inst_id: state.underlying + "-USD",
-          watcher_id: panelCache.quoteWatcherId,
-        }),
-      });
-    } catch (_) {
-      /* ignore watch errors; soft poll fallback remains */
-    }
-  }
-
-  function patchListRowDom(instId, c) {
-    const tr = document.querySelector('#opt-strike-tbody tr.opt-strike-row[data-inst="' + instId + '"]');
-    if (!tr || !c) return;
-    const indexPx = state.chain && state.chain.index_px;
-    const tds = tr.children;
-    if (tds.length < 8) return;
-    tds[3].textContent = "";
-    tds[3].className = "opt-px-sz";
-    tds[3].innerHTML = fmtPxSz(c.ask, c.ask_sz, c.ask_estimated);
-    tds[4].className = "opt-chain-lev";
-    tds[4].textContent = fmtChainLeverage(calcAskLeverage(indexPx, c.ask));
-    tds[5].className = "opt-px-sz";
-    tds[5].innerHTML = fmtPxSz(c.bid, c.bid_sz);
-    tds[6].textContent = c.expiry_be_px != null ? fmt(c.expiry_be_px, 0) : "—";
-    tds[7].className = distBeClass(c.dist_expiry_be);
-    tds[7].textContent = fmtDist(c.dist_expiry_be);
-  }
-
-  function patchTRowDom(instId, c) {
-    if (!c) return;
-    const callTr = document.querySelector(
-      '#opt-strike-tbody tr.opt-strike-row-t[data-call-inst="' + instId + '"]'
-    );
-    const putTr = document.querySelector(
-      '#opt-strike-tbody tr.opt-strike-row-t[data-put-inst="' + instId + '"]'
-    );
-    const tr = callTr || putTr;
-    if (!tr) return;
-    const callInst = tr.getAttribute("data-call-inst");
-    const putInst = tr.getAttribute("data-put-inst");
-    const call = callInst ? findChainContract(callInst) : null;
-    const put = putInst ? findChainContract(putInst) : null;
-    const callOk = call && (!askLiqFilterOn() || hasAskLiquidity(call)) ? call : null;
-    const putOk = put && (!askLiqFilterOn() || hasAskLiquidity(put)) ? put : null;
-    const tds = tr.children;
-    if (tds.length < 9) return;
-    tds[0].innerHTML = callOk ? fmtPxSz(callOk.ask, callOk.ask_sz, callOk.ask_estimated) : "—";
-    tds[7].innerHTML = putOk ? fmtPxSz(putOk.ask, putOk.ask_sz, putOk.ask_estimated) : "—";
-    const combined = straddleAskPerUnit(callOk && callOk.ask, putOk && putOk.ask);
-    tds[4].innerHTML = formatStraddlePremiumCell(callOk && callOk.ask, putOk && putOk.ask);
-    tds[5].innerHTML = formatStraddleBand(tr.getAttribute("data-strike"), combined);
-  }
-
-  function applyLiveQuotes(payload) {
-    if (!payload || !state.chain) return;
-    const uly = String(state.underlying || "").toUpperCase();
-    if (payload.indexes && payload.indexes[uly] != null && Number.isFinite(Number(payload.indexes[uly]))) {
-      state.chain.index_px = Number(payload.indexes[uly]);
-    } else if (payload.underlying && String(payload.underlying).toUpperCase() === uly) {
-      if (payload.index_px != null && Number.isFinite(Number(payload.index_px))) {
-        state.chain.index_px = Number(payload.index_px);
-      }
-    } else if (payload.underlying && String(payload.underlying).toUpperCase() !== uly) {
-      // 别的标的推送:仍可 patch 本页已有合约
-    }
-    const quotes = payload.quotes || [];
-    quotes.forEach(function (q) {
-      const instId = q && q.inst_id;
-      if (!instId) return;
-      if (q.underlying && String(q.underlying).toUpperCase() !== uly) return;
-      const c = findChainContract(instId);
-      if (!c) return;
-      if (q.ask !== undefined) c.ask = q.ask;
-      if (q.bid !== undefined) c.bid = q.bid;
-      if (q.ask_sz !== undefined) c.ask_sz = q.ask_sz;
-      if (q.bid_sz !== undefined) c.bid_sz = q.bid_sz;
-      if (q.mark_px !== undefined) c.mark_px = q.mark_px;
-      if (q.ask_estimated !== undefined) c.ask_estimated = !!q.ask_estimated;
-      if (q.expiry_be_px !== undefined) c.expiry_be_px = q.expiry_be_px;
-      if (q.dist_expiry_be !== undefined) c.dist_expiry_be = q.dist_expiry_be;
-      if (state.chainView === "t") patchTRowDom(instId, c);
-      else patchListRowDom(instId, c);
-    });
-    if (payload.ts) chainQuotedAt = Number(payload.ts) || Date.now();
-    else if (quotes.length || payload.index_px != null) chainQuotedAt = Date.now();
-    quoteLiveWsOk = payload.ws_ok !== false;
-    renderIndexLine();
-    if (state.selectedInst && quotes.some(function (q) { return q && q.inst_id === state.selectedInst; })) {
-      const now = Date.now();
-      if (now - lastOrderQuoteLiveAt >= ORDER_QUOTE_LIVE_MIN_MS) {
-        lastOrderQuoteLiveAt = now;
-        void selectContract(state.selectedInst, null, true);
-      }
-    }
-  }
-
-  function stopQuoteLiveStream() {
-    if (quoteLiveReconnectTimer) {
-      clearTimeout(quoteLiveReconnectTimer);
-      quoteLiveReconnectTimer = null;
-    }
-    if (quoteLiveEs) {
-      try { quoteLiveEs.close(); } catch (_) {}
-      quoteLiveEs = null;
-    }
-    quoteLiveOk = false;
-    quoteLiveWsOk = false;
-  }
-
-  function startQuoteLiveStream() {
-    if (quoteLiveEs) return;
-    if (typeof EventSource === "undefined") return;
-    try {
-      quoteLiveEs = new EventSource("/api/options/quotes/stream");
-    } catch (_) {
-      quoteLiveOk = false;
-      return;
-    }
-    quoteLiveEs.addEventListener("quotes", function (ev) {
-      try {
-        const data = JSON.parse(ev.data || "{}");
-        quoteLiveOk = true;
-        if (data.reason === "connect") {
-          quoteLiveWsOk = !!data.ws_ok;
-          renderIndexLine();
-          return;
-        }
-        applyLiveQuotes(data);
-      } catch (_) {}
-    });
-    quoteLiveEs.onopen = function () {
-      quoteLiveOk = true;
-      renderIndexLine();
-      void watchCurrentExpiryQuotes();
-    };
-    quoteLiveEs.onerror = function () {
-      quoteLiveOk = false;
-      quoteLiveWsOk = false;
-      renderIndexLine();
-      stopQuoteLiveStream();
-      quoteLiveReconnectTimer = setTimeout(function () {
-        quoteLiveReconnectTimer = null;
-        startQuoteLiveStream();
-      }, 8000);
-    };
-  }
-
   function softRefreshChainThrottled(force) {
     if (document.hidden) return;
     if (!document.getElementById("options-root")) return;
-    // WS 推送正常时不靠 REST 刷卖一,避免 50011;仅结构兜底可 force
-    if (!force && quoteLiveOk && quoteLiveWsOk) return;
+    if (chainLoadInFlight) return;
     const now = Date.now();
     if (!force && now - lastChainSoftAt < CHAIN_SOFT_POLL_MS) return;
     lastChainSoftAt = now;
@@ -1447,16 +1241,15 @@
 
   async function loadChain(opts) {
     const soft = !!(opts && opts.soft);
+    // soft 门禁必须在 seq++ 之前,否则叠刷会抬高 seq 导致 inFlight 永不清理
+    if (chainLoadInFlight && soft) return;
     const uly = state.underlying;
     const seq = ++chainLoadSeq;
     const btn = document.getElementById("opt-load-chain");
     const hadChain = chainHasExpiries(state.chain) && state.chain.underlying === uly;
-    if (btn && !soft) {
-      btn.disabled = true;
-      if (!btn.dataset.origText) btn.dataset.origText = btn.textContent || "刷新链";
-      btn.textContent = "刷新中…";
-    }
-    // 已有链时不先清空表格,避免「白屏等很久」的体感
+    chainLoadInFlight = true;
+    if (btn && !soft) btn.disabled = true;
+    // 已有链时不先清空,避免刷新白屏
     if (!soft && !hadChain) {
       setExpirySelectStatus("加载到期日中…");
       const tbody = document.getElementById("opt-strike-tbody");
@@ -1468,18 +1261,11 @@
     try {
       let d = null;
       let lastMsg = "";
-      const expMs = (document.getElementById("opt-exp-select") || {}).value || "";
-      // WS 已热时走 fast,跳过最慢的整家族 REST tickers
-      const useFast = soft || quoteLiveWsOk || hadChain;
-      const maxAttempts = soft ? 2 : 3;
+      // soft 只试 1 次,避免与 15s 轮询叠加重试打爆 OKX
+      const maxAttempts = soft ? 1 : 3;
       for (let attempt = 0; attempt < maxAttempts; attempt++) {
         if (seq !== chainLoadSeq) return;
-        let url =
-          "/api/options/chain?underlying=" +
-          encodeURIComponent(uly) +
-          (useFast ? "&fast=1" : "");
-        if (expMs) url += "&exp_time=" + encodeURIComponent(expMs);
-        d = await apiJson(url);
+        d = await apiJson("/api/options/chain?underlying=" + encodeURIComponent(uly));
         if (seq !== chainLoadSeq) return;
         if (d && d.ok && chainHasExpiries(d)) break;
         lastMsg = (d && (d.msg || d.chain_error)) || "暂无到期日";
@@ -1494,7 +1280,7 @@
             );
           }
           await new Promise(function (resolve) {
-            setTimeout(resolve, rateLimited ? 1200 * (attempt + 1) : 300);
+            setTimeout(resolve, rateLimited ? 1200 * (attempt + 1) : 400);
           });
         }
       }
@@ -1504,7 +1290,6 @@
           if (!soft) {
             renderExpiries();
             renderStrikes();
-            void watchCurrentExpiryQuotes();
           }
           return;
         }
@@ -1526,7 +1311,7 @@
         alert(friendly);
         return;
       }
-      const keepExp = soft || hadChain ? (document.getElementById("opt-exp-select") || {}).value : "";
+      const keepExp = soft ? (document.getElementById("opt-exp-select") || {}).value : "";
       state.chain = d;
       panelCache.chain = d;
       panelCache.underlying = uly;
@@ -1534,7 +1319,7 @@
       chainQuotedAt = Date.now();
       lastChainSoftAt = chainQuotedAt;
       syncAskLiqFilterFromChain(d);
-      if (!soft && !hadChain) {
+      if (!soft) {
         state.selectedInst = null;
         resetMoneyFilterToAll();
         state.strikeExpandAll = false;
@@ -1544,19 +1329,16 @@
       }
       updateUnderlyingLabel();
       renderExpiries();
-      if (keepExp) {
+      if (soft && keepExp) {
         const sel = document.getElementById("opt-exp-select");
         if (sel && Array.from(sel.options).some(function (o) { return o.value === keepExp; })) {
           sel.value = keepExp;
         }
       }
-      // soft/已有链时保留 selectedInst；renderStrikes 会先 park 再按 prevSelected 静默重挂下单面板
+      // soft 时保留 selectedInst；renderStrikes 会先 park 再按 prevSelected 静默重挂下单面板
       renderStrikes();
-      void watchCurrentExpiryQuotes();
-      startQuoteLiveStream();
     } catch (e) {
       if (seq !== chainLoadSeq || soft) return;
-      if (hadChain) return;
       setExpirySelectStatus("选择到期日");
       const tbody = document.getElementById("opt-strike-tbody");
       if (tbody) {
@@ -1566,9 +1348,9 @@
           "</td></tr>";
       }
     } finally {
-      if (seq === chainLoadSeq && btn) {
-        btn.disabled = false;
-        if (btn.dataset.origText) btn.textContent = btn.dataset.origText;
+      if (seq === chainLoadSeq) {
+        chainLoadInFlight = false;
+        if (btn) btn.disabled = false;
       }
     }
   }
@@ -2466,7 +2248,6 @@
     const expandCb = document.getElementById("opt-strike-expand-all");
     if (expandCb) expandCb.checked = false;
     renderStrikes();
-    void watchCurrentExpiryQuotes();
   }
 
   function bootOptionsPanel() {
@@ -2478,7 +2259,6 @@
     refreshPendingOrders();
     startPendingOrdersPoll();
     startChainSoftPoll();
-    startQuoteLiveStream();
     const hasCache =
       chainHasExpiries(panelCache.chain) &&
       panelCache.underlying === state.underlying &&
@@ -2488,8 +2268,7 @@
       renderExpiries();
       renderStrikes();
       refreshAllPositions();
-      void watchCurrentExpiryQuotes();
-      // 后台静默刷新结构;卖一优先走 WS
+      // 后台静默刷新,避免缓存过期后到期日变空 / 卖一过期
       softRefreshChainThrottled(true);
       return;
     }
@@ -2614,7 +2393,7 @@
   window.OptionsPanelLive = {
     refreshSoft: function () {
       refreshAllPositions();
-      // 有 WS 实时报价时不再 REST 刷链;断开时才兜底
+      // embed SSE 只通知「该拉了」,不推送链报价;这里节流拉新鲜卖一/买一
       softRefreshChainThrottled(false);
     },
     refreshChain: loadChain,

lib/exchange/okx_options_lib.py

@@ -32,7 +32,7 @@ _INSTRUMENTS_CACHE_TTL_SEC = 90.0
 _INSTRUMENTS_STALE_SEC = 600.0
 _TICKERS_CACHE: dict[str, dict[str, Any]] = {}
 _TICKERS_CACHE_LOCK = threading.Lock()
-_TICKERS_CACHE_TTL_SEC = 8.0
+_TICKERS_CACHE_TTL_SEC = float(os.getenv("OKX_OPTIONS_TICKERS_CACHE_SEC", "10") or "10")
 
 
 def invalidate_options_balance_cache() -> None:
@@ -730,7 +730,7 @@ def fetch_option_tickers(
     if (
         not force
         and cached
-        and now - float(cached.get("updated_at") or 0) < _TICKERS_CACHE_TTL_SEC
+        and now - float(cached.get("updated_at") or 0) < max(1.0, _TICKERS_CACHE_TTL_SEC)
         and isinstance(cached.get("rows"), dict)
         and cached["rows"]
     ):
@@ -771,9 +771,6 @@ def build_option_chain(
     itm_only: bool = True,
     itm_max_dist_usd: float = 30.0,
     index_px: float | None = None,
-    tickers_override: dict[str, dict[str, Any]] | None = None,
-    fetch_tickers: bool = True,
-    force_tickers: bool = False,
 ) -> dict[str, Any]:
     u = (underlying or "ETH").upper()
     family = f"{u}-USD_UM"
@@ -794,13 +791,7 @@ def build_option_chain(
         rate_limited = _is_okx_rate_limit(e)
         if rate_limited:
             instruments_err = "OKX 请求过于频繁(50011),请稍后点「刷新链」重试"
-    tickers: dict[str, dict[str, Any]] = {}
-    if fetch_tickers:
-        tickers = fetch_option_tickers(ex, family, force=force_tickers)
-    if tickers_override:
-        for iid, row in tickers_override.items():
-            if isinstance(row, dict) and iid:
-                tickers[str(iid)] = {**(tickers.get(str(iid)) or {}), **row}
+    tickers = fetch_option_tickers(ex, family)
     expiries: dict[str, list[dict[str, Any]]] = {}
     skipped_no_index = 0
     for meta in instruments:

lib/exchange/okx_public_ws_lib.py

@@ -1,199 +0,0 @@
-"""OKX 公共 WebSocket(同步线程):订阅 tickers / index-tickers,自动重连."""
-from __future__ import annotations
-
-import json
-import logging
-import threading
-import time
-from collections.abc import Callable
-from typing import Any
-
-logger = logging.getLogger(__name__)
-
-OKX_PUBLIC_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
-_SUBSCRIBE_CHUNK = 40
-_APP_PING_SEC = 20.0
-
-
-class OkxPublicWs:
-    """单连接公共 WS;set_subscriptions 全量对齐目标频道."""
-
-    def __init__(
-        self,
-        *,
-        on_data: Callable[[dict[str, Any]], None],
-        url: str = OKX_PUBLIC_WS_URL,
-        name: str = "okx-public-ws",
-    ) -> None:
-        self._on_data = on_data
-        self._url = url
-        self._name = name
-        self._lock = threading.RLock()
-        self._desired: dict[str, dict[str, str]] = {}
-        self._active: set[str] = set()
-        self._stop = threading.Event()
-        self._thread: threading.Thread | None = None
-        self._ws: Any = None
-        self._connected = False
-        self._last_msg_at = 0.0
-
-    @property
-    def connected(self) -> bool:
-        return self._connected
-
-    @property
-    def last_msg_at(self) -> float:
-        return self._last_msg_at
-
-    def start(self) -> None:
-        if self._thread and self._thread.is_alive():
-            return
-        self._stop.clear()
-        self._thread = threading.Thread(target=self._run_loop, name=self._name, daemon=True)
-        self._thread.start()
-
-    def stop(self) -> None:
-        self._stop.set()
-        ws = self._ws
-        if ws is not None:
-            try:
-                ws.close()
-            except Exception:
-                pass
-        if self._thread and self._thread.is_alive():
-            self._thread.join(timeout=3.0)
-
-    def set_subscriptions(self, args: list[dict[str, str]]) -> None:
-        desired: dict[str, dict[str, str]] = {}
-        for raw in args:
-            if not isinstance(raw, dict):
-                continue
-            channel = str(raw.get("channel") or "").strip()
-            inst_id = str(raw.get("instId") or "").strip()
-            if not channel or not inst_id:
-                continue
-            key = f"{channel}:{inst_id}"
-            desired[key] = {"channel": channel, "instId": inst_id}
-        with self._lock:
-            self._desired = desired
-            ws = self._ws
-            connected = self._connected
-            active = set(self._active)
-        if connected and ws is not None:
-            self._sync_subs(ws, active, desired)
-
-    def _sync_subs(
-        self,
-        ws: Any,
-        active: set[str],
-        desired: dict[str, dict[str, str]],
-    ) -> None:
-        unsub_args: list[dict[str, str]] = []
-        for key in active - set(desired.keys()):
-            channel, _, inst_id = key.partition(":")
-            if channel and inst_id:
-                unsub_args.append({"channel": channel, "instId": inst_id})
-        sub_args = [desired[k] for k in (set(desired.keys()) - active)]
-        if unsub_args:
-            self._send_op(ws, "unsubscribe", unsub_args)
-        if sub_args:
-            self._send_op(ws, "subscribe", sub_args)
-        with self._lock:
-            self._active = set(desired.keys())
-
-    def _send_op(self, ws: Any, op: str, args: list[dict[str, str]]) -> None:
-        for i in range(0, len(args), _SUBSCRIBE_CHUNK):
-            chunk = args[i : i + _SUBSCRIBE_CHUNK]
-            try:
-                ws.send(json.dumps({"op": op, "args": chunk}, ensure_ascii=False))
-            except Exception as e:
-                logger.warning("%s %s failed: %s", self._name, op, e)
-                return
-            if i + _SUBSCRIBE_CHUNK < len(args):
-                time.sleep(0.08)
-
-    def _run_loop(self) -> None:
-        try:
-            import websocket
-        except ImportError:
-            logger.error("%s: websocket-client not installed", self._name)
-            return
-        backoff = 1.0
-        while not self._stop.is_set():
-            opened = False
-            try:
-                self._connected = False
-                with self._lock:
-                    self._active.clear()
-
-                def on_open(ws: Any) -> None:
-                    nonlocal opened
-                    opened = True
-                    self._connected = True
-                    self._last_msg_at = time.time()
-                    with self._lock:
-                        desired = dict(self._desired)
-                    self._sync_subs(ws, set(), desired)
-
-                def on_message(_ws: Any, message: str) -> None:
-                    self._last_msg_at = time.time()
-                    if message == "pong":
-                        return
-                    try:
-                        payload = json.loads(message)
-                    except Exception:
-                        return
-                    if not isinstance(payload, dict):
-                        return
-                    if payload.get("event") in ("subscribe", "unsubscribe", "error"):
-                        if payload.get("event") == "error":
-                            logger.warning("%s event error: %s", self._name, payload)
-                        return
-                    if payload.get("arg") and payload.get("data") is not None:
-                        try:
-                            self._on_data(payload)
-                        except Exception:
-                            logger.exception("%s on_data failed", self._name)
-
-                def on_error(_ws: Any, error: Any) -> None:
-                    logger.warning("%s error: %s", self._name, error)
-
-                def on_close(_ws: Any, *_args: Any) -> None:
-                    self._connected = False
-
-                self._ws = websocket.WebSocketApp(
-                    self._url,
-                    on_open=on_open,
-                    on_message=on_message,
-                    on_error=on_error,
-                    on_close=on_close,
-                )
-                ping_stop = threading.Event()
-
-                def ping_loop() -> None:
-                    while not self._stop.is_set() and not ping_stop.is_set():
-                        ws = self._ws
-                        if ws is not None and self._connected:
-                            try:
-                                ws.send("ping")
-                            except Exception:
-                                pass
-                        if ping_stop.wait(_APP_PING_SEC):
-                            break
-
-                ping_thread = threading.Thread(
-                    target=ping_loop, name=f"{self._name}-ping", daemon=True
-                )
-                ping_thread.start()
-                self._ws.run_forever(ping_interval=0)
-                ping_stop.set()
-            except Exception as e:
-                logger.warning("%s run failed: %s", self._name, e)
-            finally:
-                self._connected = False
-                self._ws = None
-            if self._stop.is_set():
-                break
-            time.sleep(backoff)
-            backoff = 1.0 if opened else min(30.0, backoff * 1.7)
-

lib/options/options_hub_lib.py

@@ -22,7 +22,8 @@ def build_options_hub_snapshot(cfg: dict[str, Any]) -> dict[str, Any]:
         raw = cfg["fetch_option_positions"](ex)
         if raw is None:
             return {"ok": False, "enabled": True, "msg": "获取期权持仓失败"}
-        positions = build_display_option_positions(cfg, ex, raw)
+        # 中控看板不拉逐仓 books(易超 HUB_FLASK_TIMEOUT);实例页仍走完整 preview
+        positions = build_display_option_positions(cfg, ex, raw, with_close_preview=False)
         target_monitors: list[dict[str, Any]] = []
         try:
             conn = cfg["get_db"]()

lib/options/options_positions_lib.py

@@ -145,8 +145,10 @@ def build_display_option_positions(
     cfg: dict[str, Any],
     ex: Any,
     raw_positions: list[dict[str, Any]],
+    *,
+    with_close_preview: bool = True,
 ) -> list[dict[str, Any]]:
-    """与实例 /api/options/positions 相同 enrichment + close_preview."""
+    """与实例 /api/options/positions 相同 enrichment;中控可关 close_preview 避免逐仓拉盘口超时."""
     meta_cache: dict[str, dict[str, Any] | None] = {}
     rows: list[dict[str, Any]] = []
     conn = cfg["get_db"]()
@@ -162,7 +164,8 @@ def build_display_option_positions(
                 meta_cache=meta_cache,
                 premium_override=premium_override,
             )
-            attach_close_preview(cfg, ex, row, premium_paid=_safe_float(row.get("premium_paid")))
+            if with_close_preview:
+                attach_close_preview(cfg, ex, row, premium_paid=_safe_float(row.get("premium_paid")))
             rows.append(row)
     finally:
         conn.close()

lib/options/options_quote_live_lib.py

@@ -1,509 +0,0 @@
-"""期权链实时报价: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})
-        # 订阅可能分片 sleep,不能堵 Flask 请求线程
-        threading.Thread(
-            target=self._ws.set_subscriptions,
-            args=(args,),
-            name="okx-options-quote-sub",
-            daemon=True,
-        ).start()
-
-    def as_okx_tickers(self, underlying: str | None = None) -> dict[str, dict[str, Any]]:
-        """转成 build_option_chain 可用的 OKX ticker 字段."""
-        u = (underlying or "").upper()
-        out: dict[str, dict[str, Any]] = {}
-        with self._lock:
-            for inst_id, q in self._tickers.items():
-                if u and str(q.get("underlying") or "").upper() not in ("", u):
-                    continue
-                row: dict[str, Any] = {"instId": inst_id}
-                if q.get("ask") is not None and not q.get("ask_estimated"):
-                    row["askPx"] = q.get("ask")
-                    row["askSz"] = q.get("ask_sz")
-                if q.get("bid") is not None:
-                    row["bidPx"] = q.get("bid")
-                    row["bidSz"] = q.get("bid_sz")
-                if q.get("mark_px") is not None:
-                    row["markPx"] = q.get("mark_px")
-                out[inst_id] = row
-        return out
-
-    def index_px_for(self, underlying: str) -> float | None:
-        u = (underlying or "").upper()
-        with self._lock:
-            return self._index_by_uly.get(u)
-
-    def is_ws_fresh(self, *, max_age_sec: float = 15.0) -> bool:
-        if not self._ws.connected:
-            return False
-        last = float(self._ws.last_msg_at or 0)
-        return last > 0 and (time.time() - last) <= max_age_sec
-
-    def schedule_seed_from_chain(
-        self,
-        chain: dict[str, Any],
-        *,
-        exp_time: str | int | None = None,
-        watcher_id: str | None = None,
-    ) -> None:
-        threading.Thread(
-            target=self.seed_from_chain,
-            kwargs={"chain": chain, "exp_time": exp_time, "watcher_id": watcher_id},
-            name="options-quote-seed",
-            daemon=True,
-        ).start()
-
-    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,14 +61,6 @@ 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)
 
 
@@ -374,24 +366,6 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
         u = (request.args.get("underlying") or cfg["default_underly"]).upper()
         # 热更新:链展示天数每次读 env,保存后刷新链即可
         chain_max_dte = _env_float("OKX_OPTIONS_CHAIN_MAX_DTE_DAYS", float(cfg.get("chain_max_dte_days") or 14))
-        fast = (request.args.get("fast") or "").strip().lower() in ("1", "true", "yes")
-        force_tickers = (request.args.get("force_tickers") or "").strip().lower() in ("1", "true", "yes")
-        watch_exp = (request.args.get("exp_time") or "").strip() or None
-        live_index = None
-        live_tickers = None
-        ws_fresh = False
-        try:
-            from lib.options.options_quote_live_lib import options_quote_live
-
-            ws_fresh = options_quote_live.is_ws_fresh()
-            live_index = options_quote_live.index_px_for(u)
-            live_tickers = options_quote_live.as_okx_tickers(u) or None
-        except Exception:
-            pass
-        # fast: WS 已热则跳过整家族 REST tickers(最慢的一步),用 WS 缓存覆盖
-        fetch_tickers = True
-        if fast and ws_fresh and not force_tickers:
-            fetch_tickers = False
         try:
             chain = cfg["build_option_chain"](
                 ex,
@@ -399,10 +373,6 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
                 max_dte_days=chain_max_dte,
                 itm_only=False,
                 itm_max_dist_usd=cfg["itm_max_dist"],
-                index_px=live_index,
-                tickers_override=live_tickers,
-                fetch_tickers=fetch_tickers,
-                force_tickers=force_tickers,
             )
         except Exception as e:
             return jsonify({"ok": False, "msg": f"加载期权链失败: {e}"})
@@ -411,13 +381,6 @@ 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
-
-                options_quote_live.schedule_seed_from_chain(chain, exp_time=watch_exp)
-            except Exception:
-                pass
         if not expiries:
             return jsonify(
                 {
@@ -428,8 +391,6 @@ 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"],
-                    "chain_fast": fast,
-                    "ws_fresh": ws_fresh,
                 }
             )
         return jsonify(
@@ -440,10 +401,6 @@ 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,
-                "chain_fast": fast,
-                "ws_fresh": ws_fresh,
-                "tickers_fetched": fetch_tickers,
             }
         )
 

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=59"></script>
+<script src="/static/options_panel.js?v=60"></script>

manual_trading_hub/hub.py

@@ -2028,7 +2028,8 @@ async def _fetch_flask_json(
             return parsed
         return _parse_http_json_body(r)
     except Exception as e:
-        return {"ok": False, "error": str(e)}
+        err = str(e)
+        return {"ok": False, "error": err, "msg": err}
 
 
 async def _notify_instance_user_close(
@@ -2567,8 +2568,8 @@ def _merge_flask_exchange_tpsl(agent_row: dict, snap: dict | None, hub_mon: dict
 
 async def _fetch_exchange_flask_bundle(
     client: httpx.AsyncClient, ex: dict, *, trading_day: str | None = None
-) -> tuple[dict | None, dict | None, list | None, dict | None, dict | None, dict | None]:
-    """单所 Flask:monitor / meta / price_snapshot / account / trades/today(有 flask_url 时)并行拉取."""
+) -> tuple:
+    """单所 Flask:monitor / meta / price_snapshot / account / trades/today / options 并行拉取."""
     caps = ex.get("capabilities") or []
     tasks = [
         _fetch_flask_json(client, ex, "/api/hub/monitor"),
@@ -2576,6 +2577,7 @@ async def _fetch_exchange_flask_bundle(
     ]
     has_flask = bool((ex.get("flask_url") or "").strip())
     day = (trading_day or "").strip()
+    want_options = has_flask and "options" in caps
     if has_flask:
         tasks.extend(
             [
@@ -2592,15 +2594,26 @@ async def _fetch_exchange_flask_bundle(
                     params={"trading_day": day},
                 )
             )
+        if want_options:
+            tasks.append(_fetch_flask_json(client, ex, "/api/hub/options/snapshot"))
     results = await asyncio.gather(*tasks)
     hub_mon = results[0]
     meta = results[1]
-    snap = results[2] if has_flask and len(results) > 2 else None
-    account = results[3] if has_flask and len(results) > 3 else None
-    trades_today = results[4] if has_flask and day and len(results) > 4 else None
+    idx = 2
+    snap = None
+    account = None
+    trades_today = None
     options_snap = None
-    if has_flask and "options" in caps:
-        options_snap = await _fetch_flask_json(client, ex, "/api/hub/options/snapshot")
+    if has_flask:
+        snap = results[idx]
+        idx += 1
+        account = results[idx]
+        idx += 1
+        if day:
+            trades_today = results[idx]
+            idx += 1
+        if want_options:
+            options_snap = results[idx]
     key_prices = None
     want_prices = HUB_BOARD_KEY_PRICES and "key" in caps
     if want_prices and isinstance(snap, dict):

manual_trading_hub/static/app.js

@@ -4022,20 +4022,26 @@
   function renderOptionsSectionBody(row, opts) {
     const options = opts || {};
     const layout = options.layout || "table";
-    const opt = row.options || {};
+    const caps = Array.isArray(row.capabilities) ? row.capabilities : [];
+    const wantsOptions = caps.indexOf("options") >= 0;
+    const opt = row.options;
     let html = "";
-    if (opt.enabled === false) {
-      html += renderOptionsAccountStatRow(opt);
+    if (wantsOptions && (opt == null || typeof opt !== "object")) {
+      html += '<div class="section-title hub-options-title">期权持仓</div>';
+      html += `<div class="err">期权数据不可用</div>`;
+    } else if ((opt || {}).enabled === false) {
+      html += renderOptionsAccountStatRow(opt || {});
       html += '<div class="section-title hub-options-title">期权持仓</div>';
       html += '<div class="empty-hint">期权未启用(OKX_OPTIONS_ENABLED)</div>';
-    } else if (opt.ok === false) {
-      html += renderOptionsAccountStatRow(opt);
+    } else if ((opt || {}).ok === false) {
+      html += renderOptionsAccountStatRow(opt || {});
       html += '<div class="section-title hub-options-title">期权持仓</div>';
-      html += `<div class="err">${esc(opt.msg || "期权数据不可用")}</div>`;
+      html += `<div class="err">${esc((opt && (opt.msg || opt.error)) || "期权数据不可用")}</div>`;
     } else {
-      const pos = Array.isArray(opt.positions) ? opt.positions : [];
-      const targets = Array.isArray(opt.target_monitors) ? opt.target_monitors : [];
-      html += renderOptionsAccountStatRow(opt);
+      const optSafe = opt || {};
+      const pos = Array.isArray(optSafe.positions) ? optSafe.positions : [];
+      const targets = Array.isArray(optSafe.target_monitors) ? optSafe.target_monitors : [];
+      html += renderOptionsAccountStatRow(optSafe);
       html += `<div class="section-title hub-options-title">期权持仓 · ${pos.length} 仓</div>`;
       html +=
         layout === "cards"

requirements.txt

@@ -4,7 +4,6 @@
 flask>=3.0,<4
 requests>=2.31,<3
 ccxt>=4.2,<5
-websocket-client>=1.6,<2
 werkzeug>=3.0,<4
 PySocks>=1.7,<2
 Pillow>=10.0,<12

tests/test_options_quote_live_lib.py

@@ -1,64 +0,0 @@
-"""options_quote_live_lib 单元测试."""
-from __future__ import annotations
-
-import json
-
-from lib.options.options_quote_live_lib import OptionsQuoteLive
-
-
-class _FakeWs:
-    connected = True
-    last_msg_at = 0.0
-
-    def start(self) -> None:
-        return None
-
-    def stop(self) -> None:
-        return None
-
-    def set_subscriptions(self, args) -> None:
-        self.last_args = list(args)
-
-
-def test_ticker_patch_and_flush():
-    live = OptionsQuoteLive()
-    live._ws = _FakeWs()  # type: ignore[assignment]
-    live._started = True
-    live.watch(
-        underlying="ETH",
-        exp_time="1",
-        contracts=[{"inst_id": "ETH-USD-260811-2500-C", "opt_type": "C", "strike": 2500}],
-        index_inst_id="ETH-USD",
-    )
-    live._on_ws_data(
-        {
-            "arg": {"channel": "tickers", "instId": "ETH-USD-260811-2500-C"},
-            "data": [
-                {
-                    "instId": "ETH-USD-260811-2500-C",
-                    "askPx": "12.5",
-                    "askSz": "3",
-                    "bidPx": "11.0",
-                    "bidSz": "2",
-                    "markPx": "12.0",
-                }
-            ],
-        }
-    )
-    live._on_ws_data(
-        {
-            "arg": {"channel": "index-tickers", "instId": "ETH-USD"},
-            "data": [{"idxPx": "2600"}],
-        }
-    )
-    raw = live._build_flush_event()
-    assert raw is not None
-    payload = json.loads(raw)
-    assert payload["index_px"] == 2600.0
-    assert payload["quotes"]
-    q = next(x for x in payload["quotes"] if x["inst_id"] == "ETH-USD-260811-2500-C")
-    assert q["ask"] == 12.5
-    assert q["ask_sz"] == 3.0
-    assert q["expiry_be_px"] == 2512.5
-    st = live.status()
-    assert st["watch_count"] == 1