feat: receive funding payments via chain streamer

examples/chain_client/7_ChainStream.py

@@ -8,7 +8,8 @@
 
 
 async def chain_stream_event_processor(event: Dict[str, Any]):
-    print(event)
+    for funding_update in event["marketFundingUpdates"]:
+        print(funding_update)
 
 
 def stream_error_processor(exception: RpcError):
@@ -20,60 +21,24 @@ def stream_closed_processor():
 
 
 async def main() -> None:
-    network = Network.testnet()
+    network = Network.mainnet()
 
     client = AsyncClient(network)
     composer = await client.composer()
 
-    subaccount_id = "0xbdaedec95d563fb05240d6e01821008454c24c36000000000000000000000000"
+    btc_usdc_perp_market = "0x0ee7ca44147bab6ec81ac293b5fe7915488e612af59964b2d663d6008d861dee"
+    inj_usdc_perp_market = "0x790aee464fbbd02cf4476444554c71d1225f7edfe15e6dc7f874c455fd883d31"
 
-    inj_usdt_market = "0x0611780ba69656949525013d947713300f56c37b6175e02f26bffa495c3208fe"
-    inj_usdt_perp_market = "0x17ef48032cb24375ba7c2e39f384e56433bcab20cbee9a7357e4cba2eb00abe6"
-
-    bank_balances_filter = composer.chain_stream_bank_balances_filter(
-        accounts=["inj1hkhdaj2a2clmq5jq6mspsggqs32vynpk228q3r"]
-    )
-    subaccount_deposits_filter = composer.chain_stream_subaccount_deposits_filter(subaccount_ids=[subaccount_id])
-    spot_trades_filter = composer.chain_stream_trades_filter(subaccount_ids=["*"], market_ids=[inj_usdt_market])
-    derivative_trades_filter = composer.chain_stream_trades_filter(
-        subaccount_ids=["*"], market_ids=[inj_usdt_perp_market]
-    )
-    spot_orders_filter = composer.chain_stream_orders_filter(
-        subaccount_ids=[subaccount_id], market_ids=[inj_usdt_market]
-    )
-    derivative_orders_filter = composer.chain_stream_orders_filter(
-        subaccount_ids=[subaccount_id], market_ids=[inj_usdt_perp_market]
-    )
-    spot_orderbooks_filter = composer.chain_stream_orderbooks_filter(market_ids=[inj_usdt_market])
-    derivative_orderbooks_filter = composer.chain_stream_orderbooks_filter(market_ids=[inj_usdt_perp_market])
-    positions_filter = composer.chain_stream_positions_filter(
-        subaccount_ids=[subaccount_id], market_ids=[inj_usdt_perp_market]
-    )
-    oracle_price_filter = composer.chain_stream_oracle_price_filter(symbols=["INJ", "USDT"])
-    order_failures_filter = composer.chain_stream_order_failures_filter(
-        accounts=["inj1hkhdaj2a2clmq5jq6mspsggqs32vynpk228q3r"]
-    )
-    conditional_order_trigger_failures_filter = composer.chain_stream_conditional_order_trigger_failures_filter(
-        subaccount_ids=[subaccount_id], market_ids=[inj_usdt_perp_market]
+    market_funding_filter = composer.chain_stream_market_funding_filter(
+        market_ids=[btc_usdc_perp_market, inj_usdc_perp_market]
     )
 
     task = asyncio.get_event_loop().create_task(
         client.listen_chain_stream_updates(
             callback=chain_stream_event_processor,
             on_end_callback=stream_closed_processor,
             on_status_callback=stream_error_processor,
-            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,
-            order_failures_filter=order_failures_filter,
-            conditional_order_trigger_failures_filter=conditional_order_trigger_failures_filter,
+            market_funding_filter=market_funding_filter,
         )
     )
 
@@ -82,4 +47,4 @@ async def main() -> None:
 
 
 if __name__ == "__main__":
-    asyncio.get_event_loop().run_until_complete(main())
+    asyncio.run(main())

pyinjective/async_client_v2.py

@@ -963,6 +963,7 @@ async def listen_chain_stream_updates(
         conditional_order_trigger_failures_filter: Optional[
             chain_stream_v2_query.ConditionalOrderTriggerFailuresFilter
         ] = None,
+        market_funding_filter: Optional[chain_stream_v2_query.MarketFundingFilter] = None,
     ):
         return await self.chain_stream_api.stream_v2(
             callback=callback,
@@ -980,6 +981,7 @@ async def listen_chain_stream_updates(
             oracle_price_filter=oracle_price_filter,
             order_failures_filter=order_failures_filter,
             conditional_order_trigger_failures_filter=conditional_order_trigger_failures_filter,
+            market_funding_filter=market_funding_filter,
         )
 
     # region IBC Transfer module
@@ -1457,9 +1459,11 @@ async def _initialize_tokens_and_markets(self):
                     Decimal(market_info["minQuantityTickSize"])
                 ),
                 min_notional=Token.convert_value_from_extended_decimal_format(Decimal(market_info["minNotional"])),
-                settlement_price=None
-                if market_info["settlementPrice"] == ""
-                else Token.convert_value_from_extended_decimal_format(Decimal(market_info["settlementPrice"])),
+                settlement_price=(
+                    None
+                    if market_info["settlementPrice"] == ""
+                    else Token.convert_value_from_extended_decimal_format(Decimal(market_info["settlementPrice"]))
+                ),
             )
 
             binary_option_markets[market.id] = market

pyinjective/client/chain/grpc_stream/chain_grpc_chain_stream.py

@@ -79,6 +79,7 @@ async def stream_v2(
         conditional_order_trigger_failures_filter: Optional[
             chain_stream_v2_pb.ConditionalOrderTriggerFailuresFilter
         ] = None,
+        market_funding_filter: Optional[chain_stream_v2_pb.MarketFundingFilter] = None,
     ):
         request = chain_stream_v2_pb.StreamRequest(
             bank_balances_filter=bank_balances_filter,
@@ -93,6 +94,7 @@ async def stream_v2(
             oracle_price_filter=oracle_price_filter,
             order_failures_filter=order_failures_filter,
             conditional_order_trigger_failures_filter=conditional_order_trigger_failures_filter,
+            market_funding_filter=market_funding_filter,
         )
 
         await self._assistant.listen_stream(

pyinjective/composer_v2.py

@@ -1670,6 +1670,13 @@ def chain_stream_conditional_order_trigger_failures_filter(
             subaccount_ids=subaccount_ids, market_ids=market_ids
         )
 
+    def chain_stream_market_funding_filter(
+        self,
+        market_ids: Optional[List[str]] = None,
+    ) -> chain_stream_v2_query.MarketFundingFilter:
+        market_ids = market_ids or ["*"]
+        return chain_stream_v2_query.MarketFundingFilter(market_ids=market_ids)
+
     # endregion
 
     # ------------------------------------------------

tests/client/chain/stream_grpc/configurable_chain_stream_query_servicer.py

@@ -22,7 +22,9 @@ class ConfigurableChainStreamV2QueryServicer(chain_stream_v2_grpc.StreamServicer
     def __init__(self):
         super().__init__()
         self.stream_responses = deque()
+        self.stream_requests = deque()
 
     async def StreamV2(self, request: chain_stream_v2_pb.StreamRequest, context=None, metadata=None):
+        self.stream_requests.append(request)
         for event in self.stream_responses:
             yield event

tests/client/chain/stream_grpc/test_chain_grpc_chain_stream.py

@@ -9,7 +9,11 @@
 from pyinjective.core.network import DisabledCookieAssistant, Network
 from pyinjective.proto.cosmos.base.v1beta1 import coin_pb2 as coin_pb
 from pyinjective.proto.injective.exchange.v1beta1 import exchange_pb2 as exchange_pb
-from pyinjective.proto.injective.exchange.v2 import exchange_pb2 as exchange_v2_pb, order_pb2 as order_v2_pb
+from pyinjective.proto.injective.exchange.v2 import (
+    exchange_pb2 as exchange_v2_pb,
+    market_pb2 as market_v2_pb,
+    order_pb2 as order_v2_pb,
+)
 from pyinjective.proto.injective.stream.v1beta1 import query_pb2 as chain_stream_pb
 from pyinjective.proto.injective.stream.v2 import query_pb2 as chain_stream_v2_pb
 from tests.client.chain.stream_grpc.configurable_chain_stream_query_servicer import (
@@ -818,6 +822,7 @@ async def test_stream_v2(
                     "errorDescription": conditional_order_trigger_failure_update.error_description,
                 },
             ],
+            "marketFundingUpdates": [],
         }
 
         asyncio.get_event_loop().create_task(
@@ -845,6 +850,67 @@ async def test_stream_v2(
         assert first_update == expected_update
         assert end_event.is_set()
 
+    @pytest.mark.asyncio
+    async def test_stream_v2_market_funding_updates(
+        self,
+        chain_stream_servicer,
+        chain_stream_v2_servicer,
+    ):
+        market_id = "0x790aee464fbbd02cf4476444554c71d1225f7edfe15e6dc7f874c455fd883d31"
+        funding = market_v2_pb.PerpetualMarketFunding(
+            cumulative_funding="0.00125",
+            cumulative_price="123.45",
+            last_timestamp=1708099200,
+        )
+        market_funding_update = chain_stream_v2_pb.MarketFundingUpdate(
+            market_id=market_id,
+            funding=funding,
+            is_hourly_funding=True,
+            funding_rate="0.000125",
+            mark_price="25.42",
+        )
+        chain_stream_v2_servicer.stream_responses.append(
+            chain_stream_v2_pb.StreamResponse(
+                block_height=19114391,
+                block_time=1701457189786,
+                market_funding_updates=[market_funding_update],
+            )
+        )
+
+        api = self._api_instance(servicer=chain_stream_servicer, servicer_v2=chain_stream_v2_servicer)
+        events = asyncio.Queue()
+        end_event = asyncio.Event()
+        market_funding_filter = chain_stream_v2_pb.MarketFundingFilter(market_ids=[market_id])
+
+        asyncio.get_event_loop().create_task(
+            api.stream_v2(
+                callback=lambda update: events.put_nowait(update),
+                on_end_callback=lambda: end_event.set(),
+                on_status_callback=lambda exception: pytest.fail(str(exception)),
+                market_funding_filter=market_funding_filter,
+            )
+        )
+
+        first_update = await asyncio.wait_for(events.get(), timeout=1)
+
+        assert first_update["blockHeight"] == "19114391"
+        assert first_update["blockTime"] == "1701457189786"
+        assert first_update["marketFundingUpdates"] == [
+            {
+                "marketId": market_id,
+                "funding": {
+                    "cumulativeFunding": funding.cumulative_funding,
+                    "cumulativePrice": funding.cumulative_price,
+                    "lastTimestamp": str(funding.last_timestamp),
+                },
+                "isHourlyFunding": True,
+                "fundingRate": market_funding_update.funding_rate,
+                "markPrice": market_funding_update.mark_price,
+            }
+        ]
+        assert chain_stream_v2_servicer.stream_requests.popleft().market_funding_filter == market_funding_filter
+        assert end_event.is_set()
+
     def _api_instance(self, servicer, servicer_v2):
         network = Network.devnet()
         channel = grpc.aio.insecure_channel(network.grpc_endpoint)

tests/test_composer_v2.py

@@ -2771,3 +2771,34 @@ def test_chain_stream_conditional_order_trigger_failures_filter_default(self, ba
             always_print_fields_with_no_presence=True,
         )
         assert dict_message == expected_message
+
+    def test_chain_stream_market_funding_filter(self, basic_composer):
+        market_ids = [
+            "0x0ee7ca44147bab6ec81ac293b5fe7915488e612af59964b2d663d6008d861dee",
+            "0x790aee464fbbd02cf4476444554c71d1225f7edfe15e6dc7f874c455fd883d31",
+        ]
+
+        filter_result = basic_composer.chain_stream_market_funding_filter(market_ids=market_ids)
+
+        expected_message = {
+            "marketIds": market_ids,
+        }
+
+        dict_message = json_format.MessageToDict(
+            message=filter_result,
+            always_print_fields_with_no_presence=True,
+        )
+        assert dict_message == expected_message
+
+    def test_chain_stream_market_funding_filter_default(self, basic_composer):
+        filter_result = basic_composer.chain_stream_market_funding_filter()
+
+        expected_message = {
+            "marketIds": ["*"],
+        }
+
+        dict_message = json_format.MessageToDict(
+            message=filter_result,
+            always_print_fields_with_no_presence=True,
+        )
+        assert dict_message == expected_message