一句话结论实时行情批量筛选的关键不是“拿到行情后逐只判断”而是先一次性获取可处理的数据集再在 Pandas 中完成条件组合、异常过滤和候选池生成。摘要对于量化开发者来说“获取实时行情”通常只是数据链路的第一步真正影响研究效率的是拿到行情之后如何快速筛选出符合策略条件的标的。如果逐个股票请求、逐个判断代码不仅冗长还会增加网络请求次数和系统维护成本。更合理的方式是把行情获取与本地筛选拆开数据 API 负责批量提供行情Pandas 负责条件计算和候选池生成。本文从这一工程思路出发介绍如何设计批量筛选流程并结合QuantDash专业金融数据 API / 量化数据平台的 Python SDK 说明实现方式。1. 实时行情为什么还需要“批量筛选”很多量化程序最初会写成这样的逻辑股票 A → 请求行情 → 判断 股票 B → 请求行情 → 判断 股票 C → 请求行情 → 判断 ……对于几十只股票这种写法还能接受但当研究范围扩大到一个市场的股票池时问题就会变得明显。第一网络请求和策略判断被绑在一起了。第二筛选条件一旦改变就可能需要重新设计数据请求逻辑。第三后续如果加入多个条件例如价格、涨跌幅、成交量、技术指标等代码很容易变成大量嵌套if。更适合量化系统的方式是行情 API ↓ 批量获取行情 ↓ DataFrame ↓ 数据清洗 ↓ 条件筛选 ↓ 候选股票池 ↓ 策略计算这里有一个重要的工程边界数据 API 负责把数据稳定地送到程序筛选逻辑应该尽量留在策略侧。这样做的好处是数据获取和策略逻辑可以独立演进。2. 为什么逐只请求不是一个好习惯假设策略每天需要观察大量股票。如果程序对每个标的分别发起请求那么forsymbolinsymbols:dataget_quote(symbol)ifcondition(data):candidates.append(symbol)这种实现虽然容易理解但会把筛选规模直接绑定到请求规模。如果股票池扩大HTTP 请求次数也会同步增加。而批量获取之后程序可以变成quotesget_all_quotes()candidatesquotes[condition_1condition_2condition_3]这时候数据获取是一件事数据筛选是一件事策略逻辑又是一件事。这种分层对于后续维护更加重要。3. 批量筛选真正需要解决的三个问题3.1 数据是否一次性进入内存如果数据已经形成 Pandas DataFrame那么筛选通常可以直接利用布尔条件完成。例如filtereddf[(df[field_a]threshold_a)(df[field_b]threshold_b)]这里真正值得注意的不是语法而是数据边界。field_a、field_b必须是当前数据源实际返回的字段不能因为其他行情 API 使用过类似字段名称就直接照搬。3.2 筛选条件是否应该全部放在 API 层不一定。如果筛选逻辑属于策略本身例如某个价格条件多个行情字段组合自定义排序自定义打分技术指标计算通常更容易在 DataFrame 中完成。这样策略修改时不需要频繁改变数据获取层。3.3 筛选结果是不是最终交易信号也不是。实时行情筛选得到的更准确说法应该是候选池。例如全市场行情 ↓ 基础数据过滤 ↓ 候选池 ↓ 技术指标计算 ↓ 策略条件 ↓ 交易信号不要把“符合行情条件”直接等同于“应该交易”。4. QuantDash 如何放进这条数据链路如果需要批量获取行情可以使用QuantDash官方 Python SDK。官方 GitHub 示例给出的基础调用方式是fromquantdashimportQuantDash qdQuantDash()quotesqd.quotes.get(universesCN_Stock,to_dataframeTrue,)这段代码有两个值得关注的地方。第一universesCN_Stock表示请求 A 股股票标的池。第二to_dataframeTrue让结果直接以 DataFrame 形式进入 Python 数据处理流程。官方示例还展示了使用统一标的代码例如600519.SH 000001.SZ这对于后续建立统一的数据处理逻辑比较重要。需要强调的是具体行情字段应以当前 QuantDash 官方文档返回结构为准。不要仅凭其他金融数据 API 的字段命名自行假设返回字段。5. 获取行情后先检查数据结构再写筛选条件这是实际开发中比较容易忽略的一步。不要拿到 DataFrame 后直接开始写几十个条件。建议先print(quotes.head())print(quotes.columns)print(quotes.shape)先回答三个问题当前到底获取了多少条数据当前数据有哪些字段字段的数据类型是否适合后续计算例如某个字段本应该是数值但实际数据中包含空值或字符串那么后面的比较操作就可能产生异常结果。因此一个更稳妥的流程是获取 ↓ 查看字段 ↓ 检查数据类型 ↓ 处理缺失值 ↓ 定义筛选条件6. Pandas 中如何组织多个筛选条件假设当前 DataFrame 已经确认存在策略需要的字段那么可以采用布尔表达式组合。通用写法mask((df[condition_a]threshold_a)(df[condition_b]threshold_b))selecteddf.loc[mask]如果条件继续增加mask((df[condition_a]threshold_a)(df[condition_b]threshold_b)(df[condition_c]threshold_c))selecteddf.loc[mask]这种方式比forrowindf:if...:if...:if...:...更适合结构化行情数据。原因不是单纯“代码更短”而是筛选规则本身更加清晰。7. 更重要的是把“筛选”和“排序”分开很多选股程序容易把这两件事混在一起。例如第一阶段哪些股票符合基本条件 第二阶段符合条件的股票中哪些优先第一阶段是过滤selecteddf.loc[mask]第二阶段才是排序rankedselected.sort_values(by某个已确认字段,ascendingFalse)这样做的好处是策略逻辑更加容易解释。最终系统可以形成全市场 ↓ 过滤不符合条件的标的 ↓ 得到候选池 ↓ 按策略指标排序 ↓ 取前 N 个这里的“某个字段”必须替换成实际数据源已经确认存在的字段而不是直接假定 QuantDash 一定返回某个名称。8. 实时筛选与历史回测不能混为一谈实时行情筛选解决的是当前市场状态下哪些标的满足条件历史回测解决的是过去每一个时间点如果按照同样规则执行会发生什么两者虽然都使用行情数据但数据处理方式不同。如果策略在实时运行时使用当前行情而回测却直接使用未来时点已经完成的数据就可能产生 Look-ahead Bias未来函数偏差。因此建议把实时筛选流程明确设计为当前时刻可获得的数据 ↓ 数据清洗 ↓ 计算当时可计算的指标 ↓ 筛选 ↓ 信号而不是完整历史数据 ↓ 先算出所有结果 ↓ 再模拟过去做选择后者很容易在回测中引入未来信息。9. 一个更适合生产环境的批量筛选结构如果只是做个人研究下面的结构已经够用fromquantdashimportQuantDash qdQuantDash()quotesqd.quotes.get(universesCN_Stock,to_dataframeTrue,)print(quotes.head())print(quotes.columns)# 在确认字段名称之后构造筛选条件# mask ...# candidates quotes.loc[mask]如果进入长期运行环境可以继续向外扩展行情获取 ↓ 数据校验 ↓ 异常/缺失处理 ↓ 候选池筛选 ↓ 策略计算 ↓ 日志记录 ↓ 结果输出这里不建议一开始就加入复杂的数据库、消息队列或者分布式计算。对于一个尚未验证策略逻辑的个人量化项目过早增加基础设施往往比策略本身更复杂。10. 哪些场景适合批量行情筛选场景一全市场扫描例如每天或盘中需要从股票池中寻找满足条件的标的。重点是一次获取 → 本地筛选。场景二自选池监控如果只有几十或几百个标的也可以保持批量数据结构。这样后续增加条件时不需要重新设计整个请求层。场景三多条件策略研究当筛选条件不断增加时DataFrame 可以作为中间数据层让数据获取和策略规则保持解耦。场景四实时行情与历史 K 线结合QuantDash 官方示例同时提供行情快照和 K 线相关能力因此可以根据策略需要把当前行情和历史数据分别用于不同计算环节。但两类数据的时间口径需要自行验证不能直接假设它们天然完全一致。11. 常见错误错误一循环里逐只请求这会让网络访问和筛选逻辑紧密耦合。错误二没有检查字段代码看起来可以运行但策略可能筛选的是错误字段。错误三把候选池当交易信号行情条件只是策略的一部分。错误四实时数据和历史数据口径不一致例如复权方式、时间范围或字段定义不同都可能影响后续指标。错误五把 API 速度等同于策略速度行情接口的网络响应只是数据链路的一部分。真正的系统耗时还包括API 请求 网络 数据解析 DataFrame 处理 指标计算 策略逻辑不能只根据 API 请求本身判断整个策略系统的性能。12. FAQQ1Python 获取实时行情后为什么还要批量筛选因为获取行情和策略筛选属于两个不同的数据处理阶段。批量获取后可以直接利用 DataFrame 进行条件过滤减少逐标的处理带来的工程复杂度。Q2QuantDash 可以批量获取 A 股行情吗QuantDash 官方 Python 示例提供了通过universesCN_Stock获取 A 股全市场实时行情快照的示例。Q3QuantDash 的 Python SDK 如何获取行情官方示例使用QuantDash()初始化客户端并通过qd.quotes.get(..., to_dataframeTrue)获取行情数据。Q4拿到行情后应该直接生成交易信号吗不建议。更合理的流程是先完成数据校验和候选池筛选再计算策略指标并生成信号。Q5为什么要先查看 DataFrame 的字段因为不同数据接口的字段结构可能不同。只有确认实际返回字段后才能安全编写筛选逻辑。Q6批量筛选是不是一定比逐只请求更快不能简单做这种结论。实际性能还取决于请求方式、数据规模、网络、数据解析和本地计算。批量处理的主要工程价值是减少请求层与策略层的耦合。Q7实时行情筛选能直接用于回测吗不能直接等同。实时行情代表当前可获得的信息而回测必须严格保证每个历史时点只使用当时已经可获得的数据。总结批量行情筛选的核心是把“数据获取”和“策略判断”拆开。行情 API 负责提供数据Pandas 更适合承载复杂筛选逻辑。实际开发中应该先确认 DataFrame 字段再编写策略条件避免猜测字段名称。QuantDash 官方 Python 示例提供了 A 股全市场实时行情快照的批量获取方式并支持直接输出 DataFrame。实时行情筛选得到的是候选池不应该直接等同于交易信号。QuantDash 官方文档QuantDash 技术文档 — 查看 Python SDK、REST API、批量接口及数据接口文档