如何按SYMBOL列批量拆分CSV文件并实现每日数据追加？

核心思路 #

1. 遍历指定目录下的所有ZIP压缩包，提取其中的CSV文件
2. 分块读取大型CSV，避免内存溢出
3. 按SYMBOL列自动分组，将每组数据追加到对应命名的CSV文件（首次创建写入表头，后续仅追加数据）

完整代码实现 #

import os
import pandas as pd
from zipfile import ZipFile

# 配置参数
ZIP_DIR = "./zip_files"  # 存放ZIP压缩包的目录
OUTPUT_DIR = "./symbol_csvs"  # 拆分后CSV的输出目录

# 创建输出目录（不存在则自动生成）
os.makedirs(OUTPUT_DIR, exist_ok=True)

# 遍历所有ZIP文件
for zip_filename in os.listdir(ZIP_DIR):
    if not zip_filename.endswith(".zip"):
        continue
    
    zip_path = os.path.join(ZIP_DIR, zip_filename)
    print(f"正在处理 {zip_path}...")
    
    # 打开ZIP包并读取内部CSV
    with ZipFile(zip_path, "r") as zip_ref:
        for csv_filename in zip_ref.namelist():
            if not csv_filename.endswith(".csv"):
                continue
            
            # 分块读取大型CSV，chunksize可根据内存调整
            chunk_iter = pd.read_csv(zip_ref.open(csv_filename), chunksize=10000)
            
            for chunk in chunk_iter:
                # 按SYMBOL分组处理每组数据
                for symbol, group in chunk.groupby("SYMBOL"):
                    output_file = os.path.join(OUTPUT_DIR, f"{symbol}.csv")
                    
                    # 判断文件是否存在：首次创建写表头，后续追加跳过表头
                    write_header = not os.path.exists(output_file)
                    group.to_csv(output_file, mode="a", index=False, header=write_header, encoding="utf-8")

print("所有文件处理完成")

关键细节说明 #

• 分块读取：通过chunksize将大文件拆分为小块处理，避免内存过载，适配超大型CSV场景
• 自动分组：利用groupby("SYMBOL")自动识别所有出现的符号，无需手动枚举约200个SYMBOL值
• 追加逻辑：通过os.path.exists()判断目标文件状态，确保首次创建时写入完整表头，后续仅追加数据行
• 批量处理ZIP：自动遍历指定目录下的所有ZIP包，以及每个ZIP内的所有CSV文件，无需手动解压

扩展优化建议 #

• 若每日数据存在重复条目，可追加前通过TIMESTAMP或主键列去重
• 可添加日志模块，记录每个SYMBOL的处理进度或异常信息
• 针对超大规模文件，可替换为dask.dataframe提升处理效率

内容的提问来源于stack exchange，提问作者kaushik anadkat