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

小象实战讲义 · 量化交易Python入门：从咖啡馆到华尔街

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

💡 核心导读

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

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

基础读写操作

标准库csv模块

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

import csv
import io

# 在内存中模拟文件操作，避免实际读写文件
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(['证券简称', '年龄', '股票标的'])
writer.writerow(['小象咖啡', 28, '苹果公司'])
writer.writerow(['腾讯控股', 32, '谷歌公司'])

csv_content = output.getvalue()
print("写入的CSV内容：")
print(csv_content)

print("\n读取CSV内容：")
reader = csv.reader(io.StringIO(csv_content))
for row in reader:
    print(row)

执行后文件内容：

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


读取CSV内容：
['证券简称', '年龄', '股票标的']
['小象咖啡', '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)

print("原始DataFrame：")
print(df)
print(f"\n数据类型：\n{df.dtypes}")

# 演示读取并处理缺失值
# 模拟从CSV读取数据，使用na_values参数
df_read = pd.DataFrame(data)
# 为了演示na_values，我们创建一个包含0的副本并标记为NaN
df_with_na = df_read.replace({0: pd.NA})
print("\n读取并处理后的DataFrame（0已标记为NA）：")
print(df_with_na)
print(f"\n数据类型：\n{df_with_na.dtypes}")

输出结果：

原始DataFrame：
     产品   库存    单价
0  A101   45  12.5
1  B205    0  89.0
2  C307  128   7.2

数据类型：
产品        str
库存      int64
单价    float64
dtype: object

读取并处理后的DataFrame（0已标记为NA）：
     产品    库存    单价
0  A101    45  12.5
1  B205  <NA>  89.0
2  C307   128   7.2

数据类型：
产品        str
库存     object
单价    float64
dtype: object

高级处理技巧

大文件分块读取

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

import pandas as pd

# 构造一个示例大数据集（小规模，用于演示分块）
large_data = {
    '日期': pd.date_range('2024-01-01', periods=25000, freq='D').strftime('%Y-%m-%d'),
    '销售额': [i * 100 for i in range(1, 25001)],  # 100, 200, ... 2,500,000
    '产品代码': ['P{:03d}'.format(i % 100 + 1) for i in range(25000)]
}
df_large = pd.DataFrame(large_data)

print(f"原始数据集总行数：{len(df_large)}")

# 使用分块迭代器处理（在内存中模拟）
chunk_size = 10000
chunks = [df_large[i:i+chunk_size] for i in range(0, len(df_large), chunk_size)]

for i, chunk in enumerate(chunks):
    print(f'正在处理第{i+1}个数据块，行数：{len(chunk)}')
    # 在此处添加处理逻辑
    processed = chunk[chunk['销售额'] > 1000]
    # 演示保存结果（注释说明）
    # processed.to_csv(f'result_{i}.csv', index=False)
    print(f'  筛选后行数：{len(processed)}')

原始数据集总行数：25000
正在处理第1个数据块，行数：10000
  筛选后行数：9990
正在处理第2个数据块，行数：10000
  筛选后行数：10000
正在处理第3个数据块，行数：5000
  筛选后行数：5000

特殊字符处理

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

import csv
import pandas as pd
import io

# 构造包含特殊字符和多行字段的CSV内容
special_csv_content = '''证券代码|名称|描述
000001|平安银行|"这是一家\n大型银行"
000002|万科A|"房地产|龙头企业"
000003|PT金田A|"特殊处理\\"ST\\"股票"'''

print("原始CSV内容：")
print(special_csv_content)
print("\n" + "="*50 + "\n")

# 使用pandas读取，指定分隔符、引号字符和转义字符
df_pandas = pd.read_csv(io.StringIO(special_csv_content),
                        sep='|',
                        quotechar='"',
                        escapechar='\\')
print("Pandas读取结果：")
print(df_pandas)
print("\n" + "="*50 + "\n")

# 使用csv模块读取多行字段
multiline_content = '''"证券代码","名称","描述"
"000001","平安银行","这是一家
大型银行"
"000002","万科A","房地产龙头企业"'''

print("多行CSV内容：")
print(multiline_content)
print("\n" + "="*50 + "\n")

reader = csv.reader(io.StringIO(multiline_content),
                    quoting=csv.QUOTE_ALL,
                    skipinitialspace=True)
print("csv.reader逐行读取结果：")
for row in reader:
    print(row)

数据结构转换

CSV与数据结构互转

原始CSV内容：
证券代码|名称|描述
000001|平安银行|"这是一家
大型银行"
000002|万科A|"房地产|龙头企业"
000003|PT金田A|"特殊处理\"ST\"股票"

==================================================

Pandas读取结果：
   证券代码     名称          描述
0     1   平安银行  这是一家\n大型银行
1     2    万科A    房地产|龙头企业
2     3  PT金田A  特殊处理"ST"股票

==================================================

多行CSV内容：
"证券代码","名称","描述"
"000001","平安银行","这是一家
大型银行"
"000002","万科A","房地产龙头企业"

==================================================

csv.reader逐行读取结果：
['证券代码', '名称', '描述']
['000001', '平安银行', '这是一家\n大型银行']
['000002', '万科A', '房地产龙头企业']

转换代码实现：

import csv
import io

def csv_to_dictlist(csv_content):
    """将CSV字符串转换为字典列表"""
    reader = csv.DictReader(io.StringIO(csv_content))
    return list(reader)

def dictlist_to_csv(data, file_path=None):
    """将字典列表转换为CSV字符串（或写入文件）"""
    if not data:
        return ""
    keys = data[0].keys()
    output = io.StringIO()
    writer = csv.DictWriter(output, fieldnames=keys)
    writer.writeheader()
    writer.writerows(data)
    result = output.getvalue()
    
    if file_path:
        # 实际应用中可写入文件，此处仅注释说明
        # with open(file_path, 'w', newline='', encoding='utf-8') as f:
        #     f.write(result)
        print(f"（模拟）数据已写入 {file_path}")
    
    return result

# 示例数据
sample_csv = """date,open,high,low,close,volume
2024-01-02,100.0,102.5,99.5,101.2,1000000
2024-01-03,101.5,103.0,100.8,102.0,1200000
2024-01-04,102.2,104.5,101.5,103.8,1500000"""

print("原始CSV：")
print(sample_csv)
print("\n" + "="*50 + "\n")

# 转换为字典列表
dict_list = csv_to_dictlist(sample_csv)
print("转换后的字典列表：")
for i, item in enumerate(dict_list):
    print(f"行{i+1}: {item}")

print("\n" + "="*50 + "\n")

# 添加一行新数据
new_row = {'date': '2024-01-05', 'open': 103.9, 'high': 105.0, 'low': 103.0, 'close': 104.5, 'volume': 1800000}
dict_list.append(new_row)

# 转换回CSV
new_csv = dictlist_to_csv(dict_list, 'updated_kline.csv')
print("更新后的CSV内容：")
print(new_csv)

原始CSV：
date,open,high,low,close,volume
2024-01-02,100.0,102.5,99.5,101.2,1000000
2024-01-03,101.5,103.0,100.8,102.0,1200000
2024-01-04,102.2,104.5,101.5,103.8,1500000

==================================================

转换后的字典列表：
行1: {'date': '2024-01-02', 'open': '100.0', 'high': '102.5', 'low': '99.5', 'close': '101.2', 'volume': '1000000'}
行2: {'date': '2024-01-03', 'open': '101.5', 'high': '103.0', 'low': '100.8', 'close': '102.0', 'volume': '1200000'}
行3: {'date': '2024-01-04', 'open': '102.2', 'high': '104.5', 'low': '101.5', 'close': '103.8', 'volume': '1500000'}

==================================================

（模拟）数据已写入 updated_kline.csv
更新后的CSV内容：
date,open,high,low,close,volume
2024-01-02,100.0,102.5,99.5,101.2,1000000
2024-01-03,101.5,103.0,100.8,102.0,1200000
2024-01-04,102.2,104.5,101.5,103.8,1500000
2024-01-05,103.9,105.0,103.0,104.5,1800000

📝 动手练一练

第 1 题 · 概念应用

你正在为一只股票策略构建数据预处理管道。原始行情 CSV 文件包含 date（日期）、open（开盘）、high（最高）、low（最低）、close（收盘）、volume（成交量）以及 turnover（成交额）、pe_ratio（市盈率）等共计 20 列。你发现该文件有 5GB 大小，而你的分析仅需要计算每日的收盘价收益率序列，后续还要直接用 Pandas/Numpy 做滚动统计与向量化计算。以下哪种做法能在最省内存的同时最高效地衔接后续 Pandas 分析？

• A. 使用 pd.read_csv('data.csv', dtype={'close': 'float32'}) 读取全部数据，然后只保留 date 和 close 列进行计算。
• B. 使用 pd.read_csv('data.csv', usecols=['date', 'close'], dtype={'close': 'float32'}) 仅读取所需列，并指定更小的数据类型。
• C. 使用 csv.DictReader 逐行读取文件，在循环中手动提取 date 和 close 字段。
• D. 将 CSV 文件导入 SQLite 数据库，然后用 SQL 查询 date 和 close 列。

第 2 题 · 动手题（代码填空）

你从数据供应商处获得了一个格式“非标准”的股票行情 CSV 字符串 dirty_csv_content：

• 使用竖线 | 作为分隔符；
• 部分文本字段（如描述）本身也包含竖线 |，这些字段被双引号 " 包裹，包裹内的竖线不应被当作分隔符。

请补全下方代码中的分隔符参数，使用 Pandas 正确读取这个内存中的 CSV 字符串。

import pandas as pd
import io

# 模拟从文件读取的"脏"CSV内容
dirty_csv_content = """symbol|name|description
600519.SH|贵州茅台|"白酒|龙头企业"
000858.SZ|五粮液|白酒生产销售
300750.SZ|宁德时代|"新能源|动力电池"
"""

print("原始'脏'CSV内容：")
print(dirty_csv_content)
print("\n" + "="*50)

# 指定竖线为分隔符；引号包裹字段内的竖线不作为分隔符（quotechar 默认就是双引号）
df = pd.read_csv(io.StringIO(dirty_csv_content),
                 sep=______,       # 指定分隔符
                 quotechar='"')

print("\n解析后的整洁 DataFrame：")
print(df)
print(f"\n列名: {df.columns.tolist()}")
print(f"第一行描述字段: {df.loc[0, 'description']}")

import pandas as pd
import io

# 模拟从文件读取的"脏"CSV内容
dirty_csv_content = """symbol|name|description
600519.SH|贵州茅台|"白酒|龙头企业"
000858.SZ|五粮液|白酒生产销售
300750.SZ|宁德时代|"新能源|动力电池"
"""

print("原始'脏'CSV内容：")
print(dirty_csv_content)
print("\n" + "="*50)

# 指定竖线为分隔符；引号包裹字段内的竖线不作为分隔符（quotechar 默认就是双引号）
df = pd.read_csv(io.StringIO(dirty_csv_content),
                 sep='|',
                 quotechar='"')

print("\n解析后的整洁 DataFrame：")
print(df)
print(f"\n列名: {df.columns.tolist()}")
print(f"第一行描述字段: {df.loc[0, 'description']}")

原始'脏'CSV内容：
symbol|name|description
600519.SH|贵州茅台|"白酒|龙头企业"
000858.SZ|五粮液|白酒生产销售
300750.SZ|宁德时代|"新能源|动力电池"


==================================================

解析后的整洁 DataFrame：
      symbol  name description
0  600519.SH  贵州茅台     白酒|龙头企业
1  000858.SZ   五粮液      白酒生产销售
2  300750.SZ  宁德时代    新能源|动力电池

列名: ['symbol', 'name', 'description']
第一行描述字段: 白酒|龙头企业

本章小结

📋 行动清单

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

—— 小象教研组