大盤分析訂閱

生產環境：https://stream.valuescan.ai

---

一、概述

大盤分析是 ValueAgent 透過多維度資料，對 BTC/ETH 的行情進行綜合分析的輸出結果。大盤訂閱是一種伺服端主動推送機制，使用者訂閱後，當大盤行情發生變化時，伺服端透過 SSE 通道主動推送分析文字到用戶端。

---

二、訂閱位址

GET https://stream.valuescan.ai/stream/market/subscribe

鑑權參數（Query String）：

參數	類型	必填	說明
apiKey	string	是	ValueScan Access Key
sign	string	是	HMAC-SHA256 簽名
timestamp	long	是	毫秒級 Unix 時間戳
nonce	string	是	隨機字串（防重放）

簽名演算法：

sign = HMAC-SHA256(SK, timestamp 毫秒 + nonce)

---

三、SSE 事件格式

連線成功

event: connected
data: subscribed

心跳（維持連線）

: heartbeat

event: heartbeat
data: ping

伺服端每 20 秒發送一次心跳。

大盤分析推送

event: market
data: {"ts":1775604300144,"uniqueId":"大盤分析_20260408103000","content":"BTC 主導市場，資金集中度高，建議關注 BTC 走勢。"}

market 事件 payload 欄位：

欄位	類型	說明
ts	long	推送時間（毫秒時間戳）
uniqueId	string	唯一識別（用於去重）
content	string	大盤分析純文字內容

---

四、訂閱範例

Python

import hashlib
import hmac
import json
import time
import uuid
import urllib.request
from urllib.parse import urlencode

AK = "your-access-key"
SK = "your-secret-key"

ts = int(time.time() * 1000)
nonce = uuid.uuid4().hex
sign = hmac.new(
    SK.encode(),
    (str(ts) + nonce).encode(),
    hashlib.sha256,
).hexdigest()

params = urlencode({
    "apiKey": AK,
    "sign": sign,
    "timestamp": ts,
    "nonce": nonce,
})

url = f"https://stream.valuescan.ai/stream/market/subscribe?{params}"
req = urllib.request.Request(url, headers={"Accept": "text/event-stream"})

with urllib.request.urlopen(req, timeout=300) as resp:
    event_name = ""
    data_str = ""
    for raw in resp:
        line = raw.decode("utf-8").rstrip("\r\n")

        if line.startswith(":"):
            print("♥ 心跳")
            continue

        if line == "":
            if data_str:
                handle_event(event_name, data_str)
            event_name = ""
            data_str = ""
        elif line.startswith("event:"):
            event_name = line[6:].strip()
        elif line.startswith("data:"):
            data_str = line[5:].strip()


def handle_event(event_name: str, data_str: str) -> None:
    if event_name == "connected":
        print(f"[連線] {data_str}")
    elif event_name == "market":
        payload = json.loads(data_str)
        ts = payload["ts"]
        content = payload["content"]
        dt = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts / 1000))
        print(f"[大盤分析 {dt}] {content}")

cURL（快速驗證）

python3 -c "
import hashlib, hmac, time, uuid, urllib.parse
SK='your-secret-key'
ts=int(time.time()*1000)
nonce=uuid.uuid4().hex
sign=hmac.new(SK.encode(), (str(ts)+nonce).encode(), hashlib.sha256).hexdigest()
params=urllib.parse.urlencode({'apiKey':'your-access-key','sign':sign,'timestamp':ts,'nonce':nonce})
print(f"curl -N 'https://stream.valuescan.ai/stream/market/subscribe?{params}' -H 'Accept: text/event-stream'")
'"

---

五、斷線重連

與訊號訂閱一致，建議實作指數退避重連：

def listen_with_retry(url: str, max_retries: int = 10) -> None:
    retries = 0
    while retries <= max_retries:
        try:
            req = urllib.request.Request(url, headers={"Accept": "text/event-stream"})
            with urllib.request.urlopen(req, timeout=300) as resp:
                _read_sse(resp)
                break
        except KeyboardInterrupt:
            print("使用者中斷，結束。")
            break
        except Exception as e:
            retries += 1
            wait = min(2 ** retries, 60)
            print(f"連線中斷（{e}），{wait} 秒後第 {retries} 次重新連線...")
            time.sleep(wait)
    else:
        print("重試次數已用盡，結束。")

---

六、錯誤碼

HTTP 狀態碼	說明
200	連線成功
400	參數缺失或格式錯誤
401	簽名驗證失敗或時間戳超限
404	訂閱位址錯誤
503	伺服端不可用（鑑權後端連線失敗）