2021 年 1 月，GameStop（GME）逼空行情在美股盘后继续发酵。交易所已收盘，散户只能盯着隔夜期货猜测走势；而持有 GMEUSDT 股票代币的交易者，却能在链上继续买卖，捕捉盘后的价格发现机会。

这就是股票代币的核心价值：把传统股票搬到区块链上，让它像加密货币一样 24 小时不间断交易。对于想要构建全天候行情监控或交易系统的开发者来说，股票代币行情接口是一个值得关注的新数据品类。

---

一、什么是股票代币

股票代币（Tokenized Stocks）是由链上机构发行的、锚定真实上市公司股价的数字资产。每枚代币对应一定比例的真实股票，价格在美股交易时段与标的股票高度联动，盘后则由链上供需决定。

与传统股票行情接口相比，股票代币行情有几个显著特点：

• 7×24 小时连续交易：不受交易所开盘时间限制，盘前、盘后、节假日均有成交
• USDT 计价：价格以 USDT 结算，无需换汇
• 盘口为一档：股票代币的买卖盘口只提供最优一档（买一/卖一），不同于 A 股、港股的五档或十档
• 覆盖股票 + ETF + 贵金属：除个股外，还包括 SPYUSDT（标普 500 ETF）、QQQUSDT（纳斯达克 100 ETF）、XAUUSDT（现货黄金）等

---

二、股票代币品种

Infoway API 目前支持 64 个股票代币，覆盖科技、金融、能源、消费等主要板块，同时包含主流 ETF 和贵金属：

科技股：AAPLUSDT（苹果）、NVDAUSDT（英伟达）、MSFTUSDT（微软）、GOOGLUSDT（谷歌）、METAUSDT（Meta）、TSLAUSDT（特斯拉）、AMDUSDT（AMD）、INTCUSDT（英特尔）、QCOMUSDT（高通）、TSMUSDT（台积电）、ASML（阿斯麦）…

金融股：JPMUSDT（摩根大通）、BACUSDT（美国银行）、BRKBUSDT（伯克希尔 B 类）、COINUSDT（Coinbase）、HOODUSDT（Robinhood）…

能源 / 消费：CVXUSDT（雪佛龙）、OXYUSDT（西方石油）、AMZNUSDT（亚马逊）、COSTUSDT（好市多）、NFLXUSDT（Netflix）…

ETF：SPYUSDT（标普 500）、QQQUSDT（纳斯达克 100）、SOXLUSDT（三倍做多半导体）、EWJUSDT（MSCI 日本）、EWYUSDT（MSCI 韩国）

贵金属：XAUUSDT（现货黄金）、XAGUSDT（现货白银）、XCUUSDT（现货铜）、XPTUSDT（铂金）、XPDUSDT（钯金）

通过接口获取完整列表：

import requests

API_KEY = "your_api_key"
BASE_URL = "https://data.infoway.io"

resp = requests.get(
    f"{BASE_URL}/common/basic/symbols",
    headers={"apiKey": API_KEY},
    params={"type": "CRYPTO"}
)
all_symbols = resp.json()["data"]
# 股票代币的 symbol 均以 USDT 结尾，过滤即可
token_stocks = [s for s in all_symbols if s["symbol"].endswith("USDT")]
print(f"股票代币品种数量：{len(token_stocks)}")

---

三、REST API 快速上手

股票代币与加密货币共用同一套接口路径，business=crypto 即可访问。

3.1 实时成交明细

def get_token_trades(symbols: list[str]) -> list[dict]:
    """查询股票代币最新成交明细"""
    codes = ",".join(symbols)
    resp = requests.get(
        f"{BASE_URL}/crypto/batch_trade/{codes}",
        headers={"apiKey": API_KEY}
    )
    resp.raise_for_status()
    return resp.json()["data"]

trades = get_token_trades(["TSLAUSDT", "NVDAUSDT", "AAPLUSDT"])
for t in trades:
    direction = {0: "中性", 1: "买入", 2: "卖出"}.get(t["td"], "未知")
    print(f"{t['s']}: 价格={t['p']} USDT  成交量={t['v']} 股  方向={direction}")

注意：v 字段单位为股票份额数（股），而非"手"。v="6.251" 表示成交 6.251 股 AAPL 代币。

3.2 实时买卖盘口（一档）

股票代币的盘口只有最优一档，返回结构与多档盘口相同，但数组长度为 1：

def get_token_depth(symbols: list[str]) -> list[dict]:
    """查询股票代币一档买卖盘口"""
    codes = ",".join(symbols)
    resp = requests.get(
        f"{BASE_URL}/crypto/batch_depth/{codes}",
        headers={"apiKey": API_KEY}
    )
    resp.raise_for_status()
    return resp.json()["data"]

depth_list = get_token_depth(["TSLAUSDT", "PDDUSDT"])
for item in depth_list:
    ask_price = item["a"][0][0]  # 卖一价
    ask_vol   = item["a"][1][0]  # 卖一量
    bid_price = item["b"][0][0]  # 买一价
    bid_vol   = item["b"][1][0]  # 买一量
    spread = float(ask_price) - float(bid_price)
    print(f"{item['s']}: 卖一={ask_price}({ask_vol}股)  买一={bid_price}({bid_vol}股)  价差={spread:.5f}")

3.3 K 线数据

import time

def get_token_kline(symbol: str, kline_type: int = 1, num: int = 100) -> list[dict]:
    """
    获取股票代币 K 线
    kline_type: 1=1分钟, 5=1小时, 8=日K
    """
    resp = requests.post(
        f"{BASE_URL}/crypto/v2/batch_kline",
        headers={"apiKey": API_KEY},
        json={"klineType": kline_type, "klineNum": num, "codes": symbol}
    )
    resp.raise_for_status()
    data = resp.json()["data"]
    return data[0]["respList"] if data else []

# 获取特斯拉代币最近 50 根 1 小时 K 线
klines = get_token_kline("TSLAUSDT", kline_type=5, num=50)
for k in klines[:5]:
    ts = time.strftime("%m-%d %H:%M", time.localtime(int(k["t"])))
    print(f"{ts} | O:{k['o']}  H:{k['h']}  L:{k['l']}  C:{k['c']}  涨跌:{k['pc']}")

---

四、WebSocket 实时订阅（含断线重连）

股票代币使用 business=crypto 的 WebSocket 地址：

wss://data.infoway.io/ws?business=crypto&apikey=YOUR_API_KEY

以下是完整 Python 客户端，同时订阅 TSLA、NVDA、AAPL 三只代币的成交明细 + 盘口 + 1 分钟 K 线：

import os, asyncio, json, uuid, logging
from typing import Optional
import websockets
from websockets.asyncio.client import ClientConnection
from websockets.exceptions import ConnectionClosed

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger("tokenstock-ws")

REQ_TRADE     = 10000
REQ_DEPTH     = 10003
REQ_KLINE     = 10006
REQ_HEARTBEAT = 10010
PUSH_TRADE, PUSH_DEPTH, PUSH_KLINE = 10002, 10005, 10008
ACK_CODES = {10001, 10004, 10007}

SUBSCRIBE_SYMBOLS = "TSLAUSDT,NVDAUSDT,AAPLUSDT"


class TokenStockWSClient:
    """股票代币行情 WebSocket 客户端（指数退避重连）"""

    def __init__(self, api_key: str):
        self.ws_url = f"wss://data.infoway.io/ws?business=crypto&apikey={api_key}"
        self.ws: Optional[ClientConnection] = None
        self.running = True
        self.reconnect_base, self.reconnect_max = 5, 60
        self.heartbeat_interval = 30
        self.heartbeat_task: Optional[asyncio.Task] = None

    async def _send(self, msg: dict) -> None:
        await self.ws.send(json.dumps(msg))

    async def _subscribe_all(self) -> None:
        t = lambda: str(uuid.uuid4())
        await self._send({"code": REQ_TRADE,     "trace": t(), "data": {"codes": SUBSCRIBE_SYMBOLS}})
        await self._send({"code": REQ_DEPTH,     "trace": t(), "data": {"codes": SUBSCRIBE_SYMBOLS}})
        await self._send({"code": REQ_KLINE,     "trace": t(), "data": {"arr": [{"type": 1, "codes": SUBSCRIBE_SYMBOLS}]}})
        logger.info("已订阅股票代币：%s", SUBSCRIBE_SYMBOLS)

    def _start_heartbeat(self) -> None:
        self._cancel_heartbeat()
        async def _loop():
            try:
                while True:
                    await asyncio.sleep(self.heartbeat_interval)
                    if self.ws is None or self.ws.close_code is not None:
                        break
                    await self._send({"code": REQ_HEARTBEAT, "trace": str(uuid.uuid4())})
            except (ConnectionClosed, asyncio.CancelledError):
                pass
        self.heartbeat_task = asyncio.create_task(_loop())

    def _cancel_heartbeat(self) -> None:
        if self.heartbeat_task and not self.heartbeat_task.done():
            self.heartbeat_task.cancel()
        self.heartbeat_task = None

    def _on_message(self, raw: str) -> None:
        try:
            msg = json.loads(raw)
        except json.JSONDecodeError:
            return
        code, data = msg.get("code"), msg.get("data", {})

        if code == PUSH_TRADE:
            direction = {1: "买入", 2: "卖出"}.get(data.get("td"), "中性")
            logger.info("[成交] %s  价格=%s USDT  量=%s股  方向=%s",
                        data.get("s"), data.get("p"), data.get("v"), direction)
        elif code == PUSH_DEPTH:
            # 股票代币只有一档，取 [0][0] 即可
            ask1 = data["a"][0][0] if data.get("a") else "N/A"
            bid1 = data["b"][0][0] if data.get("b") else "N/A"
            logger.info("[盘口] %s  卖一=%s  买一=%s", data.get("s"), ask1, bid1)
        elif code == PUSH_KLINE:
            logger.info("[K线] %s  收=%s USDT  涨跌=%s",
                        data.get("s"), data.get("c"), data.get("pfr"))
        elif code in ACK_CODES:
            logger.info("订阅确认 code=%s", code)

    async def _connect_once(self) -> None:
        async with websockets.connect(self.ws_url) as ws:
            self.ws = ws
            logger.info("WebSocket 连接成功")
            await self._subscribe_all()
            self._start_heartbeat()
            try:
                async for message in ws:
                    self._on_message(message)
            finally:
                self._cancel_heartbeat()
                self.ws = None

    async def start(self) -> None:
        backoff = self.reconnect_base
        while self.running:
            try:
                await self._connect_once()
                backoff = self.reconnect_base
            except ConnectionClosed as e:
                logger.warning("连接关闭: %s", e)
            except Exception as e:
                logger.error("连接异常: %s", e)
            if not self.running:
                break
            logger.info("%.0f 秒后重连...", backoff)
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, self.reconnect_max)


async def main():
    api_key = os.environ.get("INFOWAY_API_KEY", "YOUR_API_KEY")
    await TokenStockWSClient(api_key).start()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("退出")

---

五、使用注意事项

5.1 盘口只有一档

股票代币的深度数据中，a（卖盘）和 b（买盘）数组长度均为 1，直接取 [0][0] 即可，无需像股票接口那样遍历多档。

5.2 美股交易时段 vs 盘后时段的行为差异

美股开盘期间（北京时间 21:30–04:00，夏令时提前 1 小时），代币价格与标的股票联动紧密，价差极小；盘后时段流动性下降，价差可能扩大。构建策略时建议用时间段过滤：

import datetime, pytz

def is_us_market_open() -> bool:
    """判断当前是否处于美股正式交易时段（含夏令时自动切换）"""
    ny_tz = pytz.timezone("America/New_York")
    now_ny = datetime.datetime.now(ny_tz)
    if now_ny.weekday() >= 5:      # 周末
        return False
    market_open  = now_ny.replace(hour=9, minute=30, second=0, microsecond=0)
    market_close = now_ny.replace(hour=16, minute=0,  second=0, microsecond=0)
    return market_open <= now_ny <= market_close

5.3 跨接口联动：获取标的公司财报数据

股票代币的底层是真实上市公司，可通过财报接口查询对应股票的基本面数据，Symbol 格式切换回标准股票代码即可：

def get_underlying_financials(token_symbol: str) -> dict:
    """通过代币 Symbol（如 TSLAUSDT）查询标的股票财务数据"""
    ticker = token_symbol.replace("USDT", "")   # TSLAUSDT → TSLA
    us_symbol = f"{ticker}.US"                  # → TSLA.US
    resp = requests.get(
        f"{BASE_URL}/common/basic/financial/statistics",
        headers={"apiKey": API_KEY},
        params={"symbol": us_symbol, "type": "STOCK_US"}
    )
    resp.raise_for_status()
    return resp.json()["data"]

stats = get_underlying_financials("NVDAUSDT")
for item in stats[:3]:
    print(f"{item['itemName']}: {item.get('currentValue')}")

5.4 ETF 和贵金属代币的特殊性

SPYUSDT、QQQUSDT 等 ETF 代币没有对应的个股财报接口；XAUUSDT（黄金）、XAGUSDT（白银）等贵金属代币同样不支持财报查询，但可通过 /common/batch_trade/ 获取实时报价，逻辑与黄金现货行情一致。