通达信Level2数据解析：逐笔成交与主力资金流实战指南-北京尧图网络

简介这是一份面向职业交易员与进阶投资者的通达信Level2数据实战指南代码包围绕深度行情、逐笔成交、资金流向分析和盘口异动预警等核心功能重点解决Level2数据不会看、看得慢、用不深的问题帮助识别主力资金真实动向、识别虚假挂单以及大单拆分等关键交易信号。压缩包共3个文件以HTML说明文档为核心配套inscode代码片段与gitignore配置整包仅4KB内容小而精方便快速查阅、复制代码并在通达信定制版面和模拟实战中直接练习。目前已有1270人学习/下载适合从入门到实战逐步建立对Level2数据的理解获取市场信息优势。除了核心信号识别方法还讲解如何判断优质Level2服务包括极速行情、完整功能和专属投顾等选择要点并给出从基础认知到实战操作的系统路径能有效缩短投资者掌握Level2数据语言的学习曲线。1. 通达信Level2数据指南从代码包开始拆逐笔成交与资金流做量化研究时我常常盯住盘口不放但日线和分钟线只能看到已经聚合过的结果拆单、挂单、撤单这些动作全被揉成一团。真正让我把目光转向通达信Level2数据的是某次复盘时发现某只标的在尾盘十几分钟里连续出现大额成交而分钟K线却几乎看不出异动。这份资源围绕通达信Level2数据做了一整套可运行的代码包覆盖数据文件解析、逐笔成交还原、主力资金流计算和可视化输出。适合手里有Level2行情数据、想自己验证盘口逻辑的开发者也适合刚接触逐笔数据、想快速上手拆解的学生。它把“看盘”变成“算盘”核心价值在于给出一套能直接改的源码框架。2. 认识Level2数据文件扩展名、字段结构与选型理由2.1 文件扩展名与周期对应关系通达信类行情终端导出的Level2数据常见扩展名就那么几种我拆包时先按扩展名把文件归了类避免把不同周期的数据混在一起读。扩展名常见含义典型记录内容.dlr日线数据开高低收、成交量、成交额.lc11分钟线数据每分钟的开高低收、总量.lc55分钟线数据每5分钟聚合一次的快照.sh逐笔成交记录逐笔时间、价格、量、方向.dat通用快照或扩展数据十档买卖盘、委托队列我一般会先到行情终端的数据目录里找对应日期和代码的文件确认扩展名后再写解析逻辑。比如要分析某标的某一天的盘口微观结构就只需要把对应日期的 .sh 文件单独拎出来。这里有个经验不要通过文件大小猜内容有一天我发现一个 .sh 文件只有几 KB以为是空数据后来才发现当天有停牌记录数确实少得可怜。文件头也是值得先看的部分。不同数据商导出的文件头部可能是几字节的日期标记也可能是纯记录流。稳妥的做法是先以 16 进制模式打开文件把前几十个字节打印出来人工对一眼确认是否存在固定头再决定解析时要不要跳过。常见做法是跳过前 4 到 16 字节具体偏移量以样例文件实测为准。2.2 逐笔成交与快照行情字段差异决定分析深度逐笔成交和快照是两类完全不同的数据。十档快照记录的是某一个瞬间的买卖十档报价与委托量一般每隔 3 秒左右生成一条。逐笔成交则记录每一笔实际成交的时间、价格、数量、金额、成交编号和主动买卖方向时间精度能达到 0.01 秒级别。分析主力行为时逐笔数据明显更有价值因为它能还原大单是怎么拆成一笔笔小单成交的快照数据则更适合分析盘口挂单的厚度变化。逐笔成交里最关键的字段我认为有三个成交时间、成交编号、买卖方向。成交编号用于判断连续性如果编号跳跃说明中间可能有若干笔被过滤掉了买卖方向字段用来区分这笔成交是主动买还是主动卖这直接决定资金流计算的口径。有些存量数据年代久远可能不包含方向字段只能靠价格与前一笔对比来推断这个我在第 4 章单独说。2.3 为什么用 Python 拆而不是用现成软件很多行情终端自带分时资金流统计但它是黑匣子并不会告诉你超大单阈值是多少、集合竞价怎么处理、复权有没有做。我选择用 Python 重构这套流程一是因为结构清晰读写二进制文件用 struct 模块即可二是因为解析完成后可以直接接 pandas 做聚合分析不必再导出 CSV 二次加工三是整个链路可控参数可以随时调。选择 Python 解析的另一个现实原因是生态成熟。matplotlib 出图、SQLite 存储、pandas 聚合几乎不用额外装什么重量级组件。对新手来说一个脚本就能跑通从二进制到可视化学习曲线相对平缓。如果换成 C 或者 C#性能虽然更好但调试成本会高不少尤其是不熟悉内存布局的人更容易在字节偏移上翻车。3. 解析主流程把 .sh 逐笔成交读成结构化数据3.1 文件头校验与字节序判断解析二进制行情数据第一步不是立刻逐条读取而是先做字节序判断。常见的数据文件在小端序机器上生成解析格式串以开头但偶尔会遇到大端序数据如果按小端解价格和数量会变成异常大数甚至负数。我在代码包里放了一个简单的头部校验函数读取前 8 字节并识别特征值如果特征值符合预期则继续否则直接报警提示字节序可能反了。import struct def check_header(file_path, offset0): with open(file_path, rb) as f: f.seek(offset) raw f.read(8) print(前8字节十六进制:, raw.hex()) if raw[:4] b\x01\x00\x00\x00: print(检测到小端序特征) return elif raw[:4] b\x00\x00\x00\x01: print(检测到大端序特征) return else: print(未识别到标准文件头尝试按小端序解析) return 这段代码的作用是打印文件头并根据特征值返回 struct 解析用的字节序前缀。逻辑说明我截取前 4 字节用十六进制字符串直接对比\x01\x00\x00\x00是小端序存放整数 1 的典型形态反过来的\x00\x00\x00\x01就是大端序。实际项目中有些文件头并不是固定整数这时候就退化为默认小端序。参数说明offset默认 0如果确认文件有自定义头可以传实际偏移量。3.2 struct.unpack 逐条解析拿到字节序后就是核心的逐条解包。这里以我常用的一种逐笔成交结构为例每条记录包含时间戳、成交价、成交量、成交金额、成交编号、买卖方向合计 23 字节。时间戳按整数存储数值含义类似 HHMMSSmmm比如 93001001 表示 09:30:01.001。import struct RECORD_SIZE 23 def parse_ticks(file_path, endian): ticks [] fmt endian iIhIIB # 时间戳, 价格, 量(手), 金额, 编号, 方向 size struct.calcsize(fmt) if size ! RECORD_SIZE: raise ValueError(f结构大小不匹配: {size} ! {RECORD_SIZE}) with open(file_path, rb) as f: data f.read() offset 0 while offset RECORD_SIZE len(data): values struct.unpack_from(fmt, data, offset) ts, price, vol, amount, tick_no, direction values ticks.append({ time: ts, price: price / 100.0, # 价格最小变动单位 0.01 volume: vol, amount: amount, tick_no: tick_no, direction: direction }) offset RECORD_SIZE return ticks # 使用示例 ticks parse_ticks(sh600000_20240102.sh, ) print(解析条数:, len(ticks)) print(第一条:, ticks[0])逻辑说明struct.unpack_from在指定偏移处一次性解出 6 个字段能避免反复读文件造成性能浪费。volume我用的是手数单位但不同数据商可能用股需要看字段说明确认price除以 100 是因为价格存储为整数分转成元更直观。参数说明endian由第 3.1 节的校验函数传入RECORD_SIZE必须和结构格式串严格对应如果中间有多余字节会导致从某条记录开始错位后面全解错。遇到最后剩余不足一条记录长度时我选择直接丢弃这通常说明文件被截断或末尾有填充字节。3.3 落盘方案先 CSV 后 SQLite解析出来的列表直接放内存里做小样本分析没问题但真实数据一天可能几十万笔聚合查询就会变慢。我的习惯是先落一份 CSV 方便肉眼检查再用 SQLite 做正经的数据管理和条件筛选。CSV 的好处是任何工具都能打开出了问题方便人工对账SQLite 的好处是支持 SQL按时间范围、方向筛选非常快。import sqlite3 import csv def save_csv(ticks, csv_path): with open(csv_path, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnames[time, price, volume, amount, tick_no, direction]) writer.writeheader() writer.writerows(ticks) def save_sqlite(ticks, db_path): conn sqlite3.connect(db_path) conn.execute(CREATE TABLE IF NOT EXISTS ticks ( time INTEGER, price INTEGER, volume INTEGER, amount INTEGER, tick_no INTEGER, direction INTEGER )) conn.execute(CREATE INDEX IF NOT EXISTS idx_time ON ticks(time)) with conn: # 自动开启事务 conn.executemany( INSERT INTO ticks VALUES (:time, :price, :volume, :amount, :tick_no, :direction), ticks ) conn.close()逻辑说明executemany批量插入比逐条 execute 快一个量级with conn保证事务在最后统一提交否则默认每条 insert 都自动提交遇到几十万条数据会明显卡顿。参数说明建表语句把 price 按整数存储避免浮点误差查询后再除以 100索引加在 time 字段上因为最常见的查询条件就是按时间范围聚合成分钟序列。这一步做完数据就从二进制文件变成了可查询的结构化数据后续计算都在这个基础上展开。4. 进阶还原主力资金流与主动买卖单的计算4.1 主动买卖方向的判定规则如果数据里带了明显的方向字段比如 0 表示主动卖、1 表示主动买那么直接映射就行。但我遇到过一些历史数据只有成交价和成交编号没有方向字段。这时候我采用的是一种极简还原规则对比相邻两笔成交如果当前成交价高于上一笔则大概率是主动买低于上一笔则是主动卖价格相等则维持前一笔方向。这个方法不是百分百精确但多数行情软件都采用类似思路。def infer_direction(ticks): direction [] prev_price None prev_dir None for t in ticks: if prev_price is None: d 1 # 第一笔按主动买处理 elif t[price] prev_price: d 1 elif t[price] prev_price: d 0 else: d prev_dir if prev_dir is not None else 1 direction.append(d) prev_price t[price] prev_dir d return direction逻辑说明价格跳价是判断主动权最直观的信号。prev_price为 None 说明是第一笔此时没有参照我默认当成主动买需要单独标记避免影响总体统计。参数说明如果行情数据自身带成交编号还可以额外检查编号连续性发现跳号时把缺失段标记为不可信避免用错误的相邻关系推断方向。这种还原方法在盘口撮合连续时表现尚可但在早盘集合竞价后第一笔成交时容易误判需要配合时间过滤一起使用。4.2 简易资金流模型四档阈值与分钟聚合市场上常见的资金流划分方式是按单笔成交金额分档常见阈值是超大单 100 万以上、大单 20 万到 100 万、中单 4 万到 20 万、小单 4 万以下。不同数据商阈值略有差异代码包里把阈值做成了可配置参数方便按自己的口径调整。计算资金流时先判断单笔属于哪一档再把主动买的金额记为正向流入、主动卖的金额记为负向流出按分钟聚合。import pandas as pd def compute_money_flow(ticks, minute_interval1): df pd.DataFrame(ticks) df[amount_yuan] df[amount] * 100.0 # 假设原始金额单位是万元 df[minute] df[time] // 100000 # 取前6位作为HHMMSS # 简化处理转成小时分钟 df[minute_key] df[time] // 1000 def classify(row): amt row[amount_yuan] if amt 1000000: return 超大单 elif amt 200000: return 大单 elif amt 40000: return 中单 else: return 小单 df[order_type] df.apply(classify, axis1) df[flow] df.apply( lambda r: r[amount_yuan] if r[direction] 1 else -r[amount_yuan], axis1 ) grouped df.groupby([minute_key, order_type])[flow].sum().unstack(fill_value0) grouped[净流入] grouped.sum(axis1) return grouped逻辑说明先把时间戳转成可分组的分钟键time // 1000能把类似 93001001 变成 9300100即去掉毫秒位保留到秒再聚合时按分钟处理。amount_yuan是我按自己样例数据的单位换算成元如果你手上的原始单位就是元这步可以省略。参数说明classify函数里的四个阈值决定了分档口径实盘分析时我建议先看看当天单笔金额分布再决定阈值是否合理flow字段直接由方向决定正负方向为 1 时计为正流入方向为 0 时计为负流出。这个模型没有考虑撤单、市价单成交滑点等因素做相对趋势判断足够用。4.3 对账校验分钟合计与日线总量资金流算完之后不能直接拿去用必须做一步对账。最有效的办法是把当日所有逐笔成交量累加与日线文件里的总成交量做对比。两者差距如果在 1% 以内说明解析基本完整如果差距大就要回头检查是不是漏读了某段时间的数据或者文件本身有截断。我一般会在代码里自动输出这样一行当日逐笔总成交 1234567 手日线总成交 1235000 手吻合度 99.97%。这个数值低于 99% 时我会先检查是否过滤掉了集合竞价时段的记录再看文件末尾是否有剩余字节没解析。对账是成本最低的质量检查手段能提前暴露很多肉眼看不出的问题。5. 常见问题与排查五个实测踩坑点5.1 解析到一半程序崩溃记录结构错位现象某个文件解析到大约三分之一时抛异常报错信息指向 unpack 解出异常大的数值有时还会出现数组越界。原因我一度以为是文件本身损坏后来逐段调试发现是前几条记录之间有一个隐藏的字段长度不一致。部分记录在末尾多带了一个扩展字节导致从那条记录之后全部错位。解决不要直接改代码硬跳字节先把前几十条记录的原始十六进制打印出来人工对一遍。发现规律后把变长部分单独处理例如读取时先判断某字段值是否落在合法区间如果不合法则把它当作扩展字段跳过。从那以后我每次拿到新格式的行情数据都会先做小样本试解析打印前 20 条原始和解包结果对比后再全量跑。5.2 价格变成负数或异常小数现象解析出的某只高价股价格变成了 0.03还有一部分变成负数整体看起来完全不符合行情常识。原因这是字节序理解反了。数据文件是小端序但代码里用了大端序格式串解析导致整数的高低字节被颠倒价格自然变成莫名其妙的数值。解决用第 3.1 节的头部校验函数判断字节序或者找一个已知当日收盘价的标的做校准。解析完成后输出第一条和最后一条记录对比行情软件里的价格能快速发现这类问题。我一般还会增加断言如果解析出的价格超过该标的历史价格范围十倍直接判定为字节序错误。5.3 集合竞价那笔大单总是把净流入拉爆现象每天 09:30 那一分钟的资金流异常巨大频繁出现超大单净流入和盘面实际感受不符。原因集合竞价阶段会产生一笔或几笔成交金额远超正常连续竞价单笔如果直接参与分钟聚合会把 09:30 那一分钟的净流入拉出一个假高峰。解决在计算资金流前把时间早于 09:30:00 的成交记录单独剔除或者单独标记为“集合竞价”分组。常见做法是只取 09:30:00 之后的记录做分钟资金流集合竞价数据单独统计。这里还要注意尾盘 14:57 到 15:00 的收盘集合竞价同样需要单独处理不能简单混合进普通分钟序列。5.4 分钟量合计总是对不上日线总量现象所有分钟量加总之后比日线总成交少了一截少的数量不固定有时差几十手有时差几百手。原因逐笔数据完整性的问题。部分成交记录可能因为数据供应商的传输策略被合并或省略比如某些平台把同一秒钟内同价格的连续成交合并为一条另外大宗交易和盘后固定价格交易也可能不计入逐笔文件。解决不要硬凑数据。先把差异量级算出来如果长期稳定在固定比例考虑是不是单位换算出错如果差异来自盘后固定价格交易就单独把日线总量拆成连续竞价和盘后两部分再对账。我一般会在聚合结果里加一列“覆盖度”低于 99% 时给警告提示避免后续分析建立在缺损数据上。5.5 SQLite 写入越来越慢现象同一个脚本解析当天数据入库时只需要几秒但连续处理一周数据后入库速度明显变慢后面几天要等很久。原因每插入一条记录都自动提交事务加上没有先建索引后插入导致磁盘随机写入频繁。索引会在每次 insert 时同步更新数据量越大越慢。解决改成批量插入并先插入全部数据再单独建索引。代码包里默认用的是executemany加with conn包裹就是防止这个问题。如果数据量超过几千万条我会改用先写 CSV 再导入 SQLite 的方式SQLite的.import命令比自己逐条 insert 快一个量级。6. 把解析结果变成图表资金流曲线与一句话量化结论数据算出来是为了指导判断不是看一串数字。我习惯把资金流结果直接画成图用柱状图表示每分钟净流入用折线表示当日累计净流入这样主力和散户的进出节奏一眼就能看出来。matplotlib 是最直接的选择画完保存成 PNG 或者 SVG 都行。import matplotlib.pyplot as plt def plot_money_flow(grouped, output_image): fig, ax plt.subplots(figsize(12, 6)) ax.bar(grouped.index, grouped[净流入], color#d62728, label分钟净流入) ax2 ax.twinx() cumulative grouped[净流入].cumsum() ax2.plot(grouped.index, cumulative, color#1f77b4, label累计净流入) plt.title(Level2 逐笔资金流) plt.savefig(output_image, dpi150) plt.close()逻辑说明twinx()创建双轴柱状图展示每分钟资金流折线展示累计值两者单位相同但量级不同放一起能同时看到节奏和趋势。参数说明grouped是第 4.2 节函数的返回结果index是分钟时间键横轴会显示为形如 9300100 的数字可读性一般正式出图前我会格式化成时分字符串。除了画图我还会让脚本自动输出一句结论例如“09:30 到 14:00 主力资金持续净流出14:00 后转为净流入全天净流入占总成交额 2.3%”。这句话直接基于分组求和的结果生成省得每次人工看图再口算。整个过程跑通后每天收盘后十分钟内就能拿到一份完整的盘口资金流报告。最后说一个我自己的教训有次直接拿没处理复权的价格去算资金流结果某标的除息日那天除息前后价格跳空导致当日净流入被生生拉出一个假缺口我还以为是主力大举进场。从那以后我每次做分钟级资金流都强制走一遍复权校验先对比当日价格是否出现超过 10% 的跳空再检查日线总量与逐笔总量是否吻合两关都过了才把结果用于决策。希望帮到你。本文还有配套的精品资源点击获取

通达信Level2数据解析：逐笔成交与主力资金流实战指南

相关推荐

有关大数据，看这一篇就够了！

Boring Notch 技术指南：把 MacBook 刘海变成音乐控制中心、日历与系统 HUD

深度解析无线鼠标蓝牙芯片封装：技术演进与封测行业新机遇

更多资讯

湖仓一体深度拆解：从技术原理到落地避坑指南

七轴机械臂之字形走线实战（二）：从仿真到真机，第一次全程走线

新文档进来只能全量重建？把索引更新的账算清

429 一来就狂重试：退避时间预算怎么算

两万星标AI编程教程：项目驱动与实战踩坑经验分享

JSP房产中介管理系统部署实战：从环境搭建到二次开发全流程解析

今日更新

小波包能量谱原理与MATLAB实现：从频带分解到故障特征提取

基于LSTM的网易云音乐评论情感分析：数据清洗与模型调优实战

Alpha-Beta剪枝与极大极小值算法：博弈树搜索原理与实战优化

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？