Aggregate core market metrics

app/core/dashboard.py

@@ -328,6 +328,16 @@ def render_dashboard():
     button.primary:hover { background: #1f2937; }
     button.danger { color: var(--err); }
     button.small { padding: 5px 8px; min-height: 30px; }
+    .icon-button {
+      width: 36px;
+      min-width: 36px;
+      padding: 0;
+      display: inline-flex;
+      align-items: center;
+      justify-content: center;
+      font-size: 18px;
+      line-height: 1;
+    }
     pre {
       margin: 0;
       white-space: pre-wrap;
@@ -575,8 +585,8 @@ def render_dashboard():
     </div>
     <div class="toolbar">
       <span id="updated" class="muted">等待刷新</span>
-      <button onclick="loadStatus()">刷新状态</button>
-      <button class="primary" onclick="saveFormConfig()">保存参数</button>
+      <button class="icon-button" onclick="loadStatus()" title="刷新状态" aria-label="刷新状态">↻</button>
+      <button class="icon-button primary" onclick="saveFormConfig()" title="保存参数" aria-label="保存参数">✓</button>
       <span id="topMessage" class="message"></span>
     </div>
   </header>
@@ -616,7 +626,7 @@ def render_dashboard():
         </div>
         <div class="market-detail-grid">
           <div class="market-tile">
-            <span>24h 成交额</span>
+            <span>聚合 24h 成交额</span>
             <strong id="coreVolume24h">--</strong>
             <small id="coreVolume5m">近 5m --</small>
           </div>
@@ -631,7 +641,7 @@ def render_dashboard():
             <small id="coreFuturesSymbol">合约未启用</small>
           </div>
           <div class="market-tile">
-            <span>合约持仓量</span>
+            <span>聚合合约持仓量</span>
             <strong id="coreOpenInterest">--</strong>
             <small id="coreOpenInterestChange">变化 --</small>
           </div>
@@ -641,9 +651,9 @@ def render_dashboard():
             <small id="coreFlowMeta">等待样本</small>
           </div>
           <div class="market-tile">
-            <span>监控状态</span>
-            <strong id="coreMonitorState">--</strong>
-            <small id="coreMonitorMeta">等待数据</small>
+            <span>数据覆盖</span>
+            <strong id="coreCoverage">--</strong>
+            <small id="coreCoverageMeta">等待来源</small>
           </div>
         </div>
       </div>
@@ -985,6 +995,7 @@ def render_dashboard():
     let currentConfig = null;
     let currentCoin = null;
     let activePanel = "global";
+    let uiEvents = [];
 
     const globalFields = [
       ["poll_interval_sec", "number"],
@@ -1105,6 +1116,29 @@ def render_dashboard():
       return "等待首次成功";
     }
 
+    function sourceNames(sources, limit = 3) {
+      const list = Array.isArray(sources) ? sources : [];
+      if (!list.length) return "--";
+      const names = list.slice(0, limit).map((item) => item.name).filter(Boolean);
+      const extra = list.length > limit ? ` +${list.length - limit}` : "";
+      return names.join(" / ") + extra;
+    }
+
+    function sourceCountLabel(count, total) {
+      const n = Number(count || 0);
+      return `${n}/${total}`;
+    }
+
+    function addUiEvent(message, level = "info") {
+      uiEvents.unshift({
+        created_at: new Date().toISOString(),
+        monitor_id: "dashboard",
+        level,
+        message
+      });
+      uiEvents = uiEvents.slice(0, 12);
+    }
+
     function formatDateTime(value) {
       if (!value) return "--";
       const date = new Date(value);
@@ -1257,6 +1291,10 @@ def render_dashboard():
       const coreSnap = core.snapshot || {};
       const btcSnap = summary?.btc?.snapshot || {};
       const coreMonitor = core.monitor || null;
+      const aggregate = core.aggregate || {};
+      const aggVolume = aggregate.volume24h || {};
+      const aggOi = aggregate.openInterest || {};
+      const aggFunding = aggregate.fundingRate || {};
       const oiChange = coreSnap.openInterestChange || {};
       const flowBits = [];
       if (Number.isFinite(Number(coreSnap.volumeRatio5m))) {
@@ -1279,28 +1317,33 @@ def render_dashboard():
       setText("btcMeta", btcSnap.priceChange?.["1h"] !== undefined
         ? `1h ${formatSignedPercent(btcSnap.priceChange["1h"], 2)}`
         : "等待数据");
-      setText("coreVolume24h", formatUsd(coreSnap.quoteVolume24h));
-      setText("coreVolume5m", `近 5m ${formatUsd(coreSnap.volume5m)} · ${coreSnap.volumeRatio5m ? "x" + formatNumber(coreSnap.volumeRatio5m, 1) : "--"}`);
+      setText("coreVolume24h", formatUsd(aggVolume.total || coreSnap.quoteVolume24h));
+      setText("coreVolume5m", aggVolume.sourceCount
+        ? `${sourceCountLabel(aggVolume.sourceCount, 11)} 来源 · ${sourceNames(aggVolume.sources)}`
+        : `近 5m ${formatUsd(coreSnap.volume5m)} · ${coreSnap.volumeRatio5m ? "x" + formatNumber(coreSnap.volumeRatio5m, 1) : "--"}`);
       setText("coreRange24h", coreSnap.high24 && coreSnap.low24
         ? `${formatPrice(coreSnap.low24)} - ${formatPrice(coreSnap.high24)}`
         : "--");
-      setText("coreRsi", `RSI ${formatNumber(coreSnap.rsi, 1)}`);
-      setText("coreFunding", coreSnap.funding !== null && coreSnap.funding !== undefined
-        ? formatPercent(coreSnap.funding, 4)
+      setText("coreRsi", `Binance 24h 区间 · RSI ${formatNumber(coreSnap.rsi, 1)}`);
+      setText("coreFunding", aggFunding.value !== null && aggFunding.value !== undefined
+        ? formatPercent(aggFunding.value, 4)
         : "--");
       setText("coreFuturesSymbol", core.enable_futures
-        ? `合约 ${core.futures_symbol || "--"}`
+        ? `Binance Futures · ${aggFunding.symbol || core.futures_symbol || "--"}`
         : "合约未启用");
-      setText("coreOpenInterest", coreSnap.openInterest ? formatNumber(coreSnap.openInterest, 2) : "--");
-      setText("coreOpenInterestChange", oiChange["15m"] !== undefined
+      setText("coreOpenInterest", aggOi.total ? formatUsd(aggOi.total) : (coreSnap.openInterest ? formatNumber(coreSnap.openInterest, 2) : "--"));
+      setText("coreOpenInterestChange", aggOi.sourceCount
+        ? `${sourceCountLabel(aggOi.sourceCount, 6)} 来源 · ${sourceNames(aggOi.sources)}`
+        : oiChange["15m"] !== undefined
         ? `15m ${formatSignedPercent(oiChange["15m"], 1)}`
         : "等待持仓样本");
       setText("coreFlow", flowBits[0] || "--");
       setText("coreFlowMeta", flowBits.slice(1).join("；") || "等待资金异动样本");
       const displayStatus = !isEnabled(coreMonitor) ? "disabled" : (isPaused(coreMonitor) ? "paused" : coreMonitor?.status);
-      setText("coreMonitorState", statusText(displayStatus));
-      setText("coreMonitorMeta", formatMonitorMeta(coreMonitor));
-      setText("coreUpdated", coreSnap.updatedAt ? `更新 ${formatDateTime(coreSnap.updatedAt)}` : "等待行情");
+      const statusMeta = `${statusText(displayStatus)} · ${formatMonitorMeta(coreMonitor)}`;
+      setText("coreCoverage", `成交 ${aggVolume.sourceCount || 0} · OI ${aggOi.sourceCount || 0}`);
+      setText("coreCoverageMeta", `成交额覆盖 11 所；持仓覆盖 6 所`);
+      setText("coreUpdated", coreSnap.updatedAt ? `更新 ${formatDateTime(coreSnap.updatedAt)} · ${statusMeta}` : `等待行情 · ${statusMeta}`);
     }
 
     function renderStrategySummary(monitors) {
@@ -1391,7 +1434,7 @@ def render_dashboard():
       const data = await res.json();
       const monitors = data.monitors || [];
       const alerts = data.alerts || [];
-      const events = data.events || [];
+      const events = [...uiEvents, ...(data.events || [])];
       const errorCount = monitors.filter((m) => m.status === "error" && !isPaused(m)).length;
 
       document.getElementById("platformState").textContent = data.started ? "运行中" : "未启动";
@@ -1469,7 +1512,9 @@ def render_dashboard():
       renderCoinForm();
       renderExchangeForm();
       syncFormToJson();
-      setMessage("配置已加载", "ok");
+      setMessage("");
+      addUiEvent("配置已加载");
+      await loadStatus();
     }
 
     function renderGlobalForm() {

app/monitors/crypto_price.py

@@ -52,6 +52,12 @@ class CryptoPriceMonitor(BaseMonitor):
         self.alpha_api = legacy.BinanceAlphaAPI(self.log)
         self.bitget = legacy.BitgetAPI(self.log)
         self.okx = legacy.OKXAPI(self.log)
+        self.core_aggregator = legacy.CoreMarketAggregator(
+            self.api,
+            self.okx,
+            self.bitget,
+            self.log,
+        )
         self.okx_adapter = legacy.OKXMarketAdapter(self.okx, self.log)
         self.binance_futures_adapter = legacy.BinanceFuturesMarketAdapter(self.api)
         self.binance_futures_with_fallback = legacy.FallbackMarketAdapter(
@@ -201,10 +207,11 @@ class CryptoPriceMonitor(BaseMonitor):
             "GRAM" if "GRAM" in coins else (next(iter(coins), None)),
         )
 
-        def market(name):
+        def market(name, include_aggregate=False):
             if not name:
                 return None
             coin = coins.get(name, {})
+            snapshot = snapshots.get(name)
             monitor = self.store.get_monitor(coin_monitor_id(name))
             return {
                 "name": name,
@@ -212,12 +219,16 @@ class CryptoPriceMonitor(BaseMonitor):
                 "futures_symbol": coin.get("futures_symbol"),
                 "data_source": coin.get("data_source", "binance"),
                 "enable_futures": bool(coin.get("enable_futures")),
-                "snapshot": snapshots.get(name),
+                "snapshot": snapshot,
+                "aggregate": (
+                    self.core_aggregator.aggregate(name, coin, snapshot)
+                    if include_aggregate and snapshot else None
+                ),
                 "monitor": monitor,
             }
 
         return {
-            "core": market(core_name),
+            "core": market(core_name, include_aggregate=True),
             "btc": market("BTC") if "BTC" in coins else None,
         }
 

coinpush.py

@@ -79,6 +79,15 @@ OKX_PUBLIC_CACHE_TTL_SEC = int(os.getenv("OKX_PUBLIC_CACHE_TTL_SEC", "120"))
 OKX_HISTORY_CACHE_TTL_SEC = int(os.getenv("OKX_HISTORY_CACHE_TTL_SEC", "300"))
 OKX_429_COOLDOWN_SEC = int(os.getenv("OKX_429_COOLDOWN_SEC", "120"))
 OKX_ERROR_LOG_COOLDOWN_SEC = int(os.getenv("OKX_ERROR_LOG_COOLDOWN_SEC", "300"))
+CORE_AGG_CACHE_TTL_SEC = int(os.getenv("CORE_MARKET_AGG_CACHE_TTL_SEC", "120"))
+BYBIT_BASE = os.getenv("BYBIT_BASE", "https://api.bybit.com")
+GATE_BASE = os.getenv("GATE_BASE", "https://api.gateio.ws/api/v4")
+KUCOIN_BASE = os.getenv("KUCOIN_BASE", "https://api.kucoin.com")
+MEXC_BASE = os.getenv("MEXC_BASE", "https://api.mexc.com")
+KRAKEN_BASE = os.getenv("KRAKEN_BASE", "https://api.kraken.com")
+UPBIT_BASE = os.getenv("UPBIT_BASE", "https://api.upbit.com")
+HYPERLIQUID_BASE = os.getenv("HYPERLIQUID_BASE", "https://api.hyperliquid.xyz")
+ASTER_FAPI_BASE = os.getenv("ASTER_FAPI_BASE", "https://fapi.asterdex.com")
 
 # 合约功能全局开关（默认关闭）。币安合约接口/强平流对 US 区域可能返回 451。
 # 该开关控制现货主行情币种的合约指标与强平 websocket；若标的本身以 Binance
@@ -1309,6 +1318,246 @@ class OKXMarketAdapter:
         return [{"fundingRate": x["fundingRate"]} for x in data if "fundingRate" in x]
 
 
+class CoreMarketAggregator:
+    """Best-effort cross-exchange metrics for the dashboard core asset."""
+
+    def __init__(self, binance_api, okx_api, bitget_api, logger, ttl=CORE_AGG_CACHE_TTL_SEC):
+        self.binance = binance_api
+        self.okx = okx_api
+        self.bitget = bitget_api
+        self.logger = logger
+        self.ttl = max(30, int(ttl))
+        self.session = requests.Session()
+        self.session.headers.update(REQ_HEADERS)
+        self.cache = {}
+        self.last_error_log = {}
+
+    @staticmethod
+    def _asset(symbol, name=None):
+        text = str(symbol or name or "").strip().upper()
+        for suffix in ("-USDT-SWAP", "-USDT", "_USDT", "USDT", "USD"):
+            if text.endswith(suffix):
+                text = text[: -len(suffix)]
+                break
+        return text.replace("-", "").replace("_", "") or str(name or "").upper()
+
+    @staticmethod
+    def _num(value):
+        try:
+            n = float(value)
+            if n == n and n not in (float("inf"), float("-inf")):
+                return n
+        except (TypeError, ValueError):
+            return None
+        return None
+
+    @staticmethod
+    def _source(name, value, market="spot", note=None):
+        try:
+            value = float(value)
+        except (TypeError, ValueError):
+            return None
+        if value <= 0:
+            return None
+        out = {"name": name, "value": value, "market": market}
+        if note:
+            out["note"] = note
+        return out
+
+    def _log_error_once(self, key, msg, cooldown=600):
+        now = time.time()
+        last = self.last_error_log.get(key, 0)
+        if now - last >= cooldown:
+            self.last_error_log[key] = now
+            self.logger(msg)
+
+    def _get_json(self, base, path, params=None, method="GET", body=None, timeout=6):
+        url = base.rstrip("/") + path
+        try:
+            if method == "POST":
+                resp = self.session.post(url, json=body or {}, timeout=timeout)
+            else:
+                resp = self.session.get(url, params=params, timeout=timeout)
+            if resp.status_code != 200:
+                return None
+            return resp.json()
+        except Exception as exc:
+            self._log_error_once((base, path, "exception"), f"⚠️ 聚合行情 {base}{path} 异常: {exc}")
+            return None
+
+    def aggregate(self, coin_name, coin_cfg, snapshot=None):
+        symbol = (coin_cfg or {}).get("symbol")
+        futures_symbol = (coin_cfg or {}).get("futures_symbol") or symbol
+        asset = self._asset(futures_symbol or symbol, coin_name)
+        cache_key = (asset, str(symbol or ""), str(futures_symbol or ""))
+        cached = self.cache.get(cache_key)
+        now = time.time()
+        if cached and now - cached["ts"] < self.ttl:
+            return copy.deepcopy(cached["value"])
+
+        reference_price = self._num((snapshot or {}).get("price"))
+        value = {
+            "asset": asset,
+            "volume24h": self._volume24h(asset, symbol, snapshot),
+            "openInterest": self._open_interest(asset, futures_symbol, reference_price),
+            "fundingRate": self._binance_funding(futures_symbol or f"{asset}USDT"),
+            "updatedAt": datetime.now(timezone.utc).isoformat(),
+        }
+        self.cache[cache_key] = {"ts": now, "value": copy.deepcopy(value)}
+        return value
+
+    def _volume24h(self, asset, symbol, snapshot):
+        sources = []
+
+        def add(name, value, market="spot", note=None):
+            item = self._source(name, value, market, note)
+            if item:
+                sources.append(item)
+
+        spot_symbol = f"{asset}USDT"
+        add("Binance", (snapshot or {}).get("quoteVolume24h"), "primary")
+
+        okx = self.okx.ticker(f"{asset}-USDT")
+        if okx:
+            add("OKX", okx.get("qvol"), "spot")
+
+        bitget = self.bitget.spot_ticker(spot_symbol)
+        if bitget:
+            add("Bitget", bitget.get("qvol"), "spot")
+
+        hl = self._hyperliquid_ctx(asset)
+        if hl:
+            add("Hyperliquid", hl.get("dayNtlVlm"), "perp")
+
+        aster = self._get_json(ASTER_FAPI_BASE, "/fapi/v1/ticker/24hr", {"symbol": spot_symbol})
+        if isinstance(aster, dict):
+            add("Aster", aster.get("quoteVolume"), "perp")
+
+        bybit = self._bybit_ticker("spot", spot_symbol)
+        if bybit:
+            add("Bybit", bybit.get("turnover24h"), "spot")
+
+        gate = self._get_json(GATE_BASE, "/spot/tickers", {"currency_pair": f"{asset}_USDT"})
+        if isinstance(gate, list) and gate:
+            add("Gate", gate[0].get("quote_volume"), "spot")
+
+        kucoin = self._get_json(KUCOIN_BASE, "/api/v1/market/stats", {"symbol": f"{asset}-USDT"})
+        if isinstance(kucoin, dict) and str(kucoin.get("code")) == "200000":
+            add("KuCoin", (kucoin.get("data") or {}).get("volValue"), "spot")
+
+        mexc = self._get_json(MEXC_BASE, "/api/v3/ticker/24hr", {"symbol": spot_symbol})
+        if isinstance(mexc, dict):
+            add("MEXC", mexc.get("quoteVolume"), "spot")
+
+        kraken_pair = f"{'XBT' if asset == 'BTC' else asset}USDT"
+        kraken = self._get_json(KRAKEN_BASE, "/0/public/Ticker", {"pair": kraken_pair})
+        if isinstance(kraken, dict) and not kraken.get("error"):
+            result = kraken.get("result") or {}
+            if result:
+                row = next(iter(result.values()))
+                close = self._num((row.get("c") or [None])[0])
+                vol = self._num((row.get("v") or [None, None])[1])
+                if close and vol:
+                    add("Kraken", close * vol, "spot")
+
+        upbit = self._get_json(UPBIT_BASE, "/v1/ticker", {"markets": f"USDT-{asset}"})
+        if isinstance(upbit, list) and upbit:
+            add("Upbit", upbit[0].get("acc_trade_price_24h"), "spot")
+
+        sources.sort(key=lambda item: item["value"], reverse=True)
+        return {
+            "total": sum(item["value"] for item in sources),
+            "sources": sources,
+            "sourceCount": len(sources),
+        }
+
+    def _open_interest(self, asset, futures_symbol, reference_price=None):
+        sources = []
+        symbol = f"{asset}USDT"
+
+        def add(name, value, market="perp", note=None):
+            item = self._source(name, value, market, note)
+            if item:
+                sources.append(item)
+
+        bi = self.binance.open_interest(futures_symbol or symbol)
+        bi_oi = self._num((bi or {}).get("openInterest"))
+        if bi_oi and reference_price:
+            add("Binance", bi_oi * reference_price, "perp")
+
+        bybit = self._bybit_ticker("linear", symbol)
+        if bybit:
+            add("Bybit", bybit.get("openInterestValue"), "perp")
+
+        okx = self.okx.open_interest(f"{asset}-USDT-SWAP")
+        if okx:
+            add("OKX", okx.get("oi_usd"), "perp")
+
+        bitget = self._get_json(
+            BITGET_BASE,
+            "/api/v2/mix/market/ticker",
+            {"symbol": symbol, "productType": "USDT-FUTURES"},
+        )
+        data = (bitget or {}).get("data")
+        row = data[0] if isinstance(data, list) and data else (data if isinstance(data, dict) else None)
+        if row:
+            oi = self._num(row.get("holdingAmount"))
+            price = self._num(row.get("lastPr")) or reference_price
+            if oi and price:
+                add("Bitget", oi * price, "perp")
+
+        hl = self._hyperliquid_ctx(asset)
+        if hl:
+            oi = self._num(hl.get("openInterest"))
+            mark = self._num(hl.get("markPx")) or reference_price
+            if oi and mark:
+                add("Hyperliquid", oi * mark, "perp")
+
+        aster_oi = self._get_json(ASTER_FAPI_BASE, "/fapi/v1/openInterest", {"symbol": symbol})
+        aster_mark = self._get_json(ASTER_FAPI_BASE, "/fapi/v1/premiumIndex", {"symbol": symbol})
+        if isinstance(aster_oi, dict):
+            oi = self._num(aster_oi.get("openInterest"))
+            mark = self._num((aster_mark or {}).get("markPrice")) or reference_price
+            if oi and mark:
+                add("Aster", oi * mark, "perp")
+
+        sources.sort(key=lambda item: item["value"], reverse=True)
+        return {
+            "total": sum(item["value"] for item in sources),
+            "sources": sources,
+            "sourceCount": len(sources),
+        }
+
+    def _binance_funding(self, symbol):
+        data = self.binance.premium_index(symbol)
+        rate = self._num((data or {}).get("lastFundingRate"))
+        if rate is None:
+            return None
+        return {
+            "symbol": symbol,
+            "value": rate * 100,
+            "source": "Binance Futures",
+        }
+
+    def _bybit_ticker(self, category, symbol):
+        data = self._get_json(BYBIT_BASE, "/v5/market/tickers", {"category": category, "symbol": symbol})
+        if not isinstance(data, dict) or str(data.get("retCode")) != "0":
+            return None
+        rows = ((data.get("result") or {}).get("list") or [])
+        return rows[0] if rows else None
+
+    def _hyperliquid_ctx(self, asset):
+        data = self._get_json(HYPERLIQUID_BASE, "/info", method="POST", body={"type": "metaAndAssetCtxs"})
+        if not isinstance(data, list) or len(data) < 2:
+            return None
+        universe = ((data[0] or {}).get("universe") or [])
+        contexts = data[1] or []
+        for idx, item in enumerate(universe):
+            if str(item.get("name", "")).upper() == asset and idx < len(contexts):
+                return contexts[idx]
+        return None
+
+
 class FallbackMarketAdapter:
     """主行情源不可用时透明切到备用行情源，返回格式保持 BinanceAPI 兼容。"""