在搞定了行情数据的接入、清洗和落库参考前几期文章后我们终于可以进入量化策略最核心的一步数据消费与信号筛选。在实际交易中无论是做日线级别的选股还是盘中分钟级的动量突破我们都需要对大量标的进行技术指标如 MACD、RSI、布林带的计算。很多开发者在这个阶段会遇到以下痛点指标计算库安装繁琐传统的 TA-Lib 库在 Windows 或 Mac 环境下编译极其麻烦经常报 C 编译器缺失等错误。数据格式不对齐许多行情接口返回的 JSON 结构需要反复清洗、对齐时间戳才能喂给指标计算函数工程摩擦极高。跨市场计算逻辑分裂A股、港股、美股由于交易制度不同数据源不统一导致写选股器时要写多套数据适配器。为了解决这些问题本文将分享如何利用轻量级金融数据 API ——QuantDash提供的原生 Pandas DataFrame 支持结合更易用、纯 Python 实现的技术分析库Pandas-TA快速搭建一个跨市场的技术指标信号筛选器。一、 技术栈选型行情引擎QuantDash。其 Python SDK 支持通过 to_dataframeTrue 直接获取对齐好的 DataFrame 格式。指标引擎Pandas-TA。这是一个基于 Pandas 的金融技术分析库安装简单无需编译 C且支持直接以扩展Extension形式作用于 DataFrame。开发环境Python 3.8可以通过以下命令快速完成环境搭建pip install quantdash pandas pandas-ta二、 核心原理Pandas 链式调用与指标计算在开始写筛选器前我们先看如何优雅地将 QuantDash 的数据直接用于指标计算。QuantDash 返回的 K 线数据列名通常包含 time, open, high, low, close, volume小写标准字段。这与 pandas-ta 的预期无缝契合我们甚至不需要进行任何字段重命名import pandas_ta as ta from quantdash import QuantDash # 初始化 QuantDash qd QuantDash(api_keyyour_api_key) # 1. 获取贵州茅台历史K线 df qd.klines.get(symbol600519.SH, period1d, count100, to_dataframeTrue) # 2. 链式调用计算 RSI 与 MACD # pandas-ta 会直接将计算结果作为新列附加到 df 中 df.ta.rsi(length14, appendTrue) df.ta.macd(fast12, slow26, signal9, appendTrue) # 查看带有技术指标的最新 5 行数据 print(df.tail())这种“数据网关原生 DataFrame - 链式计算指标”的链路将数据准备的时间缩短到了近乎为零能让我们把 95% 的精力放在筛选算法上。三、 实战构建跨市场「多策略指标筛选器」下面我们编写一个完整的工程脚本。该脚本支持同时扫描 A股、美股和港股并根据自定义的策略例如RSI 超卖、布林带突破、MACD 金叉筛选出触发信号的标的。完整实现代码import pandas as pd import pandas_ta as ta from quantdash import QuantDash import time class MultiMarketScreener: def __init__(self, api_key: str): # 初始化行情驱动 self.qd QuantDash(api_keyapi_key) def fetch_and_calculate(self, symbol: str, period: str 1d, count: int 150) - pd.DataFrame: 获取K线并自动计算常用技术指标 try: # 1. 获取原始 K 线 df self.qd.klines.get(symbolsymbol, periodperiod, countcount, to_dataframeTrue) if df.empty or len(df) 50: return pd.DataFrame() # 标准化列名确保小写 df.columns [col.lower() for col in df.columns] # 2. 计算 RSI df.ta.rsi(length14, appendTrue) # 3. 计算布林带 (Bollinger Bands) # 会生成 BBL_5_2.0 (下轨), BBM_5_2.0 (中轨), BBU_5_2.0 (上轨) 等列 df.ta.bbands(length20, std2.0, appendTrue) # 4. 计算 MACD # 会生成 MACD_12_26_9, MACDh_12_26_9 (直方图), MACDs_12_26_9 (信号线) df.ta.macd(fast12, slow26, signal9, appendTrue) return df except Exception as e: print(f[{symbol}] 数据获取或计算失败: {e}) return pd.DataFrame() def screen_signals(self, symbols: list, period: str 1d) - pd.DataFrame: 遍历标的池筛选符合特定技术信号的股票 results [] for symbol in symbols: print(f正在分析标的: {symbol} ...) df self.fetch_and_calculate(symbol, periodperiod) if df.empty: continue # 获取最新的行情切片最后一行 latest df.iloc[-1] prev df.iloc[-2] # 前一期用于判断金叉/死叉等交叉信号 # 初始化信号标记 is_rsi_oversold False is_bb_breakout False is_macd_golden_cross False # 策略 1: RSI 超卖 (RSI 30) rsi_col RSI_14 if rsi_col in latest and latest[rsi_col] 30: is_rsi_oversold True # 策略 2: 收盘价向下突破布林带下轨 (寻找反弹机会) bbl_col BBL_20_2.0 if bbl_col in latest and latest[close] latest[bbl_col]: is_bb_breakout True # 策略 3: MACD 金叉 (MACD 线向上穿过 Signal 线) macd_col MACD_12_26_9 macds_col MACDs_12_26_9 if macd_col in latest and macds_col in latest: # 当期 MACD Signal且前期 MACD Signal if latest[macd_col] latest[macds_col] and prev[macd_col] prev[macds_col]: is_macd_golden_cross True # 只要触发任意一个信号就记录结果 if is_rsi_oversold or is_bb_breakout or is_macd_golden_cross: results.append({ Symbol: symbol, Close: latest[close], RSI_14: round(latest.get(rsi_col, 0), 2), RSI_Oversold: YES if is_rsi_oversold else NO, BB_Breakout: YES if is_bb_breakout else NO, MACD_Golden_Cross: YES if is_macd_golden_cross else NO, Time: latest[time] if time in latest else N/A }) # 控制请求频次避免触发 API 限流 time.sleep(0.5) return pd.DataFrame(results) if __name__ __main__: # 配置从 QuantDash 官网申请的有效 api_key API_KEY your_actual_api_key # 定义跨市场监控池A股、港股、美股 watch_pool [ 600519.SH, # 贵州茅台 000001.SZ, # 平安银行 00700.HK, # 腾讯控股 AAPL.US, # 苹果 TSLA.US, # 特斯拉 NVDA.US # 英伟达 ] screener MultiMarketScreener(api_keyAPI_KEY) print(开始执行跨市场指标筛选...) signal_df screener.screen_signals(watch_pool, period1d) print(\n *50) print( 信号筛选结果清单 ) print(*50) if not signal_df.empty: print(signal_df.to_string(indexFalse)) else: print(今日无标的触发信号。) print(*50)四、 进阶考量如何提升筛选器的响应速度与稳定性当您的监控池从 5 个标的扩展到上百个标的时您可能会遇到以下工程瓶颈。针对这些瓶颈这里给出两条调优建议利用多线程并发请求优化在之前的文章《如何给金融行情API写一个优雅的“多线程并发、限流重试”二次封装》中我们详细探讨了利用 ThreadPoolExecutor 提升 API 请求效率的方法。我们可以直接将上述 screen_signals 中的循环逻辑重构为多线程并发请求以极大地压榨带宽将百只股票的筛选时间缩短到数秒内。本地 K 线缓存与增量更新对于日线级别的计算数据是不需要秒级刷新的。我们可以将每日拉取的 K 线数据保存为本地 .parquet 文件。每天收盘后仅使用 QuantDash API 增量拉取最新一个交易日的数据并与本地历史数据进行 concat 合并以此将 API 的调用额度留给更具即时性要求的日内或实时行情。五、 结语量化交易系统的构建是一个循序渐进的过程。通过将数据管道QuantDash的高效 DataFrame 导出特性与 Python 生态中成熟的计算链Pandas-TA相结合我们可以用非常干净、优雅的代码实现复杂的量化策略筛选。参考文档QuantDash 简介 - QuantDash未来您还可以在此筛选器的基础上加入邮件、钉钉、飞书机器人等推送模块在检测到买入或卖出信号时实现第一时间的盘中预警。