Python量化交易学习笔记（46）——通达信日线数据获取

序

从2020年初开始接触量化，马上就要满一年了。在这一年里，想过去做量化，想过去做机器学习，想过去做少儿编程教育。就这样大概折腾了小半年时间，最后在CSDN上看到这样一句话：“你把时间投在专业上，两三年，你就能在圈子里小有名气。四五年，你就能靠这个专业赚钱。过了10年以上，你就能成为这个领域的专家。”觉得自己可以按这句话努力一把，于是便坚定了探索量化的决心。

python、backtrader、量化、机器学习基本上都是从零学起，对近一年来的学习进度总体还算满意。CSDN的粉丝量280+，博客升级为5级，访问量9万+，总排名5万+，微信群成员260+，QQ群也建立起来。本来想在QQ群里以任务方式同步自己的学习进度，无奈踩的坑实在太多，学习时间也不固定，所以还是等我理清思路，再做系统化的分享。

这里也特别感谢各位群友的支持与帮助。这篇文章就是基于群友阿猪和阳光分享的内容进行的整理。

通达信数据获取优势

快！就是快！
在使用baostock获取或者更新股票数据时，4000多只股票基本要耗时1个多小时。

在使用通达信获取数据时，除了首次通过通达信软件把数据下载到本地需要十几分钟时间外，后面每日更新数据在1分钟内就能下载完毕，在加上脚本解析数据存入csv文件也是在1分钟内即可完成。

因此，有必要掌握通过通达信来获取股票数据的方法。

通达信软件下载日线数据

首先，下载通达信软件，安装后打开，在菜单栏点击“系统”->“盘后数据下载”，弹出下面的面板：

勾选日线和实时行情数据，选择需要下载的日线时间范围，点击开始下载，日线数据就会被保存在本地。其中上证A股的日线数据保存在“通达信安装路径\vipdoc\sh\lday”目录下，文件名为“市场代码.day”，例如“sh600000.day”，深证A股的日线数据保存在“通达信安装路径\vipdoc\sz\lday”目录下，文件名格式与上证A股相同，例如“sz000001.day”。

解析数据

通过通达信下载的day文件是二进制文件，这里对day文件进行解析，保存为csv文件。

def transform_data():# 保存csv文件的目录target = proj_path + 'data/tdx/day'if not os.path.exists(target):os.makedirs(target)code_list = []source_list = ['C:/new_tdx/vipdoc/sz/lday', 'C:/new_tdx/vipdoc/sh/lday']for source in source_list:file_list = os.listdir(source)# 逐个文件进行解析for f in file_list:day2csv(source, f, target)# 获取所有股票/指数代码code_list.extend(list(map(lambda x: x[:x.rindex('.')], file_list)))# 保存所有代码列表pd.DataFrame(data=code_list, columns=['code']).to_csv(proj_path + 'data/tdx/all_codes.csv', index=False)

这里代码最后一行是把所有股票/指数代码保存在all_codes.csv文件中，便于后续更新日线数据时使用。

代码中的day2csv方法实现了日线数据由day文件向csv文件的转化。

# 将通达信的日线文件转换成CSV格式
def day2csv(source_dir, file_name, target_dir):# 以二进制方式打开源文件source_file = open(source_dir + os.sep + file_name, 'rb')buf = source_file.read()source_file.close()# 打开目标文件，后缀名为CSVtarget_file = open(target_dir + os.sep + file_name[: file_name.rindex('.')] + '.csv', 'w')buf_size = len(buf)rec_count = int(buf_size / 32)begin = 0end = 32header = str('date') + ',' + str('open') + ',' + str('high') + ',' + str('low') + ',' \+ str('close') + ',' + str('amount') + ',' + str('volume') + '\n'target_file.write(header)for i in range(rec_count):# 将字节流转换成Python数据格式# I: unsigned int# f: floata = unpack('IIIIIfII', buf[begin:end])# 处理date数据year = a[0] // 10000month = (a[0] % 10000) // 100day = (a[0] % 10000) % 100date = '{}-{:02d}-{:02d}'.format(year, month, day)line = date + ',' + str(a[1] / 100.0) + ',' + str(a[2] / 100.0) + ',' \+ str(a[3] / 100.0) + ',' + str(a[4] / 100.0) + ',' + str(a[5]) + ',' \+ str(a[6]) + '\n'target_file.write(line)begin += 32end += 32target_file.close()

day文件中，每32个字节存储了一根日线数据，各字节存储数据如下：

00 ~ 03 字节：年月日, 整型
04 ~ 07 字节：开盘价*100，整型
08 ~ 11 字节：最高价*100，整型
12 ~ 15 字节：最低价*100，整型
16 ~ 19 字节：收盘价*100，整型
20 ~ 23 字节：成交额（元），float型
24 ~ 27 字节：成交量（股），整型
28 ~ 31 字节：（保留）

更新数据（非全部下载）

假设在每日收盘后，我们都要更新csv文件中的股票当日的K线数据，显然没有必要把所有的day文件从头解析一遍，而只需要从day文件的文件末，解析出我们需要更新的日线数据即可。当然，这也需要先从通达信软件先下载盘后数据，下载方法参考本文开头的“通达信软件下载日线数据”章节。更新数据代码如下：

def update_data():# 读入所有股票/指数代码codes = pd.read_csv(proj_path + 'data/tdx/all_codes.csv')['code']for code in codes:data_path = proj_path + 'data/tdx/day/' + code + '.csv'# 读取当前已存在的数据exist_df = pd.read_csv(data_path)# 获取需要更新的日线开始时间from_date = pd.read_csv(proj_path + 'data/tdx/day/' + code + '.csv')['date'].iloc[-1]# 提取新数据data = extract_data(from_date, 'C:/new_tdx/vipdoc/' + code[0:2] + '/lday/' + code + '.day')if not len(data):continuedf = pd.DataFrame(data).rename(columns={0: 'date', 1: 'open', 2: 'high', 3: 'low', 4: 'close', 5: 'amount', 6: 'volume'})# 合并数据df = exist_df.append(df)# 保存文件df.to_csv(data_path, index=False)

其中，extract_data用于提取日期from_date后的数据：

def extract_data(from_date, file_name):# 以二进制方式打开源文件source_file = open(file_name, 'rb')buf = source_file.read()source_file.close()buf_size = len(buf)rec_count = int(buf_size / 32)# 从文件末开始访问数据begin = buf_size - 32end = buf_sizedata = []for i in range(rec_count):# 将字节流转换成Python数据格式# I: unsigned int# f: floata = unpack('IIIIIfII', buf[begin:end])# 处理date数据year = a[0] // 10000month = (a[0] % 10000) // 100day = (a[0] % 10000) % 100date = '{}-{:02d}-{:02d}'.format(year, month, day)if from_date == date:breakdata.append([date, str(a[1] / 100.0), str(a[2] / 100.0), str(a[3] / 100.0), \str(a[4] / 100.0), str(a[5]), str(a[6])])begin -= 32end -= 32# 反转数据data.reverse()return data

小结

通过解析通信达软件下载的日线数据，可以实现股票数据的快速获取。
解析2000年以来的所有股票日线数据，在我的机器上大概耗时50s。
更新当日K线数据，在我的机器上花费耗时大概为30s。

通信达日线数据解析全部代码如下：

import os
import sys
import time
import pandas as pd
from struct import unpack# 获取当前目录
proj_path = os.path.dirname(os.path.abspath(sys.argv[0])) + '/../'# 将通达信的日线文件转换成CSV格式
def day2csv(source_dir, file_name, target_dir):# 以二进制方式打开源文件source_file = open(source_dir + os.sep + file_name, 'rb')buf = source_file.read()source_file.close()# 打开目标文件，后缀名为CSVtarget_file = open(target_dir + os.sep + file_name[: file_name.rindex('.')] + '.csv', 'w')buf_size = len(buf)rec_count = int(buf_size / 32)begin = 0end = 32header = str('date') + ',' + str('open') + ',' + str('high') + ',' + str('low') + ',' \+ str('close') + ',' + str('amount') + ',' + str('volume') + '\n'target_file.write(header)for i in range(rec_count):# 将字节流转换成Python数据格式# I: unsigned int# f: floata = unpack('IIIIIfII', buf[begin:end])# 处理date数据year = a[0] // 10000month = (a[0] % 10000) // 100day = (a[0] % 10000) % 100date = '{}-{:02d}-{:02d}'.format(year, month, day)line = date + ',' + str(a[1] / 100.0) + ',' + str(a[2] / 100.0) + ',' \+ str(a[3] / 100.0) + ',' + str(a[4] / 100.0) + ',' + str(a[5]) + ',' \+ str(a[6]) + '\n'target_file.write(line)begin += 32end += 32target_file.close()def transform_data():# 保存csv文件的目录target = proj_path + 'data/tdx/day'if not os.path.exists(target):os.makedirs(target)code_list = []source_list = ['C:/new_tdx/vipdoc/sz/lday', 'C:/new_tdx/vipdoc/sh/lday']for source in source_list:file_list = os.listdir(source)# 逐个文件进行解析for f in file_list:day2csv(source, f, target)# 获取所有股票/指数代码code_list.extend(list(map(lambda x: x[:x.rindex('.')], file_list)))# 保存所有代码列表pd.DataFrame(data=code_list, columns=['code']).to_csv(proj_path + 'data/tdx/all_codes.csv', index=False)def extract_data(from_date, file_name):# 以二进制方式打开源文件source_file = open(file_name, 'rb')buf = source_file.read()source_file.close()buf_size = len(buf)rec_count = int(buf_size / 32)# 从文件末开始访问数据begin = buf_size - 32end = buf_sizedata = []for i in range(rec_count):# 将字节流转换成Python数据格式# I: unsigned int# f: floata = unpack('IIIIIfII', buf[begin:end])# 处理date数据year = a[0] // 10000month = (a[0] % 10000) // 100day = (a[0] % 10000) % 100date = '{}-{:02d}-{:02d}'.format(year, month, day)if from_date == date:breakdata.append([date, str(a[1] / 100.0), str(a[2] / 100.0), str(a[3] / 100.0), \str(a[4] / 100.0), str(a[5]), str(a[6])])begin -= 32end -= 32# 反转数据data.reverse()return datadef update_data():# 读入所有股票/指数代码codes = pd.read_csv(proj_path + 'data/tdx/all_codes.csv')['code']for code in codes:data_path = proj_path + 'data/tdx/day/' + code + '.csv'# 读取当前已存在的数据exist_df = pd.read_csv(data_path)# 获取需要更新的日线开始时间from_date = pd.read_csv(proj_path + 'data/tdx/day/' + code + '.csv')['date'].iloc[-1]# 提取新数据data = extract_data(from_date, 'C:/new_tdx/vipdoc/' + code[0:2] + '/lday/' + code + '.day')if not len(data):continuedf = pd.DataFrame(data).rename(columns={0: 'date', 1: 'open', 2: 'high', 3: 'low', 4: 'close', 5: 'amount', 6: 'volume'})# 合并数据df = exist_df.append(df)# 保存文件df.to_csv(data_path, index=False)def get_all_stock_codes():all_codes_file = proj_path + 'data/tdx/all_codes.csv'if not os.path.exists(all_codes_file):print('请先更新数据！')returndf = pd.read_csv(all_codes_file)df = df[((df['code'] >= 'sh600000') & (df['code'] <= 'sh605999')) | \((df['code'] >= 'sz000001') & (df['code'] <= 'sz003999')) | \((df['code'] >= 'sz300000') & (df['code'] <= 'sz300999'))]df.to_csv(proj_path + 'data/tdx/all_stock_codes.csv', index=False)if __name__ == '__main__':# 程序开始时的时间time_start = time.time()# 获取所有股票代码# get_all_stock_codes()# 转换所有数据transform_data()# 更新数据# update_data()# 程序结束时系统时间time_end = time.time()print('程序所耗时间：', time_end - time_start)

欢迎大家关注、点赞、转发、留言，感谢支持！
微信群用于学习交流，感兴趣的读者请扫码加微信！
QQ群（676186743）用于资料共享，欢迎加入！