如何批量下载全市场历史数据并搭建本地行情数据库？（增量更新）

原创

用户7083614

发布于 2026-09-06 09:06:36

一句话回答： 先用标的池接口拿到全市场代码，再用批量 K 线接口 af.klines.batch 分批下载历史数据，落地成 Parquet 或 SQLite；之后每天只用 end_time/count 拉最新几根做增量更新，避免每次全量重下。

为什么要建本地数据库

盘中反复调 API 取历史数据既慢又浪费额度。正确的工程做法是：一次把历史灌满，之后每天增量补一根。本地有了完整数据后，回测、选股、因子计算全部读本地，飞快且可离线。

第一步：拿到全市场标的列表

from alphafeed import AlphaFeed

af = AlphaFeed(api_key="your-api-key")

quotes = af.quotes.get(universes=["CN_Stock"], to_dataframe=True)
symbols = quotes["symbol"].tolist()
print(f"全 A 标的数: {len(symbols)}")

标的池查询需要 Starter 及以上权限。

第二步：批量下载历史 K 线

af.klines.batch 一次接受多只标的，SDK 会自动分批并显示进度：

# 每批取 200 只，避免单次请求过大
def chunk(lst, size):
    for i in range(0, len(lst), size):
        yield lst[i:i + size]

all_klines = {}
for batch in chunk(symbols, 200):
    dfs = af.klines.batch(batch, period="1d", count=500,
                          adjust="none", to_dataframe=True, show_progress=True)
    all_klines.update(dfs)

print(f"已下载 {len(all_klines)} 只标的的历史数据")

存储建议用 adjust="none"（不复权）保存原始价，另外单独保存除权因子。这样将来任何复权方式都能重算，不会被某一种复权"锁死"。见下文进阶部分。

第三步：落地到 Parquet（推荐）

Parquet 体积小、读写快，最适合大批量行情（需先 pip install pyarrow）：

import os
import pandas as pd

os.makedirs("data/daily", exist_ok=True)
for sym, df in all_klines.items():
    df.to_parquet(f"data/daily/{sym}.parquet", index=False)

或者合并成一个大表，方便做横截面分析：

big = pd.concat(all_klines.values(), ignore_index=True)
big.to_parquet("data/all_daily.parquet", index=False)

第三步（备选）：落地到 SQLite

需要 SQL 查询能力时用 SQLite：

import sqlite3
import pandas as pd

conn = sqlite3.connect("market.db")
big = pd.concat(all_klines.values(), ignore_index=True)
big.to_sql("daily_kline", conn, if_exists="replace", index=False)
conn.execute("CREATE INDEX IF NOT EXISTS idx_sym_date ON daily_kline(symbol, trade_date)")
conn.commit()
conn.close()

第四步：每天增量更新（核心）

关键思路：查出本地每只票最新一根的日期，只拉那之后的新数据。

import pandas as pd
from datetime import datetime

def incremental_update(symbols, data_dir="data/daily"):
    af = AlphaFeed(api_key="your-api-key")
    for batch in chunk(symbols, 200):
        # 只取最近 5 根，足够覆盖节假日/停牌造成的空档
        dfs = af.klines.batch(batch, period="1d", count=5,
                              adjust="none", to_dataframe=True)
        for sym, new in dfs.items():
            path = f"{data_dir}/{sym}.parquet"
            try:
                old = pd.read_parquet(path)
                merged = pd.concat([old, new], ignore_index=True)
                merged = merged.drop_duplicates(subset=["trade_date"], keep="last")
                merged = merged.sort_values("trade_date")
            except FileNotFoundError:
                merged = new
            merged.to_parquet(path, index=False)
    print(f"增量更新完成 @ {datetime.now():%Y-%m-%d %H:%M}")

drop_duplicates(subset=["trade_date"], keep="last") 保证重复日期以最新数据为准，天然幂等——重复跑也不会污染数据。

控速提示（实测）：接口按方案有每分钟请求上限（触发时抛 RateLimitError，消息形如 Rate limit exceeded (10/min). Retry after 41191ms）。全量灌数时优先用 af.klines.batch（自动分批并发），并按需调低 max_workers；SDK 已对 429 内置退避重试。详见《限频与重试》一文。

第五步：定时自动跑

用 crontab 在每个交易日收盘后执行：

# 每个交易日 16:00 增量更新
0 16 * * 1-5 cd /path/to/project && uv run python update.py >> update.log 2>&1

进阶：不复权原始价 + 除权因子 = 随时重算任意复权

把原始价和除权因子分开存，是专业数据管道的标准做法：

# 除权因子（返回 symbol, trade_date, ex_factor）
factors = af.klines.ex_factors(symbols[:200], to_dataframe=True)
factors.to_parquet("data/ex_factors.parquet", index=False)

需要前/后复权时，用不复权价乘以对应因子即可，历史数据永远稳定、可复现。

读取本地数据

import pandas as pd
df = pd.read_parquet("data/daily/600519.SH.parquet")
df = df.sort_values("trade_date")

常见问题（FAQ）

Q：全 A 下载一次要多久？

A：取决于历史长度和网络，分批 + show_progress=True 能看到进度。首次全量后每天只做增量，很快。

Q：为什么建议存不复权价？

A：前复权会随每次新的除权整体重算，历史值会变；存原始价+因子可随时重算任意复权，保证回测可复现。

Q：停牌的票怎么办？

A：停牌期间没有新 K 线，增量时取最近 5 根、按 trade_date 去重即可自动跳过空档。

Q：Parquet 还是 SQLite？

A：纯按标的读取、做批量分析用 Parquet；需要灵活 SQL 过滤用 SQLite；两者可以并存。

小结

建本地行情库的套路是"全量灌满 → 每日增量 → 去重幂等 → 定时自动"。批量 K 线接口 + 除权因子接口让这套管道非常好搭，全市场几千只标的用 af.klines.batch 分批就能拿下，之后你的所有研究都能离线高速运行。

参考

AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

pandas

#量化交易

#股票数据