一只原本值得继续研究的股票已经出现明显变化，你到盘后才从新闻或朋友转发里看到。
问题可能不在指标。它从来没有进入你的观察池，均线、量比、估值和预警自然都看不见它。你花了半小时调整参数，实际上只是把一个偏小的范围筛得更细。
所以，多市场选股的第一步不是继续加指标，而是先回答：
该看的市场和标的，是否已经进入同一张观察清单？
TickDB 在这里解决的也不是“替你找出会涨的股票”。它能把 A 股、港股、美股等对象放进同一条行情研究路径：先找到对象，再批量取得当前行情、研究指标和历史 K 线，最后由你自己的规则决定先研究谁。
先看这次真实运行的结果。
三个市场，先放进同一条研究链
本次运行时间为 2026年7月24日09:11:36 UTC。共发出 8 次请求，3 个跨市场样本全部通过。
| 市场与样本 | 当次目录股票数 | 现价 | 5日变化 | 10日变化 | 量比 | 日K线 |
|---|---|---|---|---|---|---|
| A股 600519.SH 贵州茅台 | 7059 | 1297.41 | 3.54% | 7.67% | 0.48 | 20根 |
| 港股 700.HK 腾讯控股 | 3156 | 434.6 | -5.85% | -5.56% | 0.58 | 20根 |
| 美股 AAPL.US Apple Inc. | 12587 | 321.66 | -3.48% | 1.72% | 0.78 | 20根 |
三只股票只是容易识别的接口验收样本，不代表推荐。表里的目录数量、价格、变化率和量比，也只对应这次运行。
这张表真正有用的地方，不是比较三家公司的好坏，更不是把不同币种的股价放在一起排名。它证明了一件更基础的事：三个市场的对象能够先进入同一条研究链，再用一致的步骤核对当前状态和历史数据。
脚本还按“五日绝对涨跌幅”排了一个人工复核顺序：
700.HK > 600519.SH > AAPL.US
这个顺序只表示本次三个样本里先检查谁，不是买卖信号，也不代表未来收益。
完整代码：从对象目录跑到日K线
下面是本次真实运行的完整代码。Key 只从环境变量读取，不会打印，也不会写入输出文件。
复现前先在本机设置：
export TICKDB_API_KEY="你自己的 Key"
然后运行：
#!/usr/bin/env python3
"""真实运行 TickDB 的 A股、港股、美股最小研究链。
运行前：
export TICKDB_API_KEY="你自己的 Key"
密钥只从环境变量读取，不打印、不写入文件。
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import urllib.parse
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
BASE_URL = "//api.tickdb.ai"
RUN_ID = "F-W30-01R-THS-EV13-LIVE-20260724"
SAMPLES = [
("A股", "CN", "600519.SH"),
("港股", "HK", "700.HK"),
("美股", "US", "AAPL.US"),
]
OUTPUT_ROOT = Path(__file__).resolve().parents[1]
def save_json(path: Path, value: Any) -> 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_json(
evidence_id: str,
path: str,
params: dict[str, Any],
api_key: str,
) -> dict[str, Any]:
query = urllib.parse.urlencode(params)
url = f"{BASE_URL}{path}?{query}"
result = subprocess.run(
[
"curl",
"--location",
"--silent",
"--show-error",
"--max-time",
"30",
"--header",
f"X-API-Key: {api_key}",
"--header",
"Accept: application/json",
"--write-out",
"\nHTTP_STATUS:%{http_code}",
url,
],
check=False,
capture_output=True,
text=True,
timeout=35,
)
body, marker, status = result.stdout.rpartition("\nHTTP_STATUS:")
http_status = int(status.strip()) if marker and status.strip().isdigit() else None
if result.returncode != 0 or http_status != 200:
raise RuntimeError(
f"request failed: path={path}, "
f"curl_exit={result.returncode}, http={http_status}"
)
payload = json.loads(body)
if payload.get("code") != 0:
raise RuntimeError(
f"API rejected request: path={path}, code={payload.get('code')}, "
f"message={payload.get('message')}"
)
record = {
"evidence_id": evidence_id,
"evidence_run_id": RUN_ID,
"retrieved_at_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
"method": "GET",
"base_url": BASE_URL,
"path": path,
"query": params,
"http_status": http_status,
"response": payload,
"sanitization": "Authentication header and API key are excluded.",
}
save_json(OUTPUT_ROOT / "raw" / f"{evidence_id}.json", record)
return payload
def rows_by_symbol(payload: dict[str, Any]) -> dict[str, dict[str, Any]]:
return {
row["symbol"]: row
for row in payload["data"]
if isinstance(row, dict) and "symbol" in row
}
def pct(value: Any) -> str:
return f"{float(value) * 100:.2f}%"
def main() -> int:
api_key = os.getenv("TICKDB_API_KEY")
if not api_key:
print("请先在环境变量中设置 TICKDB_API_KEY", file=sys.stderr)
return 2
retrieved_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
catalog_totals: dict[str, int] = {}
for _, market, _ in SAMPLES:
payload = get_json(
f"catalog_{market.lower()}_stock",
"/v1/symbols/available",
{"market": market, "type": "stock", "limit": 3, "offset": 0},
api_key,
)
catalog_totals[market] = payload["data"]["pagination"]["total"]
symbols = ",".join(symbol for _, _, symbol in SAMPLES)
ticker_payload = get_json(
"ticker_three_markets",
"/v1/market/ticker",
{"symbols": symbols, "type": "stock"},
api_key,
)
metrics_payload = get_json(
"metrics_three_markets",
"/v1/market/calc-index",
{"symbols": symbols, "type": "stock"},
api_key,
)
ticker = rows_by_symbol(ticker_payload)
metrics = rows_by_symbol(metrics_payload)
lines = [
"$ python public_tickdb_cross_market_check.py",
f"evidence_run_id: {RUN_ID}",
f"retrieved_at_utc: {retrieved_at}",
"api_key: loaded from environment (not printed)",
"flow: 3 catalogs + batch ticker + batch metrics + 3 daily K-lines",
"",
]
results: list[dict[str, Any]] = []
review_order: list[tuple[float, str]] = []
for market_name, catalog_market, symbol in SAMPLES:
kline_payload = get_json(
f"kline_{symbol.lower().replace('.', '_')}_1d_20",
"/v1/market/kline",
{"symbol": symbol, "type": "stock", "interval": "1d", "limit": 20},
api_key,
)
bars = kline_payload["data"]["klines"]
quote = ticker[symbol]
indicator = metrics[symbol]
review_order.append((abs(float(indicator["five_day_change_rate"])), symbol))
row = {
"market": market_name,
"symbol": symbol,
"name": quote["name"],
"catalog_stock_total": catalog_totals[catalog_market],
"last_price": quote["last_price"],
"price_change_percent_24h": quote["price_change_percent_24h"],
"ticker_timestamp": quote["timestamp"],
"five_day_change_rate": pct(indicator["five_day_change_rate"]),
"ten_day_change_rate": pct(indicator["ten_day_change_rate"]),
"volume_ratio": indicator["volume_ratio"],
"daily_kline_count": len(bars),
"latest_kline_time": bars[-1]["time"],
"latest_close": bars[-1]["close"],
"status": "PASS" if len(bars) == 20 else "FAIL",
}
results.append(row)
lines.extend(
[
f"[{market_name}] {symbol} {row['name']}",
(
f"目录股票数={row['catalog_stock_total']} | "
f"现价={row['last_price']} | 24h={row['price_change_percent_24h']}%"
),
(
f"5日={row['five_day_change_rate']} | "
f"10日={row['ten_day_change_rate']} | "
f"量比={row['volume_ratio']}"
),
(
f"日K线={row['daily_kline_count']}根 | "
f"最近收盘={row['latest_close']} | "
f"时间={row['latest_kline_time']} | {row['status']}"
),
"",
]
)
review_order.sort(reverse=True)
lines.extend(
[
"人工复核顺序（仅按5日绝对涨跌幅）："
+ " > ".join(symbol for _, symbol in review_order),
"SUMMARY: 3/3 samples passed",
"BOUNDARY: 本次三个样本、单次运行；不是荐股、收益排名或SLA证明。",
]
)
summary = {
"schema_version": "tickdb-live-research-acceptance-summary-1.0",
"task_id": "F-W30-01R-THS",
"platform": "tonghuashun",
"evidence_run_id": RUN_ID,
"retrieved_at_utc": retrieved_at,
"credential": "loaded from environment; value not saved",
"request_count": 8,
"passed_samples": sum(row["status"] == "PASS" for row in results),
"total_samples": len(results),
"results": results,
"manual_review_order_by_absolute_5d_move": [
symbol for _, symbol in review_order
],
"boundary": (
"This single run proves only the stated requests, samples and time. "
"It does not prove completeness, long-term stability, latency, "
"accuracy, SLA or investment performance."
),
}
save_json(
OUTPUT_ROOT / "summary" / "live_research_acceptance_summary.json",
summary,
)
transcript = "\n".join(lines) + "\n"
terminal_path = OUTPUT_ROOT / "terminal" / "public_code_run_terminal.txt"
terminal_path.parent.mkdir(parents=True, exist_ok=True)
terminal_path.write_text(transcript, encoding="utf-8")
print(transcript, end="")
return 0 if summary["passed_samples"] == len(results) else 1
if __name__ == "__main__":
raise SystemExit(main())
这段代码按下面的顺序工作：
- 查询 CN、HK、US 三个股票目录，确认对象范围；
- 一次批量取得三只样本的当前行情；
- 一次批量取得五日、十日变化率和量比；
- 分别取得三只样本的 20 根日 K 线；
- 保存脱敏原始响应、汇总和终端文本；
- 只按五日绝对涨跌幅排人工复核顺序。
同一次运行的完整终端输出
$ python public_tickdb_cross_market_check.py
evidence_run_id: F-W30-01R-THS-EV13-LIVE-20260724
retrieved_at_utc: 2026-07-24T09:11:36+00:00
api_key: loaded from environment (not printed)
flow: 3 catalogs + batch ticker + batch metrics + 3 daily K-lines
[A股] 600519.SH 贵州茅台
目录股票数=7059 | 现价=1297.41 | 24h=0.42%
5日=3.54% | 10日=7.67% | 量比=0.48
日K线=20根 | 最近收盘=1297.41 | 时间=1784822400000 | PASS
[港股] 700.HK 腾讯控股
目录股票数=3156 | 现价=434.6 | 24h=-2.38%
5日=-5.85% | 10日=-5.56% | 量比=0.58
日K线=20根 | 最近收盘=434.6 | 时间=1784822400000 | PASS
[美股] AAPL.US Apple Inc.
目录股票数=12587 | 现价=321.66 | 24h=-1.30%
5日=-3.48% | 10日=1.72% | 量比=0.78
日K线=20根 | 最近收盘=321.66 | 时间=1784779200000 | PASS
人工复核顺序（仅按5日绝对涨跌幅）：700.HK > 600519.SH > AAPL.US
SUMMARY: 3/3 samples passed
BOUNDARY: 本次三个样本、单次运行；不是荐股、收益排名或SLA证明。
8 次请求均返回 HTTP 200、接口 code 0，3/3 样本通过。这只能证明本次三个样本和这些请求跑通，不能外推为全市场完整、长期稳定、准确率、低延迟或 SLA。
TickDB 的六类核心价值，放到投资任务里怎么看
产品能力不能停在“接口能返回数据”。对投资者来说，更重要的是它能把哪一步做得更顺。
| 核心价值 | 投资者能得到什么 | 本文如何体现 |
|---|---|---|
| 看得更广 | 先把更多市场和对象放进观察范围 | 三个股票目录与统一 symbol 规则，让 A 股、港股、美股样本进入同一研究链 |
| 筛得更快 | 对池内对象排出先研究谁 | 批量行情和指标提供研究输入；排序规则由使用者定义 |
| 及时感知变化 | 让观察池的变化进入后续监控 | 当前产品能力可继续接入实时订阅；本轮没有实跑 WebSocket |
| 判断更稳 | 先核对对象、类型、时间和数据状态 | symbol、type、时间戳、K 线数量和返回状态都能成为检查项 |
| 少做重复接入 | 扩展市场时复用一套请求、鉴权和保存逻辑 | 同一脚本处理三个市场；本文不据此承诺具体成本降幅 |
| 做成研究工具或 AI 工作流 | 把结构化行情接入扫描器、研究助手或内部工具 | 结构化结果可以继续加工；本轮没有实跑 AI 判断 |
这六类价值不是六个互不相干的口号。它们更适合按任务组合。
四套更实用的能力组合
1. 普通投资者的跨市场观察池
组合：看得更广 + 筛得更快 + 判断更稳
先按市场、行业或产业链找到对象，再批量取得当前状态和历史 K 线。排序只负责告诉你“先检查谁”，最终判断仍由人完成。
适合的任务是：每天从多个市场中缩小人工阅读范围，又不想一开始就漏掉对象。
2. 量化研究的统一输入链
组合：看得更广 + 筛得更快 + 判断更稳 + 少做重复接入
先统一对象标识和数据资格检查，再把行情、指标和 K 线保存到研究环境。后面无论做因子、回测还是横截面比较，都从同一批经过检查的输入开始。
本文没有验证任何因子收益，也没有给出策略结论。这里讨论的是研究输入，不是收益结果。
3. 从扫描到持续监控
组合：筛得更快 + 及时感知变化 + 判断更稳
先用批量数据缩小范围，再把留下的对象放进持续监控。实际落地时还需要自行处理交易时段、权限、断线恢复、补查、去重和保存。
本次只跑了 REST 请求，没有把实时订阅写成已经验证的结论。
4. 研究工具或 AI 助手
组合：看得更广 + 判断更稳 + 少做重复接入 + 做成工具或 AI 工作流
对象、行情、指标和 K 线都变成结构化输入后，可以继续做内部扫描器、研究看板或 AI 研究助手。AI 负责整理和提问，人负责核对来源、时间与风险。
这条组合的价值在于减少数据准备的断点，不代表 AI 可以替代投资判断。
三个容易混淆的问题
1. 为什么代码里是 600519.SH、700.HK 和 AAPL.US？
后缀用于区分市场。统一 symbol 让程序知道对象属于哪个市场，也能减少重名和代码歧义。统一写法不代表三个市场的字段、权限、深度、交易时段和历史范围完全一样。
2. 当前行情和日 K 线为什么都要取？
当前行情回答“现在是什么状态”，日 K 线回答“此前怎样走到这里”。前者适合扫描，后者适合核对历史路径。只看其中一个，研究上下文都不够完整。
3. 能不能直接把人工复核顺序当成选股结果？
不能。本文只是用五日绝对涨跌幅演示如何在三个样本中安排检查顺序。它没有使用公司基本面、估值、流动性、交易成本、汇率、组合风险等条件，也没有验证与未来收益的关系。
想自己试，先做这四步
- 把自己的 TickDB Key 放进环境变量，不要写死在代码里；
- 把三个示例 symbol 换成自己的观察对象；
- 先核对对象、类型、返回时间和 K 线数量；
- 再加入自己的筛选条件，并把“排序”和“买卖决定”分开。
如果这四步跑通，你得到的不是一张神奇股票名单，而是一条更完整、可检查、可重复的研究输入链。
观察池决定你有机会研究谁，指标决定你怎样缩小范围。次序反了，指标越精细，也可能只是在一个过窄的池子里反复优化。