Python实战通达信Level2数据：解析、存储与盘口信号计算

简介一份面向职业交易者和进阶投资者的通达信Level2数据实操指南围绕逐笔成交、委托队列、资金流向与盘口异动等核心维度解析主力资金动向、虚假挂单和大单拆分等实战信号帮助读者建立盘口阅读能力与信息优势。资源包共3个文件包含可交互的HTML说明页面、inscode代码配置和gitignore项目配置压缩后仅4KB轻量便捷适合下载后直接打开对照学习或按需整合进自己的看盘环境。目前已有1270人学习内容兼顾Level2数据语言的解读、定制版面搭建与模拟实战路径并给出挑选优质Level2服务的关键标准如极速行情、功能完整度与专属投顾支持。整体以代码与页面组合的形式呈现既说明数据维度含义也给出落地到看盘页面的方法是一份从入门到实战的浓缩型参考资料。1. 通达信Level2数据指南把盘口从「看不全」变成「可计算」盯盘时最难受的一种场面是买一大单挂着价格却死活不涨撤了又挂、挂了又撤普通行情只到五档根本看不穿五档之外还压着多少卖单。通达信这类行情终端里的Level2数据就是冲着这类问题来的——十档行情、逐笔成交、委托队列能把「看不全」的盘口变成「可计算」的结构。这个标题里的[代码]部分目标是把Level2原始数据解析成能放进策略的结构化表格再从中算大单流向、盘口压力差一类信号。适合正在做量化、盘口研究和短线回测的开发者只想看图不想写代码的人这篇帮不到你。数据源按正规订阅渠道获取拿到手之后的解析、存储、计算才是这条主线要解决的问题。2. Level2数据到底长什么样先分清快照流、逐笔流和委托队列很多人拿到Level2数据后的第一反应是「这不就是多几档行情」。实际上刚导出的原始文件像个黑匣子二进制流、字段缩写、奇怪的单位直接开解析大概率踩坑。Level2的核心增量是三类数据十档快照、逐笔成交、委托队列。它们频率不同、字段不同、能回答的问题也不同。解析代码写崩往往是因为没先想清楚这三类数据各自的边界。2.1 快照流和逐笔成交流两个数据源的分工不同快照流是定时间隔推送的盘口状态汇总视图通常是买卖各十档的价位和委托量附带最新价、今开、最高、最低、累计成交量额。它像每隔几秒拍一张照片记录的是「此刻盘口挂单长什么样」。逐笔成交流则不同它忠实记录每一笔已经发生的成交明细包含成交时间、价格、数量、金额和主动性方向。它像流水账回答的是「刚才到底发生了什么」。两者的关系是互补而非重复快照描述状态逐笔描述事件。维度快照流逐笔成交流内容十档盘口、最新价、累计量额单笔成交明细推送方式固定间隔推送每笔成交触发推送核心用途观察盘口厚度与挂单分布还原资金行为与成交结构数据量级一天约几千到几万条/标的一天约几万到几十万条/标的委托队列是第三类数据把买一或卖一这一档的排队委托明细列出来能看到某个大单是真排队还是不断撤换。它的细节比快照丰富但通常只覆盖最优一档或若干档不是全市场深度。三类数据建议分类存储、分类处理不要混进同一张表否则后续做盘口指标时字段语义会被搅乱。2.2 三个容易错的基本单位时间、价格、数量解析Level2最容易翻车的不是逻辑是单位。不同来源给的数据时间可能是毫秒或微秒价格精度可能是0.01元或0.001元数量可能是股也可能是手。这类错肉眼几乎发现不了直到对账或回测时数字对不上才暴露。字段常见单位/精度建议统一基准时间当日0点起毫秒/微秒偏移毫秒 int另存 YYYYMMDD 日期 int价格0.01 / 0.001 元部分源放大100/1000/10000整数 实际价 × 1000数量股或手一律转成「股」成交金额元或千元整数 元 × 1000与价格精度一致为什么数据源之间单位差别这么大一个原因是上游行情协议本身就存在多套规格另一个原因是部分落地存储为了压缩空间做了缩放。这几乎是所有做Level2接入的人都会在不同来源之间遇到一次的事。我一般会在解析前先写一个单位配置字典把所有原始值统一过一遍转换绝不在后续代码里临时乘除宁可转换后多打印一次也不要在策略里猜单位。2.3 规范化统一基准能省掉八成对账麻烦规范化做得好第3章的解析代码只需要处理一种口径。时间键和代码键是两个核心快照和逐笔必须用同一种时钟基准当日偏移毫秒否则对齐时差三秒是常事证券代码统一成去掉前导0后的整数或者统一附加市场位避免「600000」和「1600000」这类市场歧义。# 建议在解析配置里统一声明字段来源和缩放系数 FIELD_SPEC { price: {raw: last_price, scale: 1000}, # 原始价格 × scale 后按整数存储 volume: {raw: volume, scale: 1}, # 原始量已经是股 amount: {raw: amount, scale: 1000}, # 金额按 元×1000 存与价格一致 }这段配置的价值在于把「单位约定」收敛到一个文件里。解析、清洗、回测三个模块都从FIELD_SPEC取值任何人改单位只需改这一处不会出现「这里除了1000、那里又乘了1000」的连锁错。实际使用中我还会在配置里加一个SCHEMA_VERSION字段解析脚本变更时同步加版本号方便追溯哪些历史文件是用旧规则生成的哪些需要重算。3. 用Python解析Level2原始数据从字节流到DataFrame3.1 为什么解析前先定义结构化数组内存和速度差出数量级很多第一次写Level2解析的人直接用list of dict解析几十万条还能忍全市场几百万条就跑不动了。常见做法是先定义NumPy结构化数组字段固定、类型紧凑内存占用往往是dict方案的几十分之一。以逐笔成交为例结构定义如下。import numpy as np import pandas as pd TICK_DTYPE np.dtype([ (ts, u8), # 时间戳当日0点起毫秒数 (date, u4), # 交易日YYYYMMDD 整数 (code, u4), # 证券代码去掉前导0后的整数 (price, u4), # 价格实际价格 × 1000 取整 (volume, u8), # 成交量单位股 (amount, u8), # 成交金额单位 元 × 1000与价格精度一致 (side, i1), # 主动性方向1主动买 -1主动卖 0集合竞价 ])这里每个字段都取了保守的整数类型。价格用无符号整型而不是浮点是因为浮点在大量累加时会有精度漂移整数完全没有这个问题。时间用无符号整型而不是datetime对象是因为datetime内存开销大、比较慢回测时按毫秒对齐用整数查数组最快。side这种取值只有三种的字段1字节有符号整型就够没必要为它花4字节。结构化数组定义好后后面所有解析函数都围绕它来填充数据格式变更时只需要改这一处dtype。3.2 用固定长度记录流的批量解析法实际拿到的Level2原始数据通常是固定长度的二进制记录流。解析思路统一确认单条记录字节数用NumPy的frombuffer按字段偏移一次性切出所有字段数组再组装成结构化数组。下面代码假设每条记录25字节、小端序、字段顺序是时间(8)代码(4)价格(4)量(8)方向(1)实际按你的数据源文档修改offset和dtype即可。def parse_tick_bin(buf: bytes): 把逐笔成交二进制流解析成结构化数组 单条记录结构小端序 ts u8 偏移0 时间戳当日0点起毫秒 code u4 偏移8 证券代码整数 price u4 偏移12 价格整数实际价×1000 volume u8 偏移16 成交量股 side i1 偏移24 主动买卖方向 合计 25 字节/条 n len(buf) // 25 if n 0: return np.zeros(0, dtypeTICK_DTYPE) arr np.zeros(n, dtypeTICK_DTYPE) arr[ts] np.frombuffer(buf, dtypeu8, countn, offset0) arr[code] np.frombuffer(buf, dtypeu4, countn, offset8) arr[price] np.frombuffer(buf, dtypeu4, countn, offset12) arr[volume] np.frombuffer(buf, dtypeu8, countn, offset16) arr[side] np.frombuffer(buf, dtypei1, countn, offset24) return arr逻辑说明先用总字节数除以单条记录长度算出条数然后按字段偏移依次切出整列数组。frombuffer是零拷贝读取比循环struct.unpack快两个数量级以上。参数说明里最关键的是偏移量它由前面字段的字节数累加得到84416是volume的起点16824是side的起点。如果你的记录带文件头或字段顺序不同只要重新累加一遍偏移量即可解析函数本体不用改。另外提醒一句如果上游用了结构体对齐单条记录的实际字节数可能大于字段宽度之和不要凭字段和猜长度先对比两条相邻记录的起始地址确认真实步长。3.3 快照流解析与逐笔的时间对齐快照流的解析方式和逐笔相同只是字段更多买一到买十的价格和量、卖一到卖十的价格和量加上最新价和累计量。解析完成后最重要的工作是时间对齐一笔成交发生在两个快照之间要判断它撞在什么盘口状态下必须找到它之前最近的一条快照。用NumPy的searchsorted一步完成不要用循环遍历。def align_ticks_to_snapshot(ticks, snaps): 给每条逐笔成交找它之前最近的一条快照 ticks: 逐笔结构化数组 snaps: 快照结构化数组需含 ts 字段 返回: 与ticks等长的快照索引数组-1表示该时刻之前还没有快照 idx np.searchsorted(snaps[ts], ticks[ts], sideright) - 1 return idx # 用法取对齐后该笔成交对应的买一价 idx align_ticks_to_snapshot(tick_arr, snap_arr) snap_bid1 snap_arr[bid_price1][idx]逻辑说明searchsorted返回的是插入位置减1就是「小于等于当前时间」的最后一条快照索引。参数sideright表示时间相等时取右侧插入位置保证对齐到最新一条快照而非更早那条。对齐之后你就能在逐笔成交上附加那一刻的十档状态后面算「大单连续打穿买一」这类信号就顺理成章了。快照字段名建议固定为bid_price1到bid_price10、bid_vol1到bid_vol10以及ask_price1到ask_price10、ask_vol1到ask_vol10从解析到回测都用同一套命名避免语义混乱。3.4 解析完先别急着用四行代码做自检解析完还不能直接用先跑四个检查价格范围是否合理、时间是否单调、价量额是否互相吻合、价格精度是否符合预期。每个检查两三行代码能拦掉八成低级错误。import pandas as pd import numpy as np df pd.DataFrame(tick_arr) # 1) 价格粗查不可能是负数也不该出现离谱极值 print(df[price].min(), df[price].max()) # 2) 时间乱序检查负值个数就是乱序记录数 print((df[ts].diff() 0).sum()) # 3) 价量额互查用相对误差数量级错会很明显 denom df[amount].replace(0, np.nan) ratio ((df[price] * df[volume] - df[amount]).abs() / denom).max() print(ratio) # 4) 精度检查如果 price 几乎全是 10 的倍数 # 说明价格精度可能是 0.01 而不是 0.001需复核缩放系数 print((df[price] % 10 0).mean())逻辑说明第三项利用了前面统一的整数单位price乘以volume应该等于amount的量纲因为两者都放大了1000倍乘积正好抵消。相对误差在千分之一以内属于正常尾差如果差出10倍或100倍基本可以断定某一列的单位配置错了。第四项更隐蔽价格按0.001元精度存时整数尾部0~9都会有但如果数据源实际精度是0.01元那么几乎所有的price都是10的倍数这一项能直接暴露缩放系数配错的问题。这套自检在每次接入新数据源、升级解析脚本时都要重跑一遍。4. 存储与管理Level2数据按天分文件别把内存当仓库4.1 Parquet比CSV和Pickle都靠谱Level2数据写回磁盘常见做法是Parquet格式。它是列式存储压缩率高读的时候可以只读部分列CSV解析慢、占空间Pickle虽然存取快但格式锁定在Python换环境就麻烦。保存和读取的代码很简单但有一个参数值得调compression建议用zstd压缩率和读取速度的平衡比gzip好不少。def save_daily(tick_df: pd.DataFrame, date: int, path: str): 按日保存逐笔数据到parquet date: YYYYMMDD 整数比如20240603 path: 存放目录目录需要已存在 tick_df.to_parquet(f{path}/{date}.parquet, indexFalse, compressionzstd) def load_daily(date: int, path: str) - pd.DataFrame: return pd.read_parquet(f{path}/{date}.parquet)逻辑说明按日分文件的模式文件名为date.parquet后面做按日期范围回测时直接生成文件路径列表即可不需要维护数据库索引。参数说明里indexFalse必须写否则pandas会把行号也存进文件白白占空间。环境里需要装pyarrow作为parquet后端标准安装命令即可。数据量再大的场景建议用pyarrow.dataset做分区读取只扫描需要的日期文件而不是每次都读整表。pip install pyarrow估算一下数据量一只活跃股票一天的逐笔成交在几万到几十万条之间全市场几千只股票全部做单日逐笔就是千万量级。快照流按3秒一档、每只股票一天约4800条全市场同样千万级。两者加起来单日原始数据几个GB不稀奇。所以存储设计不是锦上添花而是还没跑回测就会被数据量教做人的地方。4.2 按标的和按日分块别一次性加载全市场全市场Level2一天的逐笔数据放内存不是不行但会挤掉回测程序其他模块的余地。我的经验是按日加载一个标的或一个板块跑完即释放比一次性全量加载省心得多。用DataFrame按code分组处理就能实现。def process_one_day(date: int, data_path: str, code: int): 回测单日单标的数据 code: 整数证券代码 df load_daily(date, data_path) target df[df[code] code].copy() # 只保留一个标的 # 处理完后 target 离开函数作用域即可被回收 return target逻辑说明用code过滤后.copy()是为了断开与全量DataFrame的引用关系让全量数据可以尽早被垃圾回收。如果你确认后面不再用全量数据不copy直接过滤也行但多数回测脚本后面还会处理其他标的建议copy。也有必须全量加载的场景比如做全市场横截面统计、计算某个因子需要同一时刻全市场的十档状态这时建议用pyarrow的dataset只读需要的列而不是整表。另一个折中方案是先按日聚合出中间特征比如每5分钟的大单净流入把中间结果落地回测只加载中间结果——原始逐笔归档不用常驻。4.3 历史回测怎么读最省事日期迭代而不是一次性加载历史回测最容易写崩的地方就是一次性把所有天数的数据读进内存内存翻车后再回头改结构。常见做法是写一个日期迭代器按天读、按天算、按天存结果。即使数据覆盖好几年内存峰值也只是「单日数据量累积结果」可控得多。import os def iter_dates(start: int, end: int, data_path: str): 返回闭区间[start, end]内实际存在的日期文件列表 return sorted( int(f[:-8]) for f in os.listdir(data_path) if f.endswith(.parquet) and start int(f[:-8]) end ) # 回测主循环一天一天读结果累计内存峰值可控 for d in iter_dates(20240101, 20240630, data_path): daily load_daily(d, data_path) # 在这里跑当天的信号计算把结果追加到总表逻辑说明文件名去掉.parquet后就是日期用切片f[:-8]提出来转int配合闭区间过滤这是最轻量的索引。参数说明里要注意文件命名的固定长度只有day.parquet这种固定命名才有f[:-8]可切如果命名变成code_date.parquet就要用split方法分段取值。命名规范在存储设计时就要定死后面所有读取代码都依赖它。按日分文件还有一个隐藏收益清洗逻辑升级时不用全量重跑。某一批数据发现脏点只需要重新解析并覆盖对应日期的文件。代价是要保证解析脚本的确定性同一份原始数据、同一个版本脚本必须输出完全相同的文件。5. Level2数据解析避坑指南五个反复翻车的点这五条基本是我踩出来的血泪经验。它们的共同点是数据看起来「大致正常」但细节经不起对账。解析脚本写完后先用一个标的、一天的数据做全链路对账再放量跑全市场能省掉后面大量返工。5.1 价格精度对不上买一价和逐笔价差十倍现象同一时刻逐笔成交价显示12.34快照里的买一价显示1.234两个数据源看起来差了一个数量级。 原因两个数据流的精度约定不一致一个按0.01元存一个按0.001元存缩放系数还不同。 解决所有字段先归一化到同一个精度再解析统一用整数实际价×1000。核对方法很简单拉出当日最高价和最低价分别按缩放系数换算看能否对上行情软件里显示的极值再随机抽10条逐笔和行情软件的手工明细对照价格和成交量。如果精度不对通常所有价格都差同样的倍数这个规律能快速判断是整体缩放问题还是个别字段异常。5.2 逐笔总量和快照累计量永远对不上现象把某标的当日逐笔成交量加总和快照里的累计成交量一对比差一截怎么查都查不平。 原因快照的累计量包含集合竞价和盘后固定交易等非连续竞价时段的成交而逐笔流通常只覆盖连续竞价另外撤单中部分成交的记录口径也不同。 解决不要做全量核对。只比较连续竞价区间9:30-11:30、13:00-15:00内逐笔增量和快照增量的差值。做法是取逐笔流中ts落在连续竞价区间的记录求和再取相邻两个快照的累计成交量之差做对比。快照累计量一般是单调递增的如果差值为负就说明快照有遗漏或数据不连续。统计口径要在代码注释里写明「本统计仅覆盖连续竞价」防止后人误解。5.3 内存直接拉爆回测脚本卡死现象加载全市场几天数据后内存占用飙到几十G程序无响应只能杀掉重来。 原因一个典型错误是让pandas用object类型存字符串代码和datetime时间单条记录的开销比整数类型大几十倍。 解决按第3章的结构化数组存储代码和时间全用整数回测按天迭代而不是一次性load全部。先拿10只票跑通流程确认内存和耗时曲线后再按比例估算全市场。别一上来就全市场全量先用小样例把解析、存储、读取三条链路测匀速再放开范围。5.4 时间戳乱序成交顺序看起来错乱现象打印某几分钟的记录时间戳忽大忽小成交顺序明显不对。 原因数据源可能是多路并发合流断线重连或并发写入都会导致文件物理顺序不等于时间顺序。 解决解析后按ts排序如果数据源带有行情序号字段先按序号校验再按ts排序。不要假设文件里的顺序是对的所有依赖先后关系的计算都先做一次排序。乱序最常见的后果是资金流方向统计被颠倒一笔主动买因为排序错误被当成主动卖净流入直接反号所以排序不是可选项而是所有时序计算的前置步骤。5.5 除权日当天历史统计突然失真现象用历史数据验证某个大单买入信号发现在除权日前后信号频率暴增或骤减。 原因除权除息导致的标称价格跳空被当成真实成交价差来计算信号干扰统计。 解决加载除权除息表做复权处理。除权除息表一般包含除权日、每股派息、送转比例等字段向前复权的常见做法是以最近日期为基准把历史价格按复权因子逐日调整。短线盘口研究一般用前复权视角更符合盘中感知长周期跨年回测务必明确复权方式并在结果里标注。如果信号只依赖相对价位比如十档价差复权的影响可能不大但一旦涉及绝对价格、涨停价判断或收益率计算就必须处理复权。不做这一步后面的统计结论都不可信。6. 把解析好的Level2数据用起来大单流与盘口压力差的代码盘口研究里Level2数据最直接的价值是两个还原大单行为、量化盘口压力。下面给两个基础信号的计算代码都是几十行内能跑通的口径。重点不是胜率而是帮你建立「从数据到信号」的完整链路。6.1 大单资金流统计# 假设 tick_df 来自第3章解析结果字段单位与第3章一致 # amount 元 × 1000所以 50 万 500_000 × 1000 big tick_df[tick_df[amount] 500_000_000] buy_flow big.loc[big[side] 1, amount].sum() / 1000 sell_flow big.loc[big[side] -1, amount].sum() / 1000 net_flow buy_flow - sell_flow print(f大单净流入: {net_flow:.0f} 元)逻辑说明先按金额阈值筛出大单再按主动买卖方向分别求和。主动买意味着一笔成交由买方主动吃掉通常解读为买入意愿更强主动卖相反。参数说明里有两个口径要自己定一是单笔金额下限50万适合中型个股大盘股建议调到100万以上二是是否排除集合竞价side等于0的记录默认建议丢弃它们对连续竞价场景的意义不大。6.2 委买委卖压力差与十档深度# 假设 snap_df 是快照流解析结果字段为 bid_vol1..bid_vol10, ask_vol1..ask_vol10 bid_cols [fbid_vol{i} for i in range(1, 11)] ask_cols [fask_vol{i} for i in range(1, 11)] bid_total snap_df[bid_cols].sum(axis1) ask_total snap_df[ask_cols].sum(axis1) # 压力差正数表示买盘量更大负数表示卖盘量更大 pressure (bid_total - ask_total) / (bid_total ask_total)逻辑说明这个指标用十档总量替代普通行情的五档总量是Level2相对普通行情的典型增量。参数说明里sum的axis1必须写否则得到的是每列的和分母加总买卖两方总量做归一化是为了让不同标的、不同价位的绝对量可比。实际使用中建议分别算前五档和全十档两份压力差对比两者的差异能看出「五档之后藏了多少挂单」这正是普通行情看不到的部分。6.3 信号验证的一个朴素习惯盘口信号天然带一点玄学同一个大单流指标在不同标的上的表现可能完全不同。我自己的习惯是新定义一个信号后先不回测整个策略只做一件事——把过去20个交易日里这个信号的出现次数和次日开盘后的价格走向拉出来看分布。手动翻图几十次比任何统计检验都能更快建立起「这个信号到底有没有用」的感觉。警惕过拟合Level2的字段足够多想组合出漂亮的信号很容易但样本外大概率失效。最稳妥的路径是先用一个极简口径跑通全链路再逐步加过滤条件每加一个条件都必须看到样本内外表现一致才保留。希望今天的这份指南能帮到你让你的Level2数据从一坨二进制真正变成一个可以反复验证的信号源。本文还有配套的精品资源点击获取