Parquet + DuckDB 个人量化海量K线数据存储方案

Parquet + DuckDB 个人量化海量K线数据存储方案

本教程面向个人量化交易者，介绍如何使用Parquet 文件 + DuckDB这一极致轻量方案，高效存储和管理海量 K 线数据。无需安装数据库服务，所有操作在 Python 环境中完成，适合个人电脑使用。

1. 方案概述

1.1 核心特点

• 零配置：无需安装和运行数据库服务，DuckDB 作为嵌入式数据库直接运行在 Python 进程中
• 极致压缩：Parquet 列式存储 + ZSTD 压缩，1 亿根 K 线（约 10-20GB 原始 CSV）可压缩至2-5GB
• 高性能分析：DuckDB 针对 OLAP 场景优化，对 1 亿条数据的聚合查询通常在毫秒到秒级完成
• Python 原生：与 Pandas、PyArrow 无缝集成，可直接查询 Parquet 文件

1.2 适用场景

• 个人量化回测系统
• 数据探索性分析
• 每日数据增量更新
• 不需要并发写入的场景

2. 环境搭建

2.1 安装 DuckDB

# 基础安装pipinstallduckdb# 推荐安装完整依赖（包含 Parquet 支持和数据分析工具）pipinstallduckdb pandas pyarrow jupyter

2.2 验证安装

importduckdbimportpandasaspd# 创建内存数据库连接conn=duckdb.connect(':memory:')# 验证 DuckDB 版本print(conn.execute("SELECT version()").fetchone())# 输出示例: ('v1.5.0',)

2.3 可选：安装命令行工具

# 从官网下载 DuckDB CLI 工具，或使用 pippipinstallduckdb-cli# 启动命令行duckdb

CLI 工具适合快速查询和调试。

3. 数据存储架构设计

3.1 目录结构规划

data/ ├── raw/ # 原始数据（可选备份） │ ├── 2025/ │ │ ├── 000001.SZ_1min.csv │ │ ├── 600188.SH_1min.csv │ │ └── ... │ └── 2026/ ├── kline/ # 主要数据存储 │ ├── daily/ # 日线数据 │ │ ├── 000001.SZ_daily.parquet │ │ ├── 600188.SH_daily.parquet │ │ └── ... │ └── minute/ # 分钟线数据 │ ├── 000001.SZ_5min.parquet │ ├── 600188.SH_5min.parquet │ └── ... └── metadata/ # 元数据 └── symbols.json # 品种列表及元信息

3.2 分文件存储策略

推荐按品种（symbol）拆分文件，而非将所有数据存入单个大文件：

• 优点：
  • 查询单品种时只需读取对应文件，大幅减少 I/O
  • 增量更新更方便（只需更新变化的文件）
  • 便于并行处理和备份
• 文件名格式：{symbol}_{period}.parquet，如000001.SZ_1min.parquet

3.3 数据 schema 设计

每个 Parquet 文件包含以下字段：

# 定义标准 schemaKLINE_SCHEMA={'timestamp':'datetime64[ms]','open':'float32','high':'float32','low':'float32','close':'float32','volume':'int64','volume':'float64'}

4. 数据写入与存储

4.1 从 CSV 转换并写入 Parquet

importduckdbimportpandasaspdimportosfrompathlibimportPathdefconvert_csv_to_parquet(csv_path:str,output_dir:str,symbol:str,period:str='1min'):""" 将单个品种的 CSV 文件转换为 Parquet 格式 Args: csv_path: CSV 文件路径 output_dir: 输出目录 symbol: 品种代码 period: K线周期（1min, 5min, daily等） """# 读取 CSVdf=pd.read_csv(csv_path)# 转换时间列df['timestamp']=pd.to_datetime(df['timestamp'])# 按时间排序（重要：确保时间有序）df=df.sort_values('timestamp')# 数据类型优化df['open']=df['open'].astype('float32')df['high']=df['high'].astype('float32')df['low']=df['low'].astype('float32')df['close']=df['close'].astype('float32')df['volume']=df['volume'].astype('int64')df['amount']=df['amount'].astype('float64')# 输出文件路径output_path=Path(output_dir)/f"{symbol}_{period}.parquet"output_path.parent.mkdir(parents=True,exist_ok=True)# 写入 Parquet（使用 ZSTD 压缩，压缩率高）df.to_parquet(output_path,compression='zstd',compression_level=3# 1-22，3 是速度与压缩比的平衡点)print(f"已写入{len(df)}条记录到{output_path}")returnoutput_path# 使用示例convert_csv_to_parquet(csv_path='data/raw/2026/600188.SH_daily.csv',output_dir='data/kline/daily/',symbol='600188.SH',period='daily')

4.2 批量转换多个品种

importglobdefbatch_convert_csv_to_parquet(csv_dir:str,output_dir:str,period:str='1min'):"""批量转换目录下所有 CSV 文件"""csv_files=glob.glob(os.path.join(csv_dir,"*.csv"))forcsv_fileincsv_files:# 从文件名提取 symbolsymbol=os.path.basename(csv_file).replace(f'_{period}.csv','')try:convert_csv_to_parquet(csv_file,output_dir,symbol,period)exceptExceptionase:print(f"转换{csv_file}失败:{e}")# 使用示例batch_convert_csv_to_parquet(csv_dir='data/raw/2024/',output_dir='data/kline/minute/',period='1min')

4.3 增量更新数据