股票池一片下跌，为什么全球判断可能错了？我用一个多市场行情数据源查了八类资产行情

股票池一片下跌时，很多人的第一反应是减仓、收紧策略参数，或者把研究重心切到防守。先停一下。

如果所谓“全球复盘”只读取了股票清单，最该先止住的，是这份偏窄输入继续影响仓位和策略：外汇、黄金、指数、期货、加密货币没有进入比较，你可能把股票的局部状态放大成全球判断；等筛选、回测和监控都跑完了才发现漏项，还要补数据、改代码映射、整轮重跑。

直白答案是：先别扩大结论，先验收研究输入。 TickDB用一套多市场、多资产数据入口，把不同市场的对象放进同一份清单，并让它们继续进入当前行情、历史K线和实时更新。这一步先把“名单外什么都看不见”的静默问题拦在研究入口。

先看结果：8类资产，跑进同一份清单

2026年7月26日23:38，我用文章下方的完整脚本实际跑了一轮：

产品目录摘要：39,774

当前行情：8/8 返回

1d已结束K线：8/8 返回，每个对象5条

BTCUSDT实时ticker：已收到

三市场股票信息、市场指标、资金流：均为3/3

价格是本次运行快照，重跑时会变化。这里要确认的不是哪项资产该买，而是八类代表对象能否沿同一份清单继续进入研究。

只查到一个价格，仍然容易在下一步断掉

一张产品目录只能告诉你“有哪些对象”；只有当前价格，又看不出这次变化放进历史里是什么位置；研究做完后再换一套工具接实时，symbol和订阅逻辑还要重新维护。

这三个断点都会制造返工。更麻烦的是，缺失对象未必报错。你的程序可能照常结束，只是从头到尾都没看见名单之外的资产。

所以这次按同一顺序验收：

产品目录：先看研究范围
→ ticker：确认对象能进入当前观察
→ 1d K线：让同一对象进入历史比较
→ WebSocket：把选定对象继续送入盘中监控

TickDB在这里的差别，不是多给一张报价表，而是让一份跨市场清单继续往下走。REST负责需要时主动拉取目录、行情和历史数据；WebSocket负责持续推送实时更新。把两者拆到不同数据源，新增一个市场时往往就要再找一次代码表、再做一次映射、再接一次实时链路。

真实过程：下面就是本轮运行的完整脚本

这不是伪代码，也不是从另一份程序改写的示例。正文里的文件与本轮23:38实际执行的 tickdb_multi_asset_check.py 一致；运行后会自动创建输出目录，并保存脱敏原始返回、汇总结果和终端文本。

先安装三个依赖：

python3 -m pip install requests websockets certifi

把TickDB控制台生成的API Key放进环境变量，再执行：

export TICKDB_API_KEY="YOUR_TICKDB_API_KEY"
python3 tickdb_multi_asset_check.py --output-dir tickdb_check_output

完整代码：

#!/usr/bin/env python3
"""把一份多资产清单送进 TickDB 的目录、行情、日线和实时任务。"""

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"
USER_AGENT = "TickDB-PD-2026W31-01-public-check/1.0"
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 now() -> str:
    return dt.datetime.now().astimezone().isoformat(timespec="seconds")


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


def get(
    session: requests.Session,
    raw_dir: Path,
    name: str,
    path: str,
    params: dict[str, object],
    api_key: str,
) -> dict:
    started = time.perf_counter()
    response = session.get(
        f"{API}{path}",
        params=params,
        headers={
            "X-API-Key": api_key,
            "Accept": "application/json",
            "User-Agent": USER_AGENT,
        },
        timeout=40,
    )
    try:
        body = response.json()
    except requests.JSONDecodeError:
        body = {"unparsed_text": response.text}
    captured = {
        "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": now(),
    }
    save(raw_dir / f"{name}.json", captured)
    if response.status_code != 200 or body.get("code") != 0:
        raise RuntimeError(f"{name} 请求失败：HTTP {response.status_code}, body={body}")
    return body["data"]


async def get_live_ticker(api_key: str, raw_dir: Path) -> dict:
    symbol = "BTCUSDT"
    endpoint = f"{WS}?api_key={quote(api_key)}"
    subscribe = {
        "cmd": "subscribe",
        "data": {"channel": "ticker", "symbols": [symbol]},
    }
    messages = []
    ticker = None
    error = None
    try:
        async with websockets.connect(
            endpoint,
            ssl=SSL_CONTEXT,
            user_agent_header=USER_AGENT,
            open_timeout=20,
            close_timeout=5,
            ping_interval=None,
        ) as websocket:
            await websocket.send(json.dumps(subscribe))
            await websocket.send(json.dumps({"cmd": "ping"}))
            deadline = time.monotonic() + 25
            while time.monotonic() < deadline and len(messages) < 12:
                try:
                    raw = await asyncio.wait_for(
                        websocket.recv(),
                        timeout=max(0.5, deadline - time.monotonic()),
                    )
                except asyncio.TimeoutError:
                    break
                message = json.loads(raw)
                messages.append(message)
                data = message.get("data")
                if (
                    message.get("cmd") == "ticker"
                    and isinstance(data, dict)
                    and data.get("symbol") == symbol
                ):
                    ticker = message
                    break
    except Exception as exc:
        error = f"{type(exc).__name__}: {exc}"
    captured = {
        "request": {
            "endpoint": f"{WS}?api_key=REDACTED",
            "subscription": subscribe,
        },
        "messages": messages,
        "ticker_message": ticker,
        "error": error,
        "captured_at": now(),
    }
    save(raw_dir / "websocket_BTCUSDT_ticker.json", captured)
    if ticker is None:
        raise RuntimeError(f"没有收到 BTCUSDT ticker：{error or messages}")
    return ticker


def run(output_dir: Path, api_key: str) -> tuple[dict, list[str]]:
    raw_dir = output_dir / "raw"
    session = requests.Session()
    started_at = now()

    catalog = get(
        session,
        raw_dir,
        "catalog_all",
        "/v1/symbols/available",
        {"offset": 0, "limit": 5},
        api_key,
    )
    total = catalog["summary"]["total_products"]

    symbols = [symbol for _, symbol, _ in OBJECTS]
    ticker_data = get(
        session,
        raw_dir,
        "ticker_multi_asset",
        "/v1/market/ticker",
        {"symbols": ",".join(symbols)},
        api_key,
    )
    ticker = {item["symbol"]: item for item in ticker_data}

    rows = []
    for asset, symbol, asset_type in OBJECTS:
        data = 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": ticker.get(symbol, {}).get("last_price"),
                "ticker_result": "PASS" if symbol in ticker else "FAIL",
                "kline_count": len(data.get("klines", [])),
            }
        )

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

    live = asyncio.run(get_live_ticker(api_key, raw_dir))
    live_last = live["data"]["last_price"]
    ticker_count = sum(row["ticker_result"] == "PASS" for row in rows)
    kline_count = sum(row["kline_count"] == 5 for row in rows)
    overall = (
        "PASS"
        if ticker_count == 8
        and kline_count == 8
        and len(stock_info) == 3
        and len(metrics) == 3
        and flow_passed == 3
        else "PARTIAL"
    )
    finished_at = now()

    lines = [
        "TickDB 多市场研究清单验收",
        f"运行时间：{started_at}",
        "API Key：已从环境变量读取（未打印）",
        "",
        "[1/4] 产品目录",
        f"无过滤目录摘要：PASS，当前 total={total}",
        "",
        "[2/4] 当前行情 + 5条已结束日线",
    ]
    lines.extend(
        f"{row['asset']} {row['symbol']}：ticker={row['ticker_result']}，"
        f"last={row['last_price']}，1d K线={row['kline_count']}"
        for row in rows
    )
    lines.extend(
        [
            "",
            "[3/4] 股票对象继续研究",
            f"股票信息：{len(stock_info)}/3",
            f"市场指标：{len(metrics)}/3",
            f"资金流：{flow_passed}/3",
            "",
            "[4/4] BTCUSDT 实时行情",
            f"WebSocket ticker=PASS，symbol=BTCUSDT，last={live_last}",
            "",
            f"总结果：{overall}",
            f"当前行情：{ticker_count}/8；日线：{kline_count}/8",
            f"完成时间：{finished_at}",
            f"输出目录：{output_dir}",
        ]
    )
    report = {
        "schema_version": "tickdb-reader-multi-asset-check-1.0",
        "run_started_at": started_at,
        "run_finished_at": finished_at,
        "catalog_total": total,
        "objects": rows,
        "ticker_passed_count": ticker_count,
        "kline_passed_count": kline_count,
        "stock_context": {
            "symbols": STOCKS,
            "stock_info_count": len(stock_info),
            "market_metrics_count": len(metrics),
            "capital_flow_passed_count": flow_passed,
        },
        "websocket": {
            "symbol": "BTCUSDT",
            "result": "PASS",
            "last_price": live_last,
        },
        "overall": overall,
    }
    return report, lines


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-dir", default="tickdb_check_output")
    args = parser.parse_args()

    api_key = os.environ.get("TICKDB_API_KEY", "").strip()
    if not api_key:
        print("缺少 TICKDB_API_KEY。请先把 API Key 放入环境变量。", file=sys.stderr)
        return 2

    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    report, lines = run(output_dir, api_key)
    save(output_dir / "summary.json", report)
    transcript = "\n".join(lines) + "\n"
    (output_dir / "run_transcript.txt").write_text(transcript, encoding="utf-8")
    print(transcript, end="")
    return 0 if report["overall"] == "PASS" else 1


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

本轮完整终端输出

下面的价格、数量和时间都来自上面这份脚本的同一次运行：

TickDB 多市场研究清单验收
运行时间：2026-07-26T23:38:42+08:00
API Key：已从环境变量读取（未打印）

[1/4] 产品目录
无过滤目录摘要：PASS，当前 total=39774

[2/4] 当前行情 + 5条已结束日线
外汇 EURUSD：ticker=PASS，last=1.13709，1d K线=5
贵金属 XAUUSD：ticker=PASS，last=4056.39500，1d K线=5
指数 SPX：ticker=PASS，last=7411.98000，1d K线=5
美股 AAPL.US：ticker=PASS，last=333.02，1d K线=5
港股 700.HK：ticker=PASS，last=434.6，1d K线=5
A股 600519.SH：ticker=PASS，last=1297.41，1d K线=5
中国期货 BU2609：ticker=PASS，last=4260.00，1d K线=5
加密货币 BTCUSDT：ticker=PASS，last=64811.16000000，1d K线=5

[3/4] 股票对象继续研究
股票信息：3/3
市场指标：3/3
资金流：3/3

[4/4] BTCUSDT 实时行情
WebSocket ticker=PASS，symbol=BTCUSDT，last=64811.16000000

总结果：PASS
当前行情：8/8；日线：8/8
完成时间：2026-07-26T23:38:53+08:00
输出目录：tickdb_check_output

如果某个请求HTTP状态或业务码异常，脚本会直接停止；如果收不到目标实时ticker，也会返回错误。它不会用一行“请求成功”掩盖空数据。

把自己的研究清单换进去，十分钟做一次验收

不必先迁移整个系统。

打开脚本顶部的 OBJECTS，换成自己平时关注的对象。建议先放10个，故意混合股票、指数、外汇、贵金属、期货和加密货币。保留 asset、symbol、type 三项：

OBJECTS = [
    ("美股", "AAPL.US", "stock"),
    ("港股", "700.HK", "stock"),
    ("A股", "600519.SH", "stock"),
    ("外汇", "EURUSD", "forex"),
    ("贵金属", "XAUUSD", "forex"),
]

跑完后先看 summary.json：

• ticker_passed_count 是否等于对象数；
• 每个对象的 kline_count 是否符合预期；
• websocket.result 是否为 PASS；
• 失败时回到 raw/ 查看对应请求，而不是把空结果送进下一步。

这次验收要回答一个很具体的问题：当你准备根据“全球市场”调整判断时，系统究竟看见了哪些资产，又有哪些对象根本没进来。

FAQ

1. 为什么 symbol 和 type 要一起检查？

因为同一个代码字符串只有放在正确的资产类型和市场规则里才有明确含义。清单扩到新市场时，先用产品目录确认对象，再用ticker和K线逐项验收；不要只复制一份代码表就默认后续接口都能识别。

2. REST和WebSocket怎样分工，哪里最容易返工？

定时复盘、批量核对和历史拉取用REST；盘中持续更新用WebSocket。最容易返工的是研究阶段用一套symbol，实时阶段又换一套订阅标识。先让同一清单走通当前和历史，再把需要盯盘的对象送入实时订阅。

3. TickDB除了多市场行情，还能把股票研究带到哪一步？

本轮对 AAPL.US、700.HK、600519.SH 继续调用了股票信息、市场指标和资金流，结果都是3/3。范围确认以后，可以用市场指标补充初筛，再用资金流观察候选股票的盘中结构；跨市场比较前，仍要先处理币种、交易时段和口径。

第一次测试不必做大。拿自己的10个对象跑通，先把看不见的研究盲区找出来，再决定哪些数据要进入后面的比较、回测和监控。

入口：TickDB官网｜中文文档｜行情快照｜历史K线｜WebSocket快速开始

通过 TickDB API 获取实时行情数据

一个 API 接入外汇、加密货币、美股、港股、A股、贵金属和全球指数的实时行情。支持 WebSocket 低延迟推送，免费开始使用。