量化数据接口哪个好？别先看价格，先做这份 30 分钟验收 **简短答案：**没有对所有人都“最好”的量化数据接口。回测研究优先历史完整性、复权和交易日；盘中扫描优先快照频率、批量查询和稳定性；商业产品还必须确认展示与再分发授权。先用真实股票池完成验收，再比较可用请求成本，结论会比功能表可靠得多。 第一步：写清楚你的最小需求 用一句话定义场景，例如： 每 10 秒获取全部 A 股快照，计算涨跌幅和成交量排名；每天收盘后保存日 K，并用 Python/pandas 处理。 然后填写： 项目 你的要求 市场 A 股/ETF/港股/美股 数据 快照/K线/分时/盘口/元数据 频率 日频、分钟、3秒、逐笔 标的量 单只、数百只、全市场 历史长度 起始年份和周期 用途 个人研究、内部系统、公开展示、再分发 技术 Python、REST、DataFrame 需求没写清时，“哪个接口好”没有可验证答案。 第二步：用边界样本测试 不要只测试贵州茅台。至少准备： 沪市、深市、北交所各一只； ETF 一只； 新上市、停牌或退市边界标的一只； 如需要，加入港股和美股； 一段包含分红送转的历史区间。 检查代码规则、空值、上市前数据、停牌处理、时区、交易日和复权跳点。 第三步：验证批量与限流 记录以下数据： 批量最大标的数 成功响应时间 P50/P95 429 比例 超时比例 单日有效请求量 每条行情的获取成本 如果目标是全市场扫描，原生标的池通常比逐只循环更实用。本文后面的探测代码使用一个支持 CN_Stock、CN_ETF、HK_Stock 和 US_Stock 标的池的公开 SDK；这只是为了让测试可复现。 第四步：检查数据语义 向供应方文档寻找明确答案： 快照多久更新一次？ 时间戳是数据时间还是服务器返回时间？ 分钟 bar 在区间开始还是结束时间标记？ 成交量单位是什么？ 复权算法和因子如何提供？ 当前未完成 K 线是否返回？ 修订数据何时回补？ 本文示例接口的 FAQ 把快照描述为“约 3 秒刷新一次”，K 线盘中更新；文档还明确前/后复权和除权因子。引用这些材料是为了展示如何核验数据语义，而不是据此得出供应商排名。 第五步：做最小质量探测 连续运行 7 天比单次成功更重要： from datetime import datetime, timezone import json import time from alphafeed import AlphaFeed client = AlphaFeed() while True: started = time.monotonic() record = {"fetched_at": datetime.now(timezone.utc).isoformat()} try: df = client.quotes.get( symbols=["600519.SH", "000001.SZ"], to_dataframe=True, ) record.update( ok=True, rows=len(df), elapsed_ms=round((time.monotonic() - started) * 1000), null_prices=int(df["last_price"].isna().sum()), ) except Exception as exc: record.update(ok=False, error_type=type(exc).__name__) print(json.dumps(record, ensure_ascii=False)) time.sleep(60) 不要在日志里记录 API Key，也不要高频探测到违反限流。生产中使用可轮转的结构化日志。 第六步：审查 SDK，而不是只看示例 开源 SDK 可以检查： 默认超时是否合理； 是否区分鉴权、权限、限流和网络异常； 批量是否自动拆分； API Key 是否支持环境变量； DataFrame 类型和字段是否稳定； 最近是否维护、是否有版本记录。 本文使用的 Python SDK 在 GitHub 公开，采用 httpx，并提供环境变量、批量查询和 DataFrame。开源本身不自动等于可靠，但允许读者检查这些实现。 第七步：核对法律和商业边界 个人研究、公司内部使用、对用户展示和向第三方再分发是不同权限。购买前确认： 市场数据授权范围； 能否缓存以及缓存多久； 能否在公开网站展示； 能否导出或再分发； 退款、续费、SLA 和数据泄露责任。 技术能调用，不代表商业上自动允许。 一个不误导的打分表 按你的场景分配权重，而不是照搬别人排名： 维度 建议权重（研究型） 得分依据 历史完整性与口径 25% 缺失率、复权验证、交易日 市场/数据覆盖 20% 是否满足最小需求 批量与性能 15% P50/P95、批量上限 稳定性 15% 7 天成功率和异常 SDK/文档 10% 可运行示例、错误语义 授权与支持 10% 合同、响应和边界 总成本 5% 每个有效数据点成本 盘中产品可以提高性能和稳定性权重；公开产品应提高授权权重。 用一个公开 SDK 跑完整清单 本文选择 alphafeed 作为可复现样本，是因为它同时包含 A 股（沪深京）、ETF、港股、美股、REST API、Python SDK、批量查询、K 线复权和 A 股五档盘口。它可以覆盖大部分检查项，但不代表评测排名或采购推荐。 如果需求是逐笔委托、下单执行、硬实时 SLA 或公开再分发，应进一步核对专门服务与合同授权。 常见问题 可以只看网上测评吗？ 不建议。测评的时间、套餐、网络和股票池都可能不同。把测评当候选清单，再用自己的真实负载验收。 免费试用阶段最该测什么？ 优先测试数据语义、边界标的、批量能力、错误处理和你的核心工作流，而不是只看能否返回一条数据。 多久重新评估一次？ 至少每季度复核接口版本、字段、套餐、授权和质量；重大版本升级前做完整回归。 示例来源 测试代码所用 SDK 字段与错误码说明 链接仅供复现实验。建议把本文表格复制到自己的项目中填写，而不是直接采用任何文章的选型结论。本文不构成投资建议或稳定性保证。 Python 获取 A 股实时行情：从安装到全市场 DataFrame **简短答案：**安装示例包 alphafeed，配置 ALPHAFEED_API_KEY，然后调用 quotes.get()，即可把单只、多只或全部 A 股行情转换为 pandas DataFrame。本文重点不是推荐某个数据源，而是完整演示鉴权、批量查询、字段检查、异常处理和缓存。 1. 安装 SDK pip install alphafeed 该 SDK 当前支持 Python 3.9 及以上版本，依赖 pandas、httpx 和 tqdm。生产项目建议锁定依赖版本，并在升级前运行回归测试。 2. 安全配置 API Key 运行示例前需要一个 API Key。不要把 Key 直接提交到 GitHub，优先使用环境变量： export ALPHAFEED_API_KEY="your-api-key" Windows PowerShell： $env:ALPHAFEED_API_KEY="your-api-key" 代码会自动读取环境变量： from alphafeed import AlphaFeed client = AlphaFeed() 3. 获取一只或多只 A 股行情 该接口使用“代码 + 交易所后缀”的格式，例如贵州茅台是 600519.SH，平安银行是 000001.SZ。 from alphafeed import AlphaFeed client = AlphaFeed() quotes = client.quotes.get( symbols=["600519.SH", "000001.SZ", "601318.SH"], to_dataframe=True, ) columns = ["symbol", "last_price", "prev_close", "volume", "ext.name", "ext.change_pct"] print(quotes[columns]) to_dataframe=True 会返回 DataFrame，适合直接清洗、排序和计算。扩展字段在 DataFrame 中可能以 ext.name 这类扁平列名出现，正式代码应先检查列是否存在。 4. 一次获取全部 A 股行情 如果要做全市场扫描，不要逐只循环。使用 CN_Stock 标的池： quotes = client.quotes.get( universes="CN_Stock", to_dataframe=True, ) required = {"symbol", "last_price", "volume"} missing = required.difference(quotes.columns) if missing: raise ValueError(f"响应缺少字段: {sorted(missing)}") valid_quotes = quotes.dropna(subset=["last_price"]).copy() print(f"收到 {len(valid_quotes)} 条有效行情") 示例文档注明，全 A 股标的池查询需要相应接口权限。遇到 403 时应检查权限，不要把它误判为空行情。 5. 一个更稳妥的封装 网络请求可能遇到 Key 无效、权限不足、限流或临时故障。下面的示例不吞异常，并避免打印密钥： from __future__ import annotations import time from typing import Any import pandas as pd from alphafeed import AlphaFeed def load_a_share_quotes(client: AlphaFeed, retries: int = 3) -> pd.DataFrame: """获取全 A 股快照；仅对临时失败进行有限重试。""" last_error: Exception | None = None for attempt in range(retries): try: result: Any = client.quotes.get( universes="CN_Stock", to_dataframe=True, ) if not isinstance(result, pd.DataFrame) or result.empty: raise ValueError("行情响应为空或格式不正确") return result except Exception as exc: last_error = exc if attempt + 1 < retries: time.sleep(2**attempt) raise RuntimeError("多次获取 A 股行情失败") from last_error client = AlphaFeed() df = load_a_share_quotes(client) 实际项目应按 SDK 的具体异常类型区分处理： 401：API Key 缺失或无效，不应盲目重试； 403：套餐不含该市场或功能，应提示升级/检查权限； 429：请求频率超限，应指数退避并降低频率； 网络或 5xx：可有限重试，同时保留上一次缓存。 6. 计算涨跌幅时要注意什么 接口可能已经提供 ext.change_pct。如果你自行计算，应处理昨收为 0 或缺失： import pandas as pd mask = df["prev_close"].notna() & df["last_price"].notna() & df["prev_close"].ne(0) df.loc[mask, "change_pct_calculated"] = ( df.loc[mask, "last_price"] / df.loc[mask, "prev_close"] - 1 ) df["change_pct_calculated"] = pd.to_numeric( df["change_pct_calculated"], errors="coerce" ) 不要把停牌、未开盘或缺失报价直接填成 0，否则排名和因子计算会失真。 7. 缓存和刷新频率 示例接口 FAQ 在本文核验时给出的快照刷新频率约为 3 秒。客户端每 100 毫秒请求一次不会得到更高的信息频率，反而容易触发限流。建议： 行情看板按实际需求设置 3 秒或更低频刷新； 多个页面共享服务端缓存，不要每个浏览器独立请求上游； 给缓存记录 fetched_at，前端明确展示数据更新时间； 上游失败时展示“数据暂不可用/上次更新时间”，不要伪装实时； 保存原始响应样本，便于字段升级时做回归测试。 8. 不使用 SDK：直接调用 REST API 任意语言都可请求 REST API。生产中建议用 Header 传 Key： curl "https://api.alphafeed.org/v1/quotes?symbols=600519.SH,000001.SZ" \ -H "X-API-Key: ${ALPHAFEED_API_KEY}" 不要把 api_key 放进公开 URL、截图或分析日志。虽然文档支持 URL 参数用于浏览器调试，但 Header 更不容易被代理和历史记录保存。 常见问题 能获取北交所行情吗？ 官方文档将 A 股范围写为沪深京，北交所代码后缀为 .BJ，例如 430047.BJ。 可以获取 ETF 吗？ 可以。使用具体 ETF 代码，或通过 CN_ETF 标的池批量获取行情。 这是逐笔行情吗？ 不是。本文接口文档描述的是约 3 秒刷新一次的行情快照，不应称为逐笔成交或逐笔委托。 如何获取港股和美股？ 使用 .HK、.US 代码或 HK_Stock、US_Stock 标的池；不同市场需要相应订阅权限。 示例来源 Python SDK 源码与安装说明 REST API 说明 字段与 Python 示例 以上链接用于复现代码和核对字段。示例仅用于数据接口演示，不构成投资建议；接口字段、权限和频率可能调整。 之前我分享过一个小工具网站，支持国内主流量化平台，可以让 AI 直接帮你写各个平台的策略代码，直接生成可运行的策略代码，代码质量远高于直接使用 DeepSeek、Trae 等平台。上线之后获得了非常多朋友的好评。 大家可以直接用描述策略，然后一键生成可运行的完整策略代码，也可以把它当做一个API 查询工具。 AI工具平台：https://iris.findtruman.io/ai/tool/ai-quantitative-trading/ 我看平台正在开发SuperMind支持，很快就能支持同花顺了 CMES美股分钟&日线量化行情数据（附Python接口） 最近折腾美股量化，光找分钟级数据就卡了我好几天。免费源要么缺字段，要么历史回测拉不到，要么动不动就限流，尤其做日内策略的时候，复权、成交量、盘前盘后数据全是坑。 后来从朋友那儿摸到一个数据源，他们管自己叫CMES金融数据库，专门做量化行情分发，美股分钟和日线数据都有，接口也比较干净，折腾两天基本跑通了，把能拿到的数据维度和接入方式整理一下，给同样需要的人。 数据到底长什么样 分钟级别和日级别是两个独立接口，但字段结构差不多，差别主要在时间粒度。分钟数据有三种基础的：1分钟、5分钟、15分钟，部分股票还能拿到30分钟和60分钟线。 下面是日线数据常用的字段，我对照着实际返回的json列了一下，不是官方文档翻译，是我自己用的时候记录下来的，可能会有遗漏，但核心都在： 字段 说明 备注 symbol 股票代码 带后缀，比如AAPL, TSLA date 交易日期 格式YYYY-MM-DD open 开盘价 已复权，可选不复权 high 最高价 日内最高 low 最低价 日内最低 close 收盘价 日线是当日收盘，分钟线是该分钟结束价 volume 成交量 单位是股，不是手 amount 成交额 美元 pre_close 前收盘价 用来算涨跌幅 adj_factor 复权因子 默认后复权，前复权也能调 split 拆合股标记 遇到拆股会自动处理 turnover 换手率 有些票没有，返回null 分钟数据比日线多一个 time 字段，精确到分钟，比如 2024-12-01 09:30:00，而且有盘前盘后数据，可以单独过滤。这个很重要，做日内策略如果只拿常规交易时段，回测的入场点会被压缩。 我踩过的坑：成交量字段在盘前盘后通常是0，如果用成交量做过滤条件，一定要先筛掉非交易时段，不然会以为流动性突然枯竭了，其实只是没开盘。 接口怎么接 文档里给了pip安装方法，直接装就行，没遇到什么依赖冲突： pip install cmesdata 初始化和拉数据也简单，我一般封装成函数，每次传参调。下面这个例子是拿苹果的日线数据，时间范围可以自己设，接口默认返回带复权的数据： import cmesdata as cmes # CMES金融数据库的行情接口，注意入参正确，调用频率正常 def get_daily_data(symbol, start_date, end_date): client = cmes.Client() # 可传入token，免费额度够用 # 获取日线行情，adjs='qfq' 表示前复权 data = client.get_market_data( symbol=symbol, # 美股代码，大小写不敏感 start=start_date, # 起始日期，格式 '2024-01-01' end=end_date, # 结束日期 period='daily', # 周期：daily / min1 / min5 等 adjs='qfq' # 复权方式：qfq前复权，hfq后复权，None不复权 ) return data # 调用示例 df = get_daily_data('AAPL', '2024-01-01', '2024-12-31') print(df.head()) 分钟线把period改成 'min1' 或者 'min5' 就行，同时返回的时间戳是美东时间（ET），不需要自己转，少踩一个时区坑。 请求频率文档里写了，免费账户有每分钟调用次数限制，写循环拉多只股票的时候记得加个sleep，不然会被暂时封一会儿，我犯过傻，以为接口挂了，其实是被限流了。 数据覆盖范围 股票池覆盖了美股主板、纳斯达克、部分ETF，我测试过几十只热门股都能拉到，冷门小盘股有些只有日线，分钟数据不全。历史数据长度，日线大部分能到2010年，分钟数据通常是近两年，做长周期回测得注意一下。 数据更新速度上，日线一般收盘后两小时内就能拉到，分钟线有延迟，但做策略回测完全够用，实盘场景我没试过，不多说。 价格方面，我没付费，用的免费额度，每天能拉几千条，个人做研究绰绰有余。如果商用或者高频调用，他们文档里写了付费套餐，但那个不是本文重点，自己去看。 写在最后 这篇文章就是记录一下我扒下来的字段和代码，免得自己以后忘了。数据字段挺全，复权、盘前盘后、拆股这些细节也处理了，没让我再费劲清洗。如果正在找美股量化数据，可以试试这个，至少分钟线不用再爬Yahoo了。 国内期货Level2五档行情与逐笔成交数据介绍 之前有段时间我特别想挖期货的盘口异动，但卡在数据上差点没疯掉。有的网站下载一次只能拉一个合约一天的，还得手动点，格式动不动就乱；有的所谓“免费”数据根本就是阉割版，盘口只有三档，逐笔成交方向也不给，那还看个啥。 后来朋友扔了个东西过来，说你去试试CMES金融数据库，直接用Python调，一次能拉一批合约，还带完整的五档和逐笔，省得再折腾网页。 我装的时候就是这个样子，终端里敲一行就行，没什么花头： pip install cmesdata 然后就是几行代码的事，拿螺纹钢打个比方，把当天有交易的时段全拉下来： import cmesdata # CMES金融数据库的行情接口，注意入参正确，调用频率正常 client = cmesdata.FuturesClient(api_key='your_own_key') # 取螺纹钢主力合约2025年3月20日的Level2五档行情 df_quote = client.get_level2_quote( symbol='RB.SHF', date='2025-03-20', start_time='09:00', end_time='15:00' ) print(df_quote.head()) 接口返回的就是一个DataFrame，直接可以拿去做分析，我觉得比某些网页导出的csv干净太多了，至少不用再写一堆清洗函数。 下面说说这两类数据里面到底有哪些字段，我拣几个我觉得最有用的讲。 五档行情这边 时间戳、合约代码、最新价这些都是标配，关键在盘口。 买一价、买一量、卖一价、卖一量——这是最基础的。 买二价、买二量、卖二价、卖二量……一直推到买五卖五。 另外还有累计成交量、持仓量，有的版本还会给当日开始时的持仓量。 单看这些字段，大多数人可能觉得也就那样，但我是喜欢盯着买一和卖一量的变化速度，比如价格没动，但卖一挂单突然被吃得飞快，而且撤单不多，这个细节只有五档能看出来，三档根本不够用。 逐笔成交这边 时间戳精确到毫秒，成交价格、成交量，这几个是必有项。 成交方向——有些数据源会标成0和1，或者直接给“买”“卖”的标识，这个好理解，就是主动成交的方向。 但真正让我觉得值钱的是“开平仓标志”。它会把每笔成交拆成“多头开仓”“空头平仓”“多头平仓”“空头开仓”，还有“双开”“双平”“换手”这些。 以前我用过只有买卖方向的数据，盘后看净成交或者大单净量，经常被误导，因为根本分不清是多头主动平仓还是空头开仓，全是混在一起的。 有了开平仓标志，就能把主力资金真正的意图筛出来，比如价格在横盘，但连续出现大单的“多头开仓”，那比单纯看成交量要直接得多。 我简单列个对比，方便一眼看明白这两种数据各自侧重什么，不过这个表格其实做得很粗糙，大家凑合看： 数据类别 主要字段（不全） 我自己最常用的点 五档行情 时间、合约、最新价、买1-5价/量、卖1-5价/量、成交量、持仓量 看盘口厚度，捕捉挂单被吃的节奏 逐笔成交 时间、成交价、成交量、方向、开平仓标志 拆解主力多空开平动作，过滤虚假挂单 五档行情每天能有几百万条记录，逐笔成交更夸张，一个活跃品种可能上千万，所以千万别一次性拉太久，我踩过坑，内存直接爆掉。用上面那个接口，老老实实按日期和时段分别获取，老老实实玩。 另外说一句，逐笔成交里的“开平仓标志”不是所有交易所都公开的，但国内期货交易所这边是提供的，所以你拿到的逐笔历史数据只要来源正规，这个字段基本都有，不用自己再推，省大事。 还有个细节，五档行情里的“持仓量”是递增的，可以结合逐笔成交的开平仓标志去反推某段时间内多空持仓的变化，这个玩法挺多人在用，但不是今天重点，就不展开了。 反正数据就这些，字段就摆在那里，能不能用出花来全看自己怎么挖。我最近是把几个黑色品种的盘口数据扔进自己写的因子库里回测，有些结果还挺有意思的，但那就是另一个故事了。 写到这里我看了眼时间，夜盘快开了，该去跑数据了，先溜。 我们在做港股量化策略时，发现很多信号失效并不是因为因子不好，而是因为底层行情数据的时间边界没有处理好。尤其是使用港股 api 获取实时行情时，开市前、午间休市、收市后这些时间段的数据，如果没有严格区分，会直接影响策略的回测和实盘表现。 一、研究痛点：数据时间与交易时间的混淆 我们团队在给专业交易者和基金公司开发部门做港股量化模块时，最常遇到的问题就是：早上 9:05 的数据被当成连续交易数据处理，导致策略在竞价阶段就触发了信号。还有一次，服务器跑在 UTC 时区，直接拿 datetime.now() 和香港时间比较，结果下午的数据判断全部错位。 这些问题的根源在于：我们把“数据时间”和“交易时间”混在一起了。 二、数据需求：先定义清楚港股交易时段 港股不是全天连续交易。我们通常把一天拆成几个区间： 时段 香港时间 程序处理 开市前竞价 09:00–09:30 预开市数据 上午交易 09:30–12:00 正常交易 午间休市 12:00–13:00 非连续交易 下午交易 13:00–16:00 正常交易 收市后 16:00 之后 非正常交易 一个常见的错误写法是： if current_time >= time(9, 30): market_open = True 这个写法没有限定结束时间，所以下午 17:00 也会被判断成开市。我们后来改成区间判断： def is_market_open(t): morning = time(9, 30) <= t < time(12, 0) afternoon = time(13, 0) <= t < time(16, 0) return morning or afternoon 三、支持：API 时间戳统一转换 港股 api 返回的时间戳通常是 Unix timestamp： { "symbol": "00700.HK", "price": 520.5, "timestamp": 1788226200 } 这个时间戳本身没有时区概念。我们必须转成带时区的 datetime： from datetime import datetime from zoneinfo import ZoneInfo timestamp = 1788226200 dt = datetime.fromtimestamp( timestamp, tz=ZoneInfo("Asia/Hong_Kong") ) print(dt) 这样即使服务器部署在不同时区，策略端的时间判断也是一致的。我们在接入 AllTick 的港股数据时，也会先统一转换到 Asia/Hong_Kong，确保数据质量稳定。 四、开市前数据不能参与策略信号 9:00 到 9:30 虽然已有行情，但市场还在竞价阶段，不能作为连续交易信号。我们使用状态区分： def get_market_status(t): if time(9, 0) <= t < time(9, 30): return "pre_open" if time(9, 30) <= t < time(12, 0): return "morning" if time(12, 0) <= t < time(13, 0): return "break" if time(13, 0) <= t < time(16, 0): return "afternoon" return "closed" 策略端可以只接受 morning 和 afternoon 的数据，避免竞价阶段的异常波动干扰信号。 五、跨日期和交易日历 午夜判断也容易出错。比如： if current_time > time(16, 0): status = "closed" 凌晨 1 点确实会被判断成收市后，但如果还要判断“当前交易日”，就不能只看 time。我们的经验是：时间段用 datetime.time，交易日用 datetime.date，分开维护。 香港公众假期和台风休市要独立维护交易日历，不能只按周一到周五判断。 六、K 线聚合的时间边界 做 1 分钟 K 线时，时间边界更严格。09:29:59 和 09:30:00 只差一秒，但属于不同的 K 线周期。我们使用左闭右开区间： 09:30:00 <= tick_time < 09:31:00 这样每条 tick 只会进入一个周期，避免重复统计。 七、学术价值：可扩展的行情处理流程 我们现在的行情处理流程是： API实时数据 ↓ 解析 timestamp ↓ 转换为 Asia/Hong_Kong ↓ 判断交易日期 ↓ 判断交易时段 ↓ 过滤/分类行情 ↓ K线聚合或策略计算 把交易时段、时区、交易日历做成独立配置，策略代码里不再出现 09:30、16:00 这类魔法数字。这样无论是回测还是实盘，数据质量都能得到保证。时间边界处理是量化策略的基础功，值得我们花时间做好。 我们在做港股量化策略时，发现很多信号失效并不是因为因子不好，而是因为底层行情数据的时间边界没有处理好。尤其是使用港股 api 获取实时行情时，开市前、午间休市、收市后这些时间段的数据，如果没有严格区分，会直接影响策略的回测和实盘表现。 一、研究痛点：数据时间与交易时间的混淆 我们团队在给专业交易者和基金公司开发部门做港股量化模块时，最常遇到的问题就是：早上 9:05 的数据被当成连续交易数据处理，导致策略在竞价阶段就触发了信号。还有一次，服务器跑在 UTC 时区，直接拿 datetime.now() 和香港时间比较，结果下午的数据判断全部错位。 这些问题的根源在于：我们把“数据时间”和“交易时间”混在一起了。 二、数据需求：先定义清楚港股交易时段 港股不是全天连续交易。我们通常把一天拆成几个区间： 时段 香港时间 程序处理 开市前竞价 09:00–09:30 预开市数据 上午交易 09:30–12:00 正常交易 午间休市 12:00–13:00 非连续交易 下午交易 13:00–16:00 正常交易 收市后 16:00 之后 非正常交易 一个常见的错误写法是： if current_time >= time(9, 30): market_open = True 这个写法没有限定结束时间，所以下午 17:00 也会被判断成开市。我们后来改成区间判断： def is_market_open(t): morning = time(9, 30) <= t < time(12, 0) afternoon = time(13, 0) <= t < time(16, 0) return morning or afternoon 三、支持：API 时间戳统一转换 港股 api 返回的时间戳通常是 Unix timestamp： { "symbol": "00700.HK", "price": 520.5, "timestamp": 1788226200 } 这个时间戳本身没有时区概念。我们必须转成带时区的 datetime： from datetime import datetime from zoneinfo import ZoneInfo timestamp = 1788226200 dt = datetime.fromtimestamp( timestamp, tz=ZoneInfo("Asia/Hong_Kong") ) print(dt) 这样即使服务器部署在不同时区，策略端的时间判断也是一致的。我们在接入 AllTick 的港股数据时，也会先统一转换到 Asia/Hong_Kong，确保数据质量稳定。 四、开市前数据不能参与策略信号 9:00 到 9:30 虽然已有行情，但市场还在竞价阶段，不能作为连续交易信号。我们使用状态区分： def get_market_status(t): if time(9, 0) <= t < time(9, 30): return "pre_open" if time(9, 30) <= t < time(12, 0): return "morning" if time(12, 0) <= t < time(13, 0): return "break" if time(13, 0) <= t < time(16, 0): return "afternoon" return "closed" 策略端可以只接受 morning 和 afternoon 的数据，避免竞价阶段的异常波动干扰信号。 五、跨日期和交易日历 午夜判断也容易出错。比如： if current_time > time(16, 0): status = "closed" 凌晨 1 点确实会被判断成收市后，但如果还要判断“当前交易日”，就不能只看 time。我们的经验是：时间段用 datetime.time，交易日用 datetime.date，分开维护。 香港公众假期和台风休市要独立维护交易日历，不能只按周一到周五判断。 六、K 线聚合的时间边界 做 1 分钟 K 线时，时间边界更严格。09:29:59 和 09:30:00 只差一秒，但属于不同的 K 线周期。我们使用左闭右开区间： 09:30:00 <= tick_time < 09:31:00 这样每条 tick 只会进入一个周期，避免重复统计。 七、学术价值：可扩展的行情处理流程 我们现在的行情处理流程是： API实时数据 ↓ 解析 timestamp ↓ 转换为 Asia/Hong_Kong ↓ 判断交易日期 ↓ 判断交易时段 ↓ 过滤/分类行情 ↓ K线聚合或策略计算 把交易时段、时区、交易日历做成独立配置，策略代码里不再出现 09:30、16:00 这类魔法数字。这样无论是回测还是实盘，数据质量都能得到保证。时间边界处理是量化策略的基础功，值得我们花时间做好。 阅读时长：7分钟 标签：美股, Tick数据, WebSocket, 行情API, 量化研究, 回测, Python 引言 在量化研究与实盘策略开发过程中，基于API WebSocket获取美股Tick逐笔行情是较为常见的数据接入方式。开发阶段很容易形成一个固有假设：WebSocket推送数据包的到达顺序等价于市场事件真实发生顺序，接收后可直接送入指标计算、回测推演或者实盘策略引擎。 但接入真实市场数据流后会发现，跨洋网络波动、链路时延抖动、客户端消息解析压力、高频率Tick爆发等因素，会造成接收报文时序错乱。直接消费乱序Tick，会出现价格回溯、指标失真，回测与实盘结果出现不一致，给策略评估带来较大干扰。 本文从数据现象出发，分析Tick乱序的产生机理，结合不同研究与业务场景给出可落地的处理思路，提供客户端缓冲校正代码示例以及分层数据接入架构，供策略研究者做数据预处理参考。 Tick乱序的产生机理 本地测试环境样本量有限、网络条件稳定，时序异常会被掩盖。当订阅多只标的，接收高频逐笔行情时，传输环节的扰动会改变数据包抵达客户端的先后次序。 交易所原生生成3笔Tick记录示例： Tick标记 事件时间戳 成交价格 A 10:00:01.001 185.20 B 10:00:01.005 185.25 C 10:00:01.009 185.18 理想接收顺序：A → B → C 实际接收可能出现顺序：A → C → B 说明：报文乱序并不等同于上游数据源输出错误，多数属于网络传输与客户端处理带来的时序偏移。若不做校验直接消费，会造成行情展示异常、实盘策略信号误触发、落库原始数据集失真，进一步导致回测样本与实盘输入分布不一致。 建立核心数据原则：接收到Tick报文，不等于该报文可以直接参与模型、策略运算。 数据包的本机接收时间不能作为行情发生时间，API返回的事件时间戳、事件序列号，才是判断时序的可信基准。 标准处理逻辑流程： WebSocket接收原始行情报文 解析序列化得到Tick数据结构 提取报文中原生的市场事件时间戳 与已完成处理的最新行情时间做比对 根据研究场景执行缓存、丢弃、重放等处理逻辑 举例说明：系统已经处理完成时间戳10:00:01.009的Tick，后续收到滞后报文，事件时间为10:00:01.005。该条数据不能当作最新市场状态，需要标记为乱序样本，再结合场景选择处置逻辑。 不同量化场景下的处理取舍 不存在通用的万能解决方案，需要在数据时序精度、系统实时性之间做权衡，不同研究目标的处理策略存在明显差异。 行情可视化观测场景 可容忍毫秒级别的时序偏移，优先保证盘面输出稳定。无需构建过重的校正逻辑，设置短时缓冲窗口，积累少量Tick样本后，基于事件时间戳重排序，再输出用于观测。 Tick原始数据存储与回测数据集构建 原始event_time必须完整保留，不可仅存储本机接收时间。原生事件时间是后续数据集清洗、样本校验、回测复算的核心依据。 建议同步留存本机接收时间，用于评估链路传输时延，辅助区分问题发生在网络、数据源还是本地处理环节。回测数据集的时序正确性，直接决定策略历史评估结果是否具备参考价值。 实盘策略与模型计算场景 该场景对时序要求最高。Tick流入策略引擎、量化模型之前，必须校验数据流的时序完整性。滞后乱序的Tick一旦参与运算，可能生成虚假交易信号，造成实盘与回测结果出现偏差，缓冲与过滤机制属于必要环节。 客户端缓冲队列实现示例 以AllTick API WebSocket长连接为例，可在客户端实现短时内存缓冲，接收的Tick先进入缓冲区，依据事件时间戳完成排序后，再送入后续业务逻辑。 提示：示例中缓冲区大小20仅用于演示。实际使用需要结合Tick推送频率、网络抖动幅度、策略对延迟的容忍度调参。行情观测场景可适度放大缓冲区；低延迟实盘模型，需要控制缓冲区规模，平衡延迟开销与乱序容错能力。 import websocket # Tick缓冲队列初始化 buffer = [] def on_message(ws, message): tick = parse_tick(message) buffer.append(tick) # 使用行情原生时间戳对缓冲区重排序 buffer.sort(key=lambda x: x["timestamp"]) # 弹出时序靠前的Tick，交由后续逻辑处理 while len(buffer) > 20: tick = buffer.pop(0) process_tick(tick) ws = websocket.WebSocketApp( "wss://api.alltick.co/stock/websocket", on_message=on_message ) ws.run_forever() 重要提示：WebSocket协议仅保障消息可靠送达，不保障业务层面的事件时序。时序校验、乱序缓冲逻辑，需要在客户端侧自行实现。 容易忽略的时间字段问题 部分研究者会直接使用本机系统时间time.time()作为Tick的市场发生时间，这是数据预处理中常见误区。 received_at = time.time()获取的仅为程序收到报文的本机时刻，和美股市场真实成交发生时间相互独立。 数据落库建议同时保存两组时间字段： event_time：API返回，市场原始事件时间，作为时序判断基准 received_time：客户端本机接收报文时间 可直接计算端到端链路时延： latency = received_time - event_time 通过时延指标的变化，可以快速定位故障域：区分异常来源于网络链路、上游API服务，或是本地程序解析处理性能瓶颈。 面向时序敏感研究的分层接入架构 对于对Tick时序准确度要求较高的回测、实盘项目，建议将行情接收模块和下游策略、存储模块做解耦，避免网络抖动带来的时序异常直接传导至模型运算环节。 数据流结构： WebSocket原始报文 ↓ 行情接收接入层 ↓ 时间戳/事件序号校验 ↓ 短时缓冲 & 时序重排序 ↓ 分流：原始数据落库 / 策略模型运算 / 行情输出观测 分层设计的价值：网络临时抖动产生的乱序样本，在接入层完成缓冲校正，时序异常不会扩散到下游各个模块，降低数据集排查与策略调试的成本。 总结 使用美股API开展量化研究时，处理乱序Tick的核心，不是寄希望网络传输保证报文到达顺序完全正确。需要明确区分三组概念：报文到达顺序、市场事件发生顺序、业务处理顺序。 通过合理设置缓冲窗口、时间戳校验、双时间维度埋点，能够规避价格跳变、时间回溯等常见的数据异常，减少因为数据时序缺陷造成回测虚高、实盘表现偏离预期等问题。即便使用AllTick API这类成熟行情数据源，客户端的数据防护预处理逻辑，依旧是量化研究与实盘运行的重要基础。 亲测最好用的AI编写量化策略工具，可以让 AI 直接写各个平台的策略代码，直接生成可运行的策略代码，代码质量远高于直接使用 DeepSeek、Trae 等平台。 大家可以直接用描述策略，然后一键生成可运行的完整策略代码，也可以把它当做一个API 查询工具。 最新消息，已经支持SuperMind等主流量化平台啦，并且实盘亲测过了，很适合小白用户，上线之后获得了非常多朋友的好评。 **🚀️ AI工具平台：https://iris.findtruman.io/ai/tool/ai-quantitative-trading/** 我们在做外汇量化策略回测时，经常遇到一个让人头疼的问题：夏令时切换当天，历史K线的时间归属会出现偏差。这个问题对分钟级和小时级策略影响尤其大，如果时间轴错位，策略信号可能完全失真。 最近我们又处理了一批历史数据，把踩过的坑和解决方案整理出来，分享给社区里同样做个人量化交易的朋友。 案例描述：夏令时切换如何影响K线归属 先说一下我们遇到的具体情况。在复盘一批外汇小时线数据时，发现某些日期的K线排列位置不太自然，价格走势本身没有异常，但K线看起来总是错开一点。进一步检查时间字段后，才发现是夏令时切换对行情时间产生了影响。 外汇市场的数据来源比较复杂，不同接口返回的时间格式可能不同。有的返回市场本地时间，有的返回UTC时间。如果历史数据没有记录夏令时状态，后续进行指标计算或者回测时，可能会出现时间轴不一致的问题。 以美国市场时间为例，冬令时期间纽约市场09:00对应UTC时间14:00，而进入夏令时后，同样的本地时间会对应UTC时间13:00。 时间状态 本地交易时间 UTC时间 冬令时 09:00 14:00 夏令时 09:00 13:00 如果系统按照固定时间规则生成K线，就可能导致当天部分数据归属错误。例如本应该属于某个小时周期的数据，被划分到了前一个周期或者后一个周期。这种影响在分钟K线和小时K线中更加明显。 历史数据增加时间标记：保留原始时间 处理历史K线时，我们更倾向于保留原始时间，同时增加额外字段记录时间状态，而不是直接修改时间。这样后续做策略调整或者重新回测时，可以回溯到最原始的数据。 常见做法是在数据表中增加： 字段 用途 utc_time 统一时间标准 local_time 市场本地时间 timezone 所属时区 dst_status 夏令时状态 例如一条行情数据可以保存为： { "symbol": "EURUSD", "local_time": "2026-03-08 09:00:00", "utc_time": "2026-03-08T13:00:00Z", "dst_status": "active" } 这样后续查看历史行情时，可以清楚知道这根K线对应的时间环境。我们在自己的量化数据库里强制要求这些字段，避免回测时出现时间错位。 K线生成不要依赖固定时间差 很多数据处理中会直接通过增加或者减少几个小时完成转换，这种方式处理普通日期没有明显问题，但面对夏令时切换日期时容易出现偏差。 更稳定的方法是使用时区规则进行转换，让程序根据具体日期自动判断时间变化。Python处理时可以这样实现： from datetime import datetime import pytz timezone = pytz.timezone("US/Eastern") time_str = "2026-03-08 09:00:00" local_time = datetime.strptime( time_str, "%Y-%m-%d %H:%M:%S" ) local_time = timezone.localize(local_time) utc_time = local_time.astimezone(pytz.utc) print(utc_time) 这种方式不需要人工维护夏令时规则，数据跨越不同年份时也能保持一致。我们在实际回测中对比过，动态时区转换能显著降低时间轴不一致带来的虚假信号。 实时行情和历史数据保持统一：AllTick API示例 在实际行情系统中，历史K线和实时tick往往需要连接在一起。如果两部分采用不同时间标准，新生成的数据可能无法和历史数据准确衔接。 我们在处理实时行情时，会先统一时间格式，再进入K线计算流程。以 AllTick API 为例，通过 WebSocket 获取实时行情后，会将收到的时间字段转换为统一格式，再参与分钟线和小时线生成。这个外汇api服务返回的时间戳是UTC格式，适合用来做统一处理。 import websocket import json from datetime import datetime import pytz def on_message(ws, message): data = json.loads(message) trade_time = data["tradeTime"] tz = pytz.timezone("US/Eastern") dt = datetime.strptime( trade_time, "%Y-%m-%d %H:%M:%S" ) dt = tz.localize(dt) utc_time = dt.astimezone(pytz.utc) print(data["symbol"], utc_time) ws = websocket.WebSocketApp( "wss://api.alltick.co/stock/websocket", on_message=on_message ) ws.run_forever() 实操建议：量化交易者的避坑清单 结合我们的经验，列出几条关键建议： 存储阶段就增加时间元数据，不要等到回测时再补。 内部统一使用UTC时间作为标准，本地时间只用于展示。 K线生成不要依赖固定时间差，一定用时区规则动态转换。 实时与历史数据采用同一套时间处理逻辑，避免回测和实盘之间出现时间轴不一致。 对于长期保存的行情数据，时间字段设计比单纯保存价格更加重要。价格可以重新计算，但时间归属一旦错误，后续的数据分析都会受到影响。外汇api提供的数据本身只是原始信息，真正稳定的行情系统还需要在存储阶段处理好时区和夏令时规则。把UTC作为内部标准，把本地时间作为展示内容，这种方式更适合不同市场之间的数据转换，也能减少后续分析中的偏差。 希望这些内容对大家的策略开发有帮助，欢迎在社区里一起讨论。 
