1. 引言

通达信是国内主流的股票分析软件之一，其本地数据文件包含了丰富的股票历史行情数据。对于量化交易研究者和数据分析师来说，能够直接读取这些本地数据文件，可以避免网络延迟、实现离线分析，并充分利用通达信软件积累的历史数据。本文将详细介绍如何使用Python读取通达信本地数据文件，包括日线、5分钟线等不同周期的数据。

2. 通达信数据文件结构

通达信的数据文件通常存储在软件的安装目录下，主要包含以下几种类型：

2.1 日线数据文件

• 文件路径：T0002/hq_cache/*.day
• 文件名格式：sh000001.day（上证指数）、sz000001.day（平安银行）
• 文件结构：二进制格式，每条记录包含日期、开盘价、最高价、最低价、收盘价、成交量、成交额

2.2 5分钟线数据文件

• 文件路径：T0002/hq_cache/*.lc5、T0002/hq_cache/*.lc1
• 文件名格式：sh000001.lc5（5分钟线）、sh000001.lc1（1分钟线）
• 文件结构：二进制格式，记录更细粒度的分时数据

2.3 基础信息文件

• 文件路径：T0002/hq_cache/tdxhy.cfg、T0002/hq_cache/tdxzs.cfg
• 内容：板块分类、指数信息等元数据

3. 环境准备与依赖安装

在开始读取通达信数据之前，需要确保Python环境已安装必要的库：

pip install pandas numpy struct

主要依赖说明：

• pandas: 数据处理和分析的核心库
• numpy: 数值计算，处理二进制数据转换
• struct: Python标准库，用于解析二进制数据格式

4. 读取日线数据文件

4.1 日线数据结构解析

通达信的日线数据文件（.day）采用固定的二进制格式：

• 每条记录32字节
• 包含7个字段，每个字段4字节（32位整数）

数据结构定义：

# 日线数据记录结构
# 每32字节一条记录，包含：
# 1. 日期（YYYYMMDD格式的整数）
# 2. 开盘价（单位：0.01元）
# 3. 最高价（单位：0.01元）
# 4. 最低价（单位：0.01元）
# 5. 收盘价（单位：0.01元）
# 6. 成交量（单位：股）
# 7. 成交额（单位：0.01元）

4.2 Python实现代码

import struct
import pandas as pd
import numpy as np
from pathlib import Path
from datetime import datetime

def read_tdx_day_file(file_path):
    """
    读取通达信日线数据文件（.day格式）
    
    Args:
        file_path: 通达信日线文件路径，如 "sh000001.day"
        
    Returns:
        pandas.DataFrame: 包含日期、开盘、最高、最低、收盘、成交量、成交额的DataFrame
    """
    # 检查文件是否存在
    if not Path(file_path).exists():
        raise FileNotFoundError(f"文件不存在: {file_path}")
    
    # 定义数据结构格式
    # 每32字节一条记录，7个int32类型字段
    record_format = '7i'  # 7个32位有符号整数
    record_size = struct.calcsize(record_format)  # 7 * 4 = 28字节
    
    data_list = []
    
    try:
        with open(file_path, 'rb') as f:
            # 读取文件全部内容
            file_content = f.read()
            
            # 计算记录数量
            num_records = len(file_content) // record_size
            
            for i in range(num_records):
                # 计算当前记录的起始位置
                start_pos = i * record_size
                end_pos = start_pos + record_size
                
                # 解析二进制数据
                record_data = struct.unpack(record_format, file_content[start_pos:end_pos])
                
                # 提取各个字段
                date_int = record_data[0]
                open_price = record_data[1] / 100.0  # 转换为元
                high_price = record_data[2] / 100.0
                low_price = record_data[3] / 100.0
                close_price = record_data[4] / 100.0
                volume = record_data[5]  # 成交量（股）
                amount = record_data[6] / 100.0  # 成交额（元）
                
                # 转换日期格式
                date_str = str(date_int)
                if len(date_str) == 8:
                    trade_date = datetime.strptime(date_str, '%Y%m%d').date()
                else:
                    trade_date = None
                
                # 添加到列表
                data_list.append({
                    'date': trade_date,
                    'open': open_price,
                    'high': high_price,
                    'low': low_price,
                    'close': close_price,
                    'volume': volume,
                    'amount': amount
                })
        
        # 转换为DataFrame
        df = pd.DataFrame(data_list)
        
        # 按日期排序（通达信数据通常是倒序存储）
        if not df.empty and 'date' in df.columns:
            df = df.sort_values('date').reset_index(drop=True)
            
        return df
        
    except Exception as e:
        print(f"读取文件失败: {e}")
        return pd.DataFrame()

# 使用示例
if __name__ == "__main__":
    # 替换为你的通达信日线文件路径
    day_file = "T0002/hq_cache/sh000001.day"
    
    try:
        df_day = read_tdx_day_file(day_file)
        print(f"成功读取 {len(df_day)} 条日线数据")
        print(df_day.head())
        print("\n数据统计信息:")
        print(df_day.describe())
    except Exception as e:
        print(f"错误: {e}")

4.3 数据处理与验证

读取数据后，通常需要进行一些数据清洗和验证：

def validate_tdx_data(df):
    """验证通达信数据的完整性"""
    if df.empty:
        print("数据为空")
        return False
    
    # 检查必要列是否存在
    required_columns = ['date', 'open', 'high', 'low', 'close', 'volume']
    missing_columns = [col for col in required_columns if col not in df.columns]
    
    if missing_columns:
        print(f"缺少必要列: {missing_columns}")
        return False
    
    # 检查数据合理性
    # 1. 高价应不低于低价
    invalid_high_low = df[df['high'] < df['low']]
    if not invalid_high_low.empty:
        print(f"发现 {len(invalid_high_low)} 条高价低于低价的数据")
    
    # 2. 价格应为正数
    negative_prices = df[(df['open'] <= 0) | (df['close'] <= 0)]
    if not negative_prices.empty:
        print(f"发现 {len(negative_prices)} 条非正价格数据")
    
    # 3. 检查日期连续性
    if 'date' in df.columns:
        df_sorted = df.sort_values('date')
        date_diff = df_sorted['date'].diff().dt.days
        gaps = date_diff[date_diff > 1]
        if not gaps.empty:
            print(f"发现日期不连续，最大间隔: {gaps.max()} 天")
    
    return True

def enhance_tdx_data(df):
    """增强数据，添加技术指标"""
    if df.empty:
        return df
    
    # 添加涨跌幅
    df['pct_change'] = df['close'].pct_change() * 100
    
    # 添加移动平均线
    df['ma5'] = df['close'].rolling(window=5).mean()
    df['ma10'] = df['close'].rolling(window=10).mean()
    df['ma20'] = df['close'].rolling(window=20).mean()
    
    # 添加成交量移动平均
    df['volume_ma5'] = df['volume'].rolling(window=5).mean()
    
    # 添加价格区间
    df['range'] = df['high'] - df['low']
    df['range_pct'] = df['range'] / df['close'] * 100
    
    return df

5. 读取5分钟线数据

5.1 5分钟线数据结构

通达信的5分钟线文件（.lc5）结构与日线类似，但包含更多字段：

def read_tdx_minute_file(file_path, period='5min'):
    """
    读取通达信分钟线数据文件
    
    Args:
        file_path: 文件路径，如 "sh000001.lc5"
        period: 周期类型，'5min' 或 '1min'
        
    Returns:
        pandas.DataFrame: 包含时间、OHLCV等数据的DataFrame
    """
    # 检查文件是否存在
    if not Path(file_path).exists():
        raise FileNotFoundError(f"文件不存在: {file_path}")
    
    # 分钟线数据结构：每32字节一条记录
    # 包含：日期、时间、开盘、最高、最低、收盘、成交量、成交额
    record_format = '2i6i'  # 2个日期时间相关，6个价格成交量相关
    record_size = struct.calcsize(record_format)
    
    data_list = []
    
    try:
        with open(file_path, 'rb') as f:
            file_content = f.read()
            num_records = len(file_content) // record_size
            
            for i in range(num_records):
                start_pos = i * record_size
                end_pos = start_pos + record_size
                
                record_data = struct.unpack(record_format, file_content[start_pos:end_pos])
                
                # 解析日期和时间
                date_int = record_data[0]
                time_int = record_data[1]
                
                # 转换日期
                date_str = str(date_int)
                if len(date_str) == 8:
                    year = int(date_str[:4])
                    month = int(date_str[4:6])
                    day = int(date_str[6:8])
                else:
                    continue
                
                # 转换时间（HHMM格式）
                hour = time_int // 100
                minute = time_int % 100
                
                # 创建datetime对象
                dt = datetime(year, month, day, hour, minute)
                
                # 解析价格和成交量
                open_price = record_data[2] / 100.0
                high_price = record_data[3] / 100.0
                low_price = record_data[4] / 100.0
                close_price = record_data[5] / 100.0
                volume = record_data[6]
                amount = record_data[7] / 100.0
                
                data_list.append({
                    'datetime': dt,
                    'open': open_price,
                    'high': high_price,
                    'low': low_price,
                    'close': close_price,
                    'volume': volume,
                    'amount': amount,
                    'period': period
                })
        
        # 创建DataFrame
        df = pd.DataFrame(data_list)
        
        if not df.empty:
            df = df.sort_values('datetime').reset_index(drop=True)
            
        return df
        
    except Exception as e:
        print(f"读取分钟线文件失败: {e}")
        return pd.DataFrame()

6. 批量读取与数据管理

6.1 批量读取多个股票数据

import os
from concurrent.futures import ThreadPoolExecutor

def batch_read_tdx_data(data_dir, pattern="*.day", max_workers=4):
    """
    批量读取通达信数据文件
    
    Args:
        data_dir: 数据目录路径
        pattern: 文件匹配模式，如 "*.day"、"*.lc5"
        max_workers: 最大线程数
        
    Returns:
        dict: 股票代码到DataFrame的映射
    """
    data_dir = Path(data_dir)
    if not data_dir.exists():
        raise FileNotFoundError(f"目录不存在: {data_dir}")
    
    # 查找所有匹配的文件
    files = list(data_dir.glob(pattern))
    print(f"找到 {len(files)} 个数据文件")
    
    data_dict = {}
    
    def read_single_file(file_path):
        """读取单个文件"""
        try:
            if file_path.suffix == '.day':
                df = read_tdx_day_file(str(file_path))
            elif file_path.suffix in ['.lc5', '.lc1']:
                period = '5min' if file_path.suffix == '.lc5' else '1min'
                df = read_tdx_minute_file(str(file_path), period)
            else:
                return None
            
            # 从文件名提取股票代码
            stock_code = file_path.stem
            return stock_code, df
            
        except Exception as e:
            print(f"读取文件 {file_path.name} 失败: {e}")
            return None
    
    # 使用线程池并行读取
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = executor.map(read_single_file, files)
        
        for result in results:
            if result:
                stock_code, df = result
                if not df.empty:
                    data_dict[stock_code] = df
    
    return data_dict

def save_to_database(data_dict, db_path="tdx_data.db"):
    """将数据保存到SQLite数据库"""
    import sqlite3
    
    conn = sqlite3.connect(db_path)
    
    for stock_code, df in data_dict.items():
        if not df.empty:
            # 添加股票代码列
            df['code'] = stock_code
            
            # 保存到数据库
            table_name = f"stock_{stock_code}"
            df.to_sql(table_name, conn, if_exists='replace', index=False)
            print(f"已保存 {stock_code} 到表 {table_name}")
    
    conn.close()
    print(f"数据已保存到 {db_path}")

6.2 数据缓存与更新

import pickle
import hashlib
from datetime import datetime, timedelta

class TdxDataCache:
    """通达信数据缓存管理器"""
    
    def __init__(self, cache_dir="tdx_cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
    
    def get_cache_key(self, file_path, params=None):
        """生成缓存键"""
        # 基于文件路径和参数生成唯一键
        key_str = str(file_path)
        if params:
            key_str += str(params)
        
        # 添加文件修改时间
        file_mtime = os.path.getmtime(file_path) if os.path.exists(file_path) else 0
        key_str += str(file_mtime)
        
        return hashlib.md5(key_str.encode()).hexdigest()
    
    def load_from_cache(self, cache_key):
        """从缓存加载数据"""
        cache_file = self.cache_dir / f"{cache_key}.pkl"
        
        if cache_file.exists():
            # 检查缓存是否过期（默认1天）
            cache_age = datetime.now() - datetime.fromtimestamp(cache_file.stat().st_mtime)
            if cache_age < timedelta(days=1):
                try:
                    with open(cache_file, 'rb') as f:
                        return pickle.load(f)
                except:
                    pass
        
        return None
    
    def save_to_cache(self, cache_key, data):
        """保存数据到缓存"""
        cache_file = self.cache_dir / f"{cache_key}.pkl"
        
        try:
            with open(cache_file, 'wb') as f:
                pickle.dump(data, f)
            return True
        except:
            return False
    
    def read_with_cache(self, file_path, reader_func, **kwargs):
        """带缓存的读取"""
        cache_key = self.get_cache_key(file_path, kwargs)
        
        # 尝试从缓存加载
        cached_data = self.load_from_cache(cache_key)
        if cached_data is not None:
            print(f"从缓存加载数据: {file_path}")
            return cached_data
        
        # 重新读取
        print(f"重新读取数据: {file_path}")
        data = reader_func(file_path, **kwargs)
        
        # 保存到缓存
        if data is not None and not data.empty:
            self.save_to_cache(cache_key, data)
        
        return data

7. 常见问题与解决方案

7.1 文件路径问题

问题: 找不到通达信数据文件
解决方案:

def find_tdx_data_path():
    """自动查找通达信数据路径"""
    possible_paths = [
        # Windows 默认路径
        "C:/new_tdx/T0002/hq_cache",
        "D:/new_tdx/T0002/hq_cache",
        "C:/tdx/T0002/hq_cache",
        # 用户目录
        str(Path.home() / "tdx" / "T0002" / "hq_cache"),
    ]
    
    for path in possible_paths:
        if Path(path).exists():
            print(f"找到通达信数据路径: {path}")
            return path
    
    print("未找到通达信数据路径，请手动指定")
    return None

7.2 数据格式异常

问题: 读取的数据价格异常（如价格为0或负数）
解决方案:

def clean_tdx_data(df):
    """清洗通达信数据"""
    if df.empty:
        return df
    
    # 移除价格异常的数据
    df_clean = df.copy()
    
    # 1. 移除价格为0或负数的记录
    price_columns = ['open', 'high', 'low', 'close']
    for col in price_columns:
        df_clean = df_clean[df_clean[col] > 0]
    
    # 2. 移除高价低于低价的