Add PNL updates subscriptions

Author: rundef

async_rithmic/client.py

@@ -49,6 +49,10 @@ def __init__(
         self.on_historical_tick = Event()
         self.on_historical_time_bar = Event()
 
+        # PNL events
+        self.on_instrument_pnl_update = Event()
+        self.on_account_pnl_update = Event()
+
         if "://" not in url:
             url = f"wss://{url}"
 

async_rithmic/plants/pnl.py

@@ -18,9 +18,16 @@ def _ib_id(self):
 
     async def _login(self):
         await super()._login()
-        await self._subscribe_to_position_updates()
 
-    async def _subscribe_to_position_updates(self):
+        if len(self._subscriptions["pnl"]) > 0:
+            await self.subscribe_to_pnl_updates()
+
+    async def subscribe_to_pnl_updates(self):
+        """
+        Subscribe to instruments/accounts PNL updates
+        """
+        self._subscriptions["pnl"].add(1)
+
         for account in self._accounts:
             await self._send_request(
                 template_id=400,
@@ -30,16 +37,52 @@ async def _subscribe_to_position_updates(self):
                 request=pb.request_pnl_position_updates_pb2.RequestPnLPositionUpdates.Request.SUBSCRIBE
             )
 
+    async def unsubscribe_from_pnl_updates(self):
+        """
+        Unsubscribe from instruments/accounts PNL updates
+        """
+        self._subscriptions["pnl"].discard(1)
+
+        for account in self._accounts:
+            await self._send_request(
+                template_id=400,
+                fcm_id=self._fcm_id,
+                ib_id=self._ib_id,
+                account_id=account.account_id,
+                request=pb.request_pnl_position_updates_pb2.RequestPnLPositionUpdates.Request.UNSUBSCRIBE
+            )
+
     async def list_positions(self, **kwargs):
+        """
+        Instrument PNL snapshots
+        """
         return await self._send_and_collect(
             template_id=402,
             expected_response=dict(template_id=450, is_snapshot=True),
             **kwargs
         )
 
     async def list_account_summary(self, **kwargs):
+        """
+        Account PNL snapshots
+        """
         return await self._send_and_collect(
             template_id=402,
             expected_response=dict(template_id=451, is_snapshot=True),
             **kwargs
         )
+
+    async def _process_response(self, response):
+        if await super()._process_response(response):
+            return True
+
+        if response.template_id == 450:
+            # Instrument PNL position update
+            await self.client.on_instrument_pnl_update.call_async(response)
+
+        elif response.template_id == 451:
+            # Account PNL position update
+            await self.client.on_account_pnl_update.call_async(response)
+
+        else:
+            self.logger.warning(f"Unhandled inbound message with template_id={response.template_id}")

docs/pnl.rst

@@ -13,3 +13,45 @@ Use `list_account_summary()` to retrieve the PNL snapshot of an account.
     accounts = await client.list_account_summary(account_id="1234")
 
 The result is a list which contains a single object. See the `account_pnl_position_update.proto <https://github.com/rundef/async_rithmic/blob/main/async_rithmic/protocol_buffers/source/account_pnl_position_update.proto>`_ definition for field details.
+
+
+Streaming PNL updates
+---------------------
+
+Here's an example that gets pnl updates in a streaming fashion:
+
+.. code-block:: python
+
+    import asyncio
+    from async_rithmic import RithmicClient
+
+    async def on_instrument_pnl_update(data):
+        print("instrument_pnl_update", data)
+
+    async def on_account_pnl_update(data):
+        print("account_pnl_update", data)
+
+    async def main():
+        client = RithmicClient(
+            user="",
+            password="",
+            system_name="Rithmic Test",
+            app_name="my_test_app",
+            app_version="1.0",
+            url="rituz00100.rithmic.com:443"
+        )
+        await client.connect()
+
+        client.on_instrument_pnl_update += on_instrument_pnl_update
+        client.on_account_pnl_update += on_account_pnl_update
+
+        await client.subscribe_to_pnl_updates()
+
+        # Wait 10 seconds, unsubscribe and disconnect
+        await asyncio.sleep(10)
+        await client.unsubscribe_from_pnl_updates()
+        await client.disconnect()
+
+    asyncio.run(main())
+
+See the `account_pnl_position_update.proto <https://github.com/rundef/async_rithmic/blob/main/async_rithmic/protocol_buffers/source/account_pnl_position_update.proto>`_ and `instrument_pnl_position_update.proto <https://github.com/rundef/async_rithmic/blob/main/async_rithmic/protocol_buffers/source/instrument_pnl_position_update.proto>`_ definitions for field details.