supermind logo
同花顺logo

Python 批量获取几百只股票的历史 K 线，最快的方法

用户头像sh_*2176oo
2026-08-29 发布

Python 批量获取几百只股票的历史 K 线，最快的方法是什么？

做量化选股或者回测，经常需要同时拉几百只股票的 K 线数据。比如：

  • 从沪深 300 成分股里筛选符合技术条件的标的
  • 对自选股列表做批量回测
  • 全市场 5000 只 A 股扫描一遍

这时候最大的痛点不是"能不能拿到数据"，而是要等多久。

最常见的做法：循环 + sleep

大多数 Python 股票数据库（akshare、Tushare 等）没有原生批量接口，只能循环请求：

import akshare as ak
import time

symbols = ["600519", "000001", "300750", ...]  # 假设 200 只

all_data = {}
for i, sym in enumerate(symbols):
    try:
        df = ak.stock_zh_a_hist(symbol=sym, period="daily",
                                 start_date="20260101", end_date="20260828",
                                 adjust="qfq")
        all_data[sym] = df
    except Exception as e:
        print(f"{sym} 失败: {e}")

    if i % 5 == 0:
        time.sleep(1)  # 不加 sleep 会被封 IP

print(f"获取完成，共 {len(all_data)} 只")

200 只票，每 5 只暂停 1 秒，总耗时大约 40-80 秒。 如果中间有几只失败了，还要手动重试。

Tushare 类似，也需要循环调用 pro.daily()，而且有频率限制（免费用户 200 次/分钟）。

这种方式的问题

问题 影响
串行请求 200 只票 = 200 次 HTTP 请求，一个接一个
必须加 sleep 不加就被封，加了就更慢
失败无自动重试 某只票网络超时了，需要手动 try/except 并记录
代码量大 循环 + 异常处理 + sleep + 进度显示 = 20-30 行
数据格式不一 不同函数返回的列名可能不同

自己写多线程？

有人会想到用 concurrent.futures 加速：

from concurrent.futures import ThreadPoolExecutor
import akshare as ak

def fetch_one(sym):
    return sym, ak.stock_zh_a_hist(symbol=sym, period="daily",
                                    start_date="20260101", end_date="20260828",
                                    adjust="qfq")

with ThreadPoolExecutor(max_workers=5) as pool:
    results = dict(pool.map(lambda s: fetch_one(s), symbols))

理论上可以加速，但实际上：

  • 爬虫类接口并发容易触发反爬，5 个线程同时请求东方财富，大概率被封 IP
  • 需要自己处理线程安全、异常捕获、重试逻辑
  • 代码复杂度翻倍

最快的方法：用带原生批量接口的数据源

AlphaFeed 的 Python SDK 内置了 batch 方法，专门解决批量获取问题：

from alphafeed import AlphaFeed

af = AlphaFeed()

symbols = ["600519.SH", "000001.SZ", "300750.SZ", "601318.SH", "002594.SZ",
           # ... 假设 200 只
           ]

dfs = af.klines.batch(
    symbols,
    period="1d",
    count=60,            # 最近 60 根 K 线
    adjust="forward",    # 前复权
    to_dataframe=True,
    show_progress=True   # 显示进度条
)

print(f"获取完成，共 {len(dfs)} 只")
# dfs 是一个 dict: {symbol: DataFrame}

200 只票，几秒钟完成。

batch 内部做了什么

你不需要关心细节，但了解一下有助于理解为什么它快：

  1. 自动分块 — 把 200 个 symbol 分成若干小批次
  2. 多线程并发 — 默认 5 个线程同时请求（是官方 API 并发，不是爬虫并发，不会被封）
  3. 自动重试 — 某个请求因网络波动失败，自动重试 3 次（只对网络错误、超时、5xx 状态码重试）
  4. 结果合并 — 全部完成后返回一个 {symbol: DataFrame} 的字典

你要写的代码：3 行。 循环、sleep、try/except、线程池——全部不用写。

实际对比：200 只票的 K 线获取

指标 akshare 循环 Tushare 循环 自写多线程 AlphaFeed batch
代码行数 ~20 行 ~20 行 ~30 行 3 行
耗时（200 只） 40-80 秒 30-60 秒 15-30 秒 3-5 秒
需要 sleep ✅ ✅ ⚠️ 看接口 ❌
失败重试 手动写 手动写 手动写 内置
进度显示 手动写 手动写 手动写 内置 show_progress
被封风险 高 中 高 无

500 只、1000 只、5000 只呢？

差距只会更大：

# 全市场 A 股扫描：先拿代码列表，再批量拉 K 线

# 第一步：获取全市场标的
all_quotes = af.quotes.get(universes="CN_Stock", to_dataframe=True)
symbols = all_quotes["symbol"].tolist()
print(f"全市场共 {len(symbols)} 只")

# 第二步：批量拉最近 20 天日 K 线（全部）
dfs = af.klines.batch(
    symbols,
    period="1d",
    count=20,
    adjust="forward",
    to_dataframe=True,
    show_progress=True
)
print(f"成功获取 {len(dfs)} 只")

5000 只 A 股全拉一遍，AlphaFeed batch 大约 30-60 秒完成。循环 + sleep 的方式？你可以自己算：5000 × 0.5 秒 = 至少 40 分钟。

不只是 K 线——所有数据类型都支持 batch

# 批量获取五档盘口
depths = af.depth.batch(symbols_50)

# 批量获取日内分钟线
intraday = af.klines.intraday_batch(symbols_50, period="1m", to_dataframe=True)

# 批量获取标的信息
instruments = af.instruments.batch(symbols_50)

同样的模式：传一个列表进去，自动并发 + 重试，返回结果。

拿到数据之后能干什么

批量 K 线的典型用途：

批量计算技术指标

import pandas as pd

for symbol, df in dfs.items():
    if len(df) < 20:
        continue
    df["ma5"] = df["close"].rolling(5).mean()
    df["ma20"] = df["close"].rolling(20).mean()
    df["vol_ratio"] = df["volume"] / df["volume"].rolling(10).mean()

批量选股

selected = []
for symbol, df in dfs.items():
    if len(df) < 20:
        continue
    latest = df.iloc[-1]
    ma20 = df["close"].rolling(20).mean().iloc[-1]

    if (latest["close"] > ma20 and
        latest["volume"] > df["volume"].iloc[-11:-1].mean() * 2):
        selected.append(symbol)

print(f"符合条件：{len(selected)} 只")

批量回测一个策略

results = {}
for symbol, df in dfs.items():
    if len(df) < 20:
        continue
    df["ma5"] = df["close"].rolling(5).mean()
    df["ma20"] = df["close"].rolling(20).mean()
    df["signal"] = (df["ma5"] > df["ma20"]).astype(int)
    df["ret"] = df["close"].pct_change() * df["signal"].shift(1)
    results[symbol] = df["ret"].sum()

results_df = pd.Series(results).sort_values(ascending=False)
print("策略收益 Top 10：")
print(results_df.head(10))

数据获取只花几秒，剩下的时间全部用在分析上。 这才是正确的时间分配。

怎么开始

pip install alphafeed

注册 AlphaFeed 账号，免费版就可以使用 klines.batch 获取 A 股日 K 线。

把你现在循环 + sleep 的代码替换成 batch 调用，感受一下差距。

评论