ValueScan real-time token signal subscription

Production environment: https://stream.valuescan.ai

1. Overview

After a user subscribes to specified tokens, when a token generates a signal (opportunity, risk, or funds movement), the server pushes messages to the client proactively.

Key features:

Sub-second real-time push: Signals are delivered to the client immediately after they are generated
Multiple signal types: Supports opportunity signals (OPPORTUNITY), risk signals (RISK), and funds movement signals (FUNDS)
Automatic reconnect on disconnect: The client handles disconnects and retries automatically

2. Subscription URL

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

Authentication parameters (query string):

Parameter	Type	Required	Description
`apiKey`	string	Yes	ValueScan Access Key
`sign`	string	Yes	HMAC-SHA256 signature
`timestamp`	long	Yes	Unix timestamp in milliseconds
`nonce`	string	Yes	Random string (replay protection)
`tokens`	string	Yes	Token IDs to subscribe; separate multiple IDs with commas, e.g. `1,2,3`. Empty value defaults to all tokens

Signing algorithm:

sign = HMAC-SHA256(SK, timestampMs + nonce)

Where timestampMs is the millisecond Unix timestamp as a string concatenated with nonce.

Example (Python):

python

import hashlib
import hmac
import time
import uuid

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

Timestamp validity: ±300 seconds (5 minutes); outside this window returns 401.

3. SSE event format

Connected

event: connected
data: subscribed

Heartbeat (keep-alive)

: heartbeat

event: heartbeat
data: ping

The server sends a heartbeat every 20 seconds.

Signal push

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

signal event payload fields:

Field	Type	Description
`tokenId`	long	Token ID
`type`	string	Signal type
`content`	string	Signal detail as a JSON string
`ts`	long	Millisecond timestamp
`uniqueKey`	string	Deduplication key

4. Three signal types

1. Opportunity signal (OPPORTUNITY)

Meaning: AI-driven real-time follow-up messages for tokens that, across multiple data dimensions, show upside potential in the current market.

Related query API: POST /open/v1/ai/getChanceCoinMessageList

content shape:

Field	Type	Description
`vsTokenId`	string	Token ID
`symbol`	string	Token symbol, e.g. BTC
`name`	string	Token name, e.g. Bitcoin
`chanceMessageType`	int	Opportunity message type (major-player action type)
`scoring`	number	Opportunity score; higher means a stronger signal
`grade`	int	Signal grade
`price`	string	Current price
`percentChange24h`	number	24h change (%)
`gains`	number	Upside move
`decline`	number	Downside move
`updateTime`	long	Update time (millisecond timestamp)

2. Risk signal (RISK)

Meaning: AI-driven real-time follow-up messages for tokens that, across multiple data dimensions, show downside risk in the current market.

Related query API: POST /open/v1/ai/getRiskCoinMessageList

content shape:

Field	Type	Description
`vsTokenId`	string	Token ID
`symbol`	string	Token symbol
`name`	string	Token name
`riskMessageType`	int	Risk message type
`scoring`	number	Risk score (may be empty)
`grade`	int	Signal grade (may be empty)
`price`	string	Current price
`percentChange24h`	number	24h change (%)
`riskDecline`	number	Risk drawdown (price decline magnitude)
`rebound`	number	Rebound magnitude
`updateTime`	long	Update time (millisecond timestamp)

3. Funds movement signal (FUNDS)

Meaning: Real-time follow-up messages for major capital flows in spot or derivatives trading on centralized exchanges.

Related query API: POST /open/v1/ai/getFundsCoinMessageList

content shape:

Field	Type	Description
`vsTokenId`	string	Token ID
`symbol`	string	Token symbol
`name`	string	Token name
`tradeType`	int	Trade type (1: spot, 2: futures, 3: delivery futures)
`price`	string	Current price
`percentChange24h`	number	24h change (%)
`fundsMovementType`	int	Funds flow type (1: spot; 2: futures; etc.)
`updateTime`	long	Update time (millisecond timestamp)

5. Subscription examples

Python (recommended)

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"          # Subscribed token IDs

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("♥ Heartbeat")
            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"[Signal] tokenId={token_id} type={signal_type} content={content}")
    elif event_name == "connected":
        print(f"[Connected] {data_str}")

cURL (quick check)

bash

# Generate signed curl command
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\"')
'"

Copy and run the printed command.

6. Reconnect on disconnect

SSE connections may drop due to network issues or server restarts. Implement exponential backoff on the client:

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:
                # Clean exit (user interrupt) — do not reconnect
                _read_sse(resp)
                break
        except KeyboardInterrupt:
            print("User interrupted, exiting.")
            break
        except Exception as e:
            retries += 1
            wait = min(2 ** retries, 60)
            print(f"Disconnected ({e}), reconnect attempt {retries} in {wait}s...")
            time.sleep(wait)
    else:
        print("Max retries reached, exiting.")

7. Parsing `content`

The content field is a JSON string and must be parsed twice:

python

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

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

if payload["type"] == "OPPORTUNITY":
    print(f"Opportunity signal: {content['name']} ({content['symbol']}) "
          f"price={content['price']} score={content['scoring']}")
elif payload["type"] == "RISK":
    print(f"Risk signal: {content['name']} ({content['symbol']}) "
          f"price={content['price']} risk drawdown={content['riskDecline']}%")
elif payload["type"] == "FUNDS":
    trade = "Spot" if content["tradeType"] == 1 else "Futures"
    print(f"Funds movement: {content['name']} ({content['symbol']}) "
          f"{trade} price={content['price']}")

8. SDK integration (Python)

The official SDK is recommended; subscription takes about 20 lines:

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?<signed query parameters>",
    on_event=on_event,
)
client.start()

The SDK handles heartbeat parsing, reconnect with exponential backoff, and event dispatch.

9. Error codes

HTTP status	Description
200	Connected successfully
400	Missing or invalid parameters
401	Signature failed or timestamp out of range
404	Wrong subscription URL
503	Service unavailable (auth backend connection failed)

ValueScan real-time token signal subscription ​

1. Overview ​

2. Subscription URL ​

3. SSE event format ​

Connected ​

Heartbeat (keep-alive) ​

Signal push ​

4. Three signal types ​

1. Opportunity signal (OPPORTUNITY) ​

2. Risk signal (RISK) ​

3. Funds movement signal (FUNDS) ​

5. Subscription examples ​

Python (recommended) ​

cURL (quick check) ​

6. Reconnect on disconnect ​

7. Parsing content ​

8. SDK integration (Python) ​

9. Error codes ​

ValueScan real-time token signal subscription

1. Overview

2. Subscription URL

3. SSE event format

Connected

Heartbeat (keep-alive)

Signal push

4. Three signal types

1. Opportunity signal (OPPORTUNITY)

2. Risk signal (RISK)

3. Funds movement signal (FUNDS)

5. Subscription examples

Python (recommended)

cURL (quick check)

6. Reconnect on disconnect

7. Parsing `content`

8. SDK integration (Python)

9. Error codes