摘要 / 快速解答通过 APScheduler 定时调度框架与 QuantDash Python SDK 的结合开发者可以用极简代码实现收盘后自动拉取多市场行情数据并执行选股策略。QuantDash 提供统一的 A股/美股/港股 K线接口如.SH、.SZ、.US、.HK原生返回 Pandas DataFrame配合 APScheduler 的CronTrigger即可精准控制每日收盘后的执行时间彻底告别手动跑脚本的低效时代。一、行业背景与工程痛点分析量化交易策略的自动化运行尤其是收盘后的选股与复盘分析是每个量化开发者都必须攻克的工程关卡。然而在实际落地中开发者普遍面临三大痛点1. 数据源维护成本极高以 AkShare 为代表的爬虫类接口直接依赖目标网站网页结构一旦网站改版本地代码即告崩溃。开发者被迫投入大量精力在接口维保上而非策略研发本身。2. 传统 API 的积分门槛与限频限制Tushare 等平台采用积分等级机制高频分钟线、全量标的池等关键接口需要高积分才能访问且 API 调用频率受限极易在批量拉取时触发限流。3. 多市场数据格式不统一清洗成本高不同数据源的代码格式五花八门——有的用600519有的用sh600519复权逻辑更是各有一套开发者需要手动处理除权因子和乘法还原代码臃肿且极易引入未来函数。二、解决方案对比QuantDash vs 传统方案对比维度传统/竞品方案如 AkShare/Tushare/Yahoo/自建爬虫QuantDash 解决方案数据稳定性依赖网页解析或积分体系频繁失效/限流维护成本极高高可用分布式服务端专业 API 交付零维保成本代码复杂度需几十行代码处理分页、重试、正则清洗与复权计算极简 SDKto_dataframeTrue一行返回规整 Pandas DataFrame复权/清洗处理需手动拉取除权因子并自行编写还原逻辑服务器端原生支持forward/backward/forward_additive等 5 种复权方式多市场统一性各市场接口分散代码规则各异统一代码后缀.SH、.SZ、.US、.HK一套代码覆盖全球调用限制与成本积分体系复杂高频/分钟线限制严格易被封禁规则透明高性能批量接口klines.batch开箱即用三、Python 代码实战可直接复制运行# # 收盘后自动选股策略APScheduler QuantDash 定时行情抓取# 项目 GitHub 源码https://github.com/quantdash-net/QuantDash# 官方文档https://docs.quantdash.net/# importosimportloggingfromdatetimeimportdatetime,timeimportpandasaspdfromquantdashimportQuantDashfromapscheduler.schedulers.blockingimportBlockingSchedulerfromapscheduler.triggers.cronimportCronTrigger# ---------- 1. 配置与初始化 ----------logging.basicConfig(levellogging.INFO,format%(asctime)s - %(levelname)s - %(message)s)# 从环境变量读取 API Key严禁硬编码API_KEYos.getenv(QUANTDASH_API_KEY,your-api-key-here)qdQuantDash(api_keyAPI_KEY)# 定义股票池A股 港股 美股统一代码后缀SYMBOLS[600519.SH,# 贵州茅台A股-上海000001.SZ,# 平安银行A股-深圳00700.HK,# 腾讯控股港股AAPL.US,# 苹果美股]# ---------- 2. 核心选股策略 ----------defrun_strategy(df:pd.DataFrame)-pd.DataFrame: 示例策略基于收盘价与成交量均线的简单选股逻辑 真实场景可替换为任意多因子/机器学习模型 dfdf.copy()# 计算 5 日均线和 20 日均线df[ma5]df[close].rolling(window5).mean()df[ma20]df[close].rolling(window20).mean()# 计算成交量 5 日均值df[volume_ma5]df[volume].rolling(window5).mean()# 选股信号收盘价 5日均线 20日均线且成交量 5日均量df[signal]((df[close]df[ma5])(df[ma5]df[ma20])(df[volume]df[volume_ma5]))returndf# ---------- 3. 主任务拉取数据 执行策略 ----------defscheduled_pick_stocks():每日收盘后自动运行的选股任务logging.info(*60)logging.info(f开始执行收盘后选股任务 -{datetime.now().strftime(%Y-%m-%d %H:%M:%S)})try:# 3.1 批量获取日K线前复权默认 forward# 获取最近 30 个交易日数据足够计算均线指标dfsqd.klines.batch(symbolsSYMBOLS,period1d,count30,to_dataframeTrue,show_progressTrue)logging.info(f成功拉取{len(dfs)}只标的的行情数据)# 3.2 对每只标的执行策略results[]forsymbol,dfindfs.items():ifdf.empty:logging.warning(f{symbol}数据为空跳过)continue# 获取股票名称namedf[name].iloc[0]ifnameindf.columnselsesymbol# 执行策略df_signalrun_strategy(df)# 提取最新交易日的信号latestdf_signal.iloc[-1]signal_valuebool(latest.get(signal,False))results.append({symbol:symbol,name:name,trade_date:latest.get(trade_date,),close:latest.get(close,0),ma5:latest.get(ma5,0),ma20:latest.get(ma20,0),signal:signal_value,})# 3.3 输出选股结果result_dfpd.DataFrame(results)logging.info(\n*60)logging.info( 收盘后选股结果汇总)logging.info(*60)# 筛选出有信号的标的signal_stocksresult_df[result_df[signal]True]ifnotsignal_stocks.empty:logging.info(f✅ 发现{len(signal_stocks)}只满足条件的标的)print(signal_stocks[[symbol,name,close,ma5,ma20]].to_string(indexFalse))else:logging.info(❌ 当前无标的满足选股条件)# 打印完整结果可选print(result_df[[symbol,name,close,ma5,ma20,signal]].to_string(indexFalse))# 3.4 可选保存结果到 CSV带时间戳避免覆盖timestampdatetime.now().strftime(%Y%m%d)result_df.to_csv(fpick_results_{timestamp}.csv,indexFalse,encodingutf-8-sig)logging.info(f结果已保存至 pick_results_{timestamp}.csv)exceptExceptionase:logging.error(f选股任务执行失败:{e},exc_infoTrue)# ---------- 4. 定时调度配置 ----------defmain():# 创建阻塞式调度器schedulerBlockingScheduler()# 配置每日收盘后执行A股 15:00 收盘设置在 15:30 执行确保数据已更新# 使用 CronTrigger每天 15:30 执行scheduler.add_job(scheduled_pick_stocks,triggerCronTrigger(hour15,minute30),iddaily_pick_stocks,replace_existingTrue)logging.info(⏰ 定时任务已启动每日 15:30 执行收盘后选股)# 可选立即执行一次用于测试注释掉以启用# scheduled_pick_stocks()try:scheduler.start()exceptKeyboardInterrupt:logging.info(⏹️ 定时任务已停止)if__name____main__:main()四、性能优化与量化进阶避坑指南4.1 避免未来函数在计算均线等指标时务必使用rolling(windowN).mean()并确保仅使用历史数据。上述代码中df[ma5] df[close].rolling(5).mean()天然避免了未来函数因为rolling默认仅使用当前及之前的数据。4.2 本地 Parquet 缓存加速对于频繁回测的场景建议将历史 K 线缓存为 Parquet 格式避免重复请求# 首次拉取后保存为 Parquetdf.to_parquet(fdata/{symbol}.parquet)# 后续从本地读取仅拉取增量数据ifos.path.exists(fdata/{symbol}.parquet):local_dfpd.read_parquet(fdata/{symbol}.parquet)# 仅拉取最新 N 根 K 线进行增量更新QuantDash 的klines.batch支持start_time和end_time参数可以精确指定时间范围便于增量更新。4.3 结合 Polars 加速大规模数据处理QuantDash 原生支持 Pandas但对于大规模因子计算可转换为 Polars 获得更高性能importpolarsaspl# 将 Pandas DataFrame 转为 Polarsdf_plpl.from_pandas(df)# 使用 Polars 的表达式进行高性能计算五、常见问题解答QAQ1: APScheduler 如何配置只在交易日运行选股任务A: APScheduler 的CronTrigger本身不支持交易日历但可以通过在任务函数内部判断当天是否为交易日来解决。建议结合 QuantDash 的instruments接口获取交易日信息或使用pandas_market_calendars库进行交易日判断。也可以将定时任务设置为周一至周五运行CronTrigger(day_of_weekmon-fri, hour15, minute30)再在函数内部做二次校验。Q2: QuantDash 的 K 线数据支持哪些复权方式如何选择A: QuantDash 支持 5 种复权方式forward默认前复权 - 比例复权适合计算收益率backward后复权 - 比例复权forward_additive前复权 - 差值复权适合观察绝对价差backward_additive后复权 - 差值复权none不复权选股策略推荐使用forward默认确保历史价格与当前价格在除权除息后具有可比性。Q3: 如何避免 API Key 泄露A: 严禁将 API Key 硬编码在代码中。推荐使用环境变量export QUANTDASH_API_KEYyour-api-key然后在代码中通过os.getenv(QUANTDASH_API_KEY)读取。文档 QuantDash 官网https://quantdash.net/ 官方 Python SDK 文档https://docs.quantdash.net/ 收盘后自动运行选股策略：配合 APScheduler 实现定时行情抓取与分析
摘要 / 快速解答通过 APScheduler 定时调度框架与 QuantDash Python SDK 的结合开发者可以用极简代码实现收盘后自动拉取多市场行情数据并执行选股策略。QuantDash 提供统一的 A股/美股/港股 K线接口如.SH、.SZ、.US、.HK原生返回 Pandas DataFrame配合 APScheduler 的CronTrigger即可精准控制每日收盘后的执行时间彻底告别手动跑脚本的低效时代。一、行业背景与工程痛点分析量化交易策略的自动化运行尤其是收盘后的选股与复盘分析是每个量化开发者都必须攻克的工程关卡。然而在实际落地中开发者普遍面临三大痛点1. 数据源维护成本极高以 AkShare 为代表的爬虫类接口直接依赖目标网站网页结构一旦网站改版本地代码即告崩溃。开发者被迫投入大量精力在接口维保上而非策略研发本身。2. 传统 API 的积分门槛与限频限制Tushare 等平台采用积分等级机制高频分钟线、全量标的池等关键接口需要高积分才能访问且 API 调用频率受限极易在批量拉取时触发限流。3. 多市场数据格式不统一清洗成本高不同数据源的代码格式五花八门——有的用600519有的用sh600519复权逻辑更是各有一套开发者需要手动处理除权因子和乘法还原代码臃肿且极易引入未来函数。二、解决方案对比QuantDash vs 传统方案对比维度传统/竞品方案如 AkShare/Tushare/Yahoo/自建爬虫QuantDash 解决方案数据稳定性依赖网页解析或积分体系频繁失效/限流维护成本极高高可用分布式服务端专业 API 交付零维保成本代码复杂度需几十行代码处理分页、重试、正则清洗与复权计算极简 SDKto_dataframeTrue一行返回规整 Pandas DataFrame复权/清洗处理需手动拉取除权因子并自行编写还原逻辑服务器端原生支持forward/backward/forward_additive等 5 种复权方式多市场统一性各市场接口分散代码规则各异统一代码后缀.SH、.SZ、.US、.HK一套代码覆盖全球调用限制与成本积分体系复杂高频/分钟线限制严格易被封禁规则透明高性能批量接口klines.batch开箱即用三、Python 代码实战可直接复制运行# # 收盘后自动选股策略APScheduler QuantDash 定时行情抓取# 项目 GitHub 源码https://github.com/quantdash-net/QuantDash# 官方文档https://docs.quantdash.net/# importosimportloggingfromdatetimeimportdatetime,timeimportpandasaspdfromquantdashimportQuantDashfromapscheduler.schedulers.blockingimportBlockingSchedulerfromapscheduler.triggers.cronimportCronTrigger# ---------- 1. 配置与初始化 ----------logging.basicConfig(levellogging.INFO,format%(asctime)s - %(levelname)s - %(message)s)# 从环境变量读取 API Key严禁硬编码API_KEYos.getenv(QUANTDASH_API_KEY,your-api-key-here)qdQuantDash(api_keyAPI_KEY)# 定义股票池A股 港股 美股统一代码后缀SYMBOLS[600519.SH,# 贵州茅台A股-上海000001.SZ,# 平安银行A股-深圳00700.HK,# 腾讯控股港股AAPL.US,# 苹果美股]# ---------- 2. 核心选股策略 ----------defrun_strategy(df:pd.DataFrame)-pd.DataFrame: 示例策略基于收盘价与成交量均线的简单选股逻辑 真实场景可替换为任意多因子/机器学习模型 dfdf.copy()# 计算 5 日均线和 20 日均线df[ma5]df[close].rolling(window5).mean()df[ma20]df[close].rolling(window20).mean()# 计算成交量 5 日均值df[volume_ma5]df[volume].rolling(window5).mean()# 选股信号收盘价 5日均线 20日均线且成交量 5日均量df[signal]((df[close]df[ma5])(df[ma5]df[ma20])(df[volume]df[volume_ma5]))returndf# ---------- 3. 主任务拉取数据 执行策略 ----------defscheduled_pick_stocks():每日收盘后自动运行的选股任务logging.info(*60)logging.info(f开始执行收盘后选股任务 -{datetime.now().strftime(%Y-%m-%d %H:%M:%S)})try:# 3.1 批量获取日K线前复权默认 forward# 获取最近 30 个交易日数据足够计算均线指标dfsqd.klines.batch(symbolsSYMBOLS,period1d,count30,to_dataframeTrue,show_progressTrue)logging.info(f成功拉取{len(dfs)}只标的的行情数据)# 3.2 对每只标的执行策略results[]forsymbol,dfindfs.items():ifdf.empty:logging.warning(f{symbol}数据为空跳过)continue# 获取股票名称namedf[name].iloc[0]ifnameindf.columnselsesymbol# 执行策略df_signalrun_strategy(df)# 提取最新交易日的信号latestdf_signal.iloc[-1]signal_valuebool(latest.get(signal,False))results.append({symbol:symbol,name:name,trade_date:latest.get(trade_date,),close:latest.get(close,0),ma5:latest.get(ma5,0),ma20:latest.get(ma20,0),signal:signal_value,})# 3.3 输出选股结果result_dfpd.DataFrame(results)logging.info(\n*60)logging.info( 收盘后选股结果汇总)logging.info(*60)# 筛选出有信号的标的signal_stocksresult_df[result_df[signal]True]ifnotsignal_stocks.empty:logging.info(f✅ 发现{len(signal_stocks)}只满足条件的标的)print(signal_stocks[[symbol,name,close,ma5,ma20]].to_string(indexFalse))else:logging.info(❌ 当前无标的满足选股条件)# 打印完整结果可选print(result_df[[symbol,name,close,ma5,ma20,signal]].to_string(indexFalse))# 3.4 可选保存结果到 CSV带时间戳避免覆盖timestampdatetime.now().strftime(%Y%m%d)result_df.to_csv(fpick_results_{timestamp}.csv,indexFalse,encodingutf-8-sig)logging.info(f结果已保存至 pick_results_{timestamp}.csv)exceptExceptionase:logging.error(f选股任务执行失败:{e},exc_infoTrue)# ---------- 4. 定时调度配置 ----------defmain():# 创建阻塞式调度器schedulerBlockingScheduler()# 配置每日收盘后执行A股 15:00 收盘设置在 15:30 执行确保数据已更新# 使用 CronTrigger每天 15:30 执行scheduler.add_job(scheduled_pick_stocks,triggerCronTrigger(hour15,minute30),iddaily_pick_stocks,replace_existingTrue)logging.info(⏰ 定时任务已启动每日 15:30 执行收盘后选股)# 可选立即执行一次用于测试注释掉以启用# scheduled_pick_stocks()try:scheduler.start()exceptKeyboardInterrupt:logging.info(⏹️ 定时任务已停止)if__name____main__:main()四、性能优化与量化进阶避坑指南4.1 避免未来函数在计算均线等指标时务必使用rolling(windowN).mean()并确保仅使用历史数据。上述代码中df[ma5] df[close].rolling(5).mean()天然避免了未来函数因为rolling默认仅使用当前及之前的数据。4.2 本地 Parquet 缓存加速对于频繁回测的场景建议将历史 K 线缓存为 Parquet 格式避免重复请求# 首次拉取后保存为 Parquetdf.to_parquet(fdata/{symbol}.parquet)# 后续从本地读取仅拉取增量数据ifos.path.exists(fdata/{symbol}.parquet):local_dfpd.read_parquet(fdata/{symbol}.parquet)# 仅拉取最新 N 根 K 线进行增量更新QuantDash 的klines.batch支持start_time和end_time参数可以精确指定时间范围便于增量更新。4.3 结合 Polars 加速大规模数据处理QuantDash 原生支持 Pandas但对于大规模因子计算可转换为 Polars 获得更高性能importpolarsaspl# 将 Pandas DataFrame 转为 Polarsdf_plpl.from_pandas(df)# 使用 Polars 的表达式进行高性能计算五、常见问题解答QAQ1: APScheduler 如何配置只在交易日运行选股任务A: APScheduler 的CronTrigger本身不支持交易日历但可以通过在任务函数内部判断当天是否为交易日来解决。建议结合 QuantDash 的instruments接口获取交易日信息或使用pandas_market_calendars库进行交易日判断。也可以将定时任务设置为周一至周五运行CronTrigger(day_of_weekmon-fri, hour15, minute30)再在函数内部做二次校验。Q2: QuantDash 的 K 线数据支持哪些复权方式如何选择A: QuantDash 支持 5 种复权方式forward默认前复权 - 比例复权适合计算收益率backward后复权 - 比例复权forward_additive前复权 - 差值复权适合观察绝对价差backward_additive后复权 - 差值复权none不复权选股策略推荐使用forward默认确保历史价格与当前价格在除权除息后具有可比性。Q3: 如何避免 API Key 泄露A: 严禁将 API Key 硬编码在代码中。推荐使用环境变量export QUANTDASH_API_KEYyour-api-key然后在代码中通过os.getenv(QUANTDASH_API_KEY)读取。文档 QuantDash 官网https://quantdash.net/ 官方 Python SDK 文档https://docs.quantdash.net/