orderpulse 0.2.34

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

fastreader — High-Performance NSE Binary Reader and Orderbook Builder

fastreader is a Rust-powered Python extension for reading NSE binary order/trade feed files and building token-wise orderbooks efficiently from Python. The library is designed for high-throughput market-data workflows where Python should control the pipeline, while Rust handles the heavy binary parsing and orderbook state management.

This README explains the architecture, each exposed class/function in lib.rs, how messages flow through the system, and how to use the library correctly from Python/Jupyter.

---

1. What this library does

The library solves three major problems:

1. Read NSE binary feed files containing order and trade messages.
2. Decode messages safely into Python-readable output or Python dictionaries.
3. Build an orderbook by pushing one decoded message at a time into OrderbookBuilder.

The latest corrected design is:

StreamingBinaryLoader.open_stream(file_path)
        ↓
StreamingBinaryLoader.get_next_msg()
        ↓
Python receives one decoded message dict
        ↓
OrderbookBuilder.orderbook_add_msg(msg)
        ↓
OrderBookManager updates bid/ask state
        ↓
OrderbookBuilder.get_snapshot(token)

This means the user controls the stream one message at a time.

Correct usage:

msg = reader.get_next_msg()
builder.orderbook_add_msg(msg)

Incorrect old usage:

builder.orderbook_add_msg(reader)

orderbook_add_msg() should receive one message, not the reader object.

---

2. Architecture overview

┌─────────────────────────────┐
│       Python / Jupyter       │
│  user controls the pipeline  │
└──────────────┬──────────────┘
               │
               │ open_stream()
               ▼
┌─────────────────────────────┐
│   StreamingBinaryLoader      │
│  - opens binary file         │
│  - validates header          │
│  - reads next raw message    │
│  - returns dict to Python    │
└──────────────┬──────────────┘
               │ get_next_msg()
               ▼
┌─────────────────────────────┐
│      Python message dict     │
│  order/trade decoded fields  │
└──────────────┬──────────────┘
               │ orderbook_add_msg(msg)
               ▼
┌─────────────────────────────┐
│       OrderbookBuilder       │
│  - validates dict            │
│  - converts dict to Message  │
│  - applies filters           │
│  - sends to manager          │
└──────────────┬──────────────┘
               ▼
┌─────────────────────────────┐
│      OrderBookManager        │
│  - one book per token        │
│  - handles N/M/X/T messages  │
│  - maintains bid/ask levels  │
└──────────────┬──────────────┘
               ▼
┌─────────────────────────────┐
│      Snapshot / Depth        │
│  get_snapshot()              │
│  get_full_depth()            │
│  get_snapshot_row()          │
└─────────────────────────────┘

---

3. Message types supported

The binary parser supports these message types:

Message Type	Meaning	Internal packet type	Orderbook action
N	New order	OrderPacket	Add order
M	Modify order	OrderPacket	Cancel old state and add new state
X	Cancel order	OrderPacket	Remove order
T	Trade	TradePacket	Reduce buy and sell order quantities

---

4. Main public Python classes

The module exposes three Python classes:

from fastreader import MessageCacheReader, StreamingBinaryLoader, OrderbookBuilder

Class	Purpose	Best use case
MessageCacheReader	Loads the full file into memory	Smaller files, repeated analysis, debugging
StreamingBinaryLoader	Reads one message at a time from disk	Large files, production pipelines, Jupyter streaming
OrderbookBuilder	Builds and queries orderbook state	Snapshot/depth generation

---

5. MessageCacheReader

MessageCacheReader is a RAM-based reader. It loads all messages into memory using load_to_cache().

Use it when:

• The file is not too large.
• You want repeated access to messages.
• You want quick debugging output.

Avoid it when:

• The binary file is very large.
• You only need sequential processing.
• You do not want high memory usage.

---

5.1 MessageCacheReader()

Creates an empty cache reader.

Example

from fastreader import MessageCacheReader

reader = MessageCacheReader()
print(reader.get_cache_summary())

Expected output

{
    'file_source': None,
    'total_messages': 0,
    'total_orders': 0,
    'total_trades': 0,
    'memory_usage_bytes': 0
}

---

5.2 load_to_cache(file_path)

Loads all order/trade messages from the binary file into memory.

Signature

count = reader.load_to_cache(file_path)

What it does internally

1. Reads the binary file.
2. Decodes supported order/trade messages.
3. Stores messages in self.messages.
4. Stores source path in self.file_path.
5. Returns total loaded message count.

Example

reader = MessageCacheReader()
count = reader.load_to_cache("/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin")
print("Loaded:", count)

Expected output

Loaded: 1250000

Actual count depends on the file.

---

5.3 get_all_messages()

Returns all cached messages as formatted strings.

Example

messages = reader.get_all_messages()
print(messages[0])

Expected output for order message

Order Message: SeqNo 1, MsgLen 38, MsgType 'N', ExchTs 1700000000000000000, LocalTs 1700000000000000100, OrderId 987654321, Token 1001, Side 'B', Price 2250000, Quantity 75, Missed 0

Expected output for trade message

Trade Message: SeqNo 2, MsgLen 45, MsgType 'T', ExchTs 1700000000000000200, LocalTs 1700000000000000300, BuyOrderId 987654321, SellOrderId 987654322, Token 1001, Price 2250100, Quantity 75, Missed 0

---

5.4 get_order_message()

Returns only order messages: N, M, and X.

Example

orders = reader.get_order_message()
print("Order messages:", len(orders))
print(orders[0])

Expected output

Order messages: 900000
Order Message: SeqNo 1, MsgLen 38, MsgType 'N', ExchTs 1700000000000000000, LocalTs 1700000000000000100, OrderId 987654321, Token 1001, Side 'B', Price 2250000, Quantity 75, Missed 0

---

5.5 get_trade_message()

Returns only trade messages: T.

Example

trades = reader.get_trade_message()
print("Trade messages:", len(trades))
print(trades[0])

Expected output

Trade messages: 350000
Trade Message: SeqNo 2, MsgLen 45, MsgType 'T', ExchTs 1700000000000000200, LocalTs 1700000000000000300, BuyOrderId 987654321, SellOrderId 987654322, Token 1001, Price 2250100, Quantity 75, Missed 0

---

5.6 get_all_trade_message()

Alias for get_trade_message().

Example

trades = reader.get_all_trade_message()
print(len(trades))

Expected output

350000

---

5.7 get_cache_summary()

Returns a Python dictionary containing cache statistics.

Example

summary = reader.get_cache_summary()
print(summary)

Expected output

{
    'file_source': '/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin',
    'total_messages': 1250000,
    'total_orders': 900000,
    'total_trades': 350000,
    'memory_usage_bytes': 80000000
}

memory_usage_bytes is approximate because it uses Rust size_of::<Message>() * total_messages.

---

6. StreamingBinaryLoader

StreamingBinaryLoader is the recommended reader for large binary files.

It does not load the whole file into RAM. It keeps the file open and reads one message at a time.

Use it when:

• The binary file is large.
• You are processing in Jupyter.
• You want to push messages one by one into the orderbook.
• You want low memory usage.

---

6.1 StreamingBinaryLoader()

Creates a streaming reader.

Example

from fastreader import StreamingBinaryLoader

reader = StreamingBinaryLoader()

No output is expected.

---

6.2 open_stream(file_path, count_messages=True)

Opens the binary file and prepares it for streaming.

Signature

count = reader.open_stream(file_path, count_messages=True)

What it does internally

1. Opens the binary file.
2. Validates that the first real message header is valid.
3. Resets the file cursor back to the beginning.
4. Optionally counts all messages.
5. Stores the opened file handle inside self.file.
6. Returns the count if count_messages=True, otherwise returns 0.

Example: fast open without counting

reader = StreamingBinaryLoader()
count = reader.open_stream(file_path, count_messages=False)
print(count)

Expected output

0

The file is still opened. Only counting is skipped.

Example: open and count messages

reader = StreamingBinaryLoader()
count = reader.open_stream(file_path, count_messages=True)
print("Total messages:", count)

Expected output

Total messages: 1250000

For a very large file, count_messages=True can take time because it scans the full file once.

---

6.3 get_next_message()

Reads the next message and returns it as a formatted string.

This is mainly for debugging and human inspection.

Example

reader = StreamingBinaryLoader()
reader.open_stream(file_path, count_messages=False)

print(reader.get_next_message())
print(reader.get_next_message())
print(reader.get_next_message())

Expected output

Order Message: SeqNo 1, MsgLen 38, MsgType 'N', ExchTs 1700000000000000000, LocalTs 1700000000000000100, OrderId 987654321, Token 1001, Side 'B', Price 2250000, Quantity 75, Missed 0
Order Message: SeqNo 2, MsgLen 38, MsgType 'N', ExchTs 1700000000000000500, LocalTs 1700000000000000600, OrderId 987654322, Token 1001, Side 'S', Price 2250200, Quantity 75, Missed 0
Trade Message: SeqNo 3, MsgLen 45, MsgType 'T', ExchTs 1700000000000000900, LocalTs 1700000000000001000, BuyOrderId 987654321, SellOrderId 987654322, Token 1001, Price 2250100, Quantity 25, Missed 0

When the file ends:

END

---

6.4 get_next_msg()

Reads the next message and returns it as a Python dictionary.

This is the correct method to use with OrderbookBuilder.orderbook_add_msg().

Example

reader = StreamingBinaryLoader()
reader.open_stream(file_path, count_messages=False)

msg = reader.get_next_msg()
print(msg)

Expected output for order message

{
    'message_kind': 'order',
    'seq_no': 1,
    'msg_len': 38,
    'stream_id': 2,
    'msg_type': 'N',
    'exch_ts': 1700000000000000000,
    'local_ts': 1700000000000000100,
    'order_id': 987654321,
    'token': 1001,
    'order_type': 'B',
    'price': 2250000,
    'quantity': 75,
    'flags': False
}

Expected output for trade message

{
    'message_kind': 'trade',
    'seq_no': 3,
    'msg_len': 45,
    'stream_id': 2,
    'msg_type': 'T',
    'exch_ts': 1700000000000000900,
    'local_ts': 1700000000000001000,
    'buy_order_id': 987654321,
    'sell_order_id': 987654322,
    'token': 1001,
    'trade_price': 2250100,
    'trade_quantity': 25,
    'flags': False
}

When file ends:

None

---

6.5 reset_cursor()

Moves the file cursor back to the start of the binary file.

Example

reader = StreamingBinaryLoader()
reader.open_stream(file_path, count_messages=False)

first = reader.get_next_msg()
second = reader.get_next_msg()

reader.reset_cursor()
again_first = reader.get_next_msg()

print(first == again_first)

Expected output

True

Use this when you want to reprocess the same file from the beginning.

---

7. OrderbookBuilder

OrderbookBuilder is the Python-facing wrapper around OrderBookManager.

It receives decoded messages, converts them into internal Rust packets, and updates the orderbook.

---

7.1 OrderbookBuilder()

Creates a fresh orderbook builder.

Example

from fastreader import OrderbookBuilder

builder = OrderbookBuilder()

No output is expected.

---

7.2 apply_filter(logic_criteria=None)

Restricts which message types should be processed.

Example: process only new and modify messages

builder.apply_filter(["N", "M"])

Example: process all messages

builder.apply_filter(None)

How it works

If filter is None, all supported message types are allowed.

If filter is a list, only the first byte/character of each string is used.

builder.apply_filter(["N", "M", "X", "T"])

means:

Allow New, Modify, Cancel, and Trade messages.

---

7.3 orderbook_add_msg(msg)

Pushes one already-decoded message dictionary into the orderbook.

This is the core corrected API.

Correct usage

msg = reader.get_next_msg()
builder.orderbook_add_msg(msg)

Signature

accepted = builder.orderbook_add_msg(msg)

Return value

Return	Meaning
True	Message was valid and applied to orderbook
False	Message was valid but skipped because of filter/business rule
Exception	Message dictionary is invalid or unsupported

Full one-by-one streaming example

from fastreader import StreamingBinaryLoader, OrderbookBuilder

file_path = "/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin"

reader = StreamingBinaryLoader()
reader.open_stream(file_path, count_messages=False)

builder = OrderbookBuilder()

read_count = 0
accepted_count = 0

while True:
    msg = reader.get_next_msg()

    if msg is None:
        break

    accepted = builder.orderbook_add_msg(msg)
    read_count += 1

    if accepted:
        accepted_count += 1

    if read_count % 100000 == 0:
        print("Read:", read_count, "Accepted:", accepted_count)

print("Finished")
print("Read:", read_count)
print("Accepted:", accepted_count)

Expected output

Read: 100000 Accepted: 98210
Read: 200000 Accepted: 196455
Read: 300000 Accepted: 294770
Finished
Read: 1250000
Accepted: 1225000

Actual counts depend on the file and filter.

---

7.4 Why this API is correct

The correct architecture is to separate reading from orderbook building.

Reader responsibility:

Read next binary message and decode it.

Orderbook responsibility:

Take one decoded message and update market state.

Python/user responsibility:

Control the loop, decide how many messages to process, inspect messages, stop early, debug, filter, or branch logic.

This gives you full control in Jupyter:

msg = reader.get_next_msg()
print(msg)
builder.orderbook_add_msg(msg)

You can also stop early:

for i in range(10000):
    msg = reader.get_next_msg()
    if msg is None:
        break
    builder.orderbook_add_msg(msg)

---

7.5 build_from_list(source)

Builds the orderbook from either:

1. A MessageCacheReader, or
2. A Python list[dict] of decoded messages.

Example with MessageCacheReader

cache = MessageCacheReader()
cache.load_to_cache(file_path)

builder = OrderbookBuilder()
count = builder.build_from_list(cache)

print("Applied:", count)

Expected output

Applied: 1225000

Example with Python list of messages

reader = StreamingBinaryLoader()
reader.open_stream(file_path, count_messages=False)

messages = []
for _ in range(5):
    msg = reader.get_next_msg()
    if msg is not None:
        messages.append(msg)

builder = OrderbookBuilder()
count = builder.build_from_list(messages)
print("Applied:", count)

Expected output

Applied: 5

If some messages are skipped by filters, count can be less than list length.

---

7.6 build_from_source(source, limit=None)

Builds the orderbook directly from a supported source.

Supported sources:

• MessageCacheReader
• StreamingBinaryLoader

Example with stream and limit

reader = StreamingBinaryLoader()
reader.open_stream(file_path, count_messages=False)

builder = OrderbookBuilder()
count = builder.build_from_source(reader, limit=100000)

print("Applied:", count)

Expected output

Applied: 98210

Important design note

build_from_source() is faster for bulk processing because the loop runs inside Rust.

But if you want user-controlled one-by-one processing, use:

msg = reader.get_next_msg()
builder.orderbook_add_msg(msg)

---

7.7 get_snapshot(token, levels=None)

Returns top bid/ask levels for one token.

Example

snapshot = builder.get_snapshot(token=1001, levels=5)
print(snapshot)

Expected output

{
    'token': 1001,
    'found': True,
    'mid_price': 2250100,
    'best_bid': (2250000, 150),
    'best_ask': (2250200, 75),
    'spread': 200,
    'bids': [(2250000, 150), (2249900, 300), (2249800, 75)],
    'asks': [(2250200, 75), (2250300, 225), (2250400, 150)]
}

If token is not found:

{
    'token': 999999,
    'found': False,
    'mid_price': 0,
    'best_bid': None,
    'best_ask': None,
    'spread': None,
    'bids': [],
    'asks': []
}

---

7.8 get_orderbook_snapshot(token, levels=None)

Alias for get_snapshot().

Example

snapshot = builder.get_orderbook_snapshot(token=1001, levels=5)
print(snapshot)

Expected output

Same as get_snapshot().

---

7.9 get_full_depth(token)

Returns all non-zero bid/ask levels for a token.

Example

depth = builder.get_full_depth(token=1001)
print(depth)

Expected output

{
    'token': 1001,
    'found': True,
    'best_bid': (2250000, 150),
    'best_ask': (2250200, 75),
    'spread': 200,
    'bids': [
        (2250000, 150),
        (2249900, 300),
        (2249800, 75)
    ],
    'asks': [
        (2250200, 75),
        (2250300, 225),
        (2250400, 150)
    ]
}

If token is not found:

{
    'token': 999999,
    'found': False,
    'best_bid': None,
    'best_ask': None,
    'spread': None,
    'bids': [],
    'asks': []
}

---

7.10 snapshot_header()

Returns CSV header for fixed 5-level snapshot row.

Example

print(builder.snapshot_header())

Expected output

local_ts,exch_ts,mid_price,bid_price_0,bid_qty_0,ask_price_0,ask_qty_0,bid_price_1,bid_qty_1,ask_price_1,ask_qty_1,bid_price_2,bid_qty_2,ask_price_2,ask_qty_2,bid_price_3,bid_qty_3,ask_price_3,ask_qty_3,bid_price_4,bid_qty_4,ask_price_4,ask_qty_4

---

7.11 get_snapshot_row(token, levels=None)

Returns a CSV-style row for a token snapshot.

Example

print(builder.snapshot_header())
print(builder.get_snapshot_row(token=1001, levels=5))

Expected output

local_ts,exch_ts,mid_price,bid_price_0,bid_qty_0,ask_price_0,ask_qty_0,bid_price_1,bid_qty_1,ask_price_1,bid_qty_1,ask_price_2,ask_qty_2,bid_price_3,bid_qty_3,ask_price_3,ask_qty_3,bid_price_4,bid_qty_4,ask_price_4,ask_qty_4
0,0,2250100,2250000,150,2250200,75,2249900,300,2250300,225,2249800,75,2250400,150,0,0,0,0,0,0,0,0

Note: local_ts and exch_ts are currently returned as 0 in this snapshot-row helper because the orderbook itself stores price/quantity state, not the latest timestamp per snapshot.

---

8. Complete recommended Jupyter workflow

from fastreader import StreamingBinaryLoader, OrderbookBuilder

file_path = "/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin"

token = 1001

reader = StreamingBinaryLoader()
reader.open_stream(file_path, count_messages=False)

builder = OrderbookBuilder()

read_count = 0
accepted_count = 0

while True:
    msg = reader.get_next_msg()

    if msg is None:
        break

    accepted = builder.orderbook_add_msg(msg)

    read_count += 1
    if accepted:
        accepted_count += 1

    if read_count % 100000 == 0:
        snapshot = builder.get_snapshot(token=token, levels=5)
        print("Read:", read_count, "Accepted:", accepted_count, "Snapshot:", snapshot)

final_snapshot = builder.get_snapshot(token=token, levels=5)
print("Final read count:", read_count)
print("Final accepted count:", accepted_count)
print("Final snapshot:", final_snapshot)

Expected output

Read: 100000 Accepted: 98210 Snapshot: {'token': 1001, 'found': True, 'mid_price': 2250100, ...}
Read: 200000 Accepted: 196455 Snapshot: {'token': 1001, 'found': True, 'mid_price': 2250150, ...}
Final read count: 1250000
Final accepted count: 1225000
Final snapshot: {'token': 1001, 'found': True, 'mid_price': 2250200, 'best_bid': (2250100, 150), 'best_ask': (2250300, 75), 'spread': 200, 'bids': [...], 'asks': [...]}

---

9. Performance notes

Recommended for large files

reader.open_stream(file_path, count_messages=False)

Why?

Because count_messages=True scans the entire file once before processing starts.

For large NSE files, this means:

count_messages=True  -> scan full file once, then process file again
count_messages=False -> start processing immediately

One-by-one Python loop vs Rust bulk loop

One-by-one Python-controlled loop:

msg = reader.get_next_msg()
builder.orderbook_add_msg(msg)

Advantages:

• Easy debugging.
• User controls every message.
• Can stop early.
• Can inspect/modify/filter messages in Python.

Rust bulk loop:

builder.build_from_source(reader, limit=1000000)

Advantages:

• Faster for production bulk processing.
• Less Python/Rust call overhead.

Use one-by-one loop for debugging and research. Use build_from_source() for high-speed batch processing.

---

10. Internal decoding logic

The binary reader works like this:

read_next_message_from_file(file)
    ↓
read first byte
    ↓
skip spaces if first byte is b' '
    ↓
read PeekStructure
    ↓
check msg_type
    ↓
if T: read TradePacket size
if N/M/X: read OrderPacket size
    ↓
parse little-endian fields
    ↓
return Message::Trade or Message::Order

The parser validates:

• Empty file
• Truncated header
• Truncated payload
• Invalid message type

Errors are converted into Python RuntimeError or TypeError depending on where they happen.

---

11. Orderbook internal logic

Internally, OrderBookManager maintains one OrderBook per token.

Each OrderBook stores:

orders: order_id -> side, price, quantity
bid_levels: price-indexed aggregate bid quantity
ask_levels: price-indexed aggregate ask quantity

When a message arrives:

New order: N

Add order_id to orders map
Add quantity to bid_levels or ask_levels

Modify order: M

Cancel old order_id state
Add new order_id state with new price/quantity

Cancel order: X

Remove order_id from orders map
Subtract its quantity from bid/ask level

Trade: T

Reduce quantity from buy_order_id
Reduce quantity from sell_order_id
Remove order if remaining quantity becomes zero

---

12. Common mistakes and fixes

Mistake 1: Passing reader directly into orderbook_add_msg()

Wrong:

builder.orderbook_add_msg(reader)

Correct:

msg = reader.get_next_msg()
builder.orderbook_add_msg(msg)

---

Mistake 2: No progress print in a huge file loop

If the binary file has millions of messages, Jupyter may look stuck.

Use progress logging:

i = 0
while True:
    msg = reader.get_next_msg()
    if msg is None:
        break

    builder.orderbook_add_msg(msg)
    i += 1

    if i % 100000 == 0:
        print("Processed:", i)

---

Mistake 3: Counting messages before processing huge file

Avoid this for large files:

reader.open_stream(file_path, count_messages=True)

Prefer:

reader.open_stream(file_path, count_messages=False)

---

Mistake 4: Querying wrong token

If snapshot says found=False, maybe that token did not appear yet or does not exist in the file.

Debug by printing messages:

for _ in range(10):
    msg = reader.get_next_msg()
    print(msg)

Check the token field.

---

13. Build and install

From the Rust project root:

maturin develop --release

Then test import:

from fastreader import StreamingBinaryLoader, OrderbookBuilder
print("fastreader imported successfully")

Expected output:

fastreader imported successfully

---

14. Minimal smoke test

from fastreader import StreamingBinaryLoader, OrderbookBuilder

file_path = "/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin"

reader = StreamingBinaryLoader()
reader.open_stream(file_path, count_messages=False)

builder = OrderbookBuilder()

for i in range(10000):
    msg = reader.get_next_msg()
    if msg is None:
        print("File ended early")
        break
    builder.orderbook_add_msg(msg)

print(builder.get_snapshot(token=1001, levels=5))

Expected output shape:

{
    'token': 1001,
    'found': True,
    'mid_price': 2250100,
    'best_bid': (2250000, 150),
    'best_ask': (2250200, 75),
    'spread': 200,
    'bids': [(2250000, 150), ...],
    'asks': [(2250200, 75), ...]
}

If the token is not present in the first 10,000 messages:

{
    'token': 1001,
    'found': False,
    'mid_price': 0,
    'best_bid': None,
    'best_ask': None,
    'spread': None,
    'bids': [],
    'asks': []
}

---

15. Summary

fastreader is built around a clean separation of responsibilities:

StreamingBinaryLoader = read and decode binary messages
OrderbookBuilder     = accept decoded messages and update orderbook
Python/Jupyter       = control the processing loop

The most important usage pattern is:

while True:
    msg = reader.get_next_msg()
    if msg is None:
        break
    builder.orderbook_add_msg(msg)

This design is explicit, debuggable, and production-ready because the user controls exactly how each message enters the orderbook pipeline.