l2data-feeder 0.1.0

A library for real-time tick-level data slice feed

l2data_feeder

一个用于实时 Level2 行情数据处理的高效 Python 库

简介

l2data_feeder 是一个专为金融市场 Level2 数据处理设计的 Python 库。它能够从 Level2 行情数据文件中读取 tick、订单和成交数据，并按照指定的时间间隔将这些数据聚合到 Slice 对象中，便于后续分析和处理。

该库基于 l2data_reader 模块实现，支持处理由 protobuf 定义的 Level2 行情数据结构，包括 SecuDepthMarketData、TransactionEntrustData 和 TransactionTradeData 等。

特性

• 数据预取模式：可以预先读取指定数量的消息，快速获取历史数据
• 实时更新模式：支持实时读取新数据，通过生成器方式返回每个时间片的数据
• 灵活的时间间隔设置：可自定义时间片的间隔（毫秒级）
• 多股票并行处理：自动按股票代码聚合数据
• 完整的数据结构：保留原始 Level2 数据的所有字段信息

依赖

本库依赖于 l2data-reader 库，用于读取 Level2 行情数据。安装时会自动安装此依赖。

安装

pip install l2data_feeder

快速开始

基本用法

from l2data_feeder import Level2SliceFeeder

# 初始化数据源
index_file = "path/to/index.idx"
data_file = "path/to/data.bin"
header_file = "path/to/header.hdr"

# 创建 Level2SliceFeeder 实例
feeder = Level2SliceFeeder(
    index_file=index_file,
    data_file=data_file,
    header_file=header_file,
    time_slice_ms=100,  # 时间片间隔为100毫秒
    prefetch_count=100  # 预取100条消息
)

# 预取数据
slices = feeder.prefetch()
print(f"预取了 {len(slices)} 个时间片")

# 打印第一个时间片的信息
first_slice = slices[0]
for symbol, snapshot in first_slice.Ticks.items():
    print(f"股票: {symbol}")
    if snapshot.Tick:
        print(f"  最新价: {snapshot.Tick.last_price}")
        print(f"  成交量: {snapshot.Tick.volume}")
    print(f"  委托数量: {len(snapshot.Orders)}")
    print(f"  成交数量: {len(snapshot.Transactions)}")

实时模式

# 创建实时模式的 Level2SliceFeeder
realtime_feeder = Level2SliceFeeder(
    index_file=index_file,
    data_file=data_file,
    header_file=header_file,
    time_slice_ms=100,
    prefetch_count=0,  # 不预取数据
    realtime=True      # 启用实时模式
)

# 实时处理数据
for slice_data in realtime_feeder.feed():
    # 处理每个时间片的数据
    for symbol, snapshot in slice_data.Ticks.items():
        if snapshot.Tick:
            # 处理 tick 数据
            process_tick(snapshot.Tick)
        
        # 处理委托数据
        for order in snapshot.Orders:
            process_order(order)
        
        # 处理成交数据
        for trade in snapshot.Transactions:
            process_trade(trade)

使用辅助函数create_feeder, 更方便地使用库：

from l2data_feeder import create_feeder

# 创建 feeder 实例
feeder = create_feeder(
    index_file="path/to/index.idx",
    data_file="path/to/data.bin",
    header_file="path/to/header.hdr",
    slice_interval_ms=100,
    realtime=True
)

# 使用 feeder
for slice_data in feeder.run():
    # 处理每个时间片的数据
    pass

完整示例

from l2data_feeder import Level2SliceFeeder
import time

def main():
    # 初始化数据源
    index_file = "path/to/index.idx"
    data_file = "path/to/data.bin"
    header_file = "path/to/header.hdr"
    
    # 创建 Level2SliceFeeder 实例
    feeder = Level2SliceFeeder(
        index_file=index_file,
        data_file=data_file,
        header_file=header_file,
        time_slice_ms=100,
        prefetch_count=100,
        realtime=True
    )
    
    # 预取数据
    print("开始预取数据...")
    slices = feeder.prefetch()
    print(f"预取完成，共 {len(slices)} 个时间片")
    
    # 打印预取数据的统计信息
    symbols = set()
    tick_count = 0
    order_count = 0
    trade_count = 0
    
    for slice_data in slices:
        for symbol, snapshot in slice_data.Ticks.items():
            symbols.add(symbol)
            if snapshot.Tick:
                tick_count += 1
            order_count += len(snapshot.Orders)
            trade_count += len(snapshot.Transactions)
    
    print(f"预取数据统计:")
    print(f"  股票数量: {len(symbols)}")
    print(f"  Tick数量: {tick_count}")
    print(f"  委托数量: {order_count}")
    print(f"  成交数量: {trade_count}")
    
    # 进入实时模式
    print("\n进入实时模式，按 'q' 退出，按 'y' 打印当前时间片详情:")
    try:
        for slice_data in feeder.feed():
            # 打印简要信息
            symbols_in_slice = list(slice_data.Ticks.keys())
            print(f"\r当前时间片包含 {len(symbols_in_slice)} 个股票: {', '.join(symbols_in_slice[:3])}{'...' if len(symbols_in_slice) > 3 else ''}", end="")
            
            # 检查用户输入
            user_input = input("\n是否打印详情? (y/n/q): ")
            if user_input.lower() == 'q':
                break
            elif user_input.lower() == 'y':
                # 打印详情
                for symbol, snapshot in slice_data.Ticks.items():
                    print(f"股票: {symbol}")
                    if snapshot.Tick:
                        print(f"  时间: {snapshot.Tick.time}")
                        print(f"  最新价: {snapshot.Tick.last_price}")
                        print(f"  开盘价: {snapshot.Tick.open_price}")
                        print(f"  最高价: {snapshot.Tick.high_price}")
                        print(f"  最低价: {snapshot.Tick.low_price}")
                        print(f"  成交量: {snapshot.Tick.volume}")
                    print(f"  委托数量: {len(snapshot.Orders)}")
                    print(f"  成交数量: {len(snapshot.Transactions)}")
            
            # 模拟处理时间
            time.sleep(0.1)
    except KeyboardInterrupt:
        print("\n程序被中断")
    finally:
        feeder.stop()
        print("数据读取已停止")

if __name__ == "__main__":
    main()

API 参考

Level2SliceFeeder

主要类，用于读取和处理 Level2 行情数据。

参数

• index_file (str): 索引文件路径
• data_file (str): 数据文件路径
• header_file (str): 头文件路径
• time_slice_ms (int, 可选): 时间片间隔，单位毫秒，默认为 100
• prefetch_count (int, 可选): 预取消息数量，0 表示不预取，-1 表示全部读取，默认为 0
• realtime (bool, 可选): 是否启用实时模式，默认为 False
• logger (logging.Logger, 可选): 日志记录器，默认为 None

方法

• prefetch(): 预取数据，返回 Slice 对象列表
• feed(): 生成器方法，实时返回每个时间片的 Slice 对象
• stop(): 停止数据读取

Slice

表示一个时间片的数据集合。

属性

• Ticks (Dict[str, Snapshot]): 按股票代码索引的 Snapshot 对象字典

Snapshot

表示一个股票在特定时间点的快照数据。

属性

• Symbol (str): 股票代码
• Tick (SecuDepthMarketData): Tick 数据
• Orders (List[TransactionEntrustData]): 委托数据列表
• Transactions (List[TransactionTradeData]): 成交数据列表

构建与发布

要构建并发布此包到 PyPI，请按照以下步骤操作：

1. 确保已安装必要的工具：

   pip install build twine
   

2. 构建包：

   python -m build
   

3. 上传到 PyPI：

   python -m twine upload dist/*
   

许可证

MIT

贡献

欢迎提交 Pull Request 或创建 Issue 来帮助改进此项目。

联系方式

如有问题，请通过 GitHub Issues 联系我们。