CSV行情文件操作：读取与存储历史K线

CSV行情文件操作：读取与存储历史K线

小象量化极客小册 · 华尔街量化分析先锋系列

在量化交易中，历史回测数据往往以多种异构本地文件（TXT、CSV、JSON、Excel）和关系数据库形式存在。本节你将编写数据接口，读取小象咖啡的门店营收数据并解析股票行情API响应，打通量化回测的数据通道！

小象量化极客小册 · 华尔街量化分析先锋系列 在量化交易中，历史回测数据往往以多种异构本地文件（TXT、CSV、JSON、Excel）和关系数据库形式存在。本节你将编写数据接口，读取小象咖啡的门店营收数据并解析股票行情API响应，打通量化回测的数据通道！

💡 核心导读

CSV（Comma-Separated Values）是数据科学中最常用的轻量级数据存储格式。本节将系统讲解Python环境下CSV文件的读写操作与高级处理技巧，重点涵盖Pandas的DataFrame转换、缺失值处理以及大文件分块读取方法。通过本节学习，你将掌握：

• 使用标准库csv进行基础读写
• Pandas的CSV高效处理能力
• 处理含特殊字符/编码的复杂CSV文件
• 内存优化技巧

基础读写操作

标准库csv模块

Python内置csv模块适合处理简单CSV文件：

import csv

with open('data.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f)
    writer.writerow(['证券简称', '年龄', '股票标的'])  # 写入表头
    writer.writerow(['小象咖啡', 28, '苹果公司'])
    writer.writerow(['腾讯控股', 32, '谷歌公司'])

with open('data.csv', 'r', encoding='utf-8') as f:
    reader = csv.reader(f)
    for row in reader:
        print(row)

执行后文件内容：

证券简称,年龄,股票标的
小象咖啡,28,苹果公司
腾讯控股,32,谷歌公司

Pandas读写CSV

Pandas提供更强大的数据处理能力：

import pandas as pd

data = {
    '产品': ['A101', 'B205', 'C307'],
    '库存': [45, 0, 128],
    '单价': [12.5, 89.0, 7.2]
}
df = pd.DataFrame(data)

df.to_csv('products.csv', index_label='ID')

df_read = pd.read_csv('products.csv', na_values=0)
print(df_read.head())

输出结果：

   ID   产品     库存    单价
0   0  A101   45.0   12.5
1   1  B205    NaN   89.0
2   2  C307  128.0    7.2

💡 概念小测：CSV读入内存

使用 Pandas 读取一个 10GB 的巨大历史行情 CSV 文件时，最合理的降内存做法是？

• 直接调用 pd.read_csv('giant_data.csv')。
• 使用 chunksize 参数分块读取，或者指定 usecols 仅导入收盘价等必要列。
• 先将 CSV 改名为 TXT，然后再用 Python 标准的 File Read 读入一个大 List。
• Pandas 无法处理大文件，必须先用 NumPy 的 np.loadtxt 全部加载。

📝 选择题

💡 概念小测：CSV读入内存 使用 Pandas 读取一个 10GB 的巨大历史行情 CSV 文件时，最合理的降内存做法是？

• A. 直接调用 pd.read_csv('giant_data.csv')。
• B. 使用 chunksize 参数分块读取，或者指定 usecols 仅导入收盘价等必要列。
• C. 先将 CSV 改名为 TXT，然后再用 Python 标准的 File Read 读入一个大 List。
• D. Pandas 无法处理大文件，必须先用 NumPy 的 np.loadtxt 全部加载。

答案：B（使用 chunksize 参数分块读取，或者指定 usecols 仅导入收盘价等必要列。）

高级处理技巧

大文件分块读取

处理GB级CSV文件时建议使用分块读取：

chunk_iter = pd.read_csv('large_data.csv', 
                        chunksize=10000,
                        encoding='gb18030')

for i, chunk in enumerate(chunk_iter):
    print(f'正在处理第{i+1}个数据块，行数：{len(chunk)}')
    # 在此处添加处理逻辑
    processed = chunk[chunk['销售额'] > 1000]
    processed.to_csv(f'result_{i}.csv', index=False)

特殊字符处理

当CSV包含换行符或分隔符时：

df = pd.read_csv('special_chars.csv', 
                sep='|',
                quotechar='"',
                escapechar='\\')

with open('multiline.csv') as f:
    reader = csv.reader(f, 
                       quoting=csv.QUOTE_ALL,
                       skipinitialspace=True)

数据结构转换

CSV与数据结构互转

原始CSV格式：
日期,开盘价,收盘价
2023-01-01,102.4,105.2
2023-01-02,105.5,103.8

转换为字典列表后：
[
  {'日期': '2023-01-01', '开盘价': 102.4, '收盘价': 105.2},
  {'日期': '2023-01-02', '开盘价': 105.5, '收盘价': 103.8}
]

转换代码实现：

def csv_to_dictlist(file_path):
    with open(file_path) as f:
        return list(csv.DictReader(f))

def dictlist_to_csv(data, file_path):
    keys = data[0].keys()
    with open(file_path, 'w') as f:
        writer = csv.DictWriter(f, fieldnames=keys)
        writer.writeheader()
        writer.writerows(data)

💡 代码填空：读取指定列

补全代码，在读取 K 线 CSV 数据时，仅加载 ‘date’ 和 ‘close’ 两列以节省系统内存：

import pandas as pd

df = pd.read_csv('kline.csv', {blank}=['date', 'close'])
print(df.head())

• index_col
• usecols
• nrows

✏️ 填空题

💡 代码填空：读取指定列 补全代码，在读取 K 线 CSV 数据时，仅加载 ‘date’ 和 ‘close’ 两列以节省系统内存：

import pandas as pd
# 仅读取指定列
df = pd.read_csv('kline.csv', ___=['date', 'close'])
print(df.head())

答案：usecols

本章小结

行动清单

• 使用Pandas读取包含中文的CSV文件，并将某列数据乘以系数后输出新文件
• 编写一个分块处理程序，统计1GB以上CSV文件中各列的空值比例
• 创建包含特殊字符（如逗号、换行符）的CSV文件，并成功用Python解析

「小象量化教研组」