orderpulse 0.2.23

High-performance exchange feed parser and orderflow analytics engine with Rust and Python bindings

fastreader

High-performance Rust + PyO3 toolkit for binary market data decoding, sequential replay, and order book snapshot construction from Python.

This project is optimized for quant and market microstructure workflows where Python ergonomics and Rust performance are both required.

What This Library Does

• Reads binary order/trade feed files in Rust.
• Exposes Python classes for cached access and stream-style access.
• Builds level snapshots from processed messages.
• Provides a compatibility wrapper for legacy Python usage.

Current Architecture

The core design is split into two layers.

1. Native layer (Rust + PyO3 extension module)

• MessageCacheReader
• StreamingBinaryLoader
• OrderbookBuilder

2. Python package layer (ergonomic wrappers and compatibility)

• BinaryDataLoader (compatibility wrapper)
• BinaryLoader (Python iterator wrapper)
• ReadMsgFromBinary (alias of BinaryDataLoader)

Data Flow

Binary File (.bin)
      |
      v
Rust parser (read_messages)
      |
      +--------------------+
      |                    |
      v                    v
MessageCacheReader   StreamingBinaryLoader
      |                    |
      +----------+---------+
                 |
                 v
          OrderbookBuilder
                 |
                 v
     Snapshot dictionary (token, mid_price, bids, asks)

Installation

Build with maturin

pip install maturin
maturin develop --release

Or wheel build:

maturin build --release

Python Import

import fastreader

Or explicit symbols:

from fastreader import MessageCacheReader, StreamingBinaryLoader, OrderbookBuilder

API Reference (Native Classes)

1) MessageCacheReader

Purpose:

• Load the full feed into memory once.
• Query summary and formatted messages quickly.

Constructor

reader = MessageCacheReader()

load_to_cache(file_path: str) -> int

What it does:

• Parses the binary file.
• Caches all decoded messages in memory.
• Returns message count.

Example:

from fastreader import MessageCacheReader

reader = MessageCacheReader()
count = reader.load_to_cache("data/feed.bin")
print(count)

Expected output pattern:

250000

get_all_messages(limit: int | None = None) -> list[str]

What it does:

• Returns formatted text rows for each decoded message.
• If limit is None, returns all rows.

Example:

rows = reader.get_all_messages(limit=3)
for r in rows:
    print(r)

Expected output pattern:

Order Message: SeqNo1, msg_len64, Msg_Type'O', Exch_ts..., local_ts..., order_id..., Token26000, order_Type'B', Price..., Quantity..., missed0
Trade Message: SeqNo2, msg_len48, Msg_Type'T', Exch_ts..., local_ts..., order_id_buy..., order_id_sell..., Token26000, Price..., Quantity..., missed0
Order Message: SeqNo3, msg_len64, Msg_Type'O', Exch_ts..., local_ts..., order_id..., Token26000, order_Type'S', Price..., Quantity..., missed0

get_cache_summary() -> dict

Returned keys:

• file_path: str | None
• total_messages: int
• total_orders: int
• total_trades: int
• memory_usage_bytes: int

Example:

summary = reader.get_cache_summary()
print(summary)

Expected output pattern:

{'file_path': 'data/feed.bin', 'total_messages': 250000, 'total_orders': 180000, 'total_trades': 70000, 'memory_usage_bytes': 12000000}

2) StreamingBinaryLoader

Purpose:

• Cursor-driven message playback.
• Useful for simulation, replay, and chunk processing.

Constructor

stream = StreamingBinaryLoader()

open_stream(file_path: str) -> None

What it does:

• Parses and stores messages.
• Resets internal cursor to start.

Example:

from fastreader import StreamingBinaryLoader

stream = StreamingBinaryLoader()
stream.open_stream("data/feed.bin")

get_next_message() -> str

What it does:

• Returns next formatted message.
• Returns END when cursor reaches final message.

Example:

for _ in range(5):
    print(stream.get_next_message())

Expected output pattern:

Order Message: SeqNo1, ...
Trade Message: SeqNo2, ...
Order Message: SeqNo3, ...
Order Message: SeqNo4, ...
Trade Message: SeqNo5, ...

At end of stream:

END

get_messages_batch(limit: int) -> list[str]

What it does:

• Returns up to limit messages from current cursor.
• Cursor advances by returned count.

Example:

batch = stream.get_messages_batch(1000)
print(len(batch))

Expected output:

1000

(or lower near stream end)

reset_cursor() -> None

What it does:

• Moves internal cursor to start again.

Example:

stream.reset_cursor()
first = stream.get_next_message()
print(first)

3) OrderbookBuilder

Purpose:

• Process decoded messages into internal order book state.
• Emit current snapshot for a token at requested depth.

Constructor

builder = OrderbookBuilder()

build_from_source(source) -> int

Accepted source types:

• MessageCacheReader
• StreamingBinaryLoader

What it does:

• Ingests message objects from source.
• Updates internal order book manager.
• Returns processed message count.

Example with MessageCacheReader:

from fastreader import MessageCacheReader, OrderbookBuilder

reader = MessageCacheReader()
reader.load_to_cache("data/feed.bin")

builder = OrderbookBuilder()
processed = builder.build_from_source(reader)
print(processed)

Expected output pattern:

250000

Example with StreamingBinaryLoader:

from fastreader import StreamingBinaryLoader, OrderbookBuilder

stream = StreamingBinaryLoader()
stream.open_stream("data/feed.bin")

builder = OrderbookBuilder()
processed = builder.build_from_source(stream)
print(processed)

Important behavior:

• When source is StreamingBinaryLoader, builder consumes from current cursor to end.

get_snapshot(token: int | None = None, depth: int | None = None) -> dict

What it does:

• Returns current top levels.
• If token is None, uses last processed token.
• If depth is None, defaults to 5.

Returned structure:

• token: int
• processed_messages: int
• mid_price: int
• bids: list[{'price': int, 'quantity': int}]
• asks: list[{'price': int, 'quantity': int}]

Example:

snap = builder.get_snapshot(token=26000, depth=3)
print(snap)

Expected output pattern:

{
  'token': 26000,
  'processed_messages': 250000,
  'mid_price': 24510,
  'bids': [
    {'price': 24500, 'quantity': 1200},
    {'price': 24495, 'quantity': 900},
    {'price': 24490, 'quantity': 600}
  ],
  'asks': [
    {'price': 24520, 'quantity': 1000},
    {'price': 24525, 'quantity': 800},
    {'price': 24530, 'quantity': 500}
  ]
}

If no levels exist for requested token:

• mid_price becomes 0
• bids and asks are empty lists

reset() -> None

What it does:

• Clears builder state and processed counters.

Example:

builder.reset()

build_from_list(messages: list[str]) -> int

Current behavior:

• Raises ValueError intentionally.

Reason:

• Builder requires internal Rust message objects, not Python strings.

Expected exception message:

build_from_list expects internal Message objects. Use build_from_source(cache_or_loader) instead.

Compatibility API (Python Wrapper Layer)

These are provided in the Python package for existing code that still uses legacy naming.

• BinaryDataLoader
• ReadMsgFromBinary (alias)
• BinaryLoader

BinaryDataLoader

Implemented in Python by combining:

• MessageCacheReader for summary and all-messages access
• StreamingBinaryLoader for next-message cursor behavior

Supported methods:

• total_messages()
• total_orders()
• total_trades()
• summary()
• reset_cursor()
• get_all_messages(limit)
• get_order_messages(limit)
• get_trade_messages(limit)
• get_next_message()

Important compatibility note:

• token argument is currently not supported in constructor and raises ValueError.

BinaryLoader

Python iterator wrapper over BinaryDataLoader.get_next_message().

Example:

from fastreader import BinaryDataLoader, BinaryLoader

loader = BinaryDataLoader("data/feed.bin")
for msg in BinaryLoader(loader):
    print(msg)

Behavior:

• Iteration stops automatically when END is reached.

End-to-End Professional Usage Example

from fastreader import MessageCacheReader, StreamingBinaryLoader, OrderbookBuilder

FILE = "data/feed.bin"
TOKEN = 26000

# 1) Cached access + summary
reader = MessageCacheReader()
total = reader.load_to_cache(FILE)
print("loaded:", total)
print("summary:", reader.get_cache_summary())

# 2) Stream-style replay
stream = StreamingBinaryLoader()
stream.open_stream(FILE)
print("first message:", stream.get_next_message())
print("batch size:", len(stream.get_messages_batch(5)))
stream.reset_cursor()

# 3) Build orderbook state and query snapshots
builder = OrderbookBuilder()
processed = builder.build_from_source(reader)
print("processed:", processed)

snapshot = builder.get_snapshot(token=TOKEN, depth=5)
print("snapshot token:", snapshot["token"])
print("snapshot mid:", snapshot["mid_price"])
print("best bid:", snapshot["bids"][0] if snapshot["bids"] else None)
print("best ask:", snapshot["asks"][0] if snapshot["asks"] else None)

Expected output pattern:

loaded: 250000
summary: {'file_path': 'data/feed.bin', 'total_messages': 250000, 'total_orders': 180000, 'total_trades': 70000, 'memory_usage_bytes': 12000000}
first message: Order Message: SeqNo1, ...
batch size: 5
processed: 250000
snapshot token: 26000
snapshot mid: 24510
best bid: {'price': 24500, 'quantity': 1200}
best ask: {'price': 24520, 'quantity': 1000}

Error Handling

Typical exceptions exposed to Python:

• RuntimeError

  • parse/read failures
  • invalid or unreadable binary file

• ValueError

  • unsupported source type in build_from_source
  • get_snapshot called with no prior token context
  • build_from_list invoked with string list
  • compatibility loader used with token filter

Performance Notes

Why this design is fast:

• Parsing is in Rust.
• In-memory message storage is native.
• Python only receives already formatted or compact dictionary output.
• Batch and cursor operations reduce Python call overhead.

Project Structure

src/
  lib.rs                     # PyO3 module exports
python/
  fastreader/
    __init__.py              # compatibility + package exports
    __init__.pyi             # typing stubs

License

MIT License