美股数据源选型指南:看盘、回测、做产品,别把钱花错地方

TickDB Research ·

摘要:给量化团队选美股数据源,或为行情产品接入API,花出去的除了订阅费,还有接入、核对、重跑和维护的时间。看盘、历史回测、客户展示,各自需要什么数据?本文从具体交付任务出发,拆清价格对象、历史口径、持续更新与使用权限,结合官方资料和真实接口返回,给出一张需求验收表与可运行代码,帮助采购负责人、研究员和产品经理在付款前对齐要求。

假设你负责一个美股研究项目的数据采购:研究员要做策略复盘,产品经理要上线盘中看板,后续还准备让客户登录查看。

候选服务都写着“实时行情、历史数据、API接入”。技术同事拿到试用权限,查到了苹果股价,图表也画出来了。现在,采购申请等你确认。

难点在于,这次演示回答了“能不能取到数据”,团队还需要回答:研究能否按要求重跑,看板能否显示对应时段的行情,承诺给客户的功能是否有合适的使用条件。

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

选型顺序可以很明确:先判断数据能否完成这次交付,再比较额外能力是否值得付费。接下来按看盘、回测和产品交付逐项拆解,最后把要求放进一张技术与采购都能使用的验收表。

本文采用TickDB作为实际数据来源。它是面向开发者、量化团队和AI应用的统一市场数据服务,提供美股、港股、A股等市场的实时与历史行情接入。研究脚本可通过REST按需查询快照与历史数据,行情产品可通过WebSocket持续订阅,AI助手还可通过MCP、CLI调用结构化市场数据。[7] 本文使用其中的按需查询路径,展示看盘与历史研究怎样共用取数入口。

image.png

图:不同用途对应不同检查重点,AI生成解释图。

一、做看板,先说明用户到底要看什么

股票有覆盖,还要问行情来自哪里

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

Alpaca的官方Market Data FAQ区分IEX与SIP:前者来自单一交易所,后者是汇总行情。[1] 都能查询同一只股票,所包含的市场范围不同。

如果产品要比较跨交易场所的成交情况,这个范围就应写进需求;若只研究某个交易场所,按相应范围取数也有其用途。数据多少本身不足以决定选择,关键是是否覆盖你要回答的问题。

价格有返回,还要问它属于哪个时段

2026年11月4日美东时间04:22,我们分别查询行情快照、历史K线与交易日历,按需求表检查返回结果。

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

响应中的位置价格(美元)该对象的美东时间
顶层价格 last_price331.9011月3日16:00:00
盘前价格 pre_market_quote.last_done332.1011月4日04:22:25

如果看板要显示盘前变化,却只把顶层的331.90贴上“最新价”标签,用户就看不到这份响应里已有的盘前报价。

问题出在取数之后的选择:页面要用哪个对象、显示哪个时间、缺数据时怎样标记,都需要产品与技术共同决定。只写“显示最新价”,两个人可能做出不同理解。

需求可以改成:

盘前页面显示盘前最新成交价和对应时间;没有盘前数据时,保留清楚的状态说明,不能把其他时段的价格直接标为当前盘前价。

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

查一次价格,还要问怎样持续更新

“查询当前快照”和“持续接收更新”是两种接入任务。

一次查询返回一个时刻的结果。要让页面持续变化,应用还需要安排后续请求,或使用合适的订阅通道,并处理连接中断、旧数据展示与恢复后的核对。只把第一次画出来的图作为验收结果,这些工作就容易漏掉。

因此,做看板至少要把价格对象、数据范围、交易时段和更新方式写清。产品经理需要的是用户看到什么,技术同事再据此选择字段和调用方式。

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

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

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

假设一只股票发生一拆二,原来一股变两股,每股价格相应减半。只看每股价格,像是跌了一半;把持股数量一起算进去,拆股本身没有让持仓价值减半。美国SEC的Investor.gov用这种关系解释拆股。[4] 这里是教学假设,不是某只股票的行情记录。

回测若把这段变化直接当作市场涨跌,研究的起点就会改变。复权处理帮助比较跨公司行动的价格序列,但仍要确认输入字段的具体含义。

Alpha Vantage的Daily接口返回原始开高低收和成交量;Daily Adjusted还提供复权收盘价及拆股、分红事件。[2] 一个接口名称里有Adjusted,不代表其中每一列都已经调整。

向供应商询问时,可以具体到:

我们用哪一列计算历史收益?开高低收和成交量分别采用什么处理?公司行动记录能否一起取得?

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

如果策略要研究几年前的市场,只用今天还能找到的股票名单,会把部分后来退市或不再符合条件的对象留在样本之外。这已经改变了研究对象,不能只检查现有股票的K线有没有缺行。

这类研究应把“当时哪些股票符合条件”写进需求:是否需要退市记录、代码变化和历史成分信息,取决于策略的选样规则。数据源提供当前目录,并不自动完成历史股票池的还原。

CRSP US Stock Databases的官方介绍就把活跃与非活跃证券、公司行动,以及贯穿证券历史的永久标识列在产品说明中。[6] 对长期研究而言,这些内容与价格序列一起决定能否还原研究对象,不能只按下载文件的行数判断价值。

因此,一份长期回测需求,至少要同时说清研究区间、股票集合和价格处理。只比较“能下载几年”,还不足以判断是否适用。

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

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

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

检查什么本次实际结果用来回答什么
周期和复权方式日K,响应标明 adjust=none拿到的是约定的价格输入吗
日期范围10月26—30日,共5根对应这次指定区间吗
日期与日历逐项一致,无重复这段样本有没有多出或少掉日期
开高低收关系5根均通过记录内部是否存在明显矛盾
查询与原始返回一并保存换同事后还能从哪里复查

这里把日历与历史K线一起使用的价值,是先给“应该检查哪些日期”一个明确参照,再看实际记录。程序负责比较,数据接口提供输入;遇到差异,就有具体位置可以继续排查。

这份五日样本用于演示方法。团队若需要多年历史、分钟线或退市股票,就将那些条件带进试用,不拿眼前的五行记录代替整个研究任务。

查询“最近几根”时还要留意末根是否仍在形成。做固定历史研究,应选已结束的区间,保存获取时间与复权方式;复权历史也可能随公司行动或数据更新而调整。[5]

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

研究员自己看数据、团队共享研究表、客户登录看行情、用户下载CSV,是不同的使用方式。

采购前只问“可不可以商用”,供应商仍可能不知道你准备交付什么。尤其是展示之外的导出、接口转发和自动分析,应在需求里描述清楚。

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

这一步与技术验证可以并行做:一边检查取到什么,一边取得适用产品和协议的书面说明。交易所政策用于帮助识别问题,自己的采购仍按实际渠道和合同确认。

产品团队还要把用户看到的异常状态算进交付。例如,页面暂时没有对应时段的数据,是保留旧价并注明时间,还是显示暂不可用?谁负责核查,怎样找到当时的返回?这些处理会影响客户能否理解页面,也影响支持人员能否解释问题。

可以把下面这段直接发给候选方:

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

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

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

如果目前只做内部研究,就删去尚不存在的客户需求。把预期扩展单独列为后续确认项,也能避免为暂时用不到的功能提前付出接入成本。

四、把钱花在哪,用这张表做决定

订阅费之外,比较方案时还应考虑首次接入、定期核对、故障定位,以及新增需求时的适配工作。它们会占用研究员和开发者的时间,应在试用阶段看清。

下面这张表既是需求清单,也是一次试用的交接依据:

具体任务必须取得什么怎样检查缺项后的动作
盘前看板盘前价格、对象与时间按目标时段检查真实返回核对获取条件;调整页面状态
买卖报价功能所需买一、卖一及适用范围查字段和对应接口,不用成交价代填补相应数据能力或收窄功能
固定区间研究周期、价格口径和约定区间对日历、重复日期及开高低收保存差异并继续查原因
长期策略回测策略实际需要的历史股票池与公司行动输入按选样规则检查历史覆盖补历史来源,或调整研究范围
持续看盘目标时段内的更新与恢复方式另做连续观察和中断恢复测试分清服务与应用各自的工作
客户产品展示、下载和再分发等使用条件核对对应产品与协议说明补确认,或调整交付方式

先标出这次交付不能缺少的项目。缺少必需项的方案,即使演示漂亮,也要先找到补充路径;额外功能很多,但近期没有实际用途的,可以留到下一阶段评估。

比如,内部周频研究暂时不需要盘中持续推送,就围绕历史输入和更新方式检查;若产品承诺盘前展示,就应提前把盘前对象、时间与页面状态一起做出来。

TickDB的统一接入能力,在这个案例中对应一条可复查的取数路径:查快照,确认价格对象与时间;查历史K线,固定区间和复权方式;查交易日历,核对实际日期。三个接口各司其职,示例程序把原始输入和检查结果保存下来。采购看需求表,技术看请求与返回,研究员看固定口径的CSV,团队可以共享取数入口与核对依据,再分别完成显示、计算和判断。后续增加市场或数据类型时,也可以按同样的方法验证新增需求。

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

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

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

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

将代码保存为 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 留下运行摘要。

若接口返回错误,或发现空K线、开高低收矛盾、日期与日历不一致,程序会停止并保留相关输入。取样中买一、卖一没有返回,也如实保留为字段缺席。技术同事能继续定位具体问题,采购负责人则能据此追问缺项怎样解决。

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

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

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

资料与实测说明

价格样本采集于2026年11月4日;读者代码于11月5日复跑。样本作为数据选型演示,不涉及投资建议。其他机构材料用于解释对应概念和条件,未调用其接口。

[1] Alpaca,Market Data FAQ,“What's the difference between IEX and SIP data?”(11月5日复核)。

[2] Alpha Vantage,API Documentation,TIME_SERIES_DAILY及TIME_SERIES_DAILY_ADJUSTED(11月5日复核)。

[3] NYSE,Market Data Policies;Non-Display Use Policy(January 2015 v2.0);External Redistribution of Real-Time NYSE Proprietary Data Products(Revised December 2019)。只取用途区分,具体适用条件按采购时的产品与协议确认。

[4] 美国SEC,Investor.gov,“Stock Split”(11月5日核对)。正文拆股关系为教学说明,没有计入本篇接口实测。

[5] 案例接口文档:行情快照、K线查询、交易日历。11月4日保存快照,11月5日复核;原始返回、脚本与检查结果留档。

[6] Morningstar Indexes,CRSP US Stock Databases(11月5日核对)。引用其历史研究数据组成,不涉及接口调用或供应商排名。

[7] 产品定位与接入路径:官方产品介绍、WebSocket最佳实践及AI工具说明,2026年11月5日核对。多市场与多入口为产品能力介绍;正文运行案例使用REST查询。

相关文章

美股行情数据源怎么选?实测四家平台:别让错乱的“最新价”毁掉你的实时看板与 AI 查询

为美股团队选数据源,最怕“演示能用,交付返工”。本文结合真实自动化实测与官方资料,聚焦 实时看板、历史回测与 AI 查询 三大场景。文章不仅拆解了盘前/盘后价格时点错乱、复权口径不一等五个选型陷阱,还 直接提供了可运行的 Python 验证代码、多市场字段对比明细和复权因子计算公式 。你将看到 TickDB 在同一响应中如何返回三个时点的价格,以及如何通过 MCP 原生协议让 AI 直接取回…

TickDB Research ·

股票 API 怎么用?报价展示与历史研究,为什么要分开处理

面向正在交付股票行情页和历史导出功能的金融产品经理,本文拆解报价快照、K 线与研究文件的区别,解释相同价格为何可能表达不同信息,以及行情时间、取数日期和研究截止日各自解决什么问题。结合 TickDB 保存的真实响应和可运行 Python 示例,生成带时间的静态报价卡片与按日期筛选的研究 CSV,帮助产品和技术同事用明确规则检查交付。

TickDB Research ·

美股行情API遇上冬令时:盘前盘后时间怎么对齐?

负责美股行情页和开盘提醒的产品经理,冬令时切换后需要一起核对交易时段、北京时间和交易日期。本文结合TickDB交易时段、交易日历的真实返回,解释盘前盘后怎样换算,为什么凌晨收盘仍属于美国前一个交易日,并提供时间对照表、可运行换算示例与页面检查清单,处理半日交易等容易遗漏的例外。

TickDB Research ·

股票API接口费用怎么算?同样5000只股票,调用量为何差100倍

准备做A股行情看板、盘中筛选或投研观察池,股票API接口费用该怎么估?本文面向行情应用和投研工具负责人,从使用用途、数据范围、取数方式和计量规则拆解预算。结合TickDB实际取回的全市场行情,用5000只股票说明请求次数为何能相差100倍,并提供可运行的估算代码、场景选择表和询价清单,帮助团队比较满足同一需求的方案。

TickDB Research ·

外汇行情还在更新吗?用 Python 订阅 EURUSD 并记录双时间

适合正在开发外汇报价监控、需要确认脚本持续收到行情的 Python 开发者。使用 TickDB 的 WebSocket 行情订阅,记录 EURUSD 报价、数据时间和本机接收时间,生成原始消息与 CSV。通过目标报价数量、接收间隔和数据时间推进情况,区分连接仍在、消息到达与数据更新。本次30秒订阅观察收到38条目标报价;代码支持有限运行和自动保存结果,方便在自己的环境复查。

TickDB Research ·