企业官网建设流程全解析

在搞定了行情数据的接入、清洗和落库（参考前几期文章）后，我们终于可以进入量化策略最核心的一步：数据消费与信号筛选。

在实际交易中，无论是做日线级别的选股，还是盘中分钟级的动量突破，我们都需要对大量标的进行技术指标（如 MACD、RSI、布林带）的计算。很多开发者在这个阶段会遇到以下痛点：

1. 指标计算库安装繁琐：传统的 TA-Lib 库在 Windows 或 Mac 环境下编译极其麻烦，经常报 C++ 编译器缺失等错误。

2. 数据格式不对齐：许多行情接口返回的 JSON 结构需要反复清洗、对齐时间戳，才能喂给指标计算函数，工程摩擦极高。

3. 跨市场计算逻辑分裂：A股、港股、美股由于交易制度不同，数据源不统一，导致写选股器时要写多套数据适配器。

为了解决这些问题，本文将分享如何利用轻量级金融数据 API ——QuantDash提供的原生 Pandas DataFrame 支持，结合更易用、纯 Python 实现的技术分析库Pandas-TA，快速搭建一个跨市场的技术指标信号筛选器。

一、 技术栈选型

• 行情引擎：QuantDash。其 Python SDK 支持通过 to_dataframe=True 直接获取对齐好的 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_key="your_api_key") # 1. 获取贵州茅台历史K线 df = qd.klines.get(symbol="600519.SH", period="1d", count=100, to_dataframe=True) # 2. 链式调用计算 RSI 与 MACD # pandas-ta 会直接将计算结果作为新列附加到 df 中 df.ta.rsi(length=14, append=True) df.ta.macd(fast=12, slow=26, signal=9, append=True) # 查看带有技术指标的最新 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_key=api_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(symbol=symbol, period=period, count=count, to_dataframe=True) if df.empty or len(df) < 50: return pd.DataFrame() # 标准化列名（确保小写） df.columns = [col.lower() for col in df.columns] # 2. 计算 RSI df.ta.rsi(length=14, append=True) # 3. 计算布林带 (Bollinger Bands) # 会生成 BBL_5_2.0 (下轨), BBM_5_2.0 (中轨), BBU_5_2.0 (上轨) 等列 df.ta.bbands(length=20, std=2.0, append=True) # 4. 计算 MACD # 会生成 MACD_12_26_9, MACDh_12_26_9 (直方图), MACDs_12_26_9 (信号线) df.ta.macd(fast=12, slow=26, signal=9, append=True) 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, period=period) 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_key=API_KEY) print("开始执行跨市场指标筛选...") signal_df = screener.screen_signals(watch_pool, period="1d") print("\n" + "="*50) print(" 信号筛选结果清单 ") print("="*50) if not signal_df.empty: print(signal_df.to_string(index=False)) else: print("今日无标的触发信号。") print("="*50)

四、 进阶考量：如何提升筛选器的响应速度与稳定性

当您的监控池从 5 个标的扩展到上百个标的时，您可能会遇到以下工程瓶颈。针对这些瓶颈，这里给出两条调优建议：

1. 利用多线程并发请求优化：
   在之前的文章《如何给金融行情API写一个优雅的“多线程并发、限流重试”二次封装》中，我们详细探讨了利用 ThreadPoolExecutor 提升 API 请求效率的方法。我们可以直接将上述 screen_signals 中的循环逻辑重构为多线程并发请求，以极大地压榨带宽，将百只股票的筛选时间缩短到数秒内。

2. 本地 K 线缓存与增量更新：
   对于日线级别的计算，数据是不需要秒级刷新的。我们可以将每日拉取的 K 线数据保存为本地 .parquet 文件。每天收盘后，仅使用 QuantDash API 增量拉取最新一个交易日的数据，并与本地历史数据进行 concat 合并，以此将 API 的调用额度留给更具即时性要求的日内或实时行情。

五、 结语

量化交易系统的构建是一个循序渐进的过程。通过将数据管道（QuantDash）的高效 DataFrame 导出特性与 Python 生态中成熟的计算链（Pandas-TA）相结合，我们可以用非常干净、优雅的代码实现复杂的量化策略筛选。

参考文档：QuantDash 简介 - QuantDash

未来，您还可以在此筛选器的基础上，加入邮件、钉钉、飞书机器人等推送模块，在检测到买入或卖出信号时，实现第一时间的盘中预警。