Order book methods

async_rithmic/client.py

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

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,
+    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,
@@ -496,10 +496,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,66 @@ async def search_symbols(self, search_text, **kwargs):
             **kwargs
         )
 
+    async def request_market_depth(
+        self,
+        symbol: str,
+        exchange: str,
+        depth_price: float
+    ):
+        responses = await self._send_and_collect(
+            template_id=115,
+            expected_response=dict(template_id=116),
+            symbol=symbol,
+            exchange=exchange,
+            depth_price=depth_price
+        )
+        # TODO: verify that there's indeed only one response
+        return responses[0]
+
+    async def subscribe_to_market_depth(
+        self,
+        symbol: str,
+        exchange: str,
+        depth_price: float
+    ):
+        """
+        Subscribes to market depth updates (L2 data)
+        """
+
+        sub = (symbol, exchange, depth_price)
+        self._subscriptions["market_depth"].add(sub)
+
+        async with self.lock:
+            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)
+        """
+
+        sub = (symbol, exchange, depth_price)
+        self._subscriptions["market_depth"].add(sub)
+
+        async with self.lock:
+            await self._send_request(
+                template_id=100,
+                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 +169,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 +189,9 @@ async def _process_response(self, response):
 
             await self.client.on_tick.call_async(data)
 
+        elif response.template_id == 160:
+            # Market depth data stream
+            await self.client.on_market_depth(response)
+
         else:
             self.logger.warning(f"Unhandled inbound message with template_id={response.template_id}")

async_rithmic/protocol_buffers/source/forced_logout.proto

@@ -3,8 +3,9 @@ syntax = "proto3";
 package rti;
 
 message ForcedLogout
-	{       
+	{
 	// PB_OFFSET = 100000 , is the offset added for each MNM field id
 
-	int32  template_id     = 154467;    // PB_OFFSET + MNM_TEMPLATE_ID
+	 int32  template_id     = 154467;    // PB_OFFSET + MNM_TEMPLATE_ID
+
 	}