A股L2逐笔、十档五档Tick快照与订单簿历史数据 之前跑回测的时候，我发现模拟的成交滑点总是不对劲，后来跟实盘对照了一下，才发现问题是出在行情精度上。大部分人用的都是5秒或者3秒的切片数据，那个已经丢掉了盘口排队、撤单这些信息。想要真正还原当时的市场状态，得看交易所原始放出来的那几类L2数据。下面就把我了解到的那几种数据具体长什么样、有哪些字段整理一下，没什么花里胡哨的，就是纯粹记录，免得自己过段时间又忘了。 在这个数据源：CMES金融数据库上，A股的L2历史数据主要分成四类：逐笔成交、逐笔委托、十档Tick快照、五档Tick快照，另外还有一个订单簿数据的下载选项。它们对应的是上交所和深交所不同的披露规则，深交所从2021年左右升级了五档快照，上交所这边有逐笔委托，深交所本身不发布逐笔委托，这一点在选数据的时候挺容易搞混的。 先看逐笔成交（Tick Execution），它是每一笔实际发生的、对外公布的成交记录。我一般拉数据会关注这几个字段：时间戳（精确到毫秒）、证券代码、成交价格、成交量、成交金额、买卖方向（买盘还是卖盘，有的源叫BS标志）、成交序号。还有一个容易忽略的是“成交类别”，比如深交所会标注正常成交、大宗交易或者盘后定价交易，这个在做量价统计的时候如果不做过滤，会把一些无效的巨量单带进来，我之前就踩过这个坑。 逐笔委托（Tick Order）是上交所独有的，记录了每一笔委托进入系统的指令，不管最后有没有成交。关键字段：委托时间、委托编号、价格、数量、买卖方向、订单类型（限价/市价）。它能看出来撤单行为、大单挂撤的节奏，很适合做订单流分析。深交所没有这个数据，就只能从逐笔成交和盘口快照里去反推挂单行为，效果总是差一截。 然后是十档快照和五档快照，这两个其实都是盘口快照数据，只不过档位数不同。上交所的Level2快照发的是十档，深交所现在也有五档快照（早些年只有三档）。快照的字段基本一样：时间、前收、开盘、最高、最低、最新价，还有委买1到10（或到5）的价格和数量、委卖1到10的价格和数量，另外会有总买量、总卖量、加权平均委买委卖价这些东西。有的数据源还会附上成交量、成交额，但不是每个都带。十档和五档的区别主要在盘口深度，做高频的时候十档明显能看到更厚的挂单墙，五档就有点不够用，但这个也看票，流动性差的票挂出十档的情况其实不多，很多时候五档也就够覆盖流动性了。 订单簿（Order Book）数据，这个经常被人和快照搞混。快照是每隔一段时间（通常是3秒）拍一次盘口瞬间的状态，订单簿严格来说是指那个时刻所有档位的全部挂单集合，但实际提供下载的“订单簿”文件往往就是快照记录按时间排列，字段和刚才说的十档五档差不多。唯一区别是文件名或者数据组织方式，有的源把盘中所有的快照切片存成一个文件，就叫订单簿数据，用的时候直接从里面按时间点抽取就行。 这些数据体量差别挺大。逐笔成交一天下来，全市场大概几千万条；逐笔委托量更大，因为委托数量比成交多得多。十档快照每3秒一条，单只股票一天大约4800条，乘上全市场5000多只，数据量也不小。下载的时候基本是按天、按市场、按股票代码分隔好的压缩包，一次拖一天的全量下来解压完几十个G很正常，机械硬盘读起来慢得想死。 对比一下几种数据的特点，我弄了个简单的表格，只列了我觉得实际用起来比较重要的差异，不是那种全字段比较，随手记的： 数据类型 更新方式 关键字段（部分） 适合做什么 逐笔成交 每笔推送 时间、代码、价格、成交量、方向、成交类别 构建逐笔价格序列、滑点计算、大小单统计 逐笔委托（仅沪市） 每笔推送 委托时间、编号、价格、量、方向、订单类型 订单流因子、撤单行为、委托不平衡分析 十档快照 3秒左右 委买/委卖10档价量、最新价、累计成交等 盘口形态、挂单异动、高频因子 五档快照 3秒左右 同十档但只有5档 深市的高频策略，足够覆盖大部分流动性 订单簿（文件） 通常为快照集合 同对应快照档位 历史盘口回放 这里说一下拉数据的具体方式。网站提供了HTTP下载链接，也有Python接口可以调，我用的那个库直接 pip install cmesdata 就能装。下面是当时从接口文档复制下来的示例，稍微改了一点点变量名。代码里那条注释是我自己加的，提醒自己传参别搞错时间格式，之前就填错过起始日期导致返回空数据集，查了半天才发现是把2024-01-02写成了20240102。 import cmesdata as cd # CMES金融数据库的行情接口，注意入参正确，调用频率正常 # 获取某只股票某日的十档快照示例 df = cd.get_snapshot( code='000001.SZ', # 股票代码，带上交易所后缀 date='2024-01-03', # 日期格式YYYY-MM-DD level=10 # 档位，可选5或10 ) print(df.head()) 其实没什么特别的技术含量，就是调个接口。但要注意这个接口如果短时间内频繁请求会被限制，我一般会加个 time.sleep(0.5)，不然 IP 被封了又得等。下载全量数据建议直接走网页的批量下载，用接口一条条拉全市场的话效率太低。 数据质量方面，用了一年多偶尔会遇到个别交易日有缺秒的情况，尤其是深交所快照，偶尔会出现相邻两条间隔大于3秒，官方解释是当时没有发生任何挂单变化就不推送。所以做对齐的时候如果按固定3秒时间轴插值，需要自己处理一下缺失点，不是数据源的问题。逐笔成交和委托的完整性还可以，对比过几次交易所官网的示例样本，没发现丢记录。 这些数据都只是原始行情记录，没有任何处理，到手之后清洗、处理全部要自己来。比如方向字段，有时候会用 B/S 标示，有时候是 1/2，每个源不一样，要注意。还有毫秒时间戳，有的是从行情时间直接解析的，晚盘时区分秒都能把人绕晕。反正用之前最好先抽几个时间点跟盘面切图对一遍，确认方向和时间没反。 一下子就说多了，大概就是这些内容。反正我现在回测里用的就是这些数据，起码盘口细节能兜住了，就不像以前那么虚。哪天硬盘又满了再去申请扩容吧。 国内期货数据源选型这件事，大多数人从第一天就做错了。他们从“哪个数据源的接口好用”开始，而不是从“国内期货的制度规则和其他市场差多少”开始。结果是：你选了一个接口很顺手的数据源，但你的回测从第一根 K 线起，就在一个错误的时间框架里运行。 2025 年全年，中国期货市场成交额超过 766 万亿元（中期协口径），商品期货成交量占全球的 72.3%（FIA 口径）。这个量级的市场里没有新手保护期。它的数据复杂性来自一个很多人没意识到的事实：上期所、大商所、郑商所、中金所，四家交易所的交易制度本身就各不相同。结算价算法不同，夜盘时段不同，合约月份设计不同。这些差异全部是明文规定，不是数据商的自由裁量。 你花三个月写的策略，实盘亏的钱可能不是策略的错，是数据源替你做的拼接决定——回测里那部分收益根本不存在，实盘一跑就蒸发。三个月推倒重来，亏的是真金白银加时间。 选国内期货数据源，第一件事不是调行情接口，是先回答三个问题：你要交易的品种，在它所属的交易所里，合约怎么设计？结算价怎么算？夜盘怎么归属？答不上来，数据源给什么你就信什么——这是最贵的信。 全文导览：第一节拆解国内期货的制度分层——这是所有数据问题的根源；第二节讲数据源应该承担的角色；第三节用真实代码验证制度信息是否被保留；第四节给选型判断框架；文末附 TickDB 期货行情的体系化能力图谱。 一、同样叫“期货”，四家交易所的制度分层 1.1 合约月份设计：换月不是数据源的问题，是制度的问题 国内期货没有“这只品种”的价格，只有“这个合约”的价格。每个合约有独立到期日，到期前必须把持仓迁到下一个合约——这就是换月。但“下一个合约”是什么，取决于交易所的合约月份设计： 品种 交易所 合约月份 最后交易日 阴极铜 上期所 1–12 月，全年有合约 合约月份第 15 日 豆粕 大商所 1/3/5/7/8/9/11/12 月 合约月份第 10 个交易日 白糖 郑商所 1/3/5/7/9/11 月（仅奇数月） 合约月份第 10 个交易日 沪深300 股指 中金所 当月 + 下月 + 两个季月 到期月第三个星期五 解读这张表：沪铜任何时候都有 12 个月份的合约同时在交易，豆粕有 4 个月份是空的，白糖隔月才有合约，股指期货是季月嵌套。同一个“主力合约切换”，在不同品种上意味着完全不同的时间节奏。 换月跳空不是数据错误，它是两个合约价格差异的直接体现。这个差异有定价依据——持仓成本：资金利息、仓储费用、便利收益的合力。数据源把新旧合约价格直接拼接后，跳空被当成趋势信号计算。策略被教着去交易展期价差，而这个价差在实盘中需要真金白银承担。回测赚的是虚拟钱，实盘亏的是真钱。 做三年回测，你面对的是 36 个不同的沪铜合约。连续合约是一个合成物——它的构造方式应该由你的策略逻辑决定，而不是由数据源在黑盒里替你决定。 1.2 结算价与收盘价：同一个“价”字，两种制度功能 股票的收盘价就是最后一笔成交价。期货不是。 国内期货有两个核心价格：收盘价——最后一笔成交价，反映最后时刻的市场均衡；结算价——交易所计算的当日成交量加权平均价，用于保证金计算和下一交易日涨跌停板基准。 关键不在于有两个价格，而在于：“结算价”这个词，在不同交易所的定义不同。 上期所和大商所的日结算价是当日全部成交的成交量加权平均；中金所股指期货的日结算价是最后一小时成交的成交量加权平均。如果品种有夜盘，上期所/大商所的结算价计算包含夜盘时段；中金所没有夜盘，结算价只覆盖日盘。 假设你做跨交易所价差策略，一手沪铜、一手股指期货。两个数据源都给你“结算价”，一个覆盖夜盘四个小时的交易，一个只覆盖日盘。拿这两个数字做价差，算出来的东西在制度上根本不成立。结算价用于保证金，收盘价用于趋势回测——混用一个，你的策略框架就是错的。 1.3 夜盘：跨日结构是制度事实，不是技术选择 国内商品期货有一个全球少见的制度设计：夜盘交易归属下一个交易日。 周三晚上 21:00 开始的夜盘，在结算上属于周四。 如果数据源按“自然日”组织数据，周三夜盘的行情会落在周三；但交易所的结算价把它归属于周四。你的 K 线和结算价数据之间，在时间对齐上错了一天。 更关键的是：夜盘时段是品种级变量，不是全市场统一的。 交易所/品种 夜盘时段 上期所黄金、白银 21:00 – 次日 2:30 上期所沪铜、铝、锌 21:00 – 次日 1:00 上期所螺纹钢、热卷 21:00 – 23:00 大商所全线夜盘品种 21:00 – 23:00 郑商所全线夜盘品种 21:00 – 23:00 中金所股指期货 无夜盘 做螺纹钢日内策略，你的“日内”是 6.5 小时；做黄金日内策略，你的“日内”是 9.5 小时。在螺纹钢上开发的日内因子，直接迁移到黄金上，时间框架就是错的。 郑商所 2019 年 12 月把夜盘从 23:30 缩短到 23:00——此前此后的数据，结算价时间窗口不同，数据源必须按时期分别处理，否则你的历史回测就是混着两种口径跑的。 国内期货数据复杂性不是数据商的问题，是制度设计本身的分层性。你唯一能做的，是确认数据源是否尊重这种分层——数据源做不到，你就是在用自己的钱包替它的粗糙买单。 二、数据源的角色：精确翻译，而不是替你决策 理解了制度分层，选型标准就变了。 一个诚实的数据源，应该做的事是：把你需要的原始数据按制度差异精确提供。 合约级价格、成交结构、时段结构、复权参数。它不应该替你做的事包括：替你构造连续合约——因为连续合约本身就是你的策略设计；替你决定夜盘归属——因为夜盘规则因品种而异；把结算价和收盘价混成一个字段——因为它们的制度功能根本不同。 但在实际操作中，很多数据源做的事情恰恰相反。它们替你“处理好”了连续合约，但从不告诉你换月逻辑是什么；给你一个统一的“交易日”字段，抹平了四家交易所的时段差异；给你一个“价格”，分不清是收盘还是结算。 数据源替你做的决策越多，你对自己回测的理解就越少。它给你的每一个字段，你都应该能找到对应的制度依据。 三、实测：三份数据，回答四个制度问题 下面用真实的接口调用，验证一个数据源能不能把制度差异翻译成结构化字段。2026 年 9 月 8 日对 TickDB 实测，三个合约：CU2609（沪铜，有夜盘）、RB2609（螺纹钢，有夜盘）、IF2609（股指期货，无夜盘）。代码可直接运行，替换你的 Key 和品种即可。 import json from urllib.parse import urlencode from urllib.request import Request, urlopen BASE = "https://api.tickdb.ai/v1" API_KEY = "YOUR_KEY" SYMBOL = "CU2609" # 格式：品种代码 + 合约年月 def get(path, **params): url = f"{BASE}{path}?{urlencode(params)}" if params else f"{BASE}{path}" request = Request(url, headers={"X-API-Key": API_KEY, "Accept": "application/json"}) with urlopen(request, timeout=30) as response: assert response.status == 200 payload = json.load(response) assert payload["code"] == 0 return payload["data"] # 制度问题一：价格是新鲜的，还是陈旧的？ ticker = get(f"/market/ticker/{SYMBOL}") assert {"last_price", "timestamp", "bid_price", "ask_price", "open", "prev_close"} <= ticker.keys() assert isinstance(ticker["timestamp"], int) and ticker["timestamp"] > 10**12 print(f"ticker 验证通过，时间戳 {ticker['timestamp']}") # 制度问题二：K 线保留了哪些制度信息？持仓量是第四维，有没有？ raw = get("/market/kline", symbol=SYMBOL, type="futures", interval="1d", limit=5, adjust="none") fwd = get("/market/kline", symbol=SYMBOL, type="futures", interval="1d", limit=5, adjust="forward") assert raw["adjust"] == "none" and fwd["adjust"] == "forward" assert {"time", "open", "high", "low", "close", "volume", "open_interest"} <= raw["klines"][-1].keys() print(f"kline 验证通过，open_interest 字段存在") # 制度问题三：成交数据能区分开仓和平仓吗？ trades = get("/market/trades", symbol=SYMBOL, type="futures", limit=20) first = trades["trades"][0] assert {"price", "quantity", "side", "timestamp", "position_effect"} <= first.keys() print(f"trades 验证通过，position_effect = {first['position_effect']}") # 制度问题四：交易时段的边界在哪里？ sessions = get("/market/trading-sessions", symbol=SYMBOL, market="CN", type="futures") assert sessions and {"begin_time", "end_time"} <= sessions[0]["trading_sessions"][0].keys() print(f"trading-sessions 验证通过，日盘基准 {sessions[0]['trading_sessions']}") 这段代码的每个 assert 都不是在验证数据源“好不好”，而是在验证它有没有把制度差异翻译成你可以核验的字段。下表汇总四个制度问题的验证结果。 制度问题 验证接口 实测字段 投资价值 风险价值 场景价值 ① 价格新鲜度 ticker timestamp（Unix 毫秒）、bid_price、ask_price、last_price、open、prev_close 用时间戳判断价格是否新鲜，用买卖价做流动性门，防止用陈旧价下单 没有时间戳的“最新价”可能是一分钟前甚至昨日价格，错用会导致错误止损、错误限价 实时盯市、止损触发前核验、限价单定价 ② 持仓量（第四维） kline open_interest、open/high/low/close、volume、adjust（none/forward/backward） 用持仓量构建四维分析矩阵，判断趋势质量；用adjust 参数自选复权口径 没有持仓量的 K 线无法区分“新钱进场”和“老钱离场”；混用复权口径会制造虚假收益 回测数据基础、因子计算、趋势确认 ③ 开平仓方向 trades position_effect（both_open / both_close / long_open / long_transfer）、side、price、quantity 用开平仓组合识别双开/双平/方向交接，定位拐点信号 只有side 没有 position_effect 时，买方/卖方不能等同于开仓/平仓，成交结构分析完全失效 成交结构分析、订单流研究、拐点识别 ④ 交易时段边界 trading-sessions begin_time、end_time（本次返回日盘基准） 确认策略运行的时间窗口，防止在非交易时段误触发信号 把自然日当交易日会导致信号错位；忽略夜盘品种差异会打乱日线合成 实时策略调度、回测时段过滤 制度问题①：一分钟足以把止损从“可控”变成“灾难”。 制度问题②：持仓量是期货区别于股票的第四维度。 没有 open_interest，你看到的“放量上涨”可能只是空头在认输平仓，根本没有新资金进场。学术界关于持仓量的预测能力存在分歧，但分歧本身不否定它的分析价值——持仓量必须放在价格和成交量的交互矩阵里看。 制度问题③：position_effect 四个枚举值直接对应期货合约的生命周期：双开创造新契约，持仓量增加；双平消灭旧契约，持仓量减少；多开空平和空开多平是方向交接，持仓量不变。没有这个字段，你对期货的理解就停留在股票框架里——看到价和量，看不到那个正在形成的拐点。 制度问题④：夜盘是品种级变量，同一交易所内部也有差异。 没有夜盘标签，意味着你需要自己维护品种级时段映射——这是一个边界，但也是一个明确的边界：数据源告诉你它覆盖了什么，把没覆盖的东西还给你自己处理。 TickDB 在这些实测里做的事很简单：它把制度差异留在数据里，不去替你压平。没有连续合约标识符，是让你知道连续合约是你自己的策略设计；没有夜盘统一标签，是让你知道你交易哪个品种就该知道哪个品种的夜盘。这不是功能缺失，是数据源在告诉你：制度的事，你自己要想清楚。 四、选型判断：把制度信息保留得最多的数据源，才是可验证的数据源 制度维度 免费 Python 库类 积分制数据源类 TickDB（2026-09-08 实测） 合约级原始数据 需自行拼接，无官方换月声明 有主力合约逻辑，不透明 合约级 K 线可用，adjust 参数支持 none/forward/backward 持仓量（第四维） 通常不提供 部分品种有 K 线含open_interest 字段 开平仓制度结构 成交数据无方向字段 部分品种有，需单独开通 position_effect 实测返回 both_open / both_close / long_open / long_transfer 交易时段制度暴露 不提供独立时段结构 无统一时段接口 返回日盘基准时段，夜盘需按品种自建 价格时间属性 部分数据无时间戳 部分有 ticker 含timestamp，单位 Unix 毫秒 连续合约没有？那是它把换月逻辑还给了你。夜盘标签没有？那是它把品种级制度差异还给了你。开平仓字段有？那是它把四维市场的第四维给了你。 你要做的判断不是谁功能全，是谁给你的数据能让你用自己的方式应对制度分层。 表格中 TickDB 列的内容全部来自 2026 年 9 月 8 日对 CU2609、RB2609、IF2609 三个合约的实测调用。其他两类数据源的内容来自其公开文档。实测结果不代表全品种覆盖，生产使用前必须对目标品种重新验证。 五、TickDB 期货行情数据服务：制度翻译的体系化画像 本文以上实测涉及 TickDB 的四个接口。但 TickDB 的期货行情能力不只这四个。以下是基于其公开 OpenAPI 规范和实测结果的体系化梳理。 产品定位：TickDB 是面向开发者、量化研究和 AI 应用的统一实时市场数据服务，以 REST 和 WebSocket 双协议提供国内期货、A 股、港股、美股、外汇和加密等市场的行情数据。国内期货品种通过合约级标识符（品种代码 + 合约年月，如 CU2609）接入，不提供黑盒式的连续合约标识符。 核心接口的投资价值、风险价值、场景价值全景： 接口 制度翻译功能 投资价值 风险价值 场景价值 GET /v1/market/ticker/{symbol} 价格时间属性 + 买卖价差 用bid_price/ask_price 做流动性门，用 timestamp 判断价格新鲜度 没有时间戳的“最新价”可能是陈旧价，错用会导致错误止损或错误限价 实时盯市、止损单触发前核验 GET /v1/market/kline K 线 + 持仓量 + 复权参数 用open_interest 构建四维分析矩阵，用 adjust 参数选复权口径 混用复权口径会制造虚假收益；没有持仓量的 K 线无法判断趋势质量 回测数据基础、因子计算、趋势确认 GET /v1/market/trades 开平仓方向 + 逐笔成交结构 用position_effect 区分双开/双平/方向交接，识别拐点信号 只有side 没有 position_effect 时，买方/卖方不能简单等同于开仓/平仓 成交结构分析、订单流研究、执行质量评估 GET /v1/market/trading-sessions 交易时段边界 确认策略运行的时间窗口，防止非交易时段触发信号 把自然日当交易日会导致信号错位；忽略夜盘品种差异会打乱日线合成 实时策略调度、回测时段过滤 其他期货相关能力（基于 OpenAPI 规范 0.9.9.4.1，截至 2026-09-02）：/v1/market/trade-days 提供交易日历查询；/v1/market/intraday 提供日内分时数据；/v1/market/depth 提供盘口深度；/v1/market/klines/history 和 /range 提供灵活的历史 K 线拉取。WebSocket /v1/realtime 提供实时行情订阅。财务基本面接口（Finance API）覆盖美股、港股和 A 股，与期货行情形成多市场统一接入。 统一接入的价值：如果你同时做国内期货和 A 股，TickDB 的 REST 接口规则和字段风格是统一的。一个 get() 函数可以同时拉取沪铜 ticker 和沪深300 ticker，不需要为每个市场维护一套独立的客户端。 AI 工具接入：TickDB 通过 MCP、CLI 和 Skill 为 AI 工作流提供结构化市场数据，让模型先取得带标的、字段和时间戳的事实，再进行分析和表达——而不是用训练数据里的历史价格回答“沪铜现在多少钱”。 能力边界（诚实声明）：本文实测未验证的项包括：WebSocket 实时通道的延迟和稳定性；全品种的历史数据深度；连续合约的具体换月方案（TickDB 不提供黑盒连续合约标识符，换月逻辑是用户自己的策略设计）；夜盘时段的品种级标注（trading-sessions 当前返回日盘基准，夜盘映射需用户按品种自建）。 TickDB 的期货行情能力，核心逻辑和本文的选型标准是一致的：它把制度差异翻译成结构化字段，把原始材料给你，把决策权留给你。它不替你构造连续合约，不替你定义夜盘，不替你区分结算价和收盘价。它只做一件事：让你能用自己的方式，精确地应对国内期货的制度分层。 国内期货从制度层面就不是一个统一的市场。四家交易所，四套规则。数据源无法替你理解这些制度，它只能帮你把制度翻译成结构化字段，把原始材料给你，让你自己去理解、去构建、去决定。选数据源之前，先打开交易所的官方规则，确认你的品种在哪套制度里运行。然后拿代码去验证数据源有没有尊重这套制度。两个都做了，你的选型就不是碰运气，是有制度依据的判断。 下一步：先查你要交易品种的交易所合约文本——上期所、大商所、郑商所、中金所官网的合约月份设置、结算价定义、夜盘时段规则。然后跑一遍上面的代码，验证数据源对这四个制度维度的翻译。做完，再谈选型。 参考文献 [1] 上海期货交易所. 上海期货交易所阴极铜期货业务细则（修订版）[S]. 上海：上海期货交易所, 现行有效. [2] 大连商品交易所. 大连商品交易所豆粕期货业务细则[S]. 大连：大连商品交易所, 现行有效. [3] 郑州商品交易所. 白糖期货合约表及业务细则[S]. 郑州：郑州商品交易所, 现行有效. [4] 中国金融期货交易所. 沪深300股指期货合约表及交易细则[S]. 上海：中国金融期货交易所, 现行有效. [5] 上海期货交易所. 上海期货交易所交易细则、结算细则[S]. 上海：上海期货交易所, 现行有效. [6] 大连商品交易所. 大连商品交易所结算管理办法[S]. 大连：大连商品交易所, 现行有效. [7] 中国金融期货交易所. 中国金融期货交易所期货结算细则[S]. 上海：中国金融期货交易所, 现行有效. [8] 上海期货交易所. 连续交易（夜盘）品种交易时间表及规则公告[Z]. 上海：上海期货交易所. [9] 大连商品交易所. 夜盘交易时段与品种清单公告[Z]. 大连：大连商品交易所, 2019. [10] 郑州商品交易所. 夜盘交易时间调整公告（2019年12月11日）[Z]. 郑州：郑州商品交易所, 2019. [11] 上海国际能源交易中心. 交易细则及夜盘相关公告[Z]. 上海：上海国际能源交易中心. [12] Gorton G, Rouwenhorst K G. Facts and Fantasies about Commodity Futures[J]. Financial Analysts Journal, 2006, 62(2): 47–68. [13] López de Prado M. Advances in Financial Machine Learning[M]. New Jersey: John Wiley & Sons, 2018. [14] Bailey D H, Borwein J M, López de Prado M, Zhu Q J. The Probability of Backtest Overfitting[J]. Journal of Computational Finance, 2017, 20(4): 39–69. [15] 中国期货业协会. 2025年全年期货市场成交统计[Z]. 北京：中国期货业协会, 2026年1月. [16] FIA. Annual Futures and Options Volume Report[R]. Washington D.C.: Futures Industry Association, 各年度. [17] 中国期货业协会. 期货市场投资者结构统计数据[Z]. 北京：中国期货业协会, 各年度. 📌 摘要 / 快速解答 (Direct Answer) 我做量化以后越来越觉得，Python 最大的优势不是“代码短”，而是能把金融数据、Pandas、指标计算和 AI 分析串成一条工作流。金融数据 SDK 负责把 A 股行情稳定地送进 Python，QuantDash 再通过统一代码格式、批量 K 线和原生 DataFrame，让“Python 提取股票 K 线 → 指标选股 → AI 辅助诊股”这条链路简单很多。 一、为什么传统方式写选股策略这么累？ 如果你只是偶尔查一只股票，数据从哪里来可能不重要。 但如果你准备真正做 A股量化选股，情况完全不一样。 我见过不少 SuperMind 社区的新手，第一版策略往往是： 找一个数据接口 ↓ 下载数据 ↓ Excel清洗 ↓ Python读取 ↓ 计算指标 ↓ 回测 然后策略一改，前面的数据处理再来一次。 折腾几轮之后，很多人会发现： 我明明是来研究选股逻辑的，怎么每天都在修数据？ 1. 单只股票能跑，不代表全市场能跑 比如： df = qd.klines.get("600519.SH", period="1d") 单只股票当然很简单。 但真正的量化策略通常不是研究一只股票。 我们真正想做的是： 5500+ A股 ↓ 第一轮筛选 ↓ 1000只 ↓ 技术指标 ↓ 100只 ↓ 更深度的策略分析 这时候如果还手动循环请求、自己处理异常和数据格式，代码很快就会变得非常臃肿。 2. 多市场策略又是另外一个坑 A股、港股、美股如果分别找不同数据源，最麻烦的其实不是“能不能拿到”。 而是： 代码、字段、时间和数据结构不统一。 QuantDash 使用统一标的代码规则： 600519.SH A股上海 000001.SZ A股深圳 920047.BJ A股北京 00700.HK 港股 AAPL.US 美股 对于 Python 策略来说，这种统一非常舒服。 你的股票池可以直接变成： symbols = [ "600519.SH", "000001.SZ", "00700.HK", "AAPL.US" ] 后面的策略逻辑不需要因为市场不同而重新设计一套接口。 二、解决方案对比：QuantDash vs 传统数据源 对比维度 传统/竞品方案（如 Tushare/AkShare/手动爬虫） QuantDash 解决方案 数据获取 常需要自己适配不同接口 Python SDK 直接调用 数据格式 可能需要转换和清洗 原生 Pandas DataFrame A股代码 不同数据源规则可能不同 .SH/.SZ/.BJ统一 海外市场 经常需要额外数据源 .US/.HK纳入统一格式 批量研究 自己写大量循环逻辑 klines.batch()支持批量 实时行情 需要单独处理实时接口 quotes.get()支持实时行情 复权 需要自行确认和处理 adjust参数直接指定 分钟数据 数据源和接口规则可能不同 支持 1m/5m/15m/30m/60m 我认为金融数据 SDK 最大的价值并不是“帮你少写几十行代码”。 而是： 把数据工程的不确定性，尽量隔离在 SDK 这一层。 这样策略研究层就可以保持干净。 三、Python代码实战：批量行情到量化候选池 这次不做单只股票，而是模拟一个更接近实盘研究的流程： 全市场 → 技术指标 → 候选股票。 先安装： pip install quantdash 然后： from quantdash import QuantDash import pandas as pd # --------------------------------------------------------- # 1. 初始化 QuantDash # --------------------------------------------------------- qd = QuantDash(api_key="your-api-key") # --------------------------------------------------------- # 2. 获取 A 股全市场实时行情 # --------------------------------------------------------- quotes = qd.quotes.get( universes=["CN_Stock"], to_dataframe=True ) print("当前行情数量：", len(quotes)) print( quotes[ ["symbol", "last_price", "prev_close", "volume"] ].head() ) # --------------------------------------------------------- # 3. 先做一个非常简单的初筛 # 这里不直接判断买卖，只缩小后续研究范围 # --------------------------------------------------------- quotes["change_pct"] = ( quotes["last_price"] / quotes["prev_close"] - 1 ) candidate = quotes[ (quotes["last_price"] > 0) & (quotes["volume"] > 0) ].copy() print("初筛后股票数量：", len(candidate)) # --------------------------------------------------------- # 4. 选择若干股票继续获取历史K线 # 实际研究时可以根据自己的因子条件扩展 # --------------------------------------------------------- symbols = candidate["symbol"].head(10).tolist() dfs = qd.klines.batch( symbols, period="1d", count=60, adjust="forward", to_dataframe=True, show_progress=True ) # --------------------------------------------------------- # 5. 对每只股票计算 MA20 # --------------------------------------------------------- results = [] for symbol, df in dfs.items(): if df.empty or len(df) < 20: continue df = df.copy() df["MA20"] = df["close"].rolling(20).mean() latest = df.iloc[-1] # 简单趋势条件 above_ma20 = latest["close"] > latest["MA20"] results.append({ "symbol": symbol, "trade_date": latest["trade_date"], "close": latest["close"], "MA20": latest["MA20"], "above_MA20": above_ma20 }) result_df = pd.DataFrame(results) # --------------------------------------------------------- # 6. 输出候选池 # --------------------------------------------------------- if not result_df.empty: selected = result_df[ result_df["above_MA20"] ].sort_values( "close", ascending=False ) print("\n技术条件通过的股票：") print(selected.to_string(index=False)) else: print("没有得到有效候选结果") 这个例子故意没有写得特别复杂。 因为我做量化比较反感一个误区： 代码越复杂，不代表策略越专业。 真正值得研究的是： 为什么选择这个因子？ 为什么选择这个窗口？ 为什么设置这个阈值？ 换市场以后还有效吗？ 交易成本加入以后还有效吗？ 而不是： 我用了多少行 Python？ 再往前一步：MACD + BOLL + AI辅助分析 如果你已经把 K 线拿到 DataFrame 里，技术指标其实就是标准的数据处理问题。 例如： df["MA20"] = df["close"].rolling(20).mean() middle = df["close"].rolling(20).mean() std = df["close"].rolling(20).std() df["BOLL_MID"] = middle df["BOLL_UPPER"] = middle + 2 * std df["BOLL_LOWER"] = middle - 2 * std ema12 = df["close"].ewm(span=12, adjust=False).mean() ema26 = df["close"].ewm(span=26, adjust=False).mean() df["DIF"] = ema12 - ema26 df["DEA"] = df["DIF"].ewm(span=9, adjust=False).mean() 接下来再生成一个结构化的 AI 分析材料： analysis_data = df[ [ "trade_date", "close", "volume", "MA20", "BOLL_UPPER", "BOLL_LOWER", "DIF", "DEA" ] ].tail(10) ai_prompt = f""" 你是A股量化研究助手。 请基于以下客观数据进行技术面描述： 1. 当前价格与MA20的关系； 2. BOLL位置； 3. DIF与DEA关系； 4. 成交量变化。 不要使用未来数据，不要承诺收益， 不要把技术指标直接等同于买卖结论。 数据： {analysis_data.to_string(index=False)} """ print(ai_prompt) 我比较推荐这种 QuantDash + Pandas + DeepSeek 的组合方式。 不是让 AI 自己找数据，而是： QuantDash 负责真实行情 ↓ Pandas 负责指标计算 ↓ DeepSeek 负责语言化分析 ↓ 交易员 负责最终判断 这个分工比“直接问 AI 哪只股票明天涨”靠谱得多。 而且 AI 出现分析错误时，也更容易定位到底是哪一层出了问题。 四、交易员避坑指南：AI再强也不能替你解决数据问题 1. AI诊股首先要防止数据穿越 如果给 DeepSeek 的数据里混入未来日期，AI 再聪明也会得到错误结论。 所以我建议： 数据截止时间 ↓ 计算指标 ↓ 生成AI输入 ↓ 分析 不要把未来行情偷偷塞进 prompt。 2. 前复权不是“万能正确答案” QuantDash 默认支持 forward 前复权。 对于趋势研究和收益率计算，它非常方便。 但如果你的策略研究的是： 实际成交价格； 除权除息事件； 某一天真实盘口价格； 那么就应该根据研究目的选择合适的 adjust 参数。 复权方式必须和研究目的对应。 3. 不要用单只股票回测结果证明策略有效 这是新手最容易犯的错误之一。 比如： 600519.SH 过去5年 收益率300% 然后宣布： MACD + MA20 是有效策略。 这远远不够。 至少还要考虑： 不同股票； 不同时间区间； 牛市； 熊市； 震荡市； 交易成本； 滑点； 样本外测试。 否则很可能只是过拟合。 五、常见问题解答（Q&A / FAQ） Q1：Python提取股票K线为什么推荐使用金融数据SDK？ A：因为 Python 的优势在于数据分析和策略研究，而不是自己维护数据抓取链路。QuantDash 可以直接通过 qd.klines.get() 获取 K 线，并原生返回 Pandas DataFrame，后面可以直接接 MA、MACD、KDJ、BOLL 等计算。 Q2：QuantDash适合做SuperMind里的A股量化研究吗？ A：如果你的 Python 工作流需要获取 A 股行情、批量 K 线、复权数据或全市场实时行情，QuantDash 的 SDK 结构比较适合这种研究流程。尤其是 CN_Stock 标的池和 klines.batch()，可以减少大量数据搬运代码。 Q3：Python量化能不能结合DeepSeek做AI智能诊股？ A：可以把两者组合起来，但应该做好职责划分。QuantDash 提供行情数据，Pandas 负责指标计算，再把结构化结果交给已经配置好的 DeepSeek 工作流进行辅助分析。这样可以避免让 AI 自己猜测行情数据。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：https://quantdash.net/ 📖 官方 Python SDK 文档：https://docs.quantdash.net/zh-Hans ⭐ GitHub 开源仓库：https://github.com/quantdash-net/QuantDash 💡 API Key：https://quantdash.net/dashboard/keys/ QuantDash 官方 GitHub 仓库提供了 Python 示例、快速开始代码以及 SDK 使用说明，建议准备长期做量化的朋友直接 Star / Fork 后再研究。 📌 摘要 / 快速解答 (Direct Answer) Python 量化开发真正难的，往往不是写 MA、MACD、KDJ，而是稳定拿到格式统一、可直接计算的数据。金融数据 SDK 可以把行情获取、复权、代码格式和 Pandas 数据处理串起来，而 QuantDash 通过 pip install quantdash 接入，日K、分钟K、批量行情都可以直接返回 DataFrame，让我把精力放回策略本身。 一、为什么传统方式写选股策略这么累？ 我刚开始做 A 股量化的时候，最容易低估的一件事就是：数据工程本身就是一个坑。 很多新手以为量化选股就是： 找股票 → 拉K线 → 算指标 → 回测 真正写起来以后，经常变成： 找接口 → 注册 → 看权限 → 攒积分 → 接口报错 → 换接口 → 清洗字段 → 处理复权 → 对齐交易日 → 再开始算指标 尤其是自己拼数据源的时候，一个简单的 MACD 都可能被数据质量拖住。 比如我想做一个很朴素的条件： MA20 向上 + MACD 金叉 + 成交量放大 策略逻辑其实没多少代码，但前面的行情数据如果格式不统一，就得不断处理。 1. A股量化选股，第一关其实是数据 Tushare 这类方案大家都熟悉，但对刚入门的朋友来说，积分、权限和接口使用门槛经常会影响研究效率。 AkShare 的优势是接口丰富，但实际做长期策略研究时，我更在意的是：批量运行的时候数据链路是否稳定。 还有一些朋友直接爬网页。 这就更容易出现： 网页结构变化； 字段名称变化； 请求失败； 数据缺失； 回测跑到一半中断。 量化策略最怕的不是代码报错，而是你不知道数据什么时候出了问题。 2. 复权问题，比很多人想象中更重要 比如我们计算历史收益率，直接使用不复权价格，很容易在分红、送转之后得到不连续的价格序列。 QuantDash 的 K 线接口支持： forward：前复权； backward：后复权； forward_additive：前复权差值； backward_additive：后复权差值； none：不复权。 我个人做趋势和收益率类策略时，通常会优先考虑前复权数据。 重点不是“哪个复权永远最好”，而是研究过程中要明确自己到底用了哪一种价格序列。 二、解决方案对比：QuantDash vs 传统数据源 对比维度 传统/竞品方案（如 Tushare/AkShare/手动爬虫） QuantDash 解决方案 数据稳定性 接口较多，实际使用时可能遇到权限、网络或接口变化问题 面向量化研究的数据 API，官方提供自动重试、限流保护等能力 使用门槛 可能需要处理权限、积分、字段清洗等问题 pip install quantdash，SDK 原生支持 DataFrame 跨市场支持 不同数据源代码规则、字段结构容易不一致 A股/美股/港股统一采用.SH/.SZ/.BJ/.US/.HK等后缀 数据格式 经常需要自己转换成 DataFrame 原生返回 Pandas DataFrame K线周期 不同数据源接口规则不统一 支持日、周、月、季、年以及 1m/5m/15m/30m/60m 批量研究 往往需要自己写循环和异常处理 提供klines.batch()等批量获取方式 复权处理 可能需要自行处理 K线接口直接提供多种adjust参数 这里我特别喜欢 QuantDash 的一点，就是数据拿回来以后直接就是 Pandas 思维。 对于 Python 量化开发者来说，这个细节非常重要。 因为后面无论是计算 MA、MACD、KDJ，还是接自己的回测框架，本质上都是在处理 DataFrame。 三、Python代码实战：从K线到技术指标 先安装： pip install quantdash QuantDash 官方 GitHub 也提供了 Python 示例和快速开始代码，建议大家实际使用前直接看官方仓库。 下面这份代码，我用一个非常典型的 A 股技术指标筛选场景来演示： from quantdash import QuantDash import pandas as pd # 初始化 QuantDash # 更推荐把 API Key 放到环境变量 QUANTDASH_API_KEY 中 qd = QuantDash(api_key="your-api-key") # --------------------------------------------------------- # 1. 获取贵州茅台前复权日K # forward：前复权比例方式 # --------------------------------------------------------- df = qd.klines.get( "600519.SH", period="1d", count=120, adjust="forward", to_dataframe=True ) # --------------------------------------------------------- # 2. 计算常用技术指标 # --------------------------------------------------------- # MA：均线 df["MA5"] = df["close"].rolling(5).mean() df["MA20"] = df["close"].rolling(20).mean() df["MA60"] = df["close"].rolling(60).mean() # MACD ema12 = df["close"].ewm(span=12, adjust=False).mean() ema26 = df["close"].ewm(span=26, adjust=False).mean() df["DIF"] = ema12 - ema26 df["DEA"] = df["DIF"].ewm(span=9, adjust=False).mean() df["MACD"] = (df["DIF"] - df["DEA"]) * 2 # 成交量5日均值 df["VOL_MA5"] = df["volume"].rolling(5).mean() # --------------------------------------------------------- # 3. 构造一个简单的趋势筛选条件 # --------------------------------------------------------- latest = df.iloc[-1] trend_ok = ( latest["close"] > latest["MA20"] and latest["MA20"] > latest["MA60"] ) macd_ok = latest["DIF"] > latest["DEA"] volume_ok = latest["volume"] > latest["VOL_MA5"] print("最新交易日：", latest["trade_date"]) print("收盘价：", latest["close"]) print("MA20：", round(latest["MA20"], 2)) print("MA60：", round(latest["MA60"], 2)) print("DIF：", round(latest["DIF"], 4)) print("DEA：", round(latest["DEA"], 4)) print("成交量：", latest["volume"]) print("5日均量：", round(latest["VOL_MA5"], 2)) if trend_ok and macd_ok and volume_ok: print("技术条件：通过") else: print("技术条件：暂不通过") 这段代码有个很实际的意义： 行情数据和策略计算彻底分开了。 QuantDash 负责： 行情 → DataFrame Pandas 负责： DataFrame → 指标 → 条件 后面你要换成 MACD、KDJ、BOLL，或者自己设计因子，都不用重新折腾行情接口。 再进一步：批量做 A 股量化选股 如果不想只看一只股票，可以直接使用标的池。 例如： from quantdash import QuantDash qd = QuantDash(api_key="your-api-key") # 获取 A 股全市场实时行情 df = qd.quotes.get( universes=["CN_Stock"], to_dataframe=True ) print("股票数量：", len(df)) print(df.head()) 这时候思路就可以升级： 全市场行情 ↓ 成交量/价格初筛 ↓ 提取候选股票 ↓ QuantDash 批量拉K线 ↓ 计算 MA/MACD/KDJ/BOLL ↓ 生成候选池 ↓ 再交给人工或 AI 做进一步研究 这比一开始就让 AI “猜哪只股票会涨”靠谱得多。 关于 DeepSeek，我建议这样接 这里我反而提醒一句：不要为了文章看起来高级，就随便编一个 DeepSeek API 调用接口。 QuantDash 官方文档明确的是金融数据 SDK 接口，并没有把 DeepSeek API 定义为 QuantDash 的一部分。 所以我的实战方式是： QuantDash 获取标准化行情； Pandas 计算 MA/MACD/KDJ/BOLL； 把结构化指标和最近若干交易日数据整理成文本； 再交给你已经配置好的 DeepSeek 工作流进行辅助分析。 例如可以生成这样的 AI 诊股输入： recent = df[ ["trade_date", "close", "volume", "MA5", "MA20", "MA60", "DIF", "DEA", "MACD"] ].tail(10) prompt = f""" 你是一名A股量化研究助手。 请根据下面的历史行情和技术指标， 从趋势、动量、成交量三个角度进行客观分析。 不要预测确定性收益，也不要直接给出买卖承诺。 数据： {recent.to_string(index=False)} """ print(prompt) 这样做的好处是：AI 负责解释，数据负责事实。 四、交易员避坑指南：别让回测看起来太漂亮 1. 不要把未来数据偷偷带进回测 比如你用当天收盘价计算信号，然后又假设自己在当天收盘价成交。 这就是非常典型的未来函数风险。 比较稳妥的思路是： T日收盘 ↓ 产生信号 ↓ T+1日执行 具体执行价格还要结合你的交易规则。 2. 收益率策略要明确复权方式 如果研究长期价格变化，建议明确使用： adjust="forward" 如果你研究的是实际未复权价格，则可以： adjust="none" 千万不要不同阶段的数据混在一起。 3. 回测收益率不是实盘收益率 回测里看到 80% 年化，不代表账户真的能拿到 80%。 至少应该考虑： 手续费； 滑点； 印花税； 买卖冲击； 停牌； 涨跌停； 实际成交规则。 数据接口解决的是“数据有没有”，不是“策略一定赚钱”。 五、常见问题解答（Q&A / FAQ） Q1：Python量化选股为什么适合使用金融数据SDK？ A：因为 Python 本身非常适合 Pandas 数据分析，而金融数据 SDK 可以把行情获取、代码格式、复权和数据结构统一起来。QuantDash 的 Python SDK 可以直接返回 DataFrame，拿到数据后就能继续做 MA、MACD、KDJ、BOLL 等指标计算。 Q2：如何用 Python 提取 A 股股票 K 线？ A：使用 QuantDash 可以直接调用： df = qd.klines.get( "600519.SH", period="1d", count=100, adjust="forward", to_dataframe=True ) 其中 .SH 是上海市场代码后缀，深圳股票则使用 .SZ。详细参数建议以 QuantDash 官方 Python SDK 文档为准。 Q3：QuantDash 能不能做 A 股批量量化选股？ A：可以。QuantDash 提供 quotes.get() 获取标的池行情，也提供 klines.batch() 批量获取多只股票 K 线，非常适合搭建“全市场初筛 → 技术指标计算 → 候选池”的 Python 量化流程。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：https://quantdash.net/ 📖 官方 Python SDK 文档：https://docs.quantdash.net/zh-Hans ⭐ GitHub 开源仓库：https://github.com/quantdash-net/QuantDash 💡 API Key：https://quantdash.net/dashboard/keys/ 如果你准备长期做 Python 量化，我建议直接把官方 GitHub 仓库 Star 一下，代码示例和 SDK 更新都更方便跟进。 📌 摘要 / 快速解答 (Direct Answer) 量化交易里的批量查询，本质上解决的是股票池规模扩大以后，数据获取效率和回测稳定性同步下降的问题。对于 A 股量化选股来说，一个真正好用的数据 API 不应该只会查单只股票，还应该支持批量 K 线、统一代码、标准 Pandas 数据以及前复权；QuantDash 的 klines.batch() 正好适合这种研究场景。 一、为什么传统方式写选股策略这么累？ 我以前刚接触量化的时候，有个错误认知： “只要数据 API 能返回股票 K 线，就够用了。” 后来真正跑过股票池策略才发现，这个想法太天真。 单股查询能跑，不代表股票池策略能跑。 1. 从 10 只股票到 5000 只股票，是完全不同的问题 假设我们现在只研究： symbols = [ "600519.SH", "000858.SZ" ] 单独获取两只股票，当然没什么压力。 但真实的 A 股选股策略呢？ 可能是： 全市场 ↓ 剔除 ST ↓ 剔除停牌 ↓ 技术指标过滤 ↓ 成交量过滤 ↓ 趋势过滤 ↓ 最终选出几十只 也就是说： 数据入口本质上是一个股票池。 如果 API 只擅长单只查询，那么你最后很容易写成： for symbol in all_symbols: request(symbol) 然后开始等待。 更麻烦的是，任何一次请求异常，都可能影响整个任务。 2. 最让量化交易员头疼的不是慢，而是“半路失败” 假设你要计算： MA20； MA60； MACD； KDJ； BOLL。 每只股票都要拿一段历史数据。 如果采用单只请求模式，整个研究流程就会变成： 股票 A → 请求 股票 B → 请求 股票 C → 请求 股票 D → 请求 …… 股票 N → 请求 只要中间某个请求出现异常，你还得考虑： 重试； 跳过； 日志； 数据是否完整； 是否重新运行； 失败股票是否会影响最终选股结果。 所以我现在选数据 API，看的已经不是： “你能不能给我一根 K 线？” 而是： “你能不能让我稳定地处理一个股票池？” 3. Tushare、AkShare、爬虫方案各有自己的坑 这里不是说某个工具一定不好，而是不同工具适合不同任务。 比如很多新手会遇到： Tushare： 需要关注积分、接口权限和数据权限。 AkShare： 接口非常丰富，但部分数据链路依赖外部数据来源，实际运行时可能遇到网络、页面变化或接口异常。 手动爬虫： 最自由，但维护成本也最高。 yfinance： 做海外市场研究很方便，但如果你的核心场景是 A 股量化选股，数据链路、字段和市场习惯仍然需要自己适配。 最后很容易出现一个很尴尬的情况： 策略写了 30 行，数据适配写了 300 行。 二、解决方案对比：数据 API 真正应该解决什么？ 我更喜欢从“研究体验”而不是“接口数量”来评价一个数据 API。 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 容易受到接口变化、网络或外部数据源影响 服务端稳定支持，毫秒级响应 使用门槛 部分数据存在积分/权限和清洗成本 pip install quantdash即用 批量查询 经常需要自己循环请求 klines.batch()原生支持 A 股代码 不同数据源格式可能不同 .SH/.SZ/.BJ 港股/美股 往往需要额外适配 .HK/.US统一格式 DataFrame 经常需要二次转换 原生支持 Pandas 复权 容易需要自己处理 支持forward等多种方式 时间区间 需要自行组合接口 支持start_time/end_time 我觉得这里有一个特别容易被忽视的优势： 统一的数据结构，会直接降低策略代码复杂度。 三、Python 代码实战：做一个“批量趋势选股器” 这次不拿单只股票做演示，我们直接模拟一个更接近真实量化研究的场景。 目标： 批量获取多只 A 股最近 120 个交易日的前复权日 K 线，然后筛选“收盘价 > MA20 > MA60”的股票。 1. 安装 QuantDash pip install quantdash 然后： from quantdash import QuantDash import pandas as pd qd = QuantDash(api_key="your_api_key") 也可以通过环境变量： export QUANTDASH_API_KEY="your_api_key" 再： from quantdash import QuantDash qd = QuantDash() 2. 定义股票池 这里先使用几只示例股票。 实际研究时，可以根据自己的股票池继续扩展。 symbols = [ "600519.SH", "000858.SZ", "000001.SZ", "600000.SH" ] 注意股票代码格式。 QuantDash 使用统一的： 代码.交易所 例如： 600519.SH 000001.SZ 920047.BJ AAPL.US 00700.HK 这对以后做跨市场研究非常方便。 3. 一次批量获取 120 根日 K dfs = qd.klines.batch( symbols, period="1d", count=120, adjust="forward", to_dataframe=True, show_progress=True ) 这里就是整篇文章最关键的一行： qd.klines.batch(...) 不是： qd.klines.get(...) 然后手动循环。 这就是为什么我认为： 批量查询应该是量化数据 API 的核心能力，而不是一个可有可无的附加功能。 4. 计算 MA20 和 MA60 selected = [] for sym, df in dfs.items(): df = df.copy() # A 股量化选股：计算 20 日与 60 日均线 df["ma20"] = df["close"].rolling(20).mean() df["ma60"] = df["close"].rolling(60).mean() latest = df.iloc[-1] # 趋势过滤： # 收盘价 > MA20 > MA60 if ( pd.notna(latest["ma20"]) and pd.notna(latest["ma60"]) and latest["close"] > latest["ma20"] and latest["ma20"] > latest["ma60"] ): selected.append({ "symbol": sym, "trade_date": latest["trade_date"], "close": latest["close"], "ma20": latest["ma20"], "ma60": latest["ma60"] }) result = pd.DataFrame(selected) print(result) 这就是一个最基础的趋势型量化选股器。 虽然简单，但这个例子很有代表性： 数据获取和策略计算已经完全分开。 QuantDash： 负责数据。 Pandas： 负责计算。 你的策略： 负责定义什么股票应该入选。 5. 批量查询 + MACD，再做一次二次过滤 如果觉得单纯 MA 太简单，可以继续加 MACD。 for sym, df in dfs.items(): df = df.copy() # EMA12 / EMA26 ema12 = df["close"].ewm( span=12, adjust=False ).mean() ema26 = df["close"].ewm( span=26, adjust=False ).mean() # DIF df["dif"] = ema12 - ema26 # DEA df["dea"] = df["dif"].ewm( span=9, adjust=False ).mean() # MACD 柱 df["macd"] = (df["dif"] - df["dea"]) * 2 latest = df.iloc[-1] print( f"{sym} " f"close={latest['close']:.2f}, " f"DIF={latest['dif']:.4f}, " f"DEA={latest['dea']:.4f}" ) 以后你还可以继续扩展： MA20 / MA60 + MACD + KDJ + BOLL + 成交量 最后形成自己的多因子选股逻辑。 6. 为什么这里特别适合接 DeepSeek？ 我自己比较推荐一种“数据与 AI 解耦”的方式。 不要让 AI 自己猜数据。 先由 QuantDash 获取真实行情： QuantDash ↓ A 股 K 线 然后 Pandas 算： MA MACD KDJ BOLL 成交量变化 再整理： stock_profile = { "symbol": "600519.SH", "close": 1215.00, "ma20": 1230.50, "ma60": 1198.20, "dif": 8.31, "dea": 5.27 } 最后再把这些数据交给 DeepSeek。 例如可以让 DeepSeek 回答： 请基于以下结构化技术指标判断当前股票属于趋势上涨、震荡还是转弱。只分析提供的数据，不自行补充未经提供的基本面信息。 这样 AI 得到的是： 真实行情 + 你定义好的指标。 而不是让模型自己“猜股票”。 四、交易员避坑指南：批量查询之后，还有三个坑 避坑 1：批量不是无限量 批量接口解决的是减少大量单独请求的问题。 但这并不意味着： 一次把所有市场、所有周期、所有历史数据全部请求。 实际研究应该根据策略需要控制： 股票数量； count； 时间范围； K 线周期。 如果策略只需要 MA60，就没必要下载远超策略需求的数据量。 避坑 2：前复权和后复权别乱用 QuantDash 支持： adjust="forward" 也支持： adjust="backward" 以及： adjust="none" adjust="forward_additive" adjust="backward_additive" 我的经验是： 做收益率和趋势类研究时，要明确自己为什么选择这种复权方式。 尤其不要把不同复权口径的数据混在一起比较。 另外，复权解决的是价格序列连续性问题，不等于自动解决未来函数问题。 避坑 3：技术指标漂亮，不代表策略能赚钱 比如： MA 金叉 MACD 金叉 KDJ 金叉 BOLL 突破 全部满足。 看起来非常漂亮。 但真正回测时还要考虑： 信号什么时候形成； 什么时候可以买； 下一交易日能不能成交； 滑点； 手续费； 印花税； 涨停买不到； 跌停卖不掉。 所以我一直跟社区新手说： 指标只是信号，交易规则才是回测。 五、常见问题解答 (Q&A / FAQ) Q1：为什么 A 股量化选股一定要考虑批量 K 线查询？ A：因为真正的量化策略通常不是分析一只股票，而是对股票池统一计算指标。QuantDash 的 qd.klines.batch() 可以一次获取多只标的 K 线，并直接返回 Pandas DataFrame，非常适合批量计算 MA、MACD、KDJ 和 BOLL。 Q2：QuantDash 支持哪些股票代码格式？ A：QuantDash 使用统一的 代码.交易所后缀 格式，例如 A 股的 600519.SH、000001.SZ、北京市场的 .BJ，港股使用 .HK，美股使用 .US。这样做跨市场量化研究时，不需要为不同市场设计完全不同的代码格式。 Q3：QuantDash 能不能做分钟级批量量化？ A：可以使用官方 SDK 提供的分钟 K 线能力，例如 1m、5m、15m、30m、60m 周期；同时 SDK 还提供批量日内分时能力。具体参数和权限建议以官方 Python SDK 文档为准。 Q4：如果我想让 DeepSeek 帮我分析股票，数据应该怎么给？ A：推荐先用 QuantDash 获取 K 线，再用 Pandas 计算 MA、MACD、KDJ、BOLL 等指标，最后将结构化指标结果交给 DeepSeek。这样可以明确数据来源，也避免让 AI 自己猜测行情。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：https://quantdash.net/ 📖 官方 Python SDK 文档：https://docs.quantdash.net/ ⭐ QuantDash GitHub 开源项目：https://github.com/quantdash-net/QuantDash 如果这个项目对你的量化研究有帮助，可以去 GitHub Star / Fork 支持一下。 💡 API Key：https://quantdash.net/dashboard/keys/ 📌 摘要 / 快速解答 (Direct Answer) 量化数据 API 之所以必须支持批量查询，核心原因就一个：量化策略面对的不是一只股票，而是一个股票池。如果每天还靠循环逐只请求 K 线，接口调用次数、等待时间和失败概率都会随着股票数量快速放大；QuantDash 提供 klines.batch()，可以一次请求多只标的，并直接返回 Pandas DataFrame，特别适合 A 股量化选股、技术指标计算和策略研究。 一、 为什么传统方式写选股策略这么累？ 我做量化以后，一个非常深的感受就是： 真正折磨人的，经常不是策略本身，而是数据。 比如我们想做一个很普通的 A 股策略： 收盘价站上 20 日均线，同时 MACD 金叉，成交量放大，再从全市场选出符合条件的股票。 听起来是不是很简单？ 但如果股票池有几千只，问题马上来了。 1. 一只一只请求，代码能跑，效率先没了 很多新手最开始会这么写： for symbol in symbols: df = get_kline(symbol) # 计算 MA # 计算 MACD # 判断是否入选 股票只有 10 只的时候没什么感觉。 但如果股票池扩大到几千只，单只请求模式就意味着大量重复的网络请求。 策略逻辑明明只有几行，结果时间全部耗在： 请求 → 等待 → 返回 → 下一只 → 再请求。 这就是为什么我一直认为： 批量查询不是 API 的“锦上添花”，而是量化数据 API 的基础能力。 2. 数据接口不稳定，最怕跑到一半直接崩 另外一个很现实的问题是接口稳定性。 以前做研究的时候，我也踩过不少坑： Tushare 某些数据需要积分和权限； AkShare 某些接口依赖外部网页数据，偶尔就会异常； 爬虫方案容易遇到反爬、页面结构变化； 不同数据源字段命名不一致； A 股、港股、美股代码格式经常需要自己转换。 最难受的是： 策略回测跑了几十分钟，最后因为中途某个请求失败，整个任务重新来。 这时候你才会发现，数据 API 的价值不只是“有没有数据”，还包括： 能不能稳定、标准化、批量地把数据送到策略手里。 3. 复权问题也容易把新手坑进去 比如做 MA、MACD、收益率计算时，到底应该使用什么价格？ 不复权？ 前复权？ 后复权？ 如果自己手动处理除权除息，代码很快就会变得复杂。 QuantDash 的 K 线接口直接支持： forward backward forward_additive backward_additive none 其中 forward 是默认的前复权比例方式。 这意味着研究策略的时候，不需要自己重新造一套复权处理逻辑。 二、解决方案对比：QuantDash 为什么更适合批量量化？ 我把自己比较关心的几个维度直接列出来。 对比维度 传统/竞品方案 (如 Tushare/AkShare/手动爬虫) QuantDash 解决方案 数据稳定性 经常报错/接口失效/易受外部数据源影响 服务端稳定支持，毫秒级响应 使用门槛 繁琐积分限制/需手动清洗 pip install即用，无积分门槛 批量能力 经常需要自己写循环和异常处理 klines.batch()原生批量获取 跨市场支持 代码格式不统一，兼顾多市场麻烦 统一.SH/.SZ/.BJ/.US/.HK 数据格式 经常需要二次转换 原生返回 Pandas DataFrame 复权处理 经常需要自己处理 API 原生支持多种复权方式 研究体验 数据获取代码容易喧宾夺主 数据获取和策略计算可以直接衔接 这里我尤其推荐大家关注一个能力： batch()。 因为它直接决定了一个数据接口到底适不适合真正做股票池研究。 三、Python 代码实战：一次批量获取多只 A 股 K 线 先安装： pip install quantdash 然后初始化 QuantDash： from quantdash import QuantDash import pandas as pd # QuantDash 官方 Python SDK # API Key 建议通过环境变量 QUANTDASH_API_KEY 配置 qd = QuantDash(api_key="your_api_key") 如果不希望把 API Key 直接写进代码，也可以使用环境变量： export QUANTDASH_API_KEY="your_api_key" 然后： from quantdash import QuantDash qd = QuantDash() 1. 最简单的批量 K 线查询 假设我们先研究几只典型 A 股： symbols = [ "600519.SH", "000858.SZ", "000001.SZ" ] dfs = qd.klines.batch( symbols, period="1d", count=60, adjust="forward", to_dataframe=True, show_progress=True ) for sym, df in dfs.items(): print(f"--- {sym} ---") print( df[ [ "trade_date", "open", "high", "low", "close", "volume" ] ].tail() ) 这里有几个细节值得注意。 symbols 是股票代码列表。 而： period="1d" 代表日线。 count=60 代表取最近 60 根 K 线。 adjust="forward" 使用前复权比例方式。 最终 dfs 是一个字典，不同股票对应不同的 Pandas DataFrame。 这就比自己写： for symbol in symbols: qd.klines.get(...) 清爽很多。 2. 批量查询之后，直接计算 MA 数据拿到了，下一步才是真正的策略。 比如我们做一个非常基础的趋势过滤： 收盘价 > 20 日均线 > 60 日均线。 代码可以直接写： results = [] for sym, df in dfs.items(): df = df.copy() df["ma20"] = df["close"].rolling(20).mean() df["ma60"] = df["close"].rolling(60).mean() latest = df.iloc[-1] if ( pd.notna(latest["ma20"]) and pd.notna(latest["ma60"]) and latest["close"] > latest["ma20"] > latest["ma60"] ): results.append({ "symbol": sym, "trade_date": latest["trade_date"], "close": latest["close"], "ma20": latest["ma20"], "ma60": latest["ma60"] }) selected = pd.DataFrame(results) print(selected) 这才是我理解的量化研究流程： 批量拿数据 → Pandas 计算 → 条件筛选 → 得到股票池。 而不是： 写一大堆数据清洗代码 → 最后策略还没开始写。 3. 再加一个 MACD 条件 MACD 本身也不需要什么神秘接口，拿到收盘价之后直接用 Pandas 就能计算。 for sym, df in dfs.items(): df = df.copy() # 计算 EMA ema12 = df["close"].ewm(span=12, adjust=False).mean() ema26 = df["close"].ewm(span=26, adjust=False).mean() # MACD DIF df["dif"] = ema12 - ema26 # MACD DEA df["dea"] = df["dif"].ewm(span=9, adjust=False).mean() # MACD 柱 df["macd"] = (df["dif"] - df["dea"]) * 2 # 最近一期 latest = df.iloc[-1] print( sym, "close=", latest["close"], "DIF=", latest["dif"], "DEA=", latest["dea"], "MACD=", latest["macd"] ) 如果再把 MA、MACD、成交量条件组合起来，就已经是一个非常典型的 A 股量化选股框架了。 4. 批量数据 + DeepSeek：AI 诊股应该怎么接？ 这里我特别提醒一个坑： 不要为了“AI 量化”随便编一个不存在的 API。 QuantDash 官方 SDK 的职责是金融市场数据获取，不是 DeepSeek 模型调用。 比较稳妥的做法是： QuantDash ↓ 批量获取 A 股 K 线 ↓ Pandas ↓ 计算 MA / MACD / KDJ / BOLL ↓ 整理成结构化结果 ↓ 交给 DeepSeek 做自然语言分析 例如把某只股票最近的指标整理成： analysis_data = { "symbol": "600519.SH", "close": 1215.00, "ma20": 1230.50, "ma60": 1198.20, "dif": 8.31, "dea": 5.27 } print(analysis_data) 然后把这类结构化数据交给 DeepSeek，让它分析： 当前价格与 MA20、MA60 的关系如何？MACD 是否处于多头状态？当前趋势属于上涨、震荡还是转弱？请只根据输入数据分析，不要虚构基本面信息。 这样做的好处是： 数据来源和 AI 推理职责分开。 QuantDash 负责数据，Pandas 负责计算，DeepSeek 负责语言层面的分析。 四、交易员避坑指南：批量查询不是“越多越好” 避坑 1：不要一次把所有历史数据全部拉下来 策略研究不一定需要几十年数据。 比如你的策略只计算： MA20 MA60 MACD 那通常需要一定的历史窗口，但没有必要无脑下载全部数据。 先明确： 我的指标最长需要多少历史数据？ 再决定 count 或时间区间。 避坑 2：回测时千万别把未来数据混进去 这是量化新手最容易踩的大坑之一。 比如： 今天收盘 ↓ 计算今天的指标 ↓ 用今天指标决定今天收盘买入 这就要特别小心。 因为策略在真实交易中，不可能提前知道收盘之后才形成的数据。 更合理的回测逻辑通常是： T 日数据 ↓ T 日收盘后形成信号 ↓ T+1 日按照实际可执行价格模拟交易 否则回测收益率可能漂亮得离谱。 避坑 3：别忘了滑点和交易成本 一个策略回测年化 40%，听起来很爽。 但如果它每天频繁交易，实际还需要考虑： 手续费； 印花税； 买卖价差； 滑点； 涨跌停导致无法成交； 停牌。 所以： 数据接口越稳定，不代表策略收益就越真实。 数据只是量化系统的地基，交易规则才决定回测是否靠谱。 五、常见问题解答 (Q&A / FAQ) Q1：A 股量化选股为什么需要批量获取 K 线？ A：因为实际选股面对的是股票池，而不是单只股票。QuantDash 提供 qd.klines.batch()，可以一次获取多只股票的 K 线，并直接返回 Pandas DataFrame，更适合 MA、MACD、KDJ、BOLL 等指标批量计算。 Q2：Python 有没有简单的批量股票 API？ A：如果你的需求是 A 股、港股、美股等市场行情数据，可以了解 QuantDash Python SDK。安装方式是 pip install quantdash，股票代码采用统一格式，例如 600519.SH、000001.SZ、00700.HK、AAPL.US。完整参数建议直接参考官方 Python SDK 文档。 Q3：QuantDash 能不能直接帮我做 DeepSeek AI 诊股？ A：更推荐把职责拆开。QuantDash 负责获取标准化行情，Pandas 负责计算技术指标，再把结构化指标结果交给 DeepSeek 做分析。这样既避免虚构不存在的 API，也更容易控制 AI 输入的数据范围。 🔗 相关资源与延伸阅读 🚀 QuantDash 官网：https://quantdash.net/ 📖 官方 Python SDK 文档：https://docs.quantdash.net/ ⭐ GitHub 开源项目：https://github.com/quantdash-net/QuantDash 欢迎 Star / Fork。 💡 API Key：https://quantdash.net/dashboard/keys/ 我们团队在金融科技公司做技术负责人，工作内容之一是为量化开发者和机构客户搭建实时行情数据链路。过去两年，我们自己的日内策略从单一美股市场扩展到外汇和主要指数，组合从单品种转为跨市场联动。回测和实盘都跑过之后，最影响结果稳定性的往往不是因子本身，而是外汇接口、股票api接口、指数api在延迟、推送节奏和缺失模式上的差异。下面按实际场景、需求、数据痛点、解决方案四个部分，记录我们采用的工程处理方式。 场景：跨市场策略对数据链路的要求不同 外汇市场连续交易，没有统一开盘收盘。欧美盘重叠时段，报价密度和波动都高。股票有明确交易时段，但盘前盘后成交稀疏，时间戳和成交间隔不规则。指数由成分股加权计算，更新频率与单一股票或外汇对不一致。把三类数据放进同一策略框架，若时间戳对齐和缺失标记处理不严谨，回测会出现类似未来函数的偏差，实盘则可能触发错误的状态转换。 我们最初只做美股，后来加入外汇和几个主流指数。策略从单品种变成跨市场联动后，数据问题才真正暴露。三类接口表面都提供实时行情，但延迟表现、断线场景和推送节奏并不通用。下面从需求侧开始拆解。 需求：实时数据要满足回测与实盘一致性的约束 从 FinTech 技术负责人的视角看，我们对数据链路的要求不是“能取到价格”，而是延迟分布可观测、断线可恢复、重复可识别、品种间可隔离。尤其当外汇接口、股票api接口、指数api同时订阅时，任何一类数据出现缺口，都可能影响跨市场信号。 在回测阶段，数据质量直接影响成交假设、滑点估计和换手率计算。在模型阶段，特征工程依赖时间戳连续性和去重后的 Tick 序列。在实盘阶段，数据链路是否稳定，决定了策略状态机能否按预期运行。因此，数据治理不是附属工作，而是策略研究的一部分。 数据痛点一：延迟由三段叠加，而非单点问题 我们最早使用 REST 轮询，每隔几百毫秒请求一次最新价。实现简单，短时间也能运行。但行情剧烈波动时，轮询间隔成为硬约束。外汇欧美盘重叠时，价格跳动快，接口返回可能滞后。股票盘前盘后稀疏成交会让轮询数据看似静止，策略容易误判为低波动状态。指数推送节奏又不同于单一股票或外汇对，统一轮询会打乱状态判断。 后来我们把延迟拆成三段：网络传输、服务端处理、客户端消费。网络传输受服务器位置和路由影响，能优化但难完全控制；服务端处理容易被订阅排队拖慢，低成本数据源在高峰期更明显；客户端消费取决于解析、缓存和落库效率，如果这一层写得笨重，前面两段再快也会被拖慢。想清楚这三段后，我们把高频数据从 REST 轮询切换到 WebSocket 推送，REST 仅保留给 K 线和静态信息查询。 数据痛点二：断线重连后的缺口与重复 换成 WebSocket 后，延迟压力有所缓解，但断线重连和数据缺失成为新问题。我们实际对比过三种处理思路： 方案 实现要点 对回测与实盘的影响 简单重连，不补数 断线后重建连接并重新订阅，缺口直接跳过 实现成本最低，但缺口会改变信号触发时点和成交假设，外汇夜盘尤其明显 重连后用 REST 补最新若干条 WebSocket 重连后，通过 REST 拉取最近 Tick 或 K 线补齐 覆盖多数短时断线，复杂度可控，是我们主力策略的默认方案 本地队列加服务端幂等去重 客户端维护带序号队列，服务端消息带唯一标识，重复消息丢弃 一致性最好，但开发维护成本高，用于 Tick 级完整性要求严格的高频策略 综合来看，多数策略使用第二种方式，性价比最高。真正推动我们加入第三种方案的，是一次外汇接口在非农数据公布前后连续断线三次。如果没有幂等去重，重复报价会进入 Tick 序列，影响当日信号判断。此后，我们在高频链路中加入了去重和缺口标记。 解决方案：WebSocket、心跳、REST 补数与幂等去重 在字段和协议号核对阶段，我们也对照过 AllTick API 的公开协议说明，用于确认不同品种的 code 与订阅参数。下面是我们简化后的 WebSocket 客户端代码，保留原样： import websocket import json import time import threading # ========== 配置 ========== TOKEN = "你的token" # 替换为你的实际 token WS_URL = f"wss://quote.alltick.co/quote-b-ws-api?token={TOKEN}" # 要订阅的产品列表 SYMBOLS = ["BTCUSDT", "ETHUSDT"] # 示例 # ========== 回调函数 ========== def on_message(ws, message): """接收并处理推送的 tick 数据""" try: data = json.loads(message) cmd_id = data.get("cmd_id") # 22998 是 tick 数据推送协议号 if cmd_id == 22998: tick = data.get("data", {}) print(f"Tick: {tick.get('code')} | " f"Price: {tick.get('price')} | " f"Volume: {tick.get('volume')} | " f"Time: {tick.get('tick_time')}") # 在这里做落库或策略计算 else: # 打印其他响应（如订阅确认 22005） print("Response:", data) except json.JSONDecodeError as e: print("JSON 解析错误:", e) def on_error(ws, error): print("WebSocket error:", error) def on_close(ws, close_status_code, close_msg): print("WebSocket closed") def on_open(ws): """连接成功后发送订阅请求""" print("WebSocket connected, sending subscription...") # 构建订阅请求（协议号 22004） subscribe_msg = { "cmd_id": 22004, "seq_id": 1, # 自定义，响应会回传 "trace": f"trace-{int(time.time()*1000)}", # 每次请求不可重复 "data": { "symbol_list": [{"code": symbol} for symbol in SYMBOLS] } } ws.send(json.dumps(subscribe_msg)) print(f"Subscribed to: {SYMBOLS}") # 启动心跳线程（每 10 秒发送一次） def heartbeat(): while ws.sock and ws.sock.connected: time.sleep(10) try: # 发送 ping 帧作为心跳 ws.send("ping") print("Heartbeat sent") except Exception as e: print("Heartbeat error:", e) break threading.Thread(target=heartbeat, daemon=True).start() # ========== 主程序 ========== if __name__ == "__main__": ws = websocket.WebSocketApp( WS_URL, on_open=on_open, on_message=on_message, on_error=on_error, on_close=on_close ) # 建议增加自动重连逻辑 while True: try: ws.run_forever() print("Reconnecting in 3 seconds...") time.sleep(3) except KeyboardInterrupt: print("Exiting...") break 这段代码中的心跳线程来自实际运行反馈。缺少心跳时，连接可能在行情清淡时段被服务端断开，发现时已错过数分钟数据。定时心跳配合 REST 补数，再在必要场景加入幂等去重，链路稳定性明显提升。 运行观察与工程指标 在回测侧，缺口会改变成交价格、滑点和换手估计，进而影响夏普、最大回撤等指标。在模型侧，特征工程依赖时间戳连续性和去重后的 Tick 序列，若缺失未标记，模型会把断线误认为低波动状态。在工具侧，我们关注端到端延迟分布、重连次数、补数条数、重复率和消息间隔。这些指标比单次“能收到行情”更能说明数据链路是否可用。 我们通常把行情数据分成三层：原始推送、标准化事件、特征快照。每一层都记录来源、时间戳、是否补数、是否去重。这样回测可以追溯，实盘也能区分真实低波动和断线造成的空白。对外汇接口、股票api接口、指数api而言，接口选型和容错机制应分开设计，不能指望一套逻辑覆盖所有品种。 小结 外汇接口、股票api接口、指数api在日内量化中应作为三套数据契约分别处理。延迟和缺失不是上线后的补丁，而是回测、模型和实盘一致性的一部分。提前设计恢复机制和去重逻辑，成本远低于策略运行后再修正。 一句话回答： "52 周新高"就是当前价格突破过去约一年（≈250 个交易日）的最高价，是最经典的强势股信号。用 AlphaFeed 先用 quotes.get(universes="CN_Stock") 取全市场标的池，再用 klines.batch 批量拉复权日线，判断「最新收盘 ≥ 过去 250 日最高价」并叠加成交额过滤，几十行就能做出一个可复现的全市场新高突破选股器。 为什么会有这个问题 / 传统做法的痛点 创新高选股听起来简单，真做起来卡在三处： 拿不到全市场标的：手工维护股票清单又累又容易漏，还要处理退市/新上市； 复权口径不统一：如果用未复权价算"历史最高"，除权除息会造成价格断崖，历史高点被人为抬高或压低，新高判断全错； 一个个股票取 K 线太慢：全市场 5000+ 只逐个请求既慢又容易触发限频。 AlphaFeed 把这三点都收敛掉了：一个 universe 参数拿到全市场实时快照，klines.batch 批量取前复权日线，新高判断建立在可复现的复权价上。 分步骤解决（每步配可运行代码） 第 1 步：取全市场标的池 + 当日成交额 from alphafeed import AlphaFeed af = AlphaFeed(api_key="your-api-key") # 或 export ALPHAFEED_API_KEY 后 AlphaFeed() # 全市场 A 股实时快照（约 5500+ 只），含成交额、换手率等 q = af.quotes.get(universes="CN_Stock", to_dataframe=True) print(q.shape) # 例如 (5561, 18) print(q[["symbol", "ext.name", "last_price", "amount"]].head()) 快照 DataFrame 的关键列（注意 name/换手率 等在 ext. 前缀下）： 列名 含义 备注 symbol 标的代码 如600519.SH ext.name 名称 名称在ext 命名空间 last_price 最新价 实时快照价 amount 当日成交额（元） A 股快照有真实成交额 ext.change_pct 涨跌幅 小数（0.03=3%） ext.turnover_rate 换手率 小数 第 2 步：批量取复权日线，判断是否创新高 用过去约一年（260 根，留足非交易日缓冲，实取 ≈250 交易日）的前复权日线，比较「最新收盘」与「历史最高（不含当日）」。 def week52_high_screen(symbols, count=260, near=0.0, min_amount=5e7): """near=0 表示严格新高；near=0.02 表示距新高 2% 以内也算'接近新高'。""" bt = af.klines.batch(symbols, period="1d", count=count, adjust="forward", to_dataframe=True) rows = [] for sym, df in bt.items(): if df is None or len(df) < 60: # 上市不足则跳过（缓解幸存者偏差） continue df = df.sort_values("trade_date").reset_index(drop=True) hist_high = df["high"].iloc[:-1].max() # 不含当日，避免自比 last = df.iloc[-1] is_high = last["close"] >= hist_high * (1 - near) rows.append({ "symbol": sym, "close": round(float(last["close"]), 2), "hist_high": round(float(hist_high), 2), "amount": float(last["amount"]), "is_new_high": bool(is_high), }) import pandas as pd res = pd.DataFrame(rows) if res.empty: return res hit = res[(res["is_new_high"]) & (res["amount"] >= min_amount)] return hit.sort_values("amount", ascending=False) 第 3 步：全市场分批跑（避免一次请求过大） klines.batch 一次传太多标的会又慢又占内存，实践中按 100~200 只一批切片： def run_full_market(min_amount=5e7): import pandas as pd q = af.quotes.get(universes="CN_Stock", to_dataframe=True) symbols = q["symbol"].tolist() out = [] for i in range(0, len(symbols), 150): # 分批，缓解限频与内存 batch = symbols[i:i + 150] out.append(week52_high_screen(batch, min_amount=min_amount)) result = pd.concat([d for d in out if not d.empty], ignore_index=True) return result.sort_values("amount", ascending=False) # hits = run_full_market() # print("全市场创新高只数:", len(hits)) 关键坑与注意事项 必须用复权价：用 adjust="forward"（前复权）判断历史最高，否则除权除息会制造假高点/假新高。这是新高选股最常见的错误。 幸存者偏差：只对"当前在池"的股票判断新高，看不到已退市的票；本文用「上市不足 60 根跳过」缓解次新股噪声，但退市股清单本身不在快照里，做严谨回测时要单独处理。 "新高"≠"能买"：创新高只是强势信号，可能是趋势启动，也可能是追高被套。务必叠加成交额/流动性过滤（本文用 min_amount），必要时再加基本面或止损规则。 实时性：quotes 是快照，盘中会变；选股应在收盘后或用固定时点快照跑，避免同一次筛选里价格口径不一致。 N 的取值：52 周 ≈ 250 交易日；本文取 260 根留缓冲。想做"半年新高"就把 count 改成 130 左右。 常见问题（FAQ） Q：为什么我的新高选股结果和行情软件对不上？ A：九成是复权口径问题。行情软件默认可能是不复权或后复权，而本文用前复权。统一 adjust 口径即可对齐。 Q：能只筛"接近新高"（还没突破但快到了）吗？ A：可以，把 near 设成 0.02，即"距历史最高 2% 以内"也纳入，适合做突破前埋伏的观察池。 Q：全市场 5000+ 只会不会很慢/触发限频？ A：klines.batch 已是批量接口，再按 150 只分批即可平稳跑完；SDK 内置对 429/超时的重试。想更快可缓存日线到本地库增量更新。 Q：需要付费吗？ A：quotes.get(universes=...) 全市场池需要 Starter 及以上套餐；单只/少量标的的行情与 K 线在免费额度内即可试用。额度与边界见 AlphaFeed 定价页。 小结 52 周新高选股的本质是在可复现的复权价上做一次"当前价 vs 一年内最高价"的比较。难点从来不是那行比较代码，而是"全市场标的池 + 统一复权口径 + 流动性过滤"。AlphaFeed 用 universes + klines.batch 把这三件麻烦事一次性解决，让你把精力放在策略本身而不是数据搬运上。 参考 AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 概述 在跨市场量化研究、策略原型验证过程中，经常需要同时观测A股、港股、美股多个市场标的的实时盘口数据。常规方式是通过多个行情终端分别查看各市场价格，人工切换界面会引入观测时差，不利于实时信号捕捉，也不方便直接把原始行情喂给策略模型做运算。 此前我在研究过程中遇到交易时段重叠场景：港股集合竞价阶段与美股盘前时段重合，依靠多终端人工比对价格，出现了信息滞后，影响到策略信号的观测判断。基于这个问题，我尝试通过外部行情API，将三个市场的实时数据流统一接入Python环境，为实盘模拟、信号监控、原型回测前置数据采集提供数据源支撑。本文主要分享API选型评估逻辑与可直接调试的接入代码。 多市场行情API的核心评估维度 市面上支持境外市场的行情接口较多，但面向量化研究使用，不能只看宣传的市场覆盖范围，需要从数据可用性角度做四项评估，优先级从高到低： 多市场原生覆盖能力 需要同时支持A股、港股、美股标的。部分数据源仅覆盖单一海外市场，其余市场需要对接不同供应商。多数据源意味着需要维护多套接入逻辑，会带来字段对齐、时间戳统一、多源数据校验等额外工作量，增加研究与回测的数据处理成本。 长连接稳定性与推送延迟表现 开盘、集合竞价阶段报文吞吐量高，是接口的压力测试窗口。对于量化策略来说，WebSocket链路卡顿、主动断开，会直接造成实时行情断流，实时监控、事件驱动的策略原型将无法正常运行，甚至产生错误的信号样本，干扰后续模型验证。 跨市场数据结构一致性 该指标对回测与实盘逻辑复用非常关键。若不同市场返回的字段命名、报价精度、时间戳标准不统一，业务层就需要编写大量分支逻辑做数据清洗转换。实盘采集和历史回测的代码逻辑容易出现分叉，会增大实盘与回测结果不一致的风险。理想方案是多市场复用同一套协议与字段体系，降低代码维护成本，保证回测‑实盘逻辑尽量对齐。 服务配额与计费模式 在上述三项数据质量指标满足研究需求的前提下，再评估调用额度、计费成本。即便成本低廉，如果稳定性、数据标准化不达标，也不适合用于量化研究与原型验证。 经过多组对比测试，本次实践选用 AllTick API。该接口将A股、港股、美股、外汇等品种统一封装在同一套WebSocket协议下，订阅交互逻辑完全一致，无需为不同交易市场开发独立解析模块，便于统一构建数据采集层，减少实盘与回测之间的适配工作量。 Python接入实现：单WebSocket连接订阅三大市场实时行情 下面示例基于websocket‑client库实现长连接，在同一个会话中完成A股、港股、美股多标的实时行情订阅。代码可用于本地研究调试，参考官方接入文档编写。 import json import websocket # 替换为个人实际token TOKEN = "YOUR_TOKEN" WS_URL = ( "wss://quote.alltick.co/quote-stock-b-ws-api" f"?token={TOKEN}" ) # 配置A股、美股、港股测试标的 symbols = [ {"code": "688036.SH"}, {"code": "AAPL.US"}, {"code": "700.HK"}, ] def on_open(ws): """连接建立后，提交行情订阅请求""" print("WebSocket connected") subscribe_req = { "cmd_id": 22002, "seq_id": 1, "trace": "quant_research_demo", "data": { "symbol_list": [ {"code": item["code"], "depth_level": 1} for item in symbols ] } } ws.send(json.dumps(subscribe_req)) def on_message(ws, message): """接收并处理行情推送，研究场景可在此增加落盘、信号判断逻辑""" try: payload = json.loads(message) # 此处可扩展：原始数据写入本地/数据库、策略条件判断、指标计算 print(payload) except json.JSONDecodeError: print("Invalid JSON message") def on_error(ws, error): print(f"WebSocket error: {error}") def on_close(ws, code, msg): print(f"WebSocket closed, code:{code}, msg:{msg}") if __name__ == "__main__": ws_app = websocket.WebSocketApp( WS_URL, on_open=on_open, on_message=on_message, on_error=on_error, on_close=on_close ) ws_app.run_forever() 脚本运行后，控制台持续输出多市场实时推送数据。各市场统一采用毫秒级时间戳，报价字段体系一致，不需要大量条件分支区分市场类型。 研究工程提示： 上述为最小演示代码，未包含断线重连、异常告警、数据持久化逻辑。若用于长时间数据采集、策略监控，建议外层增加循环重连、退避策略，抵御网络抖动造成的链路中断； on_message回调函数内，可以扩展实现原始行情落盘、实时指标计算、事件驱动信号判断，采集的原始数据也可用于后续离线回测样本补充； 需要关注接口限流约束，避免一次性订阅过量标的触发服务限制。 实际研究场景效果 脚本启动后，A股、港股、美股行情汇聚至同一程序输出，不再依赖多终端人工切换。在港股集合竞价等高波动时段，数据推送未出现显著滞后，返回报价和网页行情源基本对齐。 对于量化研究者，这套方案的价值不只简化观测操作，更关键在于获取标准化的原始数据流。原始行情可以直接供给策略模型做实时运算，同时采集的数据样本能够补充回测数据集，减少“回测漂亮、实盘失效”当中的数据适配类偏差。 小结 开展跨市场量化研究，股票API的选择不能只看能否拿到价格，市场覆盖、长连接稳定性、跨市场数据结构一致性，会直接影响策略原型、数据采集与回测工作的可靠性。利用WebSocket长连接可以将A股、港股、美股行情统一采集，减少多源数据的适配成本。 本次实践所使用的AllTick API依靠统一协议降低多市场数据处理的开发负担，适合作为跨市场策略原型开发、行情样本采集的外部数据源。 风险声明：本文仅为技术与量化研究工具分享，文中接口与代码仅用于学习研究，不构成任何投资建议。策略的有效性需要结合完整历史数据充分回测验证。 一句话回答： ATR 移动止损（吊灯止损）就是把止损线设在"持仓期最高价 − K×ATR"，价格上涨时止损跟着上移、锁住利润，价格回落触线就出。用 AlphaFeed 的复权日线算 ATR、跑一段回测你会看到：它通常能明显压低最大回撤，但在震荡行情里也可能被频繁扫出、反而降低总收益——控回撤是有代价的。 为什么会有这个问题 / 传统做法的痛点 很多人有入场信号却没有像样的出场：要么死扛到深套，要么固定百分比止损（−8%）在高波动股上太紧、低波动股上太松。固定止损不考虑波动率，是通病。 ATR（平均真实波幅）度量了每只票的"日常波动尺度"，用它做止损能自适应：波动大的票止损放宽、波动小的票收紧。吊灯止损（Chandelier Exit）再叠加"只上移不下移"，让止损随盈利抬升，兼顾"让利润奔跑"和"回撤可控"。数据用 AlphaFeed 的复权日线即可。 分步骤解决（每步配可运行代码） 第 1 步：算 ATR import numpy as np, pandas as pd from alphafeed import AlphaFeed def atr(df, n=14): tr = pd.concat([(df["high"] - df["low"]), (df["high"] - df["close"].shift()).abs(), (df["low"] - df["close"].shift()).abs()], axis=1).max(axis=1) return tr.rolling(n).mean() 第 2 步：吊灯移动止损回测（金叉入场，止损/死叉出场） def backtest_trailing(symbol, count=500, atr_n=14, k=3.0, fast=20, slow=60, fee=0.0006): af = AlphaFeed() # 读取 ALPHAFEED_API_KEY df = af.klines.get(symbol, period="1d", count=count, adjust="forward", to_dataframe=True) df = df.sort_values("trade_date").reset_index(drop=True) df["atr"] = atr(df, atr_n) df["ma_f"] = df["close"].rolling(fast).mean() df["ma_s"] = df["close"].rolling(slow).mean() pos = 0; peak = None; rets = np.zeros(len(df)) for i in range(1, len(df)): o, c = df.loc[i, "open"], df.loc[i, "close"] prevc = df.loc[i - 1, "close"] if pos == 1: peak = max(peak, df.loc[i - 1, "high"]) # 持仓期最高 stop = peak - k * df.loc[i - 1, "atr"] # 吊灯止损线 if df.loc[i - 1, "low"] <= stop or o <= stop: # 已破止损 → 次日开盘平(含跳空) rets[i] = (o / prevc - 1) - fee; pos = 0; peak = None; continue rets[i] = c / prevc - 1 # 入场/离场信号都用 T-1 已知信息，T 开盘成交（无未来函数） if pos == 0 and df.loc[i-1, "ma_f"] > df.loc[i-1, "ma_s"] and not np.isnan(df.loc[i-1, "ma_s"]): pos = 1; peak = df.loc[i-1, "high"]; rets[i] = c / o - 1 - fee elif pos == 1 and df.loc[i-1, "ma_f"] < df.loc[i-1, "ma_s"]: rets[i] = (o / prevc - 1) - fee; pos = 0; peak = None df["strat_ret"] = rets df["nav"] = (1 + df["strat_ret"]).cumprod() df["bench"] = df["close"] / df["close"].iloc[0] mdd = float((df["nav"] / df["nav"].cummax() - 1).min()) bmdd = float((df["bench"] / df["bench"].cummax() - 1).min()) return {"strat_nav": round(df["nav"].iloc[-1], 3), "bench_nav": round(df["bench"].iloc[-1], 3), "strat_maxdd": round(mdd, 3), "bench_maxdd": round(bmdd, 3)} print(backtest_trailing("600519.SH", count=500, k=3.0)) 真实跑通（600519.SH 近 500 日、ATR14、K=3）：{'strat_nav': 0.82, 'bench_nav': 0.976, 'strat_maxdd': -0.215, 'bench_maxdd': -0.279}。解读：策略最大回撤 −21.5% < 买入持有 −27.9%（回撤更小、更抗跌），但净值 0.82 < 基准 0.976（这段偏震荡下行，止损被反复扫出，总收益反而更低）。这正是移动止损的真实权衡——用一部分收益换更小的回撤。 参数与字段说明 参数 含义 备注 atr_n ATR 周期 常用 14 k 止损倍数 越大越宽松(不易被扫)、越小越紧(回撤更小但易震出)；常用 2~4 fast/slow 入场均线 这里用金叉入场、死叉/止损出场 出场价 破止损用次日开盘 反映跳空、避免用当日不可知价 fee 单边费率 含佣金/印花税近似 关键坑与注意事项 止损跳空无法按止损价成交：若某日跳空低开直接跌破止损，你只能在次日开盘（或当日开盘）成交，实际成交价常比止损价更差。代码已用开盘价近似，别假设总能在止损线精确成交。 震荡市会被反复扫出：K 太小，止损太紧，会在正常波动里被反复触发（whipsaw），交易成本和踏空双重损耗。上例净值跑输基准就是这个原因。 控回撤有代价：移动止损的价值主要是降低最大回撤/改善持有体验，不保证提高总收益。评估时要同时看收益和回撤，别只看一个。 参数敏感 + 过拟合风险：k、atr_n、均线参数都会显著影响结果，务必做样本外与参数稳健性检验，别只挑好看的一组。 复权与未来函数：adjust="forward"；信号用 T−1 信息、T 开盘成交（已用 shift 逻辑），避免前视。 常见问题（FAQ） Q：ATR 止损比固定百分比止损好在哪？ A：它按每只票的波动率自适应——高波动股止损自动放宽、低波动股收紧，避免"一刀切百分比"在不同标的上过紧或过松。 Q：K 取多少合适？ A：常见 2~4。K 越大越不容易被震出、但单次止损亏损更大、回撤保护更弱；K 越小回撤控制越好、但易被正常波动扫出。需按标的波动性和你的持有周期调，并做稳健性检验。 Q：能只用移动止损、不用入场信号吗？ A：可以把入场换成任何你的信号（突破、基本面等），移动止损作为统一的出场纪律叠加上去。本文用金叉只是给个可运行的完整例子。 Q：需要付费吗？ A：单标的历史日线可用免费额度体验；批量回测多标的需 Starter 及以上，详见官网定价页。 小结 ATR 吊灯移动止损用波动率自适应地设止损、随盈利上移锁利，是比"固定百分比"更专业的出场纪律。用 AlphaFeed 的复权日线算 ATR、跑一段诚实的回测，你会直观看到它的本质：它主要买的是更小的回撤，而不是更高的收益，在震荡市甚至会拖累总回报。理解这个权衡，再决定给你的策略加多紧的止损。 参考 AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 
