Order book methods (rundef#8)

* Start order book endpoint implementation (wip)

* Handle market depth subscriptions (wip)

* Whoops

* Implement market depth endpoints + update docs

* whoops

* Update README

* Add doc about how to handle the different values of update_type

* Update changelog

* Bump version: 1.5.3 → 1.5.4

Author: rundef

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 1.5.3
+current_version = 1.5.4
 commit = True
 tag = True
 

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## [1.5.4] - 2025-07-28
+### Added
+- Order book/market depth methods
+
 ## [1.5.3] - 2025-07-09
 ### Added
 - Logger name suffix kwarg

README.md

@@ -18,13 +18,14 @@ Designed with reliability and extensibility in mind, `async_rithmic` is a strong
 ## ✨ Key Features
 
 - ✅ **Python 3.10+ Compatibility**: Fully tested and supported.
+- ⚡ **Async-first design**: Better scalability & responsiveness.
 - 🛠️ **Robust architecture**: Built-in reconnection & fault-tolerance.
     - [**Automatic reconnection**](https://async-rithmic.readthedocs.io/en/latest/connection.html#custom-reconnection-settings): Resilient to network interruptions with customizable backoff and retry logic.
     - [**Automatic retries**](https://async-rithmic.readthedocs.io/en/latest/connection.html#custom-retry-settings): Configure how many times a slow request will be retried and for how long, making your client more resilient to network delays and backend slowness.
 - 👥 **Multi-account support**
 - 📊 **Historical + Live Time Bars**: Ideal for time-based strategies.
 - 🎯 **Live Tick Data & Best Bid/Ask Streaming**: Fine-grained market data for real-time decision-making.
-- ⚡ **Async-first design**: Better scalability & responsiveness.
+- 🪟 **Full Order Book (L2) Streaming**: Stream real-time depth of market (all bids/asks, multiple price levels) for advanced order flow analysis.
 
 ## 📦 Installation
 

async_rithmic/__init__.py

@@ -4,4 +4,4 @@
 from .exceptions import *
 from .objects import RetrySettings, ReconnectionSettings
 
-__version__ = '1.5.3'
+__version__ = '1.5.4'

async_rithmic/client.py

@@ -39,6 +39,8 @@ def __init__(
         # Real-time market updates events
         self.on_tick = Event()
         self.on_time_bar = Event()
+        self.on_order_book = Event()
+        self.on_market_depth = Event()
 
         # Order updates events
         self.on_rithmic_order_notification = Event()

async_rithmic/enums.py

@@ -2,16 +2,20 @@
 
 from . import protocol_buffers as pb
 
-class DataType(int, enum.Enum):
+
+class DataType(enum.IntEnum):
     LAST_TRADE = 1
     BBO = 2
+    ORDER_BOOK = 4
+
 
 OrderType = pb.request_new_order_pb2.RequestNewOrder.PriceType
 OrderDuration = pb.request_new_order_pb2.RequestNewOrder.Duration
 TransactionType = pb.request_new_order_pb2.RequestNewOrder.TransactionType
 
 LastTradePresenceBits = pb.last_trade_pb2.LastTrade.PresenceBits
 BestBidOfferPresenceBits = pb.best_bid_offer_pb2.BestBidOffer.PresenceBits
+OrderBookPresenceBits = pb.order_book_pb2.OrderBook.PresenceBits
 
 ExchangeOrderNotificationType = pb.exchange_order_notification_pb2.ExchangeOrderNotification.NotifyType
 

async_rithmic/plants/base.py

@@ -45,16 +45,16 @@
     113: pb.request_front_month_contract_pb2.RequestFrontMonthContract,
     114: pb.response_front_month_contract_pb2.ResponseFrontMonthContract,
 
-    #115: pb.request_depth_by_order_snapshot_pb2.RequestDepthByOrderSnapshot,
-    #116: pb.response_depth_by_order_snapshot_pb2.ResponseDepthByOrderSnapshot,
-    #117: pb.request_depth_by_order_updates_pb2.RequestDepthByOrderUpdates,
-    #118: pb.response_depth_by_order_updates_pb2.ResponseDepthByOrderUpdates,
+    115: pb.request_depth_by_order_snapshot_pb2.RequestDepthByOrderSnapshot,
+    116: pb.response_depth_by_order_snapshot_pb2.ResponseDepthByOrderSnapshot,
+    117: pb.request_depth_by_order_updates_pb2.RequestDepthByOrderUpdates,
+    118: pb.response_depth_by_order_updates_pb2.ResponseDepthByOrderUpdates,
 
     150: pb.last_trade_pb2.LastTrade,
     151: pb.best_bid_offer_pb2.BestBidOffer,
-    #156: pb.order_book_pb2.OrderBook,
-    #160: pb.depth_by_order.DepthByOrder,
-    #161: pb.depth_by_order_end_event.DepthByOrderEndEvent,
+    156: pb.order_book_pb2.OrderBook,
+    160: pb.depth_by_order_pb2.DepthByOrder,
+    161: pb.depth_by_order_end_event_pb2.DepthByOrderEndEvent,
 
     # Order Plant Infrastructure
     300: pb.request_login_info_pb2.RequestLoginInfo,
@@ -500,10 +500,11 @@ async def _process_response(self, response):
         Handles async responses
         """
 
-        if response.template_id in [13, 19, 401]:
+        if response.template_id in [13, 19, 161, 401]:
             # Ignore
             # - logout responses
             # - heartbeat responses
+            # - market depth end event
             # - pnl subscription responses
             return True
 

async_rithmic/plants/ticker.py

@@ -13,6 +13,9 @@ async def _login(self):
         for symbol, exchange, update_bits in self._subscriptions["market_data"]:
             await self.subscribe_to_market_data(symbol, exchange, update_bits)
 
+        for symbol, exchange, depth_price in self._subscriptions["market_depth"]:
+            await self.subscribe_to_market_depth(symbol, exchange, depth_price)
+
     async def list_exchanges(self):
         return await self._send_and_collect(
             template_id=342,
@@ -98,6 +101,67 @@ async def search_symbols(self, search_text, **kwargs):
             **kwargs
         )
 
+    async def request_market_depth(
+        self,
+        symbol: str,
+        exchange: str,
+        depth_price: float
+    ):
+        """
+        Request order book data for a given price
+        """
+        responses = await self._send_and_collect(
+            template_id=115,
+            expected_response=dict(template_id=116),
+            symbol=symbol,
+            exchange=exchange,
+            depth_price=depth_price,
+            account_id=None,
+        )
+        return responses[0] if responses else None
+
+    async def subscribe_to_market_depth(
+        self,
+        symbol: str,
+        exchange: str,
+        depth_price: float
+    ):
+        """
+        Subscribes to market depth updates (L2 data) for a given price
+        """
+
+        sub = (symbol, exchange, depth_price)
+        self._subscriptions["market_depth"].add(sub)
+
+        await self._send_request(
+            template_id=117,
+            symbol=symbol,
+            exchange=exchange,
+            depth_price=depth_price,
+            request=pb.request_depth_by_order_updates_pb2.RequestDepthByOrderUpdates.Request.SUBSCRIBE,
+        )
+
+    async def unsubscribe_from_market_depth(
+        self,
+        symbol: str,
+        exchange: str,
+        depth_price: float
+    ):
+        """
+        Unsubscribes from market depth updates (L2 data) for a given price
+        """
+
+        sub = (symbol, exchange, depth_price)
+        self._subscriptions["market_depth"].discard(sub)
+
+        await self._send_request(
+            template_id=117,
+            symbol=symbol,
+            exchange=exchange,
+            depth_price=depth_price,
+            request=pb.request_depth_by_order_updates_pb2.RequestDepthByOrderUpdates.Request.UNSUBSCRIBE,
+        )
+
     async def _process_response(self, response):
         if await super()._process_response(response):
             return True
@@ -106,6 +170,10 @@ async def _process_response(self, response):
             # Market data update response
             pass
 
+        elif response.template_id == 118:
+            # Market depth data update response
+            pass
+
         elif response.template_id == 150:
             # Market data stream: Last Trade
             data = self._response_to_dict(response)
@@ -122,5 +190,13 @@ async def _process_response(self, response):
 
             await self.client.on_tick.call_async(data)
 
+        elif response.template_id == 156:
+            # Market data stream: Order Book
+            await self.client.on_order_book.call_async(response)
+
+        elif response.template_id == 160:
+            # Market depth data stream
+            await self.client.on_market_depth.call_async(response)
+
         else:
             self.logger.warning(f"Unhandled inbound message with template_id={response.template_id}")