(feat) Finished implementation to support chain streams. Added more details to the example module that shows how to use them

Author: abel

…t/48_WithdrawValidatorCommission_Rewards …8_WithdrawValidatorCommission_Rewards.pyexamples/chain_client/48_WithdrawValidatorCommission_Rewards renamed to examples/chain_client/48_WithdrawValidatorCommission_Rewards.py examples/chain_client/48_WithdrawValidatorCommission_Rewards renamed to examples/chain_client/48_WithdrawValidatorCommission_Rewards.py

@@ -0,0 +1,61 @@
+import asyncio
+
+from google.protobuf import json_format
+
+from pyinjective.composer import Composer
+from pyinjective.async_client import AsyncClient
+from pyinjective.core.network import Network
+
+
+async def main() -> None:
+    # select network: local, testnet, mainnet
+    # network = Network.devnet()
+    network = Network.custom(
+        lcd_endpoint="https://staging.lcd.injective.network:443",
+        tm_websocket_endpoint="wss://staging.tm.injective.network:443/websocket",
+        grpc_endpoint="staging.chain.grpc.injective.network:443",
+        grpc_exchange_endpoint="staging.exchange.grpc.injective.network:443",
+        grpc_explorer_endpoint="staging.explorer.grpc.injective.network:443",
+        chain_stream_endpoint="staging.stream.injective.network:443",
+        chain_id="injective-1",
+        env='mainnet',
+        use_secure_connection=True,
+    )
+
+    client = AsyncClient(network)
+    composer = Composer(network=network.string())
+
+    inj_usdt_market = "0xfbc729e93b05b4c48916c1433c9f9c2ddb24605a73483303ea0f87a8886b52af"
+
+    bank_balances_filter = composer.chain_stream_bank_balances_filter()
+    subaccount_deposits_filter = composer.chain_stream_subaccount_deposits_filter()
+    spot_trades_filter = composer.chain_stream_trades_filter()
+    derivative_trades_filter = composer.chain_stream_trades_filter()
+    spot_orders_filter = composer.chain_stream_orders_filter()
+    derivative_orders_filter = composer.chain_stream_orders_filter()
+    spot_orderbooks_filter = composer.chain_stream_orderbooks_filter()
+    derivative_orderbooks_filter = composer.chain_stream_orderbooks_filter()
+    positions_filter = composer.chain_stream_positions_filter()
+    oracle_price_filter = composer.chain_stream_oracle_price_filter()
+    stream = await client.chain_stream(
+        bank_balances_filter=bank_balances_filter,
+        subaccount_deposits_filter=subaccount_deposits_filter,
+        spot_trades_filter=spot_trades_filter,
+        derivative_trades_filter=derivative_trades_filter,
+        spot_orders_filter=spot_orders_filter,
+        derivative_orders_filter=derivative_orders_filter,
+        spot_orderbooks_filter=spot_orderbooks_filter,
+        derivative_orderbooks_filter=derivative_orderbooks_filter,
+        positions_filter=positions_filter,
+        oracle_price_filter=oracle_price_filter
+    )
+    async for event in stream:
+        print(json_format.MessageToJson(
+            message=event,
+            including_default_value_fields=True,
+            preserving_proto_field_name=True)
+        )
+
+
+if __name__ == '__main__':
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/49_ChainStream.py

@@ -939,12 +939,29 @@ async def stream_account_portfolio(self, account_address: str, **kwargs):
     async def chain_stream(
         self,
         bank_balances_filter: Optional[chain_stream_query.BankBalancesFilter] = None,
-        subaccount_deposits_filter: Optional[chain_stream_query.SubaccountDepositsFilter] = None
+        subaccount_deposits_filter: Optional[chain_stream_query.SubaccountDepositsFilter] = None,
+        spot_trades_filter: Optional[chain_stream_query.TradesFilter] = None,
+        derivative_trades_filter: Optional[chain_stream_query.TradesFilter] = None,
+        spot_orders_filter: Optional[chain_stream_query.OrdersFilter] = None,
+        derivative_orders_filter: Optional[chain_stream_query.OrdersFilter] = None,
+        spot_orderbooks_filter: Optional[chain_stream_query.OrderbookFilter] = None,
+        derivative_orderbooks_filter: Optional[chain_stream_query.OrderbookFilter] = None,
+        positions_filter: Optional[chain_stream_query.PositionsFilter] = None,
+        oracle_price_filter: Optional[chain_stream_query.OraclePriceFilter] = None,
     ):
 
         request = chain_stream_query.StreamRequest(
             bank_balances_filter=bank_balances_filter,
-            subaccount_deposits_filter=subaccount_deposits_filter)
+            subaccount_deposits_filter=subaccount_deposits_filter,
+            spot_trades_filter=spot_trades_filter,
+            derivative_trades_filter=derivative_trades_filter,
+            spot_orders_filter=spot_orders_filter,
+            derivative_orders_filter=derivative_orders_filter,
+            spot_orderbooks_filter=spot_orderbooks_filter,
+            derivative_orderbooks_filter=derivative_orderbooks_filter,
+            positions_filter=positions_filter,
+            oracle_price_filter=oracle_price_filter,
+        )
         metadata = await self.network.chain_metadata(
             metadata_query_provider=self._chain_cookie_metadata_requestor
         )

pyinjective/async_client.py

@@ -902,14 +902,57 @@ def MsgVote(
             proposal_id=proposal_id, voter=voter, option=option
         )
 
-    def chain_stream_bank_balances_filter(self, accounts: List[str]) -> chain_stream_query.BankBalancesFilter:
+    def chain_stream_bank_balances_filter(self, accounts: Optional[List[str]] = None) -> chain_stream_query.BankBalancesFilter:
+        accounts = accounts or ["*"]
         return chain_stream_query.BankBalancesFilter(accounts=accounts)
 
     def chain_stream_subaccount_deposits_filter(
-            self, subaccount_ids: List[str]
+            self, subaccount_ids: Optional[List[str]] = None,
     ) -> chain_stream_query.SubaccountDepositsFilter:
+        subaccount_ids = ["*"]
         return chain_stream_query.SubaccountDepositsFilter(subaccount_ids=subaccount_ids)
 
+    def chain_stream_trades_filter(
+        self,
+        subaccount_ids: Optional[List[str]] = None,
+        market_ids: Optional[List[str]] = None,
+    ) -> chain_stream_query.TradesFilter:
+        subaccount_ids = subaccount_ids or ["*"]
+        market_ids = market_ids or ["*"]
+        return chain_stream_query.TradesFilter(subaccount_ids=subaccount_ids, market_ids=market_ids)
+
+    def chain_stream_orders_filter(
+        self,
+        subaccount_ids: Optional[List[str]] = None,
+        market_ids: Optional[List[str]] = None,
+    ) -> chain_stream_query.OrdersFilter:
+        subaccount_ids = subaccount_ids or ["*"]
+        market_ids = market_ids or ["*"]
+        return chain_stream_query.OrdersFilter(subaccount_ids=subaccount_ids, market_ids=market_ids)
+
+    def chain_stream_orderbooks_filter(
+        self,
+        market_ids: Optional[List[str]] = None,
+    ) -> chain_stream_query.OrderbookFilter:
+        market_ids = market_ids or ["*"]
+        return chain_stream_query.OrderbookFilter(market_ids=market_ids)
+
+    def chain_stream_positions_filter(
+        self,
+        subaccount_ids: Optional[List[str]] = None,
+        market_ids: Optional[List[str]] = None,
+    ) -> chain_stream_query.PositionsFilter:
+        subaccount_ids = subaccount_ids or ["*"]
+        market_ids = market_ids or ["*"]
+        return chain_stream_query.PositionsFilter(subaccount_ids=subaccount_ids, market_ids=market_ids)
+
+    def chain_stream_oracle_price_filter(
+        self,
+        symbols: Optional[List[str]] = None,
+    ) -> chain_stream_query.PositionsFilter:
+        symbols = symbols or ["*"]
+        return chain_stream_query.OraclePriceFilter(symbol=symbols)
+
     # data field format: [request-msg-header][raw-byte-msg-response]
     # you need to figure out this magic prefix number to trim request-msg-header off the data
     # this method handles only exchange responses