1. 项目背景与本地数据全景认知1.1 申万宏源金融终端和通达信的目录关系先看一个很多人一直没搞明白的问题明明装的是申万宏源金融终端为什么 C 盘目录里到处都是 zd_swhy_gm、vipdoc 这种看起来很像通达信的路径因为申万宏源金融终端就是通达信内核的券商定制版行情内核、本地存储、公式系统都是通达信那一套。这类定制版安装之后数据目录就固定长这样C:/zd_swhy_gm/ ├── vipdoc/ │ ├── sh/lday/ # 上海市场日线 │ ├── sz/lday/ # 深圳市场日线 │ └── ... ├── T0002/ └── ...打开 C:/zd_swhy_gm/vipdoc/sh/lday你会看到一堆 .day 文件。文件名就是股票代码sh 开头是上海sz 开头是深圳。sh600000.day 对应的是浦发银行sh600036.day 是招商银行。一个文件一支股票从上市第一根 K 线开始一直写到最新收盘。这些文件是软件平时自动下载、收盘时落盘生成的所以数据质量很稳定。做量化分析、个人投研数据仓库完全可以直接利用这批文件不用自己费力去爬行情也不用花钱买数据接口。1.2 .day 文件内部是定长二进制记录如果你直接用记事本打开一个 .day 文件看到的全是乱码。这是因为通达信用了一种非常紧凑的定长二进制格式每条记录固定 32 字节一条记录就是一天的行情。字段排布如下偏移量字节数字段含义存储说明04日期整数形如 2025011544开盘价整数存储实际价格需要除以 10084最高价整数存储实际价格需要除以 100124最低价整数存储实际价格需要除以 100164收盘价整数存储实际价格需要除以 100204成交额单精度浮点数单位元244成交量整数存储注意单位在不同版本下可能有差异284保留字段暂不使用这种设计的核心思路是省空间、读取快。价格用整数存储而不是浮点是为了避免浮点误差所以存储时先乘以 100解析时再除以 100。日期字段也用整数而不是字符串解析出来直接转成字符串就是 YYYYMMDD排序天然正确。理解了这个结构解析这件事就已经没有悬念了。用 Python 的 struct 模块或者用 C/C 直接做内存映射都可以很快把文件读出来。1.3 为什么要转成 sqlite3而不是继续用原始文件直接读 .day 二进制不是不行但它有很明显的痛点不方便查询想拿一只股票某个月的数据得自己从头扫到尾没有任何索引。不方便分析pandas、Excel、BI 工具大多能直接读数据库但对这种私有二进制格式无能为力。不方便维护文件一多跨股票查询、条件筛选、排序去重全靠手写逻辑麻烦且容易错。转成 sqlite3 之后一个单文件数据库就能解决上面所有问题。sqlite3 的优势在于它是 Python 标准库自带的不需要安装任何服务单文件存储、移动备份都方便支持标准 SQL查询、联表、聚合、条件筛选唾手可得量级对个人投研完全够用几十万条 K 线数据对 sqlite3 来说毫无压力。有人会问为什么不存 CSVCSV 文件一多还得维护目录、命名、合并逻辑。数据库天然有主键索引比如用 code date 做联合主键重复导入直接跳过或覆盖数据一致性有保证。所以在这个场景下sqlite3 是性价比最高的方案。2. 数据探查与环境准备2.1 先摸清自己机器上的数据现状动手写代码之前我强烈建议先到 C:/zd_swhy_gm/vipdoc/sh/lday/ 下面看一眼实际情况。有的机器装了多套行情软件路径会不一样有的软件安装在其他盘数据目录就不在 C 盘。先用资源管理器确认 sh600000.day 确实存在再开始下一步。确认之后在文件上右键看属性记录一下文件大小。sh600000 是 1999 年上市的老股票记录数很多文件大概几十 KB。一个 .day 文件大小除以 32理论上应该能整除如果有余数说明文件可能没写完或已经损坏——这个细节在后面的代码里会做校验。另外提醒一点申万宏源金融终端默认可能只下载了你自选股和常用股票的数据不会自动下载全市场。如果你发现某只股票的文件不存在不用奇怪先在软件里打开那只看一下日 K 线让软件把数据拉全文件才会出现在目录里。2.2 用 Python 快速验证二进制格式环境要求低得不能再低了Python 3 以上标准库足够不需要第三方包。我建议先用个交互式脚本验证一下数据格式别一上来就写完整工具。下面这段是最小探测脚本import struct filepath rC:/zd_swhy_gm/vipdoc/sh/lday/sh600000.day with open(filepath, rb) as f: data f.read() print(文件总字节数:, len(data)) print(记录条数:, len(data) // 32) print(余数:, len(data) % 32, (0表示完整)) # 读取前3条记录 for i in range(3): chunk data[i * 32:(i 1) * 32] date_val, open_val, high_val, low_val, close_val, amount_val, volume_val, _ struct.unpack(IIIIIfII, chunk) print(f日期{date_val}, 开{open_val / 100:.2f}, 高{high_val / 100:.2f}, f低{low_val / 100:.2f}, 收{close_val / 100:.2f}, 金额{amount_val:.2f}, 量{volume_val})跑完之后如果手里的 sh600000.day 最近有数据倒数几条记录应该是近期交易日。拿最后一条的收盘价和申万宏源金融终端里浦发银行的日 K 线对比应该完全一致。这一步验证很重要能同时确认三件事文件是否完整、字节序是否正确、价格比例尺除以 100是否对得上。如果解析出来日期看起来像另一个年份或者价格除以 100 后明显偏大偏小先把 struct.unpack 的 改成 再试一次。通达信的某些定制版在文件字节序上偶尔有差异实测大端序虽然少见但我确实遇过。2.3 建表思路表结构怎么设计才顺手数据转进 sqlite3 之前先把表结构定好。建议是尽量贴近原始字段同时加上主键约束避免重复导入产生脏数据CREATE TABLE IF NOT EXISTS stock_daily ( code TEXT NOT NULL, -- 股票代码如 sh600000 date TEXT NOT NULL, -- 交易日如 20250115 open REAL NOT NULL, -- 开盘价 high REAL NOT NULL, -- 最高价 low REAL NOT NULL, -- 最低价 close REAL NOT NULL, -- 收盘价 amount REAL NOT NULL, -- 成交额元 volume INTEGER NOT NULL, -- 成交量注意单位 PRIMARY KEY (code, date) );联合主键是刻意选的。同一个股票同一天只可能有一条日线code date 唯一天然适合做主键。这样以后重复执行导入脚本不会因为手滑多插一遍数据只会被主键挡住或者通过 UPSERT 更新。日期字段建议就用字符串保存。因为日期在文件里本来就是一个整数转成 YYYYMMDD 字符串之后排序、比较、格式化都很方便。如果后面要用 sqlite 的日期函数通过 substr(date,1,4) 就能取出年份不需要额外转类型。3. 核心实现与完整代码3.1 为什么用面向对象来封装解析与入库这个工具本身不复杂但既然要长期用就应该考虑可维护性。我不建议把解析和入库逻辑全部堆在一起而是拆成两个类一个负责读文件、解析记录一个负责建库、入库。这样有几个好处解析器独立以后想导 5 分钟线或 1 分钟线只需要扩展新的解析器。库操作独立方便切换不同的落地方式比如以后想同时输出一个 CSV 快照也容易加。逻辑分离后单元测试好写先测解析器输出再测入库结果哪里出问题一眼就能定位。3.2 解析器实现struct 解包与容错处理先写解析器核心逻辑就是之前验证过的 struct.unpack但补上了文件完整性判断和异常保护import os import struct from dataclasses import dataclass dataclass class DailyRecord: date: str open: float high: float low: float close: float amount: float volume: int class TdxDayParser: 通达信 .day 日线文件解析器 RECORD_SIZE 32 def __init__(self, filepath: str): self.filepath filepath def parse(self) - list[DailyRecord]: if not os.path.exists(self.filepath): raise FileNotFoundError(f日线文件不存在: {self.filepath}) with open(self.filepath, rb) as f: data f.read() if not data: return [] total_size len(data) if total_size % self.RECORD_SIZE ! 0: # 文件末尾可能有多余的残段忽略掉即可但值得记一条日志 print(f[警告] {self.filepath} 大小 {total_size} 不是32的整数倍将忽略尾部残段) records: list[DailyRecord] [] for offset in range(0, total_size, self.RECORD_SIZE): chunk data[offset:offset self.RECORD_SIZE] if len(chunk) self.RECORD_SIZE: break # 尾部残段直接丢弃 date_val, open_val, high_val, low_val, close_val, amount_val, volume_val, _ \ struct.unpack(IIIIIfII, chunk) records.append(DailyRecord( datestr(date_val), openopen_val / 100.0, highhigh_val / 100.0, lowlow_val / 100.0, closeclose_val / 100.0, amountamount_val, volumevolume_val )) return records有几个实现细节值得单独说明。第一价格为什么要除以 100 而不是 1000这是通达信数据格式的约定。如果某只股票股价超过 1000 元存储时也是乘 100所以解析时统一除 100 即可不用根据股票类型动态判断。见过有人把价格当作浮点数直接开箱那解析出来的价格会差很多个数量级原因就是没搞懂整数存储的设计意图。第二成交量字段的单位。我特意在注释里提示了这个坑因为这个字段在不同版本终端下可能是股也可能是手。申万宏源金融终端实测下来 .day 文件里通常按股存储但其他券商定制版不一定一样。最稳的做法是解析完拿软件里的成交量指标对照一下如果量纲不对在导入阶段统一换算。第三字节序问题。代码里显式用了 前缀表示按小端解析。这是目前 Windows 上通达信最常见的字节序。万一你读出来的日期和价格明显异常优先把 改成 再试这是排查顺序问题的最快路径。3.3 入库模块建表、批量写入与去重更新解析器拿到记录之后接下来交给入库模块。sqlite3 的标准库用法很固定下面这段代码实现了建表、批量写入和去重更新import sqlite3 class StockDailyExporter: 将解析后的日线记录写入 sqlite3 def __init__(self, db_path: str): self.db_path db_path self._init_db() def _init_db(self): conn sqlite3.connect(self.db_path) try: conn.execute( CREATE TABLE IF NOT EXISTS stock_daily ( code TEXT NOT NULL, date TEXT NOT NULL, open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL, amount REAL NOT NULL, volume INTEGER NOT NULL, PRIMARY KEY (code, date) ) ) conn.execute(CREATE INDEX IF NOT EXISTS idx_code_date ON stock_daily(code, date)) conn.commit() finally: conn.close() def import_records(self, code: str, records: list[DailyRecord]) - int: if not records: return 0 conn sqlite3.connect(self.db_path) rows [ (code, r.date, r.open, r.high, r.low, r.close, r.amount, r.volume) for r in records ] try: conn.executemany( INSERT INTO stock_daily (code, date, open, high, low, close, amount, volume) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(code, date) DO UPDATE SET openexcluded.open, highexcluded.high, lowexcluded.low, closeexcluded.close, amountexcluded.amount, volumeexcluded.volume , rows ) conn.commit() finally: conn.close() return len(rows)这里的 INSERT ... ON CONFLICT DO UPDATE 是 SQLite 3.24 以上版本的语法意思是同一股票同一天的数据如果已经存在就更新其余字段。这个设计对经常重复运行导入脚本的场景特别友好第一次导入是初始化之后每次跑脚本等价于增量同步不会报主键冲突也不会产生重复数据。批量写入用 executemany数据量小的时候感知不明显但如果你整个目录上百个股票文件一次性导入几万条记录直接循环 execute 会有明显的性能差距。实测中单条循环插入 1 万行可能要十几秒executemany 批量提交基本一瞬间完成。3.4 完整调用示例从文件到数据库一条龙两个类写完之后调用侧就非常清爽了db_path rC:/zd_swhy_gm/mydata/stock.db exporter StockDailyExporter(db_path) filepath rC:/zd_swhy_gm/vipdoc/sh/lday/sh600000.day parser TdxDayParser(filepath) records parser.parse() count exporter.import_records(sh600000, records) print(f导入完成: sh600000 共 {count} 条记录)执行之后用命令行 sqlite3 工具或者任意 SQLite 图形客户端打开 stock.db执行SELECT date, open, high, low, close, volume, amount FROM stock_daily WHERE code sh600000 ORDER BY date DESC LIMIT 10;就能看到最近 10 个交易日的行情。如果数量和收盘价都能和软件对得上这个工具就算是真正可以投入使用了。这里没有写日志模块和命令行参数解析实际使用中你可以把文件路径改成从命令行传入或者把多个文件路径用列表管理。核心就一句话解析器负责格式转换入库器负责持久化两者解耦后面扩展起来非常顺手。4. 常见问题与排雷经验4.1 高频问题速查表我在写这类工具的过程中整理过一份问题清单很多坑其实是共通的症状可能原因解决方法报错 FileNotFoundError目录或文件不存在软件没下载过该股票先打开软件查看该股票的日 K 线等数据落盘struct.error: unpack requires a buffer of 32 bytes文件被截断或者偏移算错先打印文件大小看是否少于 32 或非 32 的倍数日期解析成 1970 年左右字节序搞反了把 改成 重新解析价格明显偏大几百倍忘记除以 100或除以了 1000统一除以 100成交量对不上软件单位不一致可能是股也可能是手用软件 K 线图对照统一换算后入库股票文件不存在但软件里能看到行情数据目录不是默认路径在软件设置里确认行情数据目录的实际位置重复执行脚本后库里出现重复记录建表时没有联合主键用 code date 做主键或使用 ON CONFLICT 更新这七条基本覆盖了从文件定位到数据校验的整个链路。遇到问题先从文件名、文件大小、第一条记录逐项排查不要一上来就改大段代码。4.2 批量导入全市场数据如果你不想只导一只 sh600000而是想把整个目录全部导入写个循环就行。要注意的是不要硬编码文件路径而是用 os.listdir 扫描目录import os base_dir rC:/zd_swhy_gm/vipdoc/sh/lday exporter StockDailyExporter(rC:/zd_swhy_gm/mydata/stock.db) for filename in os.listdir(base_dir): if not filename.endswith(.day): continue code filename[:-4] # 去掉 .day 后缀 filepath os.path.join(base_dir, filename) parser TdxDayParser(filepath) records parser.parse() count exporter.import_records(code, records) print(f{code}: {count} 条)全市场几百个文件中途如果遇到一个损坏文件就中断会很麻烦。建议在循环里加上异常处理try: records parser.parse() count exporter.import_records(code, records) print(f{code}: {count} 条) except Exception as e: print(f[跳过] {filename}: {e})这样单个文件失败不影响整体入库。全市场导入完stock.db 的大小通常不会很大个人分析完全扛得住。4.3 入库后的扩展玩法数据进了 sqlite3 之后可以发挥的空间一下就打开了。最常见的用法是配合 pandas 做分析。pandas 内置 read_sql_query一句代码就能把数据读成 DataFrameimport sqlite3 import pandas as pd conn sqlite3.connect(rC:/zd_swhy_gm/mydata/stock.db) df pd.read_sql_query( SELECT * FROM stock_daily WHERE codesh600000 ORDER BY date, conn ) print(df.tail())拿到 DataFrame 之后画 K 线图、算均线、算收益率、做回测都是常规操作。如果你关注最近很火的 MCP 方向还可以考虑写一个本地 MCP Server把 stock_daily 表暴露给大模型。大模型通过自然语言就能查询本地行情数据比如问它帮我算一下浦发银行最近 20 个交易日的涨跌幅。这比各种公开接口更灵活因为数据完全在本地不受接口频率和字段限制。还有一些细节可以继续打磨比如把日期转成 YYYY-MM-DD 标准格式或者增加一个股票基础信息表存放股票简称、上市日期等。这些都是锦上添花核心管线已经有了。最后说一句实在的体会。这类解析工具最怕的不是代码写不出来而是对源数据格式的理解不够导致解析结果看起来差不多但实际有偏差。所以我强烈建议第一步永远是用软件里的 K 线图逐字段核对日期、价格、成交量逐项验证通过后再批量处理。格式理解对了代码怎么写都是顺理成章的事。还有一个小细节要提醒如果你的 Python 环境自带的 SQLite 版本比较老尤其是某些 Linux 发行版执行 ON CONFLICT 语法可能报错。遇到这种情况最简单的办法是把语句换成 INSERT OR REPLACE语义类似只是会先删后插。主键字段不变的话影响不大。