(feat) Added low level API components for the exchange portfolio module, with unit tests. Included new functions in AsyncClient to use the low level API components and marked the old functions as deprecated. Updated the examples to use the new AsyncClient functions.

Author: abel

examples/exchange_client/portfolio_rpc/1_AccountPortfolio.py

@@ -9,7 +9,7 @@ async def main() -> None:
     network = Network.testnet()
     client = AsyncClient(network)
     account_address = "inj1clw20s2uxeyxtam6f7m84vgae92s9eh7vygagt"
-    portfolio = await client.get_account_portfolio(account_address=account_address)
+    portfolio = await client.fetch_account_portfolio(account_address=account_address)
     print(portfolio)
 
 

examples/exchange_client/portfolio_rpc/2_StreamAccountPortfolio.py

@@ -1,17 +1,40 @@
 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 account_portfolio_event_processor(event: Dict[str, Any]):
+    print(event)
+
+
+def stream_error_processor(exception: RpcError):
+    print(f"There was an error listening to account portfolio updates ({exception})")
+
+
+def stream_closed_processor():
+    print("The account portfolio updates stream has been closed")
+
+
 async def main() -> None:
     network = Network.testnet()
     client = AsyncClient(network)
     account_address = "inj1clw20s2uxeyxtam6f7m84vgae92s9eh7vygagt"
-    updates = await client.stream_account_portfolio(account_address=account_address)
-    async for update in updates:
-        print("Account portfolio Update:\n")
-        print(update)
+
+    task = asyncio.get_event_loop().create_task(
+        client.listen_account_portfolio_updates(
+            account_address=account_address,
+            callback=account_portfolio_event_processor,
+            on_end_callback=stream_closed_processor,
+            on_status_callback=stream_error_processor,
+        )
+    )
+
+    await asyncio.sleep(delay=60)
+    task.cancel()
 
 
 if __name__ == "__main__":

pyinjective/async_client.py

@@ -18,12 +18,14 @@
 from pyinjective.client.indexer.grpc.indexer_grpc_insurance_api import IndexerGrpcInsuranceApi
 from pyinjective.client.indexer.grpc.indexer_grpc_meta_api import IndexerGrpcMetaApi
 from pyinjective.client.indexer.grpc.indexer_grpc_oracle_api import IndexerGrpcOracleApi
+from pyinjective.client.indexer.grpc.indexer_grpc_portfolio_api import IndexerGrpcPortfolioApi
 from pyinjective.client.indexer.grpc.indexer_grpc_spot_api import IndexerGrpcSpotApi
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_account_stream import IndexerGrpcAccountStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_auction_stream import IndexerGrpcAuctionStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_derivative_stream import IndexerGrpcDerivativeStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_meta_stream import IndexerGrpcMetaStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_oracle_stream import IndexerGrpcOracleStream
+from pyinjective.client.indexer.grpc_stream.indexer_grpc_portfolio_stream import IndexerGrpcPortfolioStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_spot_stream import IndexerGrpcSpotStream
 from pyinjective.client.model.pagination import PaginationOption
 from pyinjective.composer import Composer
@@ -213,6 +215,12 @@ def __init__(
                 metadata_query_provider=self._exchange_cookie_metadata_requestor
             ),
         )
+        self.exchange_portfolio_api = IndexerGrpcPortfolioApi(
+            channel=self.exchange_channel,
+            metadata_provider=lambda: self.network.exchange_metadata(
+                metadata_query_provider=self._exchange_cookie_metadata_requestor
+            ),
+        )
         self.exchange_spot_api = IndexerGrpcSpotApi(
             channel=self.exchange_channel,
             metadata_provider=lambda: self.network.exchange_metadata(
@@ -250,6 +258,12 @@ def __init__(
                 metadata_query_provider=self._exchange_cookie_metadata_requestor
             ),
         )
+        self.exchange_portfolio_stream_api = IndexerGrpcPortfolioStream(
+            channel=self.exchange_channel,
+            metadata_provider=lambda: self.network.exchange_metadata(
+                metadata_query_provider=self._exchange_cookie_metadata_requestor
+            ),
+        )
         self.exchange_spot_stream_api = IndexerGrpcSpotStream(
             channel=self.exchange_channel,
             metadata_provider=lambda: self.network.exchange_metadata(
@@ -2137,10 +2151,23 @@ async def fetch_binary_options_market(self, market_id: str) -> Dict[str, Any]:
     # PortfolioRPC
 
     async def get_account_portfolio(self, account_address: str):
+        """
+        This method is deprecated and will be removed soon. Please use `fetch_account_portfolio` instead
+        """
+        warn("This method is deprecated. Use fetch_account_portfolio instead", DeprecationWarning, stacklevel=2)
         req = portfolio_rpc_pb.AccountPortfolioRequest(account_address=account_address)
         return await self.stubPortfolio.AccountPortfolio(req)
 
+    async def fetch_account_portfolio(self, account_address: str) -> Dict[str, Any]:
+        return await self.exchange_portfolio_api.fetch_account_portfolio(account_address=account_address)
+
     async def stream_account_portfolio(self, account_address: str, **kwargs):
+        """
+        This method is deprecated and will be removed soon. Please use `listen_account_portfolio_updates` instead
+        """
+        warn(
+            "This method is deprecated. Use listen_account_portfolio_updates instead", DeprecationWarning, stacklevel=2
+        )
         req = portfolio_rpc_pb.StreamAccountPortfolioRequest(
             account_address=account_address, subaccount_id=kwargs.get("subaccount_id"), type=kwargs.get("type")
         )
@@ -2149,6 +2176,24 @@ async def stream_account_portfolio(self, account_address: str, **kwargs):
         )
         return self.stubPortfolio.StreamAccountPortfolio(request=req, metadata=metadata)
 
+    async def listen_account_portfolio_updates(
+        self,
+        account_address: str,
+        callback: Callable,
+        on_end_callback: Optional[Callable] = None,
+        on_status_callback: Optional[Callable] = None,
+        subaccount_id: Optional[str] = None,
+        update_type: Optional[str] = None,
+    ):
+        await self.exchange_portfolio_stream_api.stream_account_portfolio(
+            account_address=account_address,
+            callback=callback,
+            on_end_callback=on_end_callback,
+            on_status_callback=on_status_callback,
+            subaccount_id=subaccount_id,
+            update_type=update_type,
+        )
+
     async def chain_stream(
         self,
         bank_balances_filter: Optional[chain_stream_query.BankBalancesFilter] = None,

pyinjective/client/indexer/grpc/indexer_grpc_portfolio_api.py

@@ -0,0 +1,24 @@
+from typing import Any, Callable, Dict
+
+from grpc.aio import Channel
+
+from pyinjective.proto.exchange import (
+    injective_portfolio_rpc_pb2 as exchange_portfolio_pb,
+    injective_portfolio_rpc_pb2_grpc as exchange_portfolio_grpc,
+)
+from pyinjective.utils.grpc_api_request_assistant import GrpcApiRequestAssistant
+
+
+class IndexerGrpcPortfolioApi:
+    def __init__(self, channel: Channel, metadata_provider: Callable):
+        self._stub = self._stub = exchange_portfolio_grpc.InjectivePortfolioRPCStub(channel)
+        self._assistant = GrpcApiRequestAssistant(metadata_provider=metadata_provider)
+
+    async def fetch_account_portfolio(self, account_address: str) -> Dict[str, Any]:
+        request = exchange_portfolio_pb.AccountPortfolioRequest(account_address=account_address)
+        response = await self._execute_call(call=self._stub.AccountPortfolio, request=request)
+
+        return response
+
+    async def _execute_call(self, call: Callable, request) -> Dict[str, Any]:
+        return await self._assistant.execute_call(call=call, request=request)

pyinjective/client/indexer/grpc_stream/indexer_grpc_portfolio_stream.py

@@ -0,0 +1,38 @@
+from typing import Callable, Optional
+
+from grpc.aio import Channel
+
+from pyinjective.proto.exchange import (
+    injective_portfolio_rpc_pb2 as exchange_portfolio_pb,
+    injective_portfolio_rpc_pb2_grpc as exchange_portfolio_grpc,
+)
+from pyinjective.utils.grpc_api_stream_assistant import GrpcApiStreamAssistant
+
+
+class IndexerGrpcPortfolioStream:
+    def __init__(self, channel: Channel, metadata_provider: Callable):
+        self._stub = self._stub = exchange_portfolio_grpc.InjectivePortfolioRPCStub(channel)
+        self._assistant = GrpcApiStreamAssistant(metadata_provider=metadata_provider)
+
+    async def stream_account_portfolio(
+        self,
+        account_address: str,
+        callback: Callable,
+        on_end_callback: Optional[Callable] = None,
+        on_status_callback: Optional[Callable] = None,
+        subaccount_id: Optional[str] = None,
+        update_type: Optional[str] = None,
+    ):
+        request = exchange_portfolio_pb.StreamAccountPortfolioRequest(
+            account_address=account_address,
+            subaccount_id=subaccount_id,
+            type=update_type,
+        )
+
+        await self._assistant.listen_stream(
+            call=self._stub.StreamAccountPortfolio,
+            request=request,
+            callback=callback,
+            on_end_callback=on_end_callback,
+            on_status_callback=on_status_callback,
+        )

tests/client/indexer/configurable_portfolio_query_servicer.py

@@ -0,0 +1,24 @@
+from collections import deque
+
+from pyinjective.proto.exchange import (
+    injective_portfolio_rpc_pb2 as exchange_portfolio_pb,
+    injective_portfolio_rpc_pb2_grpc as exchange_portfolio_grpc,
+)
+
+
+class ConfigurablePortfolioQueryServicer(exchange_portfolio_grpc.InjectivePortfolioRPCServicer):
+    def __init__(self):
+        super().__init__()
+        self.account_portfolio_responses = deque()
+        self.stream_account_portfolio_responses = deque()
+
+    async def AccountPortfolio(
+        self, request: exchange_portfolio_pb.AccountPortfolioRequest, context=None, metadata=None
+    ):
+        return self.account_portfolio_responses.pop()
+
+    async def StreamAccountPortfolio(
+        self, request: exchange_portfolio_pb.StreamAccountPortfolioRequest, context=None, metadata=None
+    ):
+        for event in self.stream_account_portfolio_responses:
+            yield event

tests/client/indexer/grpc/test_indexer_grpc_portfolio_api.py

@@ -0,0 +1,117 @@
+import grpc
+import pytest
+
+from pyinjective.client.indexer.grpc.indexer_grpc_portfolio_api import IndexerGrpcPortfolioApi
+from pyinjective.core.network import Network
+from pyinjective.proto.exchange import injective_portfolio_rpc_pb2 as exchange_portfolio_pb
+from tests.client.indexer.configurable_portfolio_query_servicer import ConfigurablePortfolioQueryServicer
+
+
+@pytest.fixture
+def portfolio_servicer():
+    return ConfigurablePortfolioQueryServicer()
+
+
+class TestIndexerGrpcPortfolioApi:
+    @pytest.mark.asyncio
+    async def test_fetch_account_portfolio(
+        self,
+        portfolio_servicer,
+    ):
+        coin = exchange_portfolio_pb.Coin(
+            denom="peggy0x87aB3B4C8661e07D6372361211B96ed4Dc36B1B5",
+            amount="2322098",
+        )
+        subaccount_deposit = exchange_portfolio_pb.SubaccountDeposit(
+            total_balance="0.170858923182467801",
+            available_balance="0.170858923182467801",
+        )
+        subaccount_balance = exchange_portfolio_pb.SubaccountBalanceV2(
+            subaccount_id="0xc7dca7c15c364865f77a4fb67ab11dc95502e6fe000000000000000000000000",
+            denom="peggy0x87aB3B4C8661e07D6372361211B96ed4Dc36B1B5",
+            deposit=subaccount_deposit,
+        )
+        position = exchange_portfolio_pb.DerivativePosition(
+            ticker="INJ/USDT PERP",
+            market_id="0x17ef48032cb24375ba7c2e39f384e56433bcab20cbee9a7357e4cba2eb00abe6",
+            subaccount_id="0x1383dabde57e5aed55960ee43e158ae7118057d3000000000000000000000000",
+            direction="short",
+            quantity="0.070294765766186502",
+            entry_price="15980281.340438795311756847",
+            margin="561065.540974",
+            liquidation_price="23492052.224777",
+            mark_price="16197000",
+            aggregate_reduce_only_quantity="0",
+            updated_at=1700161202147,
+            created_at=-62135596800000,
+        )
+        positions_with_upnl = exchange_portfolio_pb.PositionsWithUPNL(
+            position=position,
+            unrealized_pnl="-364.479654577777780880",
+        )
+
+        portfolio = exchange_portfolio_pb.Portfolio(
+            account_address="inj1clw20s2uxeyxtam6f7m84vgae92s9eh7vygagt",
+            bank_balances=[coin],
+            subaccounts=[subaccount_balance],
+            positions_with_upnl=[positions_with_upnl],
+        )
+
+        portfolio_servicer.account_portfolio_responses.append(
+            exchange_portfolio_pb.AccountPortfolioResponse(
+                portfolio=portfolio,
+            )
+        )
+
+        network = Network.devnet()
+        channel = grpc.aio.insecure_channel(network.grpc_exchange_endpoint)
+
+        api = IndexerGrpcPortfolioApi(channel=channel, metadata_provider=lambda: self._dummy_metadata_provider())
+        api._stub = portfolio_servicer
+
+        result_auction = await api.fetch_account_portfolio(account_address=portfolio.account_address)
+        expected_auction = {
+            "portfolio": {
+                "accountAddress": portfolio.account_address,
+                "bankBalances": [
+                    {
+                        "denom": coin.denom,
+                        "amount": coin.amount,
+                    }
+                ],
+                "subaccounts": [
+                    {
+                        "subaccountId": subaccount_balance.subaccount_id,
+                        "denom": subaccount_balance.denom,
+                        "deposit": {
+                            "totalBalance": subaccount_deposit.total_balance,
+                            "availableBalance": subaccount_deposit.available_balance,
+                        },
+                    }
+                ],
+                "positionsWithUpnl": [
+                    {
+                        "position": {
+                            "ticker": position.ticker,
+                            "marketId": position.market_id,
+                            "subaccountId": position.subaccount_id,
+                            "direction": position.direction,
+                            "quantity": position.quantity,
+                            "entryPrice": position.entry_price,
+                            "margin": position.margin,
+                            "liquidationPrice": position.liquidation_price,
+                            "markPrice": position.mark_price,
+                            "aggregateReduceOnlyQuantity": position.aggregate_reduce_only_quantity,
+                            "createdAt": str(position.created_at),
+                            "updatedAt": str(position.updated_at),
+                        },
+                        "unrealizedPnl": positions_with_upnl.unrealized_pnl,
+                    },
+                ],
+            }
+        }
+
+        assert result_auction == expected_auction
+
+    async def _dummy_metadata_provider(self):
+        return None