ValueScan 代币实时信号订阅

生产环境：https://stream.valuescan.ai

一、概述

用户订阅指定代币后，当该代币产生信号（机会信号、风险信号、资金异动信号）时，服务端主动推送消息到客户端。

主要特性：

秒级实时推送：信号产生后立即推送到客户端
多信号类型：支持机会信号（OPPORTUNITY）、风险信号（RISK）、资金异动信号（FUNDS）
断线自动重连：客户端自动处理连接断开并重试

二、订阅地址

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

鉴权参数（Query String）：

参数	类型	必需	说明
`apiKey`	string	是	ValueScan Access Key
`sign`	string	是	HMAC-SHA256 签名
`timestamp`	long	是	毫秒级 Unix 时间戳
`nonce`	string	是	随机字符串（防重放）
`tokens`	string	是	订阅的代币 ID，多个代币用逗号分隔，如 `1,2,3`，传空值默认为全部代币

签名算法：

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

示例（Python）：

python

import hashlib
import hmac
import time
import uuid

ts = int(time.time() * 1000)          # 毫秒时间戳
nonce = uuid.uuid4().hex               # 随机字符串
sign = hmac.new(
    SK.encode(),
    (str(ts) + nonce).encode(),
    hashlib.sha256
).hexdigest()

时间戳有效期：±300 秒（5 分钟），超出返回 401。

三、SSE 事件格式

连接成功

event: connected
data: subscribed

心跳（维持连接）

: heartbeat

event: heartbeat
data: ping

服务端每 20 秒发送一次心跳。

信号推送

event: signal
data: {"content":"{...}","tokenId":1,"type":"OPPORTUNITY","ts":1775604300144,"uniqueKey":"1_1775604300144"}

signal 事件 payload 字段：

字段	类型	说明
`tokenId`	long	代币 ID
`type`	string	信号类型
`content`	string	信号详情 JSON 字符串
`ts`	long	毫秒时间戳
`uniqueKey`	string	去重标识

四、三种信号类型

1. 机会信号（OPPORTUNITY）

业务含义：AI 通过多数据维度，追踪当前行情下具有上涨潜力的代币的实时追踪消息。

对应查询接口：POST /open/v1/ai/getChanceCoinMessageList

content 结构：

字段	类型	说明
`vsTokenId`	string	代币 ID
`symbol`	string	代币符号，如 BTC
`name`	string	代币名称，如 Bitcoin
`chanceMessageType`	int	机会消息类型（主力动作类型）
`scoring`	number	机会得分，分数越高信号越强
`grade`	int	信号等级
`price`	string	当前价格
`percentChange24h`	number	24 小时涨跌幅
`gains`	number	涨幅
`decline`	number	跌幅
`updateTime`	long	更新时间（毫秒时间戳）

2. 风险信号（RISK）

业务含义：AI 通过多数据维度，追踪当前行情下具有下跌风险的代币的实时追踪消息。

对应查询接口：POST /open/v1/ai/getRiskCoinMessageList

content 结构：

字段	类型	说明
`vsTokenId`	string	代币 ID
`symbol`	string	代币符号
`name`	string	代币名称
`riskMessageType`	int	风险消息类型
`scoring`	number	风险得分（可能为空）
`grade`	int	信号等级（可能为空）
`price`	string	当前价格
`percentChange24h`	number	24 小时涨跌幅
`riskDecline`	number	风险跌幅（价格下跌幅度）
`rebound`	number	反弹幅度
`updateTime`	long	更新时间（毫秒时间戳）

3. 资金异动信号（FUNDS）

业务含义：监控中心化交易所的现货或合约交易中，主力资金异动的实时追踪消息。

对应查询接口：POST /open/v1/ai/getFundsCoinMessageList

content 结构：

字段	类型	说明
`vsTokenId`	string	代币 ID
`symbol`	string	代币符号
`name`	string	代币名称
`tradeType`	int	交易类型（1:现货 2:合约 3:交割合约）
`price`	string	当前价格
`percentChange24h`	number	24 小时涨跌幅
`fundsMovementType`	int	资金流动类型（1:现货; 2:合约; 等）
`updateTime`	long	更新时间（毫秒时间戳）

五、订阅示例

Python（推荐）

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"
TOKENS = "1,2,3"          # 订阅的代币 ID

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,
    "tokens": TOKENS,
})

url = f"https://stream.valuescan.ai/stream/signal/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 == "signal":
        payload = json.loads(data_str)
        token_id = payload["tokenId"]
        signal_type = payload["type"]
        content = json.loads(payload["content"])
        print(f"[信号] tokenId={token_id} type={signal_type} content={content}")
    elif event_name == "connected":
        print(f"[连接] {data_str}")

cURL（快速验证）

bash

# 生成签名参数
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,'tokens':'1,2'})
print(f'curl -N \"https://stream.valuescan.ai/stream/signal/subscribe?{params}\" -H \"Accept: text/event-stream\"')
'"

将输出的命令粘贴执行即可。

六、断线重连

SSE 连接可能因网络波动或服务端重启断开。建议客户端实现指数退避重连：

python

import time

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}s 后第 {retries} 次重连...")
            time.sleep(wait)
    else:
        print("重试次数耗尽，退出。")

七、内容解析示例

content 字段是 JSON 字符串，需要二次解析：

python

def parse_signal_content(content_str: str) -> dict:
    return json.loads(content_str)

# 示例
payload = json.loads(data_str)
content = parse_signal_content(payload["content"])

if payload["type"] == "OPPORTUNITY":
    print(f"机会信号：{content['name']} ({content['symbol']}) "
          f"价格={content['price']} 得分={content['scoring']}")
elif payload["type"] == "RISK":
    print(f"风险信号：{content['name']} ({content['symbol']}) "
          f"价格={content['price']} 风险跌幅={content['riskDecline']}%")
elif payload["type"] == "FUNDS":
    trade = "现货" if content["tradeType"] == 1 else "合约"
    print(f"资金异动：{content['name']} ({content['symbol']}) "
          f"{trade} 价格={content['price']}")

八、SDK 接入（Python）

推荐使用官方 SDK，只需约 20 行代码即可完成订阅：

python

from agent.sdk.client import SseClient

def on_event(event_name: str, payload: dict) -> None:
    if event_name == "signal":
        import json
        content = json.loads(payload["content"])
        print(f"[{payload['type']}] {content.get('name', '')} "
              f"{content.get('price', '')}")

client = SseClient(
    url="https://stream.valuescan.ai/stream/signal/subscribe?<已签名的参数>",
    on_event=on_event,
)
client.start()

SDK 内部处理：心跳解析、断线重连（指数退避）、事件分发。

九、错误码

HTTP 状态码	说明
200	连接成功
400	参数缺失或格式错误
401	签名验证失败或时间戳超限
404	订阅地址错误
503	服务端不可用（鉴权后端连接失败）

ValueScan 代币实时信号订阅 ​

一、概述 ​

二、订阅地址 ​

三、SSE 事件格式 ​

连接成功 ​

心跳（维持连接） ​

信号推送 ​

四、三种信号类型 ​

1. 机会信号（OPPORTUNITY） ​

2. 风险信号（RISK） ​

3. 资金异动信号（FUNDS） ​

五、订阅示例 ​

Python（推荐） ​

cURL（快速验证） ​

六、断线重连 ​

七、内容解析示例 ​

八、SDK 接入（Python） ​

九、错误码 ​

ValueScan 代币实时信号订阅

一、概述

二、订阅地址

三、SSE 事件格式

连接成功

心跳（维持连接）

信号推送

四、三种信号类型

1. 机会信号（OPPORTUNITY）

2. 风险信号（RISK）

3. 资金异动信号（FUNDS）

五、订阅示例

Python（推荐）

cURL（快速验证）

六、断线重连

七、内容解析示例

八、SDK 接入（Python）

九、错误码