美股数据源选型指南：看板、回测、客户产品的检查清单与云上实践

摘要：给量化团队选美股数据源，或为行情产品接入API，花出去的除了订阅费，还有接入、核对、重跑和维护的时间。看盘、历史回测、客户展示，各自需要什么数据？本文从具体交付任务出发，拆清价格对象、历史价格处理方式、持续更新与使用权限，给出一张需求验收表、一段询证文本、一份可运行的检查脚本，以及如何在华为云上落地这套验证流程，帮助采购负责人和开发者对齐要求。（注：本文只讨论数据工程的验证方法，不构成任何投资建议。）

技术同事拿到试用权限，用苹果AAPL.US查了一次快照，图表画出来了。产品经理准备验收盘前看板，指着页面问：盘前时段这个价格对吗？

同一份快照里，顶层 last_price 是前一日16:00的331.90，pre_market_quote.last_done 是当日04:22的332.10。页面如果只贴顶层价，盘前页面显示的就是上一交易日的收盘价。这不是接口取不到数据，而是取数之后选错了价格对象。

看板、回测、客户产品，各自的检查点并不相同。演示回答了“能取到”，团队还要回答：研究能否按要求重跑，看板能否显示对应时段的行情，承诺给客户的功能是否有合适的使用条件。

这些要求如果签约后才补，增加的工作可能包括重新取数、修改图表规则、重跑研究，甚至重新采购。订阅费之外，团队的时间也在花钱。

选型顺序可以很明确：先判断数据能否完成这次交付，再比较额外能力是否值得付费。接下来按看板、回测和客户产品逐项拆解，最后把要求放进一张技术与采购都能使用的验收表，并把检查脚本落到华为云上。

本文采用TickDB作为实际数据来源。它是面向开发者、量化团队和AI应用的统一市场数据服务，提供美股、港股、A股等市场的行情接入。研究脚本可通过REST按需查询快照与历史数据。本文使用其中的按需查询路径，展示看盘与历史研究怎样共用取数入口。

一、看板上线前，先拆清价格对象和时段

“支持苹果股票”能确认代码在服务范围内，但一个盘中看板还需要知道价格与成交数据来自什么范围。

Alpaca的官方Market Data FAQ区分IEX与SIP：前者来自单一交易所，后者是汇总行情。[1] 都能查询同一只股票，所包含的市场范围不同。如果产品要比较跨交易场所的成交情况，这个范围就应写进需求。

更重要的是，价格有返回，还要问它属于哪个时段。2026年11月4日美东时间04:22，我们分别查询行情快照、历史K线与交易日历，按需求表检查返回结果。

苹果AAPL.US的同一份快照里，有两条记录：

如果看板要显示盘前变化，却只把顶层的331.90贴上“最新价”标签，用户就看不到这份响应里已有的盘前报价。需求可以改成：盘前页面显示盘前最新成交价和对应时间；没有盘前数据时，保留清楚的状态说明，不能把其他时段的价格直接标为当前盘前价。

买一、卖一也应单列。最新成交价描述已经发生的交易，买卖报价描述对应的盘口信息。这次快照没有返回买一、卖一，就继续核对所需接口和获取条件，不拿已有成交价填进这两个位置。

另外，“查询当前快照”和“持续接收更新”是两种接入任务。要让页面持续变化，应用还需要安排后续请求，或使用合适的订阅通道，并处理连接中断、旧数据展示与恢复后的核对。做看板至少要把价格对象、数据范围、交易时段和更新方式写清。

二、做回测，历史数据要回答三个不同的问题

第一，价格变化里有没有公司行动

“用复权数据”是个熟悉的要求，但还需要说清哪一列、怎样调整。

假设一只股票发生一拆二，原来一股变两股，每股价格相应减半。只看每股价格，像是跌了一半；把持股数量一起算进去，拆股本身没有让持仓价值减半。回测若把这段变化直接当作市场涨跌，研究的起点就会改变。

Alpha Vantage的Daily接口返回原始开高低收和成交量；Daily Adjusted还提供复权收盘价及拆股、分红事件。[2] 一个接口名称里有Adjusted，不代表其中每一列都已经调整。向供应商询问时，可以具体到：我们用哪一列计算历史收益？开高低收和成交量分别采用什么处理？公司行动记录能否一起取得？

第二，历史股票池里应该有谁

如果策略要研究几年前的市场，只用今天还能找到的股票名单，会把部分后来退市或不再符合条件的对象留在样本之外。数据源提供当前目录，并不自动完成历史股票池的还原。

CRSP US Stock Databases的官方介绍就把活跃与非活跃证券、公司行动，以及贯穿证券历史的永久标识列在产品说明中。[6] 因此，一份长期回测需求，至少要同时说清研究区间、股票集合和价格处理。只比较“能下载几年”，还不足以判断是否适用。

第三，这段输入能否在团队里复查

有人下载文件，有人清洗数据，另一个人运行策略。如果没有保留查询时间、参数与价格处理方式，结果变了，团队就可能先去改策略，却不知道输入已经换过一版。

本篇示例把区间固定为2026年10月26—30日，请求不复权日K，并同时取得美股交易日历。随后执行以下检查：

这里把日历与历史K线一起使用的价值，是先给“应该检查哪些日期”一个明确参照。这份五日样本用于演示方法，团队若需要多年历史、分钟线或退市股票，就将那些条件带进试用。

三、做客户产品，要把能取什么与怎样使用一起确认

研究员自己看数据、团队共享研究表、客户登录看行情、用户下载CSV，是不同的使用方式。采购前只问“可不可以商用”，供应商仍可能不知道你准备交付什么。

NYSE的市场数据政策区分历史使用、非展示用途和订阅者等类别，其专有实时数据的对外再分发也有对应说明。[3] 这些资料说明，数据使用条件会随具体产品和用途而变化，不能靠API请求成功来判断。

这一步与技术验证可以并行做：一边检查取到什么，一边取得适用产品和协议的书面说明。你可以把下面这段直接发给候选方：

我们计划将美股数据用于团队研究，并在客户登录页面展示行情。请分别说明两类用途的适用条件，以及数据范围与获取方式。

用户是否可以下载？如果后续通过我们的API返回数据，需要补充哪些条件？请提供对应产品与协议说明。

对于缺字段、缺区间或更新中断，请说明可用的查询、排查与支持方式，哪些需要由我们的应用处理。

如果目前只做内部研究，就删去尚不存在的客户需求，避免为暂时用不到的功能提前付出接入成本。

四、把钱花在哪：一次上云前的验收核对

订阅费之外，比较方案时还应考虑首次接入、定期核对、故障定位，以及新增需求时的适配工作。下面这张表既是需求清单，也是一次试用的交接依据：

你可以把上面这张表发给技术同事，作为跑脚本前的核对卡片（Checklist）。先标出这次交付不能缺少的项目。缺少必需项的方案，即使演示漂亮，也要先找到补充路径。

五、可以交给技术同事的复现代码

代码集中在这里。只负责采购或产品的读者，可以先使用上面的表格与询证文本；需要复查样本时，再把这一节交给技术同事。

脚本执行三次REST请求：苹果快照、固定历史区间日K、对应的美股交易日历。原始JSON、请求参数、获取时间与检查结果分别保存，并生成日线CSV。

使用条件是Python 3.10及以上版本、系统 curl 和IANA时区数据库。密钥从环境变量 TICKDB_API_KEY 读取；指定K线起止时间需要相应权限。

将代码保存为 check_us_sample.py。macOS默认zsh终端可隐藏输入密钥，再运行：

read -rs "TICKDB_API_KEY?请输入API Key: "
export TICKDB_API_KEY
python3 check_us_sample.py sample_run_01

sample_run_01 是尚不存在的输出目录。再次运行换新目录；调整样本日期时，同步修改日K起止时间与日历日期。

"""Python 3.10+, system curl, IANA tzdata. Uses TICKDB_API_KEY only."""
import csv, json, os, subprocess, sys, platform
from pathlib import Path
from datetime import datetime, timezone
from decimal import Decimal
from urllib.parse import urlencode
from zoneinfo import ZoneInfo
OUT = Path(sys.argv[1])
OUT.mkdir(parents=True, exist_ok=False)
for name in ['raw', 'summary', 'terminal']:
    (OUT / name).mkdir()
key = os.environ.get('TICKDB_API_KEY', '')
if not key or any((c in key for c in '\r\n"\\')):
    raise SystemExit('Invalid or missing TICKDB_API_KEY')
lines = ['COMMAND: python3 check_us_sample.py ' + str(OUT), 'OUTPUT: ' + str(OUT), 'UTC: ' + datetime.now(timezone.utc).isoformat(), 'API KEY: loaded from environment (not printed)', 'BASKET: AAPL.US; US calendar 20261026-20261030', 'FLOW: ticker -> daily candles -> calendar']
results = []

def request(name, path, params):
    before = datetime.now(timezone.utc).isoformat()
    body = OUT / 'raw' / f'{name}.json'
    url = 'https://api.tickdb.ai' + path + '?' + urlencode(params)
    config = 'header = "X-API-Key: ' + key + '"\n'
    p = subprocess.run(['curl', '--config', '-', '--silent', '--show-error', '--max-time', '20', '--output', str(body), '--write-out', '%{http_code}', url], input=config, text=True, capture_output=True, timeout=25)
    text = body.read_text() if body.exists() else ''
    if key in text:
        body.unlink(missing_ok=True)
        raise RuntimeError('Credential echoed; body removed')
    try:
        obj = json.loads(text)
    except ValueError:
        obj = {}
    meta = {'endpoint': path, 'params': params, 'http_status': p.stdout, 'curl_exit': p.returncode, 'code': obj.get('code'), 'started_utc': before, 'received_utc': datetime.now(timezone.utc).isoformat()}
    (OUT / 'raw' / f'{name}_meta.json').write_text(json.dumps(meta, indent=2))
    results.append(meta)
    lines.append(name + ': HTTP=' + p.stdout + ' API=' + str(obj.get('code')))
    if p.returncode or p.stdout != '200' or obj.get('code') != 0:
        raise RuntimeError(name + ' request failed; inspect saved response')
    return obj['data']
try:
    ticker = request('ticker', '/v1/market/ticker', {'symbols': 'AAPL.US', 'type': 'stock'})
    data = request('kline', '/v1/market/kline', {'symbol': 'AAPL.US', 'type': 'stock', 'interval': '1d', 'adjust': 'none', 'start_time': int(datetime(2026, 10, 26, tzinfo=ZoneInfo('America/New_York')).timestamp() * 1000), 'end_time': int(datetime(2026, 10, 31, tzinfo=ZoneInfo('America/New_York')).timestamp() * 1000) - 1, 'limit': 10})
    days = request('days', '/v1/market/trade-days', {'market': 'US', 'beg_day': '20261026', 'end_day': '20261030'})
    if len(ticker) != 1 or ticker[0].get('symbol') != 'AAPL.US':
        raise RuntimeError('Ticker symbol mismatch')
    if data.get('symbol') != 'AAPL.US' or data.get('interval') != '1d' or data.get('adjust') != 'none':
        raise RuntimeError('Kline metadata mismatch')
    rows = []
    for k in data['klines']:
        o, h, l, c = [Decimal(k[n]) for n in ['open', 'high', 'low', 'close']]
        if not l <= min(o, c) <= max(o, c) <= h:
            raise RuntimeError('OHLC invariant failed')
        d = datetime.fromtimestamp(k['time'] / 1000, timezone.utc).astimezone(ZoneInfo('America/New_York')).strftime('%Y%m%d')
        rows.append({'symbol': 'AAPL.US', 'date_et': d, 'time': k['time'], 'open': k['open'], 'high': k['high'], 'low': k['low'], 'close': k['close'], 'volume': k['volume'], 'adjust': data['adjust']})
    if not rows:
        raise RuntimeError('No historical candles returned; inspect saved response')
    expected = sorted(days.get('trade_days', []) + days.get('half_trade_days', []))
    if not expected:
        raise RuntimeError('Calendar schema or empty calendar needs inspection')
    actual = sorted((r['date_et'] for r in rows))
    checks = {'ohlc_valid': True, 'unique_dates': len(actual) == len(set(actual)), 'dates_match_calendar': actual == expected, 'snapshot_symbol': True, 'adjust_echo': 'none'}
    with (OUT / 'summary/daily.csv').open('w', encoding='utf-8-sig', newline='') as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0]))
        w.writeheader()
        w.writerows(rows)
    snap = ticker[0]
    fields = {n: {'present': n in snap, 'non_null': snap.get(n) is not None, 'value': snap.get(n)} for n in ['last_price', 'timestamp', 'bid_price', 'ask_price', 'pre_market_quote', 'post_market_quote', 'overnight_quote']}
    summary = {'output_directory': str(OUT), 'symbol': 'AAPL.US', 'runtime': platform.python_version(), 'requests': results, 'rows': len(rows), 'expected_dates': expected, 'actual_dates': actual, 'checks': checks, 'snapshot_fields': fields, 'snapshot_time_utc': datetime.fromtimestamp(snap['timestamp'] / 1000, timezone.utc).isoformat(), 'snapshot_time_et': datetime.fromtimestamp(snap['timestamp'] / 1000, timezone.utc).astimezone(ZoneInfo('America/New_York')).isoformat(), 'passed': all((v is True for k, v in checks.items() if k != 'adjust_echo'))}
    (OUT / 'summary/result.json').write_text(json.dumps(summary, ensure_ascii=False, indent=2))
    lines += ['DAILY: ' + str(len(rows)) + ' rows; dates=' + ','.join(actual), 'CHECKS: ' + json.dumps(checks), 'SUMMARY: ' + ('PASS' if summary['passed'] else 'FAIL'), 'BOUNDARY: finite sample, no latency/SLA/license or full backtest certification']
    if not summary['passed']:
        raise RuntimeError('Validation failed; inspect summary')
except Exception as exc:
    lines.append('FAIL: ' + str(exc))
    raise
finally:
    (OUT / 'terminal/transcript.txt').write_text('\n'.join(lines) + '\n')
    print('\n'.join(lines))

正文快照采集于2026年11月4日；上述脚本于11月5日复跑，三次请求均返回HTTP 200、业务码0，固定历史区间的结果为：

DAILY: 5 rows; dates=20261026,20261027,20261028,20261029,20261030
CHECKS: {"ohlc_valid": true, "unique_dates": true,
         "dates_match_calendar": true, "snapshot_symbol": true,
         "adjust_echo": "none"}
SUMMARY: PASS

输出目录里，raw 留原始响应和请求信息；summary/daily.csv 是历史样本；summary/result.json 列出检查结果及快照字段的存在情况；terminal/transcript.txt 留下运行摘要。若接口返回错误，程序会停止并保留相关输入，技术同事能继续定位具体问题。

六、下一步：让验证脚本在华为云上持续运行

你已经在本地验证了字段结构正确、脚本能跑通。如果希望这套检查流程不再依赖本地终端，而是能够定时触发、稳定留痕，可以考虑把它部署到华为云上。

6.1 让脚本跑起来：弹性云服务器 ECS

最直接的方式是把脚本放在一台常驻的弹性云服务器 ECS 上。脚本依赖 Python 3.10 及以上、系统 curl 和 IANA 时区数据库，选择镜像时请先确认这三项。运行目录按 sample_run_01 的方式每次新建，避免覆盖上一次原始返回。

6.2 把证据留下来：对象存储服务 OBS

在单机阶段，日志散落在本地文件夹里，换台机器就找不到了。你可以把 raw、summary、terminal 三个目录定时同步到对象存储服务 OBS，作为团队共享留痕。这样采购看需求表，技术看 OBS 里的原始 JSON，研究员看固定处理方式的 CSV，大家共享同一份证据。

6.3 出问题随时查：云日志服务 LTS 与应用运维管理 AOM

如果想知道脚本每天几点跑的、有没有成功，可以把终端摘要和检查结果投递到云日志服务 LTS 便于检索。对脚本失败和请求异常，可以通过应用运维管理 AOM 设置告警，而不是仅仅在控制台里打印一行报错。

6.4 守住密钥：数据加密服务 DEW 和统一身份认证 IAM

生产环境千万不要把 API Key 写进代码或前端。脚本从环境变量读取 Key；在华为云上，建议结合数据加密服务 DEW 管理密钥，并用统一身份认证服务 IAM 限制谁可以取用和运行。具体权限策略、凭据注入方式需按华为云官方文档确认后再落地。

6.5 接入客户产品：API 网关 APIG 与云数据库 GaussDB

如果看板后端需要把按需查询包成内部服务，可以用 API 网关 APIG 做入口，后端放在 ECS 或云容器引擎 CCE。对于固定区间的历史样本复查，可以把 daily.csv 导入云数据库 GaussDB 进行分析。这里只给承载位置，不改变前面的验证逻辑：快照看价格对象与时间，历史K线固定区间和复权方式，交易日历核对实际日期。

承载对照建议表（部署位置建议，非产品能力实测）：

七、常见问题与检查清单

Q1：这套脚本能否证明数据源适合生产？

不能。它验证的是固定区间样本的取数、字段、日期和 OHLC 关系；延迟、SLA、授权和全量回测认证不在脚本结论内。

Q2：原始 JSON 和 CSV 一定要存 OBS 吗？

不一定。你可以直接留在本地，但在团队协作中，使用 OBS 可以避免“文件只在某人电脑里”的尴尬。本文代码只写本地输出目录，云上同步是脚本外的运维步骤。

Q3：快照里 bid_price、ask_price 缺失是否正常？

本文取样中买一、卖一没有返回，脚本如实记录为字段缺席，没有用成交价代填。是否正常取决于你调用的接口和套餐范围，需求里应把买一、卖一单列为必须项。

上云前检查卡（Checklist）

• [ ] 价格对象已核对（顶层价与盘前/盘后价区分正确）
• [ ] 历史区间已核对（日期无重复、与交易日历一致）
• [ ] 复权方式已核对（响应回显的 adjust 字段符合预期）
• [ ] OHLC 逻辑关系已检查（未出现高低价颠倒）
• [ ] 环境依赖已满足（Python 3.10+、系统 curl、tzdata）
• [ ] API Key 未硬编码（通过环境变量或华为云 DEW 注入）
• [ ] 原始返回与检查结果已留存（方便团队复查）

付款前，先交付一份可检查的需求

拿一个近期真的要完成的任务，把数据对象、时间范围、价格处理方式、更新方式和使用条件写清，再让技术与候选供应商分别给出结果和说明。

这样得到的不只是一份“支持哪些功能”的介绍，还有一份团队可以共同检查的交付范围。钱花在当前需要的输入上，额外能力留到新任务出现时再验证。（本文不构成任何投资建议。）