简介：这份资源面向金融数据分析初学者与量化爱好者，聚焦同花顺日线二进制文件（.day）的解析问题。同花顺日线数据包含每日开盘价、收盘价、最高价、最低价、成交量与成交额等关键字段，但文件采用特殊二进制格式存储，直接读取并不直观。资源通过一个精简的Python脚本演示如何以二进制模式打开文件、读取头信息、按固定长度记录逐条解析数据记录，并将原始字节转换为可用的数值与日期对象，最终整理成便于后续分析的数据结构。压缩包共2个文件，包含1个day示例数据与1个py解析脚本，整体约4KB，体积轻巧，适合直接运行验证。目前已有2247人学习下载。读者可借此掌握二进制文件解析的基本思路与Python实现方法，理解字段偏移与数据转换技巧，为股票历史行情分析、回测数据准备等场景打下基础。
1. 同花顺日线文件解析：从一堆二进制里把K线抠出来
做量化回测的人迟早会撞上这件事：策略在模拟盘跑得挺顺，一换到本地历史数据就抓瞎。行情软件自带的导出功能要么字段残缺，要么只给最近几百根，想拿十年日线做长周期验证，只能去啃它本地的数据文件。同花顺日线文件解析就是干这个的——把安装目录下那批.day文件读成一张能直接喂给 pandas 的表。
它解决的不是"能不能看到K线"的问题，而是"能不能批量、可复现、带完整字段地拿到K线"的问题。适合两类人：一类是自己搭回测框架、需要干净历史数据的开发者；另一类是做数据校验，想拿本地文件和接口数据对账的从业者。难点在于这些文件没有公开格式文档，全是定长二进制记录，字段偏移错一位，整列价格就全乱。下面按我实际踩过的路子，从格式推断讲到批量落地。
2. 先搞懂 .day 文件的二进制布局：定长记录与字段偏移
2.1 为什么是定长二进制而不是文本
行情软件本地存日线，追求的是读写快、体积小。文本格式（CSV、JSON）每行都要解析分隔符，几万只股票乘以几千个交易日，开销很可观。定长二进制记录的好处是：第 N 条记录的起始位置可以直接算出来，offset = 32 * N，不需要顺序扫描。这也是为什么你打开.day文件用记事本看是一堆乱码——它本来就不是给人看的。
常见做法是每条记录固定 32 字节，字段按顺序紧密排列：日期、开、高、低、收、成交量、成交额，外加一个保留位。不同版本、不同市场（沪市/深市/北交所）在字段宽度上可能有细微差别，所以第一步永远是先确认记录长度，而不是直接套一个网上的偏移表。
判断记录长度有个笨但可靠的办法：拿文件总字节数除以你预估的交易日数量，看能不能整除。比如一个文件 320000 字节，你估计有 10000 个交易日，那 32 字节/条就对得上。如果除出来是 31.9 这种小数，说明字段宽度或者记录头尾还有额外字节。
2.2 用 Python 读出第一条记录并验证
先别急着批量，拿单文件、单记录把字段对齐了再说。下面这段代码读文件头 32 字节，按常见布局解包：
import struct # 常见 .day 记录布局：日期(int32) + 开高低收(int32*4) + 成交量(float32) + 成交额(float32) + 保留(int32) # 合计 4 + 16 + 4 + 4 + 4 = 32 字节 RECORD_SIZE = 32 FMT = "<i iiiif f i" # 小端：日期、开、高、低、收、量、额、保留 def read_first_record(path): with open(path, "rb") as f: raw = f.read(RECORD_SIZE) if len(raw) < RECORD_SIZE: raise ValueError("文件不足一条记录，可能不是日线文件") fields = struct.unpack(FMT, raw) date_int, o, h, l, c, vol, amt, _ = fields # 价格字段常见做法是放大 100 倍存整数，读出来要除回去 return { "date": date_int, "open": o / 100.0, "high": h / 100.0, "low": l / 100.0, "close": c / 100.0, "volume": vol, "amount": amt, } print(read_first_record("demo.day"))逻辑说明：struct.unpack按格式串把字节流切成 Python 值，<表示小端序，这是 x86 平台本地文件的默认字节序。价格用整数存、读出来除以 100，是行情数据的通用压缩手法，能避免浮点存储的精度漂移。
参数说明：RECORD_SIZE必须和实际文件对齐，错了会整体错位；FMT里的字段顺序一旦和真实布局不符，解出来的开高低收会互相串位。验证方法很直接——把解出来的收盘价和软件界面上同一天的收盘价对一下，对不上就说明偏移或缩放系数有问题。
提示：日期字段常见是
YYYYMMDD的整数形式，比如 20240115。如果解出来是 19700101 这种，多半是字节序反了，把<换成>再试。
2.3 日期字段的两种存法与换算
日期看着简单，其实是翻车高发区。常见两种存法：一种是直接存20240115这样的整数，肉眼可读；另一种是存"距某个基准日的天数"，类似 Unix 时间戳但基准不同。前者直接转字符串就行，后者需要加一个基准日期。
判断方法：如果解出来的日期整数在 19000101 到 21001231 之间，基本就是YYYYMMDD；如果是个几万以内的小整数，那就是天数偏移。换算时用datetime.date(1990, 12, 19) + timedelta(days=n)这类方式，基准日要靠前几条记录和软件界面对照反推，不要凭记忆写。
from datetime import date, timedelta def to_date(date_int): if 19000101 <= date_int <= 21001231: s = str(date_int) return date(int(s[:4]), int(s[4:6]), int(s[6:8])) # 天数偏移存法，基准日需按实际文件反推 return date(1990, 12, 19) + timedelta(days=date_int)这段的关键是那个基准日，它因文件来源而异，必须用已知交易日反推验证，不能照抄。
3. 批量解析与字段清洗：把单文件脚本变成能跑全市场的工具
3.1 遍历目录、按股票代码归档
单文件跑通后，下一步是批量。同花顺的日线文件一般按市场分目录，文件名就是股票代码。批量脚本要处理三件事：递归找文件、从路径或文件名提取代码、把结果汇总成一张带代码列的大表。
import os import pandas as pd def parse_one(path): rows = [] with open(path, "rb") as f: while True: raw = f.read(RECORD_SIZE) if len(raw) < RECORD_SIZE: break d, o, h, l, c, v, a, _ = struct.unpack(FMT, raw) rows.append((to_date(d), o/100, h/100, l/100, c/100, v, a)) df = pd.DataFrame(rows, columns=["date","open","high","low","close","volume","amount"]) df["code"] = os.path.splitext(os.path.basename(path))[0] return df def parse_dir(root): frames = [] for dirpath, _, files in os.walk(root): for name in files: if name.lower().endswith(".day"): frames.append(parse_one(os.path.join(dirpath, name))) return pd.concat(frames, ignore_index=True)逻辑说明：while True配合read(RECORD_SIZE)逐条读，读到不足一条就停，这样能自然处理文件尾部可能的残缺字节。os.walk递归整个目录树，兼容按市场分层的结构。
参数说明：os.path.splitext取文件名主干作为代码，如果文件名带前缀（比如SH600000.day），需要额外切片处理。pd.concat汇总时ignore_index=True避免索引重复。
3.2 清洗：停牌、除权与异常值
原始数据不能直接用。停牌日可能没有记录，也可能有记录但成交量为 0；除权除息会让价格出现跳空，回测时如果不复权，收益率会算错。清洗要做的：剔除成交量为 0 的行、按日期排序、标记除权日。
def clean(df): df = df[df["volume"] > 0].copy() df = df.sort_values(["code", "date"]).reset_index(drop=True) # 相邻收盘价跳变超过阈值，标记为疑似除权 df["prev_close"] = df.groupby("code")["close"].shift(1) df["suspect_adj"] = (df["close"] / df["prev_close"] - 1).abs() > 0.11 return df逻辑说明：volume > 0过滤停牌；groupby("code")保证每只股票内部排序，不会跨股票串行；跳变阈值 0.11 对应普通股 10% 涨跌停加一点余量，超过就怀疑是除权而非正常波动。
参数说明：阈值不能设太小，否则正常涨停会被误判；也不能太大，否则漏掉除权。ST 股涨跌幅限制不同，如果数据里混了 ST，阈值要单独调。
3.3 落盘格式选择：Parquet 还是 CSV
解析完的数据建议存 Parquet。CSV 体积大、读回来还要重新解析类型，Parquet 带 schema、压缩率高、pandas 直接读。几百万行的日线，Parquet 通常比 CSV 小一个数量级，读取快好几倍。
df.to_parquet("daily_all.parquet", index=False) # 读回 df2 = pd.read_parquet("daily_all.parquet")如果后续要接数据库，也可以直接写 SQLite，但要注意日期字段的存储类型，别存成字符串导致范围查询失效。
4. 避坑与排查：解析日线文件最容易翻车的五个地方
4.1 现象：价格全是几百倍的整数，除完还是不对
原因：缩放系数不是固定的 100。有的文件价格放大 1000 倍，有的成交量是手数而非股数。解决：拿一条已知记录，用软件界面上的真实值反推系数，系数 = 原始整数 / 真实价格，算出来是多少就用多少，别默认 100。
4.2 现象：解出来的日期全是 1970 年
原因：字节序搞反了，或者把天数偏移当成了YYYYMMDD。解决：先打印原始整数值，看数量级。几万以内是偏移，八位数是YYYYMMDD。字节序用<和>各试一次，哪个解出来合理用哪个。
4.3 现象：批量跑完发现某些股票数据缺一大段
原因：文件本身不完整，或者遍历时漏了子目录。解决：解析前先统计每个文件的记录数，和该股票的上市天数粗略对比，差太多的单独查。遍历用os.walk而不是os.listdir，避免漏掉嵌套目录。
4.4 现象：回测收益率明显偏高，像是偷了未来
原因：没复权，除权日的价格跳空被当成了真实收益。解决：解析阶段只负责还原原始价格，复权在回测框架里做，或者单独拉一份复权因子表。别在解析脚本里硬编码复权逻辑，容易和后续处理冲突。
4.5 现象：内存爆掉，几百万行一次性 concat
原因：pd.concat把所有 DataFrame 堆在内存里。解决：分批写 Parquet，或者用pyarrow的流式写入。单机内存有限时，按市场或按年份分片处理，最后再合并。
5. 进阶：用内存映射加速大文件与增量更新
单文件解析慢，瓶颈往往在 Python 的逐条read。文件大、股票多的时候，可以用numpy.memmap把整个文件映射成数组，一次性 reshape 成(N, 8)的矩阵，再按列取字段，速度能快一个数量级。
import numpy as np def parse_fast(path): # 按 int32 视图读取，8 个字段对应 32 字节 data = np.fromfile(path, dtype=np.int32) n = len(data) // 8 data = data[:n * 8].reshape(n, 8) df = pd.DataFrame({ "date": data[:, 0], "open": data[:, 1] / 100.0, "high": data[:, 2] / 100.0, "low": data[:, 3] / 100.0, "close": data[:, 4] / 100.0, "volume": data[:, 5].view(np.float32), "amount": data[:, 6].view(np.float32), }) return df逻辑说明：np.fromfile把整个文件读成一维 int32 数组，reshape(n, 8)还原成记录矩阵，按列切片比逐条 unpack 快得多。成交量和成交额如果是 float32 存的，需要用.view(np.float32)重新解释字节，不能直接当整数用。
参数说明：dtype=np.int32要求记录确实是 4 字节对齐；如果文件里有 float 字段，整体按 int32 读再 view 是可行的，但字段顺序必须和真实布局一致。n * 8截断保证 reshape 不报错，尾部残缺字节直接丢弃。
增量更新是另一个实用技巧：每天收盘后只解析当天新增的记录，而不是全量重跑。做法是记录上次解析到的文件偏移量，下次从该偏移继续读。这要求文件是追加写的，且记录定长——正好符合.day的特点。
def parse_incremental(path, last_offset): with open(path, "rb") as f: f.seek(last_offset) rows = [] while True: raw = f.read(RECORD_SIZE) if len(raw) < RECORD_SIZE: break rows.append(struct.unpack(FMT, raw)) new_offset = f.tell() return rows, new_offset把last_offset存到一个小状态文件里，下次接着读。这样每天的处理量从几百万行降到几千行，回测数据维护从"每周全量重跑"变成"每天几秒钟"。
我自己维护这套解析脚本的习惯是：格式推断阶段一定留一个debug_first_record函数，任何时候字段对不上，先跑它打印原始字节和解析结果，比对着文档猜快得多。另外，解析和复权严格分开，解析层只保证"还原文件里存了什么"，不掺任何业务逻辑，这样出问题时能快速定位是数据源的问题还是处理逻辑的问题。希望帮到你。
本文还有配套的精品资源，点击获取
