(feat) Added low level API component for exchange derivative streams. Added unit tests for the new functionality. Included new functions in AsyncClient using the low level API components, and marked the old functions as deprecated

Author: abel

examples/exchange_client/derivative_exchange_rpc/10_StreamHistoricalOrders.py

@@ -1,17 +1,41 @@
 import asyncio
+from typing import Any, Dict
+
+from grpc import RpcError
 
 from pyinjective.async_client import AsyncClient
 from pyinjective.core.network import Network
 
 
+async def order_event_processor(event: Dict[str, Any]):
+    print(event)
+
+
+def stream_error_processor(exception: RpcError):
+    print(f"There was an error listening to derivative orders history updates ({exception})")
+
+
+def stream_closed_processor():
+    print("The derivative orders history updates stream has been closed")
+
+
 async def main() -> None:
     # select network: local, testnet, mainnet
     network = Network.testnet()
     client = AsyncClient(network)
     market_id = "0x17ef48032cb24375ba7c2e39f384e56433bcab20cbee9a7357e4cba2eb00abe6"
-    orders = await client.stream_historical_derivative_orders(market_id=market_id)
-    async for order in orders:
-        print(order)
+
+    task = asyncio.get_event_loop().create_task(
+        client.listen_derivative_orders_history_updates(
+            callback=order_event_processor,
+            on_end_callback=stream_closed_processor,
+            on_status_callback=stream_error_processor,
+            market_id=market_id,
+        )
+    )
+
+    await asyncio.sleep(delay=60)
+    task.cancel()
 
 
 if __name__ == "__main__":

examples/exchange_client/derivative_exchange_rpc/12_StreamTrades.py

@@ -1,21 +1,46 @@
 import asyncio
+from typing import Any, Dict
+
+from grpc import RpcError
 
 from pyinjective.async_client import AsyncClient
 from pyinjective.core.network import Network
 
 
+async def market_event_processor(event: Dict[str, Any]):
+    print(event)
+
+
+def stream_error_processor(exception: RpcError):
+    print(f"There was an error listening to derivative trades updates ({exception})")
+
+
+def stream_closed_processor():
+    print("The derivative trades updates stream has been closed")
+
+
 async def main() -> None:
     # select network: local, testnet, mainnet
     network = Network.testnet()
     client = AsyncClient(network)
     market_ids = [
         "0x17ef48032cb24375ba7c2e39f384e56433bcab20cbee9a7357e4cba2eb00abe6",
-        "0xd5e4b12b19ecf176e4e14b42944731c27677819d2ed93be4104ad7025529c7ff",
+        "0x70bc8d7feab38b23d5fdfb12b9c3726e400c265edbcbf449b6c80c31d63d3a02",
     ]
-    subaccount_id = "0xc6fe5d33615a1c52c08018c47e8bc53646a0e101000000000000000000000000"
-    trades = await client.stream_derivative_trades(market_id=market_ids[0], subaccount_id=subaccount_id)
-    async for trade in trades:
-        print(trade)
+    subaccount_ids = ["0xc6fe5d33615a1c52c08018c47e8bc53646a0e101000000000000000000000000"]
+
+    task = asyncio.get_event_loop().create_task(
+        client.listen_derivative_trades_updates(
+            callback=market_event_processor,
+            on_end_callback=stream_closed_processor,
+            on_status_callback=stream_error_processor,
+            market_ids=market_ids,
+            subaccount_ids=subaccount_ids,
+        )
+    )
+
+    await asyncio.sleep(delay=60)
+    task.cancel()
 
 
 if __name__ == "__main__":

examples/exchange_client/derivative_exchange_rpc/14_SubaccountTradesList.py

@@ -16,7 +16,7 @@ async def main() -> None:
     skip = 10
     limit = 2
     pagination = PaginationOption(skip=skip, limit=limit)
-    trades = await client.fetch_subaccount_trades_list(
+    trades = await client.fetch_derivative_subaccount_trades_list(
         subaccount_id=subaccount_id,
         market_id=market_id,
         execution_type=execution_type,

examples/exchange_client/derivative_exchange_rpc/3_StreamMarket.py

@@ -1,16 +1,39 @@
 import asyncio
+from typing import Any, Dict
+
+from grpc import RpcError
 
 from pyinjective.async_client import AsyncClient
 from pyinjective.core.network import Network
 
 
+async def market_event_processor(event: Dict[str, Any]):
+    print(event)
+
+
+def stream_error_processor(exception: RpcError):
+    print(f"There was an error listening to derivative markets updates ({exception})")
+
+
+def stream_closed_processor():
+    print("The derivative markets updates stream has been closed")
+
+
 async def main() -> None:
     # select network: local, testnet, mainnet
     network = Network.testnet()
     client = AsyncClient(network)
-    markets = await client.stream_derivative_markets()
-    async for market in markets:
-        print(market)
+
+    task = asyncio.get_event_loop().create_task(
+        client.listen_derivative_market_updates(
+            callback=market_event_processor,
+            on_end_callback=stream_closed_processor,
+            on_status_callback=stream_error_processor,
+        )
+    )
+
+    await asyncio.sleep(delay=60)
+    task.cancel()
 
 
 if __name__ == "__main__":

examples/exchange_client/derivative_exchange_rpc/5_StreamOrderbooks.py

@@ -1,17 +1,41 @@
 import asyncio
+from typing import Any, Dict
+
+from grpc import RpcError
 
 from pyinjective.async_client import AsyncClient
 from pyinjective.core.network import Network
 
 
+async def orderbook_event_processor(event: Dict[str, Any]):
+    print(event)
+
+
+def stream_error_processor(exception: RpcError):
+    print(f"There was an error listening to derivative orderbook snapshots ({exception})")
+
+
+def stream_closed_processor():
+    print("The derivative orderbook snapshots stream has been closed")
+
+
 async def main() -> None:
     # select network: local, testnet, mainnet
     network = Network.testnet()
     client = AsyncClient(network)
-    market_id = "0x17ef48032cb24375ba7c2e39f384e56433bcab20cbee9a7357e4cba2eb00abe6"
-    markets = await client.stream_derivative_orderbook_snapshot(market_ids=[market_id])
-    async for market in markets:
-        print(market)
+    market_ids = ["0x17ef48032cb24375ba7c2e39f384e56433bcab20cbee9a7357e4cba2eb00abe6"]
+
+    task = asyncio.get_event_loop().create_task(
+        client.listen_derivative_orderbook_snapshots(
+            market_ids=market_ids,
+            callback=orderbook_event_processor,
+            on_end_callback=stream_closed_processor,
+            on_status_callback=stream_error_processor,
+        )
+    )
+
+    await asyncio.sleep(delay=60)
+    task.cancel()
 
 
 if __name__ == "__main__":

examples/exchange_client/derivative_exchange_rpc/6_StreamOrderbookUpdate.py

@@ -1,10 +1,21 @@
 import asyncio
 from decimal import Decimal
+from typing import Any, Dict
+
+from grpc import RpcError
 
 from pyinjective.async_client import AsyncClient
 from pyinjective.core.network import Network
 
 
+def stream_error_processor(exception: RpcError):
+    print(f"There was an error listening to derivative orderbook updates ({exception})")
+
+
+def stream_closed_processor():
+    print("The derivative orderbook updates stream has been closed")
+
+
 class PriceLevel:
     def __init__(self, price: Decimal, quantity: Decimal, timestamp: int):
         self.price = price
@@ -53,74 +64,91 @@ async def main() -> None:
 
     market_id = "0x17ef48032cb24375ba7c2e39f384e56433bcab20cbee9a7357e4cba2eb00abe6"
     orderbook = Orderbook(market_id=market_id)
+    updates_queue = asyncio.Queue()
+    tasks = []
+
+    async def queue_event(event: Dict[str, Any]):
+        await updates_queue.put(event)
 
     # start getting price levels updates
-    stream = await async_client.stream_derivative_orderbook_update(market_ids=[market_id])
-    first_update = None
-    async for update in stream:
-        first_update = update.orderbook_level_updates
-        break
+    task = asyncio.get_event_loop().create_task(
+        async_client.listen_derivative_orderbook_updates(
+            market_ids=[market_id],
+            callback=queue_event,
+            on_end_callback=stream_closed_processor,
+            on_status_callback=stream_error_processor,
+        )
+    )
+    tasks.append(task)
 
     # load the snapshot once we are already receiving updates, so we don't miss any
     await load_orderbook_snapshot(async_client=async_client, orderbook=orderbook)
 
-    # start consuming updates again to process them
-    apply_orderbook_update(orderbook, first_update)
-    async for update in stream:
-        apply_orderbook_update(orderbook, update.orderbook_level_updates)
+    task = asyncio.get_event_loop().create_task(
+        apply_orderbook_update(orderbook=orderbook, updates_queue=updates_queue)
+    )
+    tasks.append(task)
 
+    await asyncio.sleep(delay=60)
+    for task in tasks:
+        task.cancel()
 
-def apply_orderbook_update(orderbook: Orderbook, updates):
-    # discard old updates
-    if updates.sequence <= orderbook.sequence:
-        return
 
-    print(" * * * * * * * * * * * * * * * * * * *")
+async def apply_orderbook_update(orderbook: Orderbook, updates_queue: asyncio.Queue):
+    while True:
+        updates = await updates_queue.get()
+        update = updates["orderbookLevelUpdates"]
 
-    # ensure we have not missed any update
-    if updates.sequence > (orderbook.sequence + 1):
-        raise Exception(
-            "missing orderbook update events from stream, must restart: {} vs {}".format(
-                updates.sequence, (orderbook.sequence + 1)
-            )
-        )
+        # discard updates older than the snapshot
+        if int(update["sequence"]) <= orderbook.sequence:
+            return
 
-    print("updating orderbook with updates at sequence {}".format(updates.sequence))
+        print(" * * * * * * * * * * * * * * * * * * *")
 
-    # update orderbook
-    orderbook.sequence = updates.sequence
-    for direction, levels in {"buys": updates.buys, "sells": updates.sells}.items():
-        for level in levels:
-            if level.is_active:
-                # upsert level
-                orderbook.levels[direction][level.price] = PriceLevel(
-                    price=Decimal(level.price), quantity=Decimal(level.quantity), timestamp=level.timestamp
+        # ensure we have not missed any update
+        if int(update["sequence"]) > (orderbook.sequence + 1):
+            raise Exception(
+                "missing orderbook update events from stream, must restart: {} vs {}".format(
+                    update["sequence"], (orderbook.sequence + 1)
                 )
-            else:
-                if level.price in orderbook.levels[direction]:
-                    del orderbook.levels[direction][level.price]
-
-    # sort the level numerically
-    buys = sorted(orderbook.levels["buys"].values(), key=lambda x: x.price, reverse=True)
-    sells = sorted(orderbook.levels["sells"].values(), key=lambda x: x.price, reverse=True)
-
-    # lowest sell price should be higher than the highest buy price
-    if len(buys) > 0 and len(sells) > 0:
-        highest_buy = buys[0].price
-        lowest_sell = sells[-1].price
-        print("Max buy: {} - Min sell: {}".format(highest_buy, lowest_sell))
-        if highest_buy >= lowest_sell:
-            raise Exception("crossed orderbook, must restart")
-
-    # for the example, print the list of buys and sells orders.
-    print("sells")
-    for k in sells:
-        print(k)
-    print("=========")
-    print("buys")
-    for k in buys:
-        print(k)
-    print("====================================")
+            )
+
+        print("updating orderbook with updates at sequence {}".format(update["sequence"]))
+
+        # update orderbook
+        orderbook.sequence = int(update["sequence"])
+        for direction, levels in {"buys": update["buys"], "sells": update["sells"]}.items():
+            for level in levels:
+                if level["isActive"]:
+                    # upsert level
+                    orderbook.levels[direction][level["price"]] = PriceLevel(
+                        price=Decimal(level["price"]), quantity=Decimal(level["quantity"]), timestamp=level["timestamp"]
+                    )
+                else:
+                    if level["price"] in orderbook.levels[direction]:
+                        del orderbook.levels[direction][level["price"]]
+
+        # sort the level numerically
+        buys = sorted(orderbook.levels["buys"].values(), key=lambda x: x.price, reverse=True)
+        sells = sorted(orderbook.levels["sells"].values(), key=lambda x: x.price, reverse=True)
+
+        # lowest sell price should be higher than the highest buy price
+        if len(buys) > 0 and len(sells) > 0:
+            highest_buy = buys[0].price
+            lowest_sell = sells[-1].price
+            print("Max buy: {} - Min sell: {}".format(highest_buy, lowest_sell))
+            if highest_buy >= lowest_sell:
+                raise Exception("crossed orderbook, must restart")
+
+        # for the example, print the list of buys and sells orders.
+        print("sells")
+        for k in sells:
+            print(k)
+        print("=========")
+        print("buys")
+        for k in buys:
+            print(k)
+        print("====================================")
 
 
 if __name__ == "__main__":