Python实战：解析通达信day文件并转换为CSV，助力期货历史回测

1. 为什么需要解析通达信day文件

做期货量化交易的朋友都知道，历史数据是回测的基础。通达信的day文件包含了丰富的期货历史交易数据，但它的二进制格式让很多新手望而却步。我自己刚开始做量化时，就曾被这个数据格式困扰了很久。

day文件包含了每个交易日的开盘价、最高价、最低价、收盘价、成交量等重要数据。这些数据对于构建交易策略至关重要。但直接用Python读取这些二进制文件，对新手来说确实不太友好。这就是为什么我们需要把它转换成更易读的CSV格式。

CSV格式有几个明显优势：可以用Excel直接打开查看，可以用pandas轻松处理，可以和其他数据源无缝对接。我见过不少交易员还在手动记录数据，其实只要掌握这个转换技巧，效率能提升好几倍。

2. 理解通达信day文件结构

要正确解析day文件，首先得了解它的二进制结构。经过多次测试和验证，我发现每个交易日的数据记录固定占用32字节。具体结构是这样的：

• 0-3字节：日期（4字节整型，格式YYYYMMDD）
• 4-7字节：开盘价（4字节浮点数）
• 8-11字节：最高价（4字节浮点数）
• 12-15字节：最低价（4字节浮点数）
• 16-19字节：收盘价（4字节浮点数）
• 20-23字节：持仓量（4字节整型）
• 24-27字节：成交量（4字节整型）
• 28-31字节：结算价（4字节浮点数）

这里有个坑要注意：通达信的日期存储方式比较特殊。比如20230115会存储为整数20230115，而不是时间戳。我在第一次尝试时就被这个细节坑过，解析出来的日期全是错的。

3. 准备Python解析环境

在开始写代码前，我们需要准备好Python环境。我推荐使用Anaconda，它自带了数据分析常用的库。需要安装的依赖其实很少，主要是以下几个：

import os import struct import datetime

struct模块是关键，它负责处理二进制数据的解包。datetime用于日期格式化。如果你已经安装了Python标准库，这些模块应该都已经自带了。

建议新建一个专门的文件夹来存放day文件和转换后的CSV文件。我通常这样组织目录结构：

/project /raw_data # 存放原始day文件 /csv_data # 存放转换后的CSV converter.py # 转换脚本

4. 完整代码实现与解析

下面是我优化过的完整转换代码，比原始版本增加了错误处理和日志输出：

def convert_day_to_csv(input_path, output_dir): """ 将通达信day文件转换为CSV格式 :param input_path: 输入的day文件路径 :param output_dir: 输出的CSV文件目录 """ try: with open(input_path, 'rb') as f: filename = os.path.basename(input_path).split('.')[0] output_path = os.path.join(output_dir, f"{filename}.csv") with open(output_path, 'w', encoding='utf-8') as csv_file: # 写入CSV表头 csv_file.write("Date,Open,High,Low,Close,OpenInterest,Volume,Settlement\n") while True: # 读取32字节的数据块 data = f.read(32) if not data: break # 解析各个字段 date = struct.unpack('i', data[0:4])[0] open_price = struct.unpack('f', data[4:8])[0] high = struct.unpack('f', data[8:12])[0] low = struct.unpack('f', data[12:16])[0] close = struct.unpack('f', data[16:20])[0] open_interest = struct.unpack('i', data[20:24])[0] volume = struct.unpack('i', data[24:28])[0] settlement = struct.unpack('f', data[28:32])[0] # 格式化日期 date_str = datetime.datetime.strptime(str(date), '%Y%m%d').strftime('%Y-%m-%d') # 写入CSV行 csv_line = f"{date_str},{open_price:.2f},{high:.2f},{low:.2f},{close:.2f},{open_interest},{volume},{settlement:.2f}\n" csv_file.write(csv_line) except Exception as e: print(f"转换失败: {input_path}, 错误: {str(e)}")

这个版本有几个改进点：

1. 使用了更规范的函数参数命名
2. 增加了异常处理
3. 输出价格保留了两位小数
4. 日期格式化更准确
5. 添加了详细的注释

5. 批量转换与自动化处理

实际使用时，我们通常需要批量转换多个day文件。这里分享一个我常用的批量处理脚本：

def batch_convert(input_dir, output_dir): """ 批量转换目录下的所有day文件 :param input_dir: 输入目录 :param output_dir: 输出目录 """ if not os.path.exists(output_dir): os.makedirs(output_dir) count = 0 for filename in os.listdir(input_dir): if filename.endswith('.day'): input_path = os.path.join(input_dir, filename) try: convert_day_to_csv(input_path, output_dir) count += 1 print(f"成功转换: {filename}") except Exception as e: print(f"转换失败: {filename}, 错误: {str(e)}") print(f"转换完成，共处理{count}个文件")

使用示例：

if __name__ == '__main__': # 配置路径 raw_data_dir = 'C:/tdx/vipdoc/ds/lday' # 通达信day文件目录 csv_output_dir = './csv_data' # CSV输出目录 # 执行批量转换 batch_convert(raw_data_dir, csv_output_dir)

我建议把这个脚本设置为定时任务，每天收盘后自动更新数据。这样就能始终保持数据是最新的，回测时直接用最新数据即可。

6. 数据验证与常见问题

转换完成后，一定要验证数据的准确性。我总结了几种常见的验证方法：

1. 基础检查：

   • 确保日期是连续的
   • 检查最高价是否≥最低价
   • 确认收盘价在最高价和最低价之间

2. 对比验证：

   • 随机抽取几天数据，与通达信软件显示的数据对比
   • 检查成交量是否合理

3. 统计分析：

   • 计算基本统计量（均值、标准差）
   • 检查价格变动范围是否合理

常见问题及解决方案：

问题1：转换后的CSV文件乱码

• 解决方案：在打开CSV文件时指定encoding='utf-8'

问题2：日期格式错误

• 解决方案：检查struct.unpack的参数是否正确，确保使用'i'而不是'l'

问题3：价格显示为科学计数法

• 解决方案：在写入CSV时使用f"{price:.2f}"格式化

7. 在回测中的应用技巧

有了CSV格式的历史数据，我们就可以用pandas轻松加载和分析：

import pandas as pd def load_csv_data(filepath): """ 加载CSV数据到DataFrame """ df = pd.read_csv(filepath) df['Date'] = pd.to_datetime(df['Date']) df.set_index('Date', inplace=True) return df

回测时的几个实用技巧：

1. 数据清洗：

   # 处理缺失值 df.fillna(method='ffill', inplace=True) # 去除异常值 df = df[(df['High'] >= df['Low']) & (df['Volume'] > 0)]

2. 计算技术指标：

   # 计算20日均线 df['MA20'] = df['Close'].rolling(20).mean() # 计算MACD exp12 = df['Close'].ewm(span=12, adjust=False).mean() exp26 = df['Close'].ewm(span=26, adjust=False).mean() df['MACD'] = exp12 - exp26 df['Signal'] = df['MACD'].ewm(span=9, adjust=False).mean()

3. 回测框架集成： 我通常会把转换好的数据导入Backtrader或Zipline等回测框架。以Backtrader为例：

   from backtrader.feeds import GenericCSVData class TdxCSVData(GenericCSVData): params = ( ('dtformat', '%Y-%m-%d'), ('datetime', 0), ('open', 1), ('high', 2), ('low', 3), ('close', 4), ('volume', 6), ('openinterest', 5), )

8. 性能优化建议

当处理大量历史数据时，性能就变得很重要。我总结了几点优化经验：

1. 使用多进程：

   from multiprocessing import Pool def parallel_convert(file_list, output_dir): with Pool() as pool: pool.starmap(convert_day_to_csv, [(f, output_dir) for f in file_list])

2. 内存优化：

   • 逐行处理而不是一次性读取整个文件
   • 使用生成器减少内存占用

3. 缓存机制：

   • 对已转换的文件做标记，避免重复转换
   • 使用hash检查文件是否修改过

4. 使用更高效的数据结构：

   import numpy as np # 使用numpy数组存储数据 data = np.zeros((record_count, 8), dtype=np.float32)

对于超大规模数据，我建议考虑使用Dask或者PySpark这样的分布式计算框架。不过对于一般的期货回测需求，上面的优化方法已经足够了。