当股票池一起波动时，最容易出错的不是代码，而是把“我看见的股票”当成“我已经覆盖的全球市场”。

如果外汇、黄金、指数、期货或加密资产根本不在研究输入里，后面无论如何解释行情，结论的适用范围都已经被悄悄缩小了。

本文不提供买卖建议，也不判断任何资产的涨跌。它只完成一个工程任务：用一份 Python 清单，验收 8 类代表对象能否依次进入当前行情、已结束日线和实时订阅。

适合谁：用 Python 维护研究清单、做量化研究输入或搭建行情接入的开发者。

你将得到：一套可替换对象清单的验收脚本、一次真实运行的冻结结果、失败时该在哪一层停下来的判断方法。

---

先看验收目标：不是“有价格”，而是数据链不断

一份多市场行情数据，至少要回答四个不同的问题：

关卡	要验证什么	本文的判定方式
产品目录	研究对象是否处于可发现的产品范围	读取目录摘要，不把摘要总数当永久能力承诺
当前行情	清单中的对象能否同轮取得快照	8 个对象在一个 ticker 请求中返回
历史数据	对象能否进入回看与研究	每个对象取得 5 条已结束的 1d K 线
实时数据	至少有一个对象能否真正收到订阅消息	收到一条 BTCUSDT ticker 消息

关键不在于所有市场共用同一种字段或同一个交易时段；它们并不相同。关键在于：你自己维护的同一份对象清单，不会在“当前—历史—实时”之间丢失 symbol 和 type 的对应关系。

---

本轮真实结果：8 类代表对象，四道关卡全部通过

下面是一次已冻结运行的结果，运行区间为 2026-07-26 23:38:42—23:38:53（UTC+8）。价格和目录总量只代表该时刻，重跑后会变化，不能视为实时行情或永久覆盖范围。

资产类别	symbol	type	当次 ticker	1d 已结束 K 线
外汇	EURUSD	forex	PASS，1.13709	5 条
贵金属	XAUUSD	forex	PASS，4056.39500	5 条
指数	SPX	indices	PASS，7411.98000	5 条
美股	AAPL.US	stock	PASS，333.02	5 条
港股	700.HK	stock	PASS，434.6	5 条
A 股	600519.SH	stock	PASS，1297.41	5 条
中国期货	BU2609	futures	PASS，4260.00	5 条
加密货币	BTCUSDT	crypto	PASS，64811.16000000	5 条

同轮还得到以下结果：

• 无过滤产品目录摘要：total_products = 39774；
• 三个股票对象 AAPL.US、700.HK、600519.SH 的股票信息、市场指标、资金流请求均为 3/3；
• WebSocket 实际收到了 BTCUSDT 的一条 cmd=ticker 消息；
• 总判定：PASS。

这里的 “PASS” 仅说明上面的样本、接口、参数和运行时刻通过了检查。它不证明所有标的、所有周期、完整历史、低延迟、稳定性或 SLA。

---

最小可跑代码：先把研究对象写成一张表

安装依赖：

pip install requests websockets certifi
export TICKDB_API_KEY='你的 API Key'

将下面代码保存为 tickdb_multi_asset_check.py，再执行：

python3 tickdb_multi_asset_check.py --output-dir tickdb_check_output

脚本把 API Key 只从环境变量读取，不会打印 Key；每次运行会保存脱敏后的请求/返回、汇总和终端输出。若任一必要请求失败，会直接报错，而不是把缺失数据伪装成成功。

#!/usr/bin/env python3
import argparse
import asyncio
import datetime as dt
import json
import os
from pathlib import Path
import ssl
import sys
import time
from urllib.parse import quote

import certifi
import requests
import websockets

API = "https://api.tickdb.ai"
WS = "wss://api.tickdb.ai/v1/realtime"
SSL_CONTEXT = ssl.create_default_context(cafile=certifi.where())

# 只改这里，即可替换自己的研究清单；type 必须与对象类别匹配。
OBJECTS = [
    ("外汇", "EURUSD", "forex"),
    ("贵金属", "XAUUSD", "forex"),
    ("指数", "SPX", "indices"),
    ("美股", "AAPL.US", "stock"),
    ("港股", "700.HK", "stock"),
    ("A股", "600519.SH", "stock"),
    ("中国期货", "BU2609", "futures"),
    ("加密货币", "BTCUSDT", "crypto"),
]
STOCKS = ["AAPL.US", "700.HK", "600519.SH"]


def timestamp():
    return dt.datetime.now().astimezone().isoformat(timespec="seconds")


def save_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def api_get(session, raw_dir, name, path, params, api_key):
    started = time.perf_counter()
    response = session.get(
        f"{API}{path}",
        params=params,
        headers={"X-API-Key": api_key, "Accept": "application/json"},
        timeout=40,
    )
    try:
        body = response.json()
    except requests.JSONDecodeError:
        body = {"unparsed_text": response.text}
    save_json(raw_dir / f"{name}.json", {
        "request": {"method": "GET", "path": path, "params": params},
        "response": {
            "status": response.status_code,
            "elapsed_ms": round((time.perf_counter() - started) * 1000, 1),
            "body": body,
        },
        "captured_at": timestamp(),
    })
    if response.status_code != 200 or body.get("code") != 0:
        raise RuntimeError(f"{name} failed: HTTP {response.status_code}, body={body}")
    return body["data"]


async def websocket_ticker(api_key, raw_dir):
    symbol = "BTCUSDT"
    subscribe = {"cmd": "subscribe", "data": {"channel": "ticker", "symbols": [symbol]}}
    messages, ticker, error = [], None, None
    try:
        async with websockets.connect(
            f"{WS}?api_key={quote(api_key)}",
            ssl=SSL_CONTEXT,
            open_timeout=20,
            close_timeout=5,
            ping_interval=None,
        ) as ws:
            await ws.send(json.dumps(subscribe))
            await ws.send(json.dumps({"cmd": "ping"}))
            deadline = time.monotonic() + 25
            while time.monotonic() < deadline and len(messages) < 12:
                try:
                    raw = await asyncio.wait_for(ws.recv(), timeout=max(0.5, deadline - time.monotonic()))
                except asyncio.TimeoutError:
                    break
                message = json.loads(raw)
                messages.append(message)
                if message.get("cmd") == "ticker" and message.get("data", {}).get("symbol") == symbol:
                    ticker = message
                    break
    except Exception as exc:
        error = f"{type(exc).__name__}: {exc}"

    save_json(raw_dir / "websocket_BTCUSDT_ticker.json", {
        "request": {"endpoint": f"{WS}?api_key=REDACTED", "subscription": subscribe},
        "messages": messages,
        "ticker_message": ticker,
        "error": error,
        "captured_at": timestamp(),
    })
    if ticker is None:
        raise RuntimeError(f"BTCUSDT ticker not received: {error or messages}")
    return ticker


def run(output_dir, api_key):
    raw_dir = output_dir / "raw"
    session = requests.Session()
    started_at = timestamp()

    catalog = api_get(session, raw_dir, "catalog_all", "/v1/symbols/available", {"offset": 0, "limit": 5}, api_key)
    ticker_rows = api_get(
        session, raw_dir, "ticker_multi_asset", "/v1/market/ticker",
        {"symbols": ",".join(symbol for _, symbol, _ in OBJECTS)}, api_key,
    )
    tickers = {row["symbol"]: row for row in ticker_rows}

    rows = []
    for asset, symbol, asset_type in OBJECTS:
        kline = api_get(
            session, raw_dir, f"kline_{symbol.replace('.', '_')}", "/v1/market/kline",
            {"symbol": symbol, "interval": "1d", "limit": 5, "type": asset_type}, api_key,
        )
        rows.append({
            "asset": asset, "symbol": symbol, "type": asset_type,
            "last_price": tickers.get(symbol, {}).get("last_price"),
            "ticker_result": "PASS" if symbol in tickers else "FAIL",
            "kline_count": len(kline.get("klines", [])),
        })

    stock_info = api_get(session, raw_dir, "stock_info_three_markets", "/v1/market/stock-info", {"symbols": ",".join(STOCKS), "type": "stock"}, api_key)
    metrics = api_get(session, raw_dir, "market_metrics_three_markets", "/v1/market/calc-index", {"symbols": ",".join(STOCKS), "type": "stock"}, api_key)
    for symbol in STOCKS:
        api_get(session, raw_dir, f"capital_flow_{symbol.replace('.', '_')}", "/v1/market/capital-flow", {"symbol": symbol, "type": "stock"}, api_key)

    live = asyncio.run(websocket_ticker(api_key, raw_dir))
    ticker_ok = sum(row["ticker_result"] == "PASS" for row in rows)
    kline_ok = sum(row["kline_count"] == 5 for row in rows)
    overall = "PASS" if (ticker_ok, kline_ok, len(stock_info), len(metrics)) == (8, 8, 3, 3) else "PARTIAL"
    report = {
        "run_started_at": started_at,
        "run_finished_at": timestamp(),
        "catalog_total": catalog["summary"]["total_products"],
        "objects": rows,
        "ticker_passed_count": ticker_ok,
        "kline_passed_count": kline_ok,
        "stock_info_count": len(stock_info),
        "market_metrics_count": len(metrics),
        "capital_flow_passed_count": len(STOCKS),
        "websocket": {"symbol": "BTCUSDT", "last_price": live["data"]["last_price"]},
        "overall": overall,
    }
    save_json(output_dir / "summary.json", report)
    (output_dir / "run_transcript.txt").write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    return report


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-dir", default="tickdb_check_output")
    output_dir = Path(parser.parse_args().output_dir)
    api_key = os.environ.get("TICKDB_API_KEY", "").strip()
    if not api_key:
        print("Missing TICKDB_API_KEY", file=sys.stderr)
        return 2
    report = run(output_dir, api_key)
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0 if report["overall"] == "PASS" else 1


if __name__ == "__main__":
    raise SystemExit(main())

说明：代码展示的是验收逻辑。你实际替换标的时，应该同步检查 symbol、type、市场交易时段、币种和字段口径；它们不能因为入口统一就被默认视为可直接横向比较。

---

这段代码到底做了什么

它没有“自动判断全球市场”，也没有生成交易信号；它只是把研究输入变成一个可核对的契约。

研究对象清单
    ├── 目录摘要：确认发现入口可用
    ├── ticker：确认当前行情快照
    ├── 1d K 线：确认对象可进入历史研究
    └── WebSocket：确认至少一个实时订阅实际有消息

股票样本在这里还有一条补充分支：股票信息、市场指标、资金流。它们可以作为候选对象的补充研究输入，但不能直接跨市场排名，更不能从单次返回推导出投资结论。

对于工程实现而言，最重要的是保存三类产物：

1. raw/：每个接口的脱敏原始响应，方便定位参数或字段问题；
2. summary.json：供后续程序读取的机器可读验收结果；
3. 终端输出：让人快速确认此次通过/失败的位置。

这比只在 Notebook 里看到一个价格更有用：一旦后续研究结果异常，你能回答“问题出在目录、当前行情、历史 K 线，还是实时订阅”。

---

四类失败，分别应该怎么处理

1. HTTP 200，但结果为空

不要只把 HTTP 状态码当成功。优先检查 type 是否与对象匹配、产品过滤条件、分页参数，以及响应中的业务状态和实际数据字段。HTTP 200 只表示请求成功到达，不代表研究对象真的被返回。

2. ticker 有了，K 线却没有

这通常不是“行情源完全不可用”的结论。先核对 symbol/type，再明确当前验证的是 1d、limit=5 的已结束 K 线。其他周期、其他日期范围、完整历史和回测可用性需要单独验收。

3. REST 正常，WebSocket 没有 ticker

把实时订阅视为独立关卡：确认订阅频道、symbol、鉴权、网络连通性和消息类型。本文只验证了某一轮收到一次 BTCUSDT ticker，未测试重连、断线恢复、延迟或持续稳定性。

4. 数据都到了，却无法解释或比较

这恰恰说明应该暂停使用结论。不同资产可能有不同币种、交易时段、价格精度、复权逻辑和字段定义。统一接入解决的是“进入同一条研究流程”，不是自动消除业务口径差异。

---

把它放进你的研究流程：先验收，再解释

每次准备扩大研究范围时，可以按下面顺序做：

• 先写出对象清单：对象、资产类别、symbol、type；
• 运行四关检查；
• 只对通过的对象进入下游分析；
• 对失败对象保留原始响应和失败原因；
• 在报告里明确本次覆盖了什么、没有覆盖什么。

这样做不会让市场变得可预测，但能避免一个更基础的错误：用股票局部的输入，替代一个本应覆盖多资产的研究问题。

---

发布前/使用前检查表

• API Key 只在环境变量或密钥管理中，不进入代码、截图和日志；
• 每个对象都显式写了 symbol 与 type；
• ticker、历史和实时检查的对象集合没有悄悄分叉；
• 明确 K 线验证的周期、数量和结束状态；
• 将目录总量、价格和实时消息标记为运行时快照；
• 对空结果检查业务码、过滤条件、分页和字段，而不是只看 HTTP 200；
• 不把一次样本通过外推为全量覆盖、低延迟、完整性或 SLA；
• 不把研究输入、市场指标或资金流写成荐股、收益承诺或自动交易结论。

---

常见问题

多市场行情数据怎样避免把对象映射错？

把 symbol 和 type 放在同一份版本化清单中，并让 ticker、K 线和订阅都从这份清单派生。不要在三个任务里手写三套 symbol。

REST 和 WebSocket 应该如何分工？

周期性验收、当前快照和历史 K 线适合走 REST；持续观察实时变化才需要 WebSocket。两者是任务分工，不是自动连续执行的工作流。

股票信息、市场指标、资金流可以直接生成选股结论吗？

不可以。它们只是研究输入。还需要明确字段含义、适用市场、时间口径和你自己的研究方法；本文也不提供任何投资建议。

---

结语

真正值得先验证的，往往不是某个价格是否变化，而是你的研究清单是否足以支撑你准备回答的问题。

先让对象进入可追溯的当前、历史和实时数据链，再讨论如何解释市场。这是多市场研究里成本最低、也最容易被忽略的一道质量门。

本文的实测结果来自 2026-07-26 23:38（UTC+8）的一次冻结运行。目录总量、价格与实时消息会随重跑变化；本文不构成投资建议。