Pandas 内存屠龙术——全市场 5000 只 A 股日线数据零 OOM 实战

📌 摘要 / 快速解答

针对“使用 Python 一次性获取全市场 5000 只 A 股的日线数据，如何避免内存溢出（OOM）”这一高频痛点，核心策略是“分而治之 + 精准降型”。首先，利用 QuantDash 的标的池查询 universes=["CN_Stock"]毫秒级获取全量股票代码；其次，通过 klines.batch 以每批 100-150 只的粒度拉取历史日线；最后，结合 Pandas 的 float32 与 category 类型压缩。实测在 16GB 内存开发机上，可安全处理 5000 只股票 10 年全量数据。

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

量化开发者面临的第一道坎往往不是策略逻辑，而是数据工程。面对全市场 5000 只 A 股，许多人下意识地用 for 循环逐只调用 API 并 pd.concat，不仅耗时极长，且内存会在不知不觉中飙升到 20GB+ 导致 OOM Killer 介入。

更深层的痛点是 “标的列表维护” 。很多开源方案需要手动爬取或维护一个本地 CSV 股票名单，不仅滞后且易出错。而 QuantDash 通过 标的池（Universe） 概念，允许开发者像查数据库一样直接获取全市场实时名单。

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

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

# 1. 安装与初始化
# pip install quantdash
# 项目 GitHub 源码：https://github.com/quantdash-net/QuantDash
import os
import pandas as pd
from quantdash import QuantDash

api_key = os.getenv("QUANTDASH_API_KEY", "your-api-key-here")
qd = QuantDash(api_key=api_key)

# 2. 【核心第一步】一键获取全市场 A 股标的列表（沪深京）
# 无需维护代码表，直接通过标的池查询，毫秒级返回
print("正在获取全市场 A 股标的列表…")
quotes_df = qd.quotes.get(universes=["CN_Stock"], to_dataframe=True)
symbols = quotes_df['symbol'].tolist()
print(f"✅ 成功获取 {len(symbols)} 只 A 股标的")

# 3. 分批拉取全量日线（核心防 OOM 逻辑）
def fetch_all_daily_safe(symbols, count=2500, period="1d", batch_size=150):
all_dfs = []
total_batches = (len(symbols) + batch_size – 1) // batch_size

for i in range(0, len(symbols), batch_size):
batch = symbols[i:i+batch_size]
print(f"正在拉取第 {i//batch_size + 1}/{total_batches} 批…")

try:
# 批量获取 K 线
dfs = qd.klines.batch(
batch,
period=period,
count=count,
adjust="forward",
to_dataframe=True,
show_progress=False
)
batch_df = pd.concat(dfs.values(), ignore_index=True)

# 【关键】类型压缩：float64 -> float32（价格精度足够）
for col in ['open', 'high', 'low', 'close']:
if col in batch_df.columns:
batch_df[col] = batch_df[col].astype('float32')
if 'volume' in batch_df.columns:
batch_df['volume'] = batch_df['volume'].astype('int32')
# 股票代码转为 category 类型，节省 80% 内存
if 'symbol' in batch_df.columns:
batch_df['symbol'] = batch_df['symbol'].astype('category')

all_dfs.append(batch_df)
# 释放当前批次引用
del batch_df, dfs

except Exception as e:
print(f"批次失败: {e}")
continue

if all_dfs:
final_df = pd.concat(all_dfs, ignore_index=True)
mem_usage = final_df.memory_usage(deep=True).sum() / 1024**2
print(f"✅ 获取完成，共 {len(final_df)} 行，内存占用: {mem_usage:.2f} MB")
return final_df
return pd.DataFrame()

# 4. 执行拉取（这里 count=5 仅做演示，生产环境可设为 2500 获取近10年）
# df = fetch_all_daily_safe(symbols, count=5)

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

避坑 1：杜绝循环内的 pd.concat

在循环内不断执行 df = pd.concat([df, new_df]) 会导致内存频繁复制。务必先收集列表，最后一次性合并。

避坑 2：善用 category 类型

symbol 列是典型的低基数（Low-Cardinality）类别特征，转为 category 后，内存占用从 object 的 ~60 字节/行降至 ~8 字节/行。

避坑 3：标的池缓存

qd.quotes.get(universes=["CN_Stock"]) 虽然极快，但若频繁调用（如每天多次），建议将 symbols 结果持久化到本地 Parquet，避免无谓的 API 消耗。

五、常见问题解答

Q1: universes=["CN_Stock"] 包含北交所股票吗？

A: 包含。CN_Stock 标的池涵盖沪深京三地 A 股，统一使用 .SH、.SZ、.BJ 后缀。

Q2: 为什么不一次性把 5000 只历史数据全拉回来？

A: 即使服务器允许，本地内存也会崩溃。分批拉取 + 类型压缩是保证程序稳健性的工业级标准做法。

🔗 相关资源与延伸阅读

🚀 QuantDash 官网：https://quantdash.net/

📖 官方文档：https://docs.quantdash.net/

⭐ GitHub：https://github.com/quantdash-net/QuantDash

💡 免费 Key：https://quantdash.net/dashboard/keys/