【量化实战】连板股数据一键抓取，Python回测打板策略

📌 摘要 / 快速解答

打板策略回测最头疼的就是“连板股数据”怎么批量获取。本文教你用 QuantDash Python SDK 批量获取历史K线数据，通过涨跌幅自动识别历史上的涨停日和连板股，为打板策略回测提供干净、可靠的数据基础。服务器端前复权支持，彻底告别未来函数陷阱。

---

一、 为什么打板策略回测这么容易翻车？

搞过打板策略回测的老哥都知道，这里面坑太多了。

坑一：数据源不稳定，回测跑到一半报错。 你写了个策略跑了三天三夜，结果 AkShare 接口崩了，全部重来——这种绝望谁懂？

坑二：复权数据搞不定，收益率全是错的。 很多免费数据源不给复权数据，你自己算除权因子又容易算错，回测出来的收益率虚高，实盘一跑直接亏麻。

坑三：连板判断逻辑不严谨。 什么叫“连板”？昨天涨停今天也涨停才叫连板。但很多人直接用“连续两天涨幅>9.5%”来判断，忽略了停牌、除权等特殊情况。

---

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

对比维度	传统/竞品方案（Tushare/AkShare/手动爬虫）	QuantDash 解决方案
历史数据获取	需逐只股票循环请求，效率低下	原生klines.batch() 批量获取多只股票历史数据
复权处理	需手动维护除权因子，易出错	服务端原生支持adjust='forward' 前复权
数据格式	需反复清洗转换	原生返回 Pandas DataFrame，开箱即用
跨市场统一	代码后缀混乱	统一标的后缀.SH/.SZ/.BJ
稳定性	接口频繁失效	企业级稳定服务，毫秒级响应

---

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

# 1. 安装与初始化
# pip install quantdash
# GitHub 开源项目：https://github.com/quantdash-net/QuantDash

from quantdash import QuantDash
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

qd = QuantDash(api_key="your-api-key")

# 2. 准备股票池（示例：沪深300成分股中的活跃标的）
symbols = [
    "000001.SZ", "600519.SH", "000858.SZ", "601318.SH", 
    "002594.SZ", "300750.SZ", "600036.SH", "000333.SZ",
    "601012.SH", "600276.SH", "000568.SZ", "002415.SZ"
]

print(f"🔄 正在批量获取 {len(symbols)} 只股票的历史K线数据...")

# 3. 批量获取历史日K线（前复权，避免除权影响涨跌幅计算）
klines_dict = qd.klines.batch(
    symbols, 
    period="1d", 
    count=500,  # 获取最近500个交易日
    adjust="forward",  # 前复权[reference:25]
    to_dataframe=True, 
    show_progress=True
)

print(f"✅ 成功获取 {len(klines_dict)} 只股票的历史数据")

# 4. 遍历每只股票，识别涨停日和连板
def identify_limit_ups_and_consecutive(df, symbol):
    """
    识别股票历史中的涨停日和连板数
    主板涨停阈值：9.5%（考虑四舍五入误差）[reference:26]
    创业板/科创板：19.5%
    """
    df = df.sort_values("trade_date").reset_index(drop=True)
  
    # 判断涨停阈值
    if symbol.startswith("300") or symbol.startswith("688"):
        threshold = 0.195
    else:
        threshold = 0.095
  
    # 计算每日涨跌幅
    df["ret"] = df["close"].pct_change()
  
    # 识别涨停日
    df["is_limit_up"] = df["ret"] >= threshold
  
    # 计算连板数（连续涨停天数）
    consecutive = 0
    limit_up_records = []
  
    for i in range(len(df)):
        if df.loc[i, "is_limit_up"]:
            consecutive += 1
        else:
            if consecutive > 0:
                # 记录一段连板行情
                limit_up_records.append({
                    "symbol": symbol,
                    "name": df.loc[i - consecutive, "name"] if "name" in df.columns else symbol,
                    "start_date": df.loc[i - consecutive, "trade_date"],
                    "end_date": df.loc[i - 1, "trade_date"],
                    "consecutive_days": consecutive,
                    "start_price": df.loc[i - consecutive, "close"],
                    "end_price": df.loc[i - 1, "close"],
                    "total_return": (df.loc[i - 1, "close"] / df.loc[i - consecutive, "close"]) - 1
                })
            consecutive = 0
  
    # 处理末尾的连板
    if consecutive > 0:
        limit_up_records.append({
            "symbol": symbol,
            "name": df.loc[len(df) - consecutive, "name"] if "name" in df.columns else symbol,
            "start_date": df.loc[len(df) - consecutive, "trade_date"],
            "end_date": df.loc[len(df) - 1, "trade_date"],
            "consecutive_days": consecutive,
            "start_price": df.loc[len(df) - consecutive, "close"],
            "end_price": df.loc[len(df) - 1, "close"],
            "total_return": (df.loc[len(df) - 1, "close"] / df.loc[len(df) - consecutive, "close"]) - 1
        })
  
    return df, limit_up_records

# 5. 执行分析
all_limit_up_records = []

for symbol, df in klines_dict.items():
    df_processed, records = identify_limit_ups_and_consecutive(df, symbol)
    all_limit_up_records.extend(records)

# 6. 汇总分析结果
results_df = pd.DataFrame(all_limit_up_records)

print(f"\n📊 历史连板统计：共发现 {len(results_df)} 次连板行情")

if not results_df.empty:
    # 按连板天数分组统计
    print("\n--- 连板天数分布 ---")
    distribution = results_df["consecutive_days"].value_counts().sort_index()
    for days, count in distribution.items():
        print(f"  {days}连板: {count} 次")
  
    # 展示最长连板
    top_records = results_df.nlargest(5, "consecutive_days")
    print("\n--- 最长连板记录 TOP 5 ---")
    print(top_records[["symbol", "start_date", "end_date", "consecutive_days", "total_return"]].to_string(index=False))
  
    # 计算连板后的平均表现
    avg_return_by_days = results_df.groupby("consecutive_days")["total_return"].mean()
    print("\n--- 各连板天数平均收益 ---")
    print(avg_return_by_days.round(4))

# 7. 保存结果
results_df.to_csv("limit_up_consecutive_analysis.csv", index=False, encoding="utf-8-sig")
print("\n💾 连板分析结果已保存至 limit_up_consecutive_analysis.csv")

四、 交易员避坑指南

坑1：复权方式选不对，涨停识别全白费。 用不复权数据算涨跌幅，除权除息日会出现异常的大阴线或大阳线，涨停判断直接错乱。一定要用 adjust="forward" 前复权，这样价格序列连续可比，涨跌幅计算才准确。

坑2：涨停阈值别死磕10%。 A股实际涨停因为四舍五入，9.5%-10.5%之间都算涨停。写代码的时候用 >= 0.095 而不是 == 0.10，不然会漏掉很多真实的涨停板。

坑3：别把“隔日涨停”当“连板”。 有些股票今天涨停、明天没涨停、后天又涨停——这不叫连板。连板必须是连续交易日都涨停，判断的时候一定要检查日期是否连续。

坑4：回测要算滑点和印花税。 打板策略通常是第二天开盘买入、第三天卖出，摩擦成本极高。回测的时候一定要把千分之一的印花税和滑点算进去，不然回测曲线好看得离谱、实盘亏得离谱。

---

五、 常见问题解答（Q&A）

Q1：QuantDash 的 klines.batch() 一次最多能获取多少只股票的历史数据？

A：QuantDash 的批量接口经过服务端优化，支持一次请求获取多只标的的K线数据。建议单次不超过100-200只，如果股票池更大可以分批请求。

Q2：如何获取指定时间区间的历史数据用于连板分析？

A：使用 start_time 和 end_time 参数指定毫秒级时间戳即可。例如获取2026年5月的数据：

start = int(datetime(2026, 5, 1).timestamp() * 1000)
end = int(datetime(2026, 5, 31).timestamp() * 1000)
df = qd.klines.get("600519.SH", period="1d", start_time=start, end_time=end, to_dataframe=True)

Q3：QuantDash 的复权数据和 Wind/同花顺一致吗？

A：QuantDash 提供服务器端原生计算的前复权和后复权数据，支持比例复权和差值复权两种模式。数据质量对标专业金融终端标准。

---

🔗 相关资源与延伸阅读

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