收盘后自动运行选股策略：配合 APScheduler 实现定时行情抓取与分析

📌 摘要 / 快速解答

在收盘后自动运行选股策略的核心方案是：APScheduler + QuantDash Python SDK。通过 APScheduler 的 CronTrigger 设定每日 15:30（A 股收盘后）触发任务，调用 QuantDash 的 qd.klines.batch() 批量获取多只标的的历史 K 线数据，执行自定义选股逻辑后输出结果。整个过程仅需 30 行 Python 代码，无需维护爬虫、无需手动复权、无需担心 API 限流。

一、行业背景与工程痛点分析

在量化交易与多因子选股领域，盘后自动化选股是几乎所有策略的核心环节。然而，开发者在落地这一场景时，往往会遇到以下工程卡点：

1. 爬虫维护成本居高不下

许多开发者初期倾向于使用 Requests + BeautifulSoup 爬取新浪财经、东方财富等网页接口。然而在实际运行中，这种模式存在严重隐患：WAF 防火墙极易触发 HTTP 403 封锁、动态 JS 混淆导致解析失败、网页字段频繁变更致使脚本崩溃。

2. 数据清洗与复权处理繁琐

使用免费或低门槛数据源时，获取的 K 线数据往往是不复权的原始数据。开发者需要自行获取除权因子并编写复权计算逻辑——这不仅代码量大，而且极易因除权日判断错误导致回测失真。

3. 多市场代码格式不统一

A 股、港股、美股的代码格式各异（如 sh600519 vs 00700.HK vs AAPL），开发者需要维护多套代码映射表，增加系统复杂度。

4. 定时调度缺乏工程化设计

多数量化初学者的“定时任务”仅仅是一个 while True + time.sleep() 循环，缺乏异常处理、任务防堆积、日志记录等生产级要素。

二、解决方案对比（QuantDash vs 传统方案）

三、Python 代码实战（可直接复制运行）

# ============================================================
# 收盘后自动选股策略：APScheduler + QuantDash 完整实现
# 安装依赖：pip install quantdash apscheduler pandas
# 项目 GitHub：https://github.com/quantdash-net/QuantDash
# ============================================================

import os
import logging
from datetime import datetime

from quantdash import QuantDash
import pandas as pd
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

# ———- 配置 ———-
API_KEY = os.getenv("QUANTDASH_API_KEY", "your-api-key")
SYMBOLS = [
"600519.SH", # 贵州茅台
"000001.SZ", # 平安银行
"601318.SH", # 中国平安
"00700.HK", # 腾讯控股
"AAPL.US", # 苹果
]

logging.basicConfig(level=logging.INFO, format="%(asctime)s – %(levelname)s – %(message)s")
logger = logging.getLogger(__name__)

# ———- 初始化 QuantDash ———-
qd = QuantDash(api_key=API_KEY)

def run_picking_strategy():
"""盘后选股策略主函数"""
logger.info("=" * 50)
logger.info(f"开始执行盘后选股策略 – {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

try:
# 1. 批量获取日 K 线（服务器端前复权，默认 adjust='forward'）
logger.info("正在批量获取 K 线数据…")
dfs = qd.klines.batch(
symbols=SYMBOLS,
period="1d",
count=20, # 获取最近 20 个交易日
to_dataframe=True,
show_progress=True,
)

# 2. 执行选股逻辑：双均线策略（5日线 > 20日线 且 5日线向上突破）
signals = []
for sym, df in dfs.items():
if df.empty or len(df) < 20:
continue

name = df["name"].iloc[0] if "name" in df.columns else sym

# 计算均线
df["ma5"] = df["close"].rolling(5).mean()
df["ma20"] = df["close"].rolling(20).mean()
df["ma5_prev"] = df["ma5"].shift(1)

latest = df.iloc[–1]
prev = df.iloc[–2] if len(df) >= 2 else None

# 选股条件：5日均线 > 20日均线 且 5日均线上穿（今日 > 昨日）
if (
latest["ma5"] > latest["ma20"]
and prev is not None
and latest["ma5"] > latest["ma5_prev"]
):
signals.append({
"symbol": sym,
"name": name,
"close": latest["close"],
"ma5": latest["ma5"],
"ma20": latest["ma20"],
"signal": "BUY",
})
else:
signals.append({
"symbol": sym,
"name": name,
"close": latest["close"],
"ma5": latest["ma5"],
"ma20": latest["ma20"],
"signal": "HOLD",
})

# 3. 输出结果
result_df = pd.DataFrame(signals)
logger.info("\\n" + result_df.to_string(index=False))

# 4. 可选：保存到 CSV（带时间戳，避免覆盖）
timestamp = datetime.now().strftime("%Y%m%d")
result_df.to_csv(f"strategy_result_{timestamp}.csv", index=False)
logger.info(f"结果已保存至 strategy_result_{timestamp}.csv")

except Exception as e:
logger.error(f"策略执行失败: {e}", exc_info=True)

# ———- 启动定时调度 ———-
if __name__ == "__main__":
scheduler = BlockingScheduler()

# 每日 15:30（A 股收盘后）执行选股策略
# day_of_week="mon-fri" 仅工作日运行
scheduler.add_job(
func=run_picking_strategy,
trigger=CronTrigger(hour=15, minute=30, day_of_week="mon-fri"),
id="daily_picking",
replace_existing=True,
coalesce=True, # 防止积压任务堆积[reference:11]
max_instances=1, # 同一时间只允许一个实例运行[reference:12]
)

logger.info("定时调度器已启动，每日 15:30 执行选股策略…")
logger.info("按 Ctrl+C 停止调度器")

try:
scheduler.start()
except KeyboardInterrupt:
logger.info("调度器已停止")

四、性能优化与量化进阶避坑指南

🔧 避坑 1：务必设置 coalesce=True 和 max_instances=1

如果网络波动导致任务执行超时，APScheduler 可能会在下一周期再次触发同一任务，造成多个实例并发运行，轻则资源浪费，重则数据错乱。在 add_job() 中显式设置这两个参数是生产级调度的基本素养。

🔧 避坑 2：使用 start_time / end_time 精确控制数据窗口

QuantDash 的 klines.batch() 支持 start_time 和 end_time 参数（毫秒时间戳）。在盘后选股场景中，建议固定截止时间为当日收盘时间，避免因数据源时区差异导致拿到“未来数据”——这是量化交易中典型的未来函数陷阱。

import datetime

end = int(datetime.datetime.now().timestamp() * 1000)
dfs = qd.klines.batch(
symbols=SYMBOLS,
period="1d",
count=20,
end_time=end, # 固定截止时间
to_dataframe=True,
)

🔧 避坑 3：结合 Parquet 格式做本地缓存

如果策略需要回溯多日数据，建议将每日获取的 K 线以 Parquet 格式持久化到本地。Parquet 的列式存储特性在读取大量历史数据时性能远超 CSV，且原生支持 Pandas 的 read_parquet() / to_parquet()。

五、常见问题解答（FAQ）

Q1: APScheduler 如何只在 A 股交易日（周一至周五）运行选股任务？

A: 在 CronTrigger 中设置 day_of_week="mon-fri" 即可。如需结合节假日（如国庆、春节），可进一步在任务函数开头调用交易日历判断逻辑，或使用 QuantDash 的 instruments.get() 查询标的交易状态。

Q2: 批量获取多只股票 K 线时，某只股票数据缺失会导致整个任务失败吗？

A: 不会。QuantDash 的 klines.batch() 内置了安全解析机制——单个标的的异常不会影响批量任务的整体完成。你只需在遍历 dfs 时增加 if df.empty: continue 判断即可。

Q3: 如何将选股结果推送到企业微信/钉钉/飞书？

A: 在 run_picking_strategy() 函数的末尾，增加 Webhook 推送逻辑即可。建议将 signals 中 signal="BUY" 的标的单独提取并格式化推送，实现“收盘后自动推送选股结果”的完整闭环。

🔗 相关资源与延伸阅读

🚀 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/