orderpulse 0.2.51

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

OrderPulse / fastreader

High-performance NSE binary feed parsing and order book construction for Python, powered by Rust and PyO3.

OrderPulse is designed for users who need fast, reliable parsing of NSE order/trade feed files and a clean Python API for analytics, prototyping, and production pipelines. The Rust core handles binary decoding efficiently, while Python users work with familiar dictionaries and builder classes.

What This Library Does

• Parses NSE binary order and trade feed files in Rust.
• Exposes Python bindings through PyO3.
• Supports two read modes:
  • MessageCacheReader: load entire feed into memory for repeated analysis.
  • StreamingBinaryLoader: read one message at a time for large files.
• Builds per-token order books with OrderbookBuilder.
• Enriches token metadata with SymbolMaster.
• Builds standard feed file paths with FeedPathBuilder.

Architecture Overview

Flow of data:

binary feed file -> Rust parser -> Python message dictionaries -> optional symbol master enrichment -> order book builder -> snapshot/depth output

flowchart LR
		A[Binary Feed File .bin] --> B[Rust Parser]
		B --> C[Python Dict Messages]
		C --> D{SymbolMaster attached?}
		D -- Yes --> E[Enriched Messages\n token_symbol strike option expiry lot size name]
		D -- No --> F[Raw Messages]
		E --> G[OrderbookBuilder]
		F --> G
		G --> H[Snapshots]
		G --> I[Full Depth]
		G --> J[CSV Snapshot Rows]

Installation

1) Install from PyPI

pip install OrderPulse

2) Verify import

from fastreader import (
		MessageCacheReader,
		StreamingBinaryLoader,
		OrderbookBuilder,
		SymbolMaster,
		FeedPathBuilder,
)

print("fastreader import successful")

Expected output:

fastreader import successful

Required Input File Formats

1) Binary feed file

• Must be an NSE-style binary feed containing order/trade packets.
• First valid message type must be one of T, N, M, X.
• Message types:
  • N: new order
  • M: modify order
  • X: cancel/delete order
  • T: trade

2) Symbol master CSV

SymbolMaster.load() expects CSV columns:

• Required:
  • FinInstrmId
  • TckrSymb
  • XpryDt
  • StrkPric
  • OptnTp
  • StockNm
• Optional lot column:
  • NewBrdLotQty (preferred), or
  • MinLot

If required columns are missing, loading fails with a runtime error.

Typical End-to-End Workflow

1. Build or supply binary feed file path.
2. Open feed with cache mode or stream mode.
3. Read messages.
4. Optionally load symbol master and enrich messages.
5. Build order book.
6. Fetch top levels, full depth, active tokens, or CSV snapshot rows.

Public API Reference

All examples below are notebook-cell friendly Python snippets.

Important:

• Example outputs are realistic placeholders.
• Actual outputs always depend on your input feed file and token master CSV.

---

Class: MessageCacheReader

Loads the entire feed into memory for fast repeated queries.

MessageCacheReader()

What it does:

• Creates an empty cache reader.

When to use:

• Start of cache-mode workflow.

Parameters:

• None.

Returns:

• MessageCacheReader instance.

Possible errors:

• None.

from fastreader import MessageCacheReader

reader = MessageCacheReader()
print(type(reader).__name__)

Expected output:

MessageCacheReader

load_to_cache(file_path)

What it does:

• Parses binary feed and caches all parsed messages in memory.

When to use:

• You need repeated filtering, summaries, and multiple order book passes.

Parameters:

• file_path (str): full path to .bin feed file.

Returns:

• int: total parsed messages.

Possible errors:

• missing file path / cannot open file
• invalid binary header
• truncated message payload

from fastreader import MessageCacheReader

reader = MessageCacheReader()
total = reader.load_to_cache("/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin")
print(total)

Expected output:

1245832

get_all_messages()

What it does:

• Returns all cached messages as formatted text lines.

When to use:

• Fast debugging and feed inspection.

Parameters:

• None.

Returns:

• list[str].

Possible errors:

• None (empty list if cache is empty).

lines = reader.get_all_messages()
print(len(lines))
print(lines[0])

Expected output:

1245832
Order Message: SeqNo 1, MsgLen 38, MsgType 'N', ExchTs 1700000000, LocalTs 1700000002, OrderId 8123456701, Token 40434, Side 'B', Price 2170050, Quantity 75, Missed 0

get_order_message()

What it does:

• Returns only order messages (N/M/X) as dictionaries.

When to use:

• Analyze order-side activity only.

Parameters:

• None.

Returns:

• list[dict].

Possible errors:

• None.

orders = reader.get_order_message()
print(len(orders))
print(orders[0]["message_kind"], orders[0]["msg_type"])

Expected output:

1103210
order N

get_trade_message()

What it does:

• Returns only trade messages (T) as dictionaries.

When to use:

• Analyze executions independently from order flow.

Parameters:

• None.

Returns:

• list[dict].

Possible errors:

• None.

trades = reader.get_trade_message()
print(len(trades))
print(trades[0]["message_kind"], trades[0]["msg_type"])

Expected output:

142622
trade T

get_all_trade_message()

What it does:

• Alias of get_trade_message().

When to use:

• Backward-compatible naming preference.

Parameters:

• None.

Returns:

• list[dict].

Possible errors:

• None.

same = reader.get_all_trade_message()
print(same == reader.get_trade_message())

Expected output:

True

get_cache_summary()

What it does:

• Returns source path, counts, and approximate cache memory usage.

When to use:

• Validate load result before building analytics.

Parameters:

• None.

Returns:

• dict with keys:
  • file_source
  • total_messages
  • total_orders
  • total_trades
  • memory_usage_bytes

Possible errors:

• None.

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': 1245832, 'total_orders': 1103210, 'total_trades': 142622, 'memory_usage_bytes': 79733248}

---

Class: StreamingBinaryLoader

Sequential streaming reader for large feed files.

StreamingBinaryLoader()

What it does:

• Creates an unopened stream reader.

When to use:

• Start of stream-mode workflow.

Parameters:

• None.

Returns:

• StreamingBinaryLoader instance.

Possible errors:

• None.

from fastreader import StreamingBinaryLoader

stream = StreamingBinaryLoader()
print(type(stream).__name__)

Expected output:

StreamingBinaryLoader

open_stream(file_path, count_messages=True)

What it does:

• Opens feed for sequential read.
• Optionally counts total messages by scanning file.

When to use:

• Large-file processing or one-pass consumption.

Parameters:

• file_path (str): feed path.
• count_messages (bool, default True):
  • True: returns total count (slower startup).
  • False: returns 0 (fast startup).

Returns:

• int.

Possible errors:

• missing file path / cannot open file
• invalid binary header

count = stream.open_stream(
		"/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin",
		count_messages=False,
)
print(count)

Expected output:

0

get_next_msg()

What it does:

• Reads one decoded message dictionary at current cursor.

When to use:

• Message-by-message processing in loops.

Parameters:

• None.

Returns:

• dict for next message.
• None at EOF.

Possible errors:

• truncated message payload
• read failure on underlying file

msg = stream.get_next_msg()
print(msg["message_kind"], msg["msg_type"], msg["token"])

Expected output:

order N 40434

is_end_of_msg()

What it does:

• Checks if next read would hit EOF without consuming a message.

When to use:

• Safe while-loop condition.

Parameters:

• None.

Returns:

• bool.

Possible errors:

• underlying file read/seek errors.

print(stream.is_end_of_msg())

Expected output:

False

reset_cursor()

What it does:

• Seeks stream cursor back to start.

When to use:

• Replay stream without reopening file.

Parameters:

• None.

Returns:

• None.

Possible errors:

• seek failure on underlying file.

first_before = stream.get_next_msg()
stream.reset_cursor()
first_after = stream.get_next_msg()
print(first_before["seq_no"] == first_after["seq_no"])

Expected output:

True

attach_symbol_master(master)

What it does:

• Attaches loaded SymbolMaster to auto-enrich each streamed message.

When to use:

• You want message dictionaries to include contract metadata fields.

Parameters:

• master (SymbolMaster).

Returns:

• None.

Possible errors:

• None.

from fastreader import SymbolMaster

sm = SymbolMaster()
sm.load_for_date("NSE_FO", day=27, month=5, year=2026)
stream.attach_symbol_master(sm)

msg = stream.get_next_msg()
print(msg.get("token_symbol"), msg.get("strike_price"), msg.get("expiry"))

Expected output:

FINNIFTY 21700 27-May-2026

detach_symbol_master()

What it does:

• Removes auto-enrichment source from stream.

When to use:

• You want raw messages only.

Parameters:

• None.

Returns:

• None.

Possible errors:

• None.

stream.detach_symbol_master()
msg = stream.get_next_msg()
print(msg.get("token_symbol"))

Expected output:

None

---

Class: SymbolMaster

Loads contract metadata CSV and provides token lookup/enrichment.

SymbolMaster()

What it does:

• Creates empty symbol master.

When to use:

• Before loading CSV master data.

Parameters:

• None.

Returns:

• SymbolMaster instance.

Possible errors:

• None.

from fastreader import SymbolMaster

sm = SymbolMaster()
print(sm)

Expected output:

SymbolMaster(contracts=0)

load(csv_path)

What it does:

• Loads contracts from explicit CSV path.

When to use:

• You already know CSV file location.

Parameters:

• csv_path (str).

Returns:

• int: loaded contract count.

Possible errors:

• missing file path / cannot open csv
• missing required csv columns

loaded = sm.load("/nas/50.30/CONTRACT/27_05_2026/NSE_FO_contract_27052026.csv")
print(loaded)

Expected output:

95632

load_for_date(segment, day, month, year, base_path=None)

What it does:

• Builds standard contract CSV path and loads it.

When to use:

• You follow standard directory convention.

Parameters:

• segment (str): NSE_FO, FO, NSE_CM, or CM.
• day (int)
• month (int)
• year (int)
• base_path (str | None, default /nas/50.30)

Returns:

• int: loaded contracts.

Possible errors:

• unknown segment name
• csv file not found
• missing required csv columns

loaded = sm.load_for_date("NSE_FO", day=27, month=5, year=2026)
print(loaded)

Expected output:

95632

lookup(token)

What it does:

• Looks up one token and returns metadata dict.

When to use:

• Ad-hoc inspection or validation of token mapping.

Parameters:

• token (int).

Returns:

• dict:
  • always: token, found
  • when found: symbol, name, option_type, strike, expiry, lot_size
  • when missing: above fields are None

Possible errors:

• None.

info = sm.lookup(40434)
print(info)

Expected output:

{'token': 40434, 'found': True, 'symbol': 'FINNIFTY', 'name': 'FINNIFTY27MAY21700CE', 'option_type': 'CE', 'strike': 21700, 'expiry': '27-May-2026', 'lot_size': 40}

enrich(msg)

What it does:

• Enriches one message dict in-place.

When to use:

• You control enrichment manually (without stream attachment).

Parameters:

• msg (dict from get_next_msg()).

Returns:

• bool: True if token found and enriched, else False.

Possible errors:

• input is not dict
• dict missing token

msg = {
		"message_kind": "order",
		"msg_type": "N",
		"token": 40434,
		"order_id": 1,
		"order_type": "B",
		"price": 2170050,
		"quantity": 25,
}
ok = sm.enrich(msg)
print(ok)
print(msg["token_symbol"], msg["strike_price"], msg["option_type"], msg["expiry"], msg["lot_size"], msg["name"])

Expected output:

True
FINNIFTY 21700 CE 27-May-2026 40 FINNIFTY27MAY21700CE

len(symbol_master)

What it does:

• Returns loaded contract count.

When to use:

• Sanity check after loading.

Parameters:

• None.

Returns:

• int.

Possible errors:

• None.

print(len(sm))

Expected output:

95632

repr(symbol_master)

What it does:

• Displays friendly summary string.

When to use:

• Logging / notebook display.

Parameters:

• None.

Returns:

• str.

Possible errors:

• None.

print(sm)

Expected output:

SymbolMaster(contracts=95632)

---

Class: OrderbookBuilder

Consumes order/trade messages and maintains per-token book state.

OrderbookBuilder()

What it does:

• Creates empty order book manager.

When to use:

• Beginning of book-construction workflow.

Parameters:

• None.

Returns:

• OrderbookBuilder instance.

Possible errors:

• None.

from fastreader import OrderbookBuilder

builder = OrderbookBuilder()
print(type(builder).__name__)

Expected output:

OrderbookBuilder

apply_filter(logic_criteria=None)

What it does:

• Sets message-type allowlist used during processing.

When to use:

• Restrict build to selected message types.

Parameters:

• logic_criteria (list[str] | None), examples:
  • None -> accept all types
  • ["N", "M", "X"] -> order-only book updates
  • ["T"] -> trade-only processing

Returns:

• None.

Possible errors:

• None.

builder.apply_filter(["N", "M", "X"])
print("order-only filter applied")

Expected output:

order-only filter applied

orderbook_add_msg(msg)

What it does:

• Adds one decoded message dict into builder.

When to use:

• Custom loop with stream reader and manual processing.

Parameters:

• msg (dict).

Returns:

• bool:
  • True message accepted and applied.
  • False message skipped by filter/business checks.

Possible errors:

• input is not dict
• missing required keys
• unsupported message type

ok = builder.orderbook_add_msg(
		{
				"msg_type": "N",
				"order_id": 11,
				"token": 7001,
				"order_type": "B",
				"price": 500,
				"quantity": 3,
		}
)
print(ok)

Expected output:

True

build_from_list(source)

What it does:

• Builds book from either:
  • MessageCacheReader, or
  • list[dict] decoded messages.

When to use:

• Cache-mode builds or synthetic test messages.

Parameters:

• source (MessageCacheReader | list[dict]).

Returns:

• int: number of processed messages.

Possible errors:

• wrong source type
• malformed message dictionaries
• unsupported message type

Build from cached reader:

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

Expected output:

1245832

Build from manual list:

messages = [
		{"msg_type": "N", "order_id": 1, "token": 1001, "order_type": "B", "price": 100, "quantity": 10},
		{"msg_type": "N", "order_id": 2, "token": 1001, "order_type": "S", "price": 102, "quantity": 20},
		{"msg_type": "T", "buy_order_id": 1, "sell_order_id": 2, "token": 1001, "trade_price": 101, "trade_quantity": 5},
]
builder2 = OrderbookBuilder()
print(builder2.build_from_list(messages))
print(builder2.get_snapshot(token=1001, levels=2))

Expected output:

3
{'token': 1001, 'found': True, 'mid_price': 101, 'best_bid': (100, 5), 'best_ask': (102, 15), 'spread': 2, 'bids': [(100, 5)], 'asks': [(102, 15)]}

build_from_source(source, limit=None)

What it does:

• Unified builder for MessageCacheReader or StreamingBinaryLoader.

When to use:

• Generic pipeline code that may use cache or stream source.

Parameters:

• source (MessageCacheReader | StreamingBinaryLoader)
• limit (int | None): max processed messages for stream mode.

Returns:

• int: processed count.

Possible errors:

• wrong source type
• stream read errors

Build from stream:

stream.reset_cursor()
builder3 = OrderbookBuilder()
processed = builder3.build_from_source(stream, limit=200000)
print(processed)

Expected output:

200000

get_active_tokens()

What it does:

• Returns token IDs currently present in book state.

When to use:

• Discover available instruments after build.

Parameters:

• None.

Returns:

• list[int].

Possible errors:

• None.

tokens = builder.get_active_tokens()
print(tokens[:5], len(tokens))

Expected output:

[1001, 1002, 40434, 50111, 90210] 418

get_full_depth(token)

What it does:

• Returns all bid/ask levels for one token.

When to use:

• Need complete depth, not just top N levels.

Parameters:

• token (int).

Returns:

• dict with:
  • token, found, best_bid, best_ask, spread, bids, asks

Possible errors:

• None.

depth = builder.get_full_depth(40434)
print(depth)

Expected output:

{'token': 40434, 'found': True, 'best_bid': (2170050, 150), 'best_ask': (2170100, 200), 'spread': 50, 'bids': [(2170050, 150), (2170000, 300)], 'asks': [(2170100, 200), (2170150, 100)]}

get_snapshot(token, levels=5)

What it does:

• Returns top N levels and derived metrics for a token.

When to use:

• Fast periodic snapshot display or export.

Parameters:

• token (int)
• levels (int, default 5)

Returns:

• dict with:
  • token, found, mid_price, best_bid, best_ask, spread, bids, asks

Possible errors:

• None.

snap = builder.get_snapshot(token=40434, levels=3)
print(snap)

Expected output:

{'token': 40434, 'found': True, 'mid_price': 2170075, 'best_bid': (2170050, 150), 'best_ask': (2170100, 200), 'spread': 50, 'bids': [(2170050, 150), (2170000, 300), (2169950, 75)], 'asks': [(2170100, 200), (2170150, 100), (2170200, 50)]}

snapshot_header()

What it does:

• Returns CSV header for snapshot row format.

When to use:

• Writing snapshot rows to CSV files.

Parameters:

• None.

Returns:

• str.

Possible errors:

• None.

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

get_snapshot_row(token, levels=5)

What it does:

• Returns one CSV-formatted row with top levels.

When to use:

• Compact snapshot logging.

Parameters:

• token (int)
• levels (int, default 5)

Returns:

• str CSV row.

Possible errors:

• None.

row = builder.get_snapshot_row(token=40434, levels=5)
print(row)

Expected output:

0,0,2170075,2170050,150,2170100,200,2170000,300,2170150,100,2169950,75,2170200,50,2169900,40,2170250,25,2169850,10,2170300,5

---

Class: FeedPathBuilder

Builds standard NSE feed file paths safely and consistently.

FeedPathBuilder()

What it does:

• Creates feed path builder instance.

When to use:

• Before generating feed paths.

Parameters:

• None.

Returns:

• FeedPathBuilder instance.

Possible errors:

• None.

from fastreader import FeedPathBuilder

fp = FeedPathBuilder()
print(fp)

Expected output:

FeedPathBuilder()

build(segment, stream_id, day, month, year, base_path=None)

What it does:

• Builds feed path string from date/segment/stream components.

When to use:

• Standardized path generation before opening files.

Parameters:

• segment (str): NSE_CM, CM, NSE_FO, FO.
• stream_id (int, must be > 0).
• day (int, 1..31).
• month (int, 1..12).
• year (int, 2000..2100).
• base_path (str | None, default /nas/50.30).

Returns:

• str full path.

Possible errors:

• unknown segment name
• invalid stream id/day/month/year

path = fp.build(
		segment="NSE_CM",
		stream_id=2,
		day=29,
		month=12,
		year=2025,
)
print(path)

Expected output:

/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin

build_and_verify(segment, stream_id, day, month, year, base_path=None)

What it does:

• Builds path and checks file exists on disk.

When to use:

• Fail fast before starting parse/stream pipeline.

Parameters:

• Same as build().

Returns:

• str verified path.

Possible errors:

• same validation errors as build()
• file not found

verified = fp.build_and_verify(
		segment="NSE_CM",
		stream_id=2,
		day=29,
		month=12,
		year=2025,
)
print(verified)

Expected output:

/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin

---

Common Usage Recipes

A) Load binary file into cache and inspect

from fastreader import MessageCacheReader

reader = MessageCacheReader()
reader.load_to_cache("/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin")

print(reader.get_cache_summary())
print("orders:", len(reader.get_order_message()))
print("trades:", len(reader.get_trade_message()))

B) Stream one message at a time with EOF check

from fastreader import StreamingBinaryLoader

stream = StreamingBinaryLoader()
stream.open_stream("/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin", count_messages=False)

processed = 0
while not stream.is_end_of_msg() and processed < 10:
		msg = stream.get_next_msg()
		if msg is None:
				break
		print(msg["msg_type"], msg["token"])
		processed += 1

C) Stream + enrich with SymbolMaster

from fastreader import StreamingBinaryLoader, SymbolMaster

sm = SymbolMaster()
sm.load_for_date("NSE_FO", day=27, month=5, year=2026)

stream = StreamingBinaryLoader()
stream.open_stream("/nas/50.30/NSE_FO/Feed_FO_StreamID_1_27_05_2026.bin", count_messages=False)
stream.attach_symbol_master(sm)

msg = stream.get_next_msg()
print(msg["token"], msg.get("token_symbol"), msg.get("strike_price"), msg.get("option_type"), msg.get("expiry"), msg.get("lot_size"), msg.get("name"))

D) Build order book from cached messages

from fastreader import MessageCacheReader, OrderbookBuilder

reader = MessageCacheReader()
reader.load_to_cache("/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin")

builder = OrderbookBuilder()
builder.apply_filter(["N", "M", "X", "T"])
builder.build_from_list(reader)

tokens = builder.get_active_tokens()
token = tokens[0]
print(builder.get_snapshot(token, levels=5))

E) Build order book directly from stream source

from fastreader import StreamingBinaryLoader, OrderbookBuilder

stream = StreamingBinaryLoader()
stream.open_stream("/nas/50.30/NSE_CM/Feed_CM_StreamID_2_29_12_2025.bin", count_messages=False)

builder = OrderbookBuilder()
processed = builder.build_from_source(stream, limit=500000)

print("processed:", processed)
print("active tokens:", len(builder.get_active_tokens()))

---

Troubleshooting

1) Missing file path / file not found

Symptom:

• RuntimeError with file open failure text.

Fix:

• Verify absolute path.
• Use FeedPathBuilder.build_and_verify(...) before loading.

2) Invalid binary header

Symptom:

• RuntimeError: invalid first message type ...

Fix:

• Confirm file is correct NSE binary feed.
• Ensure file is not accidentally CSV/text or compressed.

3) Truncated message payload

Symptom:

• RuntimeError: truncated message payload

Fix:

• Re-copy feed file.
• Verify upstream download/file transfer completed.

4) Missing symbol master CSV columns

Symptom:

• RuntimeError: column '...' not found in ...csv

Fix:

• Ensure required columns exist exactly:
  • FinInstrmId, TckrSymb, XpryDt, StrkPric, OptnTp, StockNm

5) Unknown segment name

Symptom:

• RuntimeError: unknown segment '...'

Fix:

• Use only NSE_FO, FO, NSE_CM, or CM.

6) Unsupported message type

Symptom:

• TypeError: unsupported msg_type: ...

Fix:

• Use only N, M, X, T when building from dictionaries.

7) Empty order book snapshot

Symptom:

• get_snapshot(...) returns found: False and empty levels.

Fix:

• Confirm messages were processed.
• Check token exists in get_active_tokens().
• Check filters are not excluding required message types.

---

Developer Notes

• Python package name on PyPI: OrderPulse
• Import path in code: fastreader
• Rust extension module: fastreader.fastreader
• Supported Python versions: >=3.9

Final Notes for Users

• This library is production-focused and optimized for performance-critical feed parsing.
• Start with cache mode for quick analysis and debugging.
• Move to stream mode for large files and memory-efficient processing.
• Always validate feed paths and CSV schema early for smoother runs.
• Keep in mind that all counts, snapshots, and enrichment values depend on your specific input feed file and token master CSV.