离线也能玩转量化：手把手教你用Python解析通达信本地股票代码文件（附完整代码）

离线量化数据实战：Python解析通达信股票代码文件的完整指南

在量化交易领域，稳定可靠的数据源是策略回测和实盘运行的基础。当网络连接不稳定或需要高频访问基础股票列表时，直接从本地软件数据文件中提取信息成为了一种高效可靠的解决方案。本文将深入讲解如何利用Python解析通达信软件本地的shm.tnf（沪市）和szm.tnf（深市）二进制文件，构建一个完全离线的股票代码数据库。

1. 准备工作与环境配置

在开始解析二进制文件之前，我们需要准备好开发环境和必要的工具。通达信的股票代码文件通常位于安装目录的T0002/hq_cache/子目录下，包含shm.tnf和szm.tnf两个文件，分别存储沪市和深市的股票信息。

基础环境要求：

• Python 3.6或更高版本
• 通达信软件（任何版本均可，文件格式保持稳定）
• 文本编辑器或IDE（推荐VS Code或PyCharm）

必要的Python库：

import struct import os import pandas as pd from pathlib import Path

安装依赖库：

pip install pandas

提示：建议在虚拟环境中进行开发，避免污染全局Python环境。可以使用python -m venv tdx_parser创建虚拟环境，然后激活它。

2. 通达信二进制文件结构解析

通达信的.tnf文件采用固定格式的二进制结构，由文件头和数据体两部分组成。理解这个结构是正确解析数据的关键。

2.1 文件头结构分析

文件头占据前50个字节，包含以下信息：

文件头解析代码示例：

def parse_file_header(file_path): with open(file_path, 'rb') as f: header_data = f.read(50) ip_address = header_data[:40].decode('ascii').rstrip('\x00') port = struct.unpack('<H', header_data[40:42])[0] date = struct.unpack('<I', header_data[42:46])[0] time = struct.unpack('<I', header_data[46:50])[0] return { 'ip_address': ip_address, 'port': port, 'date': date, 'time': time }

2.2 数据体结构详解

文件头之后是数据体部分，每个股票信息占据314个字节的固定长度。关键字段分布如下：

注意：字符串字段通常以\x00填充至指定长度，解析时需要去除这些填充字符。

3. Python实现完整解析流程

现在我们将实现一个完整的解析器，从二进制文件中提取股票代码和名称，并保存为结构化数据。

3.1 核心解析函数

def parse_tdx_stock_file(file_path): stocks = [] with open(file_path, 'rb') as f: # 跳过50字节的文件头 f.seek(50) # 计算记录总数（文件大小减去文件头后除以每条记录长度） file_size = os.path.getsize(file_path) record_count = (file_size - 50) // 314 for _ in range(record_count): record_data = f.read(314) if len(record_data) < 314: break # 解析股票代码（6字节） stock_code = record_data[:6].decode('ascii').rstrip('\x00') # 解析股票名称（从23字节开始，18字节） stock_name = record_data[23:23+18].decode('gbk').rstrip('\x00') # 解析昨收盘价（276字节开始，4字节） last_close = struct.unpack('<f', record_data[276:280])[0] stocks.append({ 'code': stock_code, 'name': stock_name, 'last_close': last_close }) return stocks

3.2 处理沪深两市数据

def parse_all_stocks(tdx_dir): sh_file = Path(tdx_dir) / 'shm.tnf' sz_file = Path(tdx_dir) / 'szm.tnf' sh_stocks = parse_tdx_stock_file(sh_file) sz_stocks = parse_tdx_stock_file(sz_file) # 为股票代码添加市场前缀 for stock in sh_stocks: stock['full_code'] = 'sh' + stock['code'] for stock in sz_stocks: stock['full_code'] = 'sz' + stock['code'] return sh_stocks + sz_stocks

3.3 数据保存与导出

def save_to_csv(stocks, output_file): df = pd.DataFrame(stocks) df.to_csv(output_file, index=False, encoding='utf_8_sig') def save_to_sqlite(stocks, db_file, table_name='stocks'): import sqlite3 conn = sqlite3.connect(db_file) df = pd.DataFrame(stocks) df.to_sql(table_name, conn, if_exists='replace', index=False) conn.close()

4. 高级应用与性能优化

基础解析功能实现后，我们可以进一步优化代码并扩展功能，使其更适合量化交易场景。

4.1 多线程解析加速

对于大型文件，可以使用多线程加速解析过程：

from concurrent.futures import ThreadPoolExecutor def parallel_parse_tdx_file(file_path, chunk_size=1000): stocks = [] file_size = os.path.getsize(file_path) record_count = (file_size - 50) // 314 def parse_chunk(start, end): chunk_stocks = [] with open(file_path, 'rb') as f: f.seek(50 + start * 314) for _ in range(end - start): record_data = f.read(314) if len(record_data) < 314: break # 解析逻辑与之前相同 # ... chunk_stocks.append(stock_info) return chunk_stocks with ThreadPoolExecutor() as executor: futures = [] for i in range(0, record_count, chunk_size): end = min(i + chunk_size, record_count) futures.append(executor.submit(parse_chunk, i, end)) for future in futures: stocks.extend(future.result()) return stocks

4.2 增量更新机制

为了避免每次全量解析，可以实现增量更新：

def get_last_modified_date(file_path): # 从文件头获取日期信息 with open(file_path, 'rb') as f: header_data = f.read(50) date = struct.unpack('<I', header_data[42:46])[0] return date def incremental_update(db_file, tdx_dir): # 检查文件修改日期，只解析新数据 # 实现略... pass

4.3 数据验证与清洗

def clean_stock_data(stocks): # 移除无效代码 valid_stocks = [s for s in stocks if s['code'].isdigit()] # 标准化股票名称 for stock in valid_stocks: stock['name'] = stock['name'].strip() return valid_stocks

5. 实际应用案例

将解析的股票数据集成到量化系统中：

class TdxStockDatabase: def __init__(self, db_path): self.conn = sqlite3.connect(db_path) def get_all_stocks(self): return pd.read_sql('SELECT * FROM stocks', self.conn) def get_stock_by_code(self, code): query = 'SELECT * FROM stocks WHERE full_code = ?' return pd.read_sql(query, self.conn, params=(code,)) def search_stocks(self, keyword): query = 'SELECT * FROM stocks WHERE name LIKE ?' return pd.read_sql(query, self.conn, params=(f'%{keyword}%',)) def close(self): self.conn.close()

提示：在实际量化交易系统中，可以将此数据库与行情接收、策略引擎等模块集成，构建完整的离线分析环境。

6. 错误处理与调试技巧

处理二进制文件时可能会遇到各种问题，完善的错误处理机制必不可少。

常见问题及解决方案：

1. 文件路径错误

   • 检查通达信安装目录
   • 确认用户有读取权限

2. 文件格式不匹配

   • 验证文件大小是否符合预期
   • 检查文件头是否有效

3. 编码问题

   • 尝试不同的字符编码（gbk/gb2312）
   • 处理特殊字符

调试代码示例：

def debug_file_structure(file_path): with open(file_path, 'rb') as f: # 打印文件头 header = f.read(50) print("File Header:", header) # 打印前几条记录 for i in range(5): record = f.read(314) print(f"Record {i}:", record)

7. 扩展思路与进阶方向

掌握了基础解析方法后，可以考虑以下扩展方向：

• 实时数据监控：结合通达信的实时行情文件，构建完整的离线行情系统
• 历史数据解析：解析通达信的历史分钟线、日线数据文件
• 跨平台支持：将解析器移植到其他语言（如C++、Rust）以获得更高性能
• GUI工具开发：使用PyQt等框架开发可视化工具

# 示例：解析分钟线数据 def parse_minute_data(file_path): # 通达信分钟线数据文件结构不同 # 实现略... pass

在实际项目中，我发现将解析结果存储为SQLite数据库最为方便，既支持快速查询，又便于与Pandas等数据分析工具集成。对于超大型文件，可以考虑使用内存映射文件技术进一步提高性能。