[History] add balance history and transactions function for address (#25

)

* [History] add balance history and transactions function for address

* update changelog

* fix merge typo

* fix lint

* address comment

* minor

* minor

* address comments

* fix merge issue

* lint

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+### Added
+
+- Support list historical balances and list transactions function for address.
+
 ## [0.0.4] - 2024-10-1
 
 ### Added

cdp/address.py

@@ -1,10 +1,13 @@
+from collections.abc import Iterator
 from decimal import Decimal
 
 from cdp.asset import Asset
 from cdp.balance import Balance
 from cdp.balance_map import BalanceMap
 from cdp.cdp import Cdp
 from cdp.faucet_transaction import FaucetTransaction
+from cdp.historical_balance import HistoricalBalance
+from cdp.transaction import Transaction
 
 
 class Address:
@@ -98,6 +101,33 @@ def balances(self):
 
         return BalanceMap.from_models(response.data)
 
+    def historical_balances(self, asset_id) -> Iterator[HistoricalBalance]:
+        """List historical balances.
+
+        Args:
+            asset_id (str): The asset ID.
+
+        Returns:
+            Iterator[HistoricalBalance]: An iterator of HistoricalBalance objects.
+
+        Raises:
+            Exception: If there's an error listing the historical balances.
+
+        """
+        return HistoricalBalance.list(network_id=self.network_id, address_id=self.address_id, asset_id=asset_id)
+
+    def transactions(self) -> Iterator[Transaction]:
+        """List transactions of the address.
+
+        Returns:
+            Iterator[Transaction]: An iterator of Transaction objects.
+
+        Raises:
+            Exception: If there's an error listing the transactions.
+
+        """
+        return Transaction.list(network_id=self.network_id, address_id=self.address_id)
+
     def __str__(self) -> str:
         """Return a string representation of the Address."""
         return f"Address: (address_id: {self.address_id}, network_id: {self.network_id})"

cdp/api_clients.py

@@ -1,11 +1,13 @@
 from cdp.cdp_api_client import CdpApiClient
 from cdp.client.api.addresses_api import AddressesApi
 from cdp.client.api.assets_api import AssetsApi
+from cdp.client.api.balance_history_api import BalanceHistoryApi
 from cdp.client.api.contract_invocations_api import ContractInvocationsApi
 from cdp.client.api.external_addresses_api import ExternalAddressesApi
 from cdp.client.api.networks_api import NetworksApi
 from cdp.client.api.smart_contracts_api import SmartContractsApi
 from cdp.client.api.trades_api import TradesApi
+from cdp.client.api.transaction_history_api import TransactionHistoryApi
 from cdp.client.api.transfers_api import TransfersApi
 from cdp.client.api.wallets_api import WalletsApi
 
@@ -46,6 +48,8 @@ def __init__(self, cdp_client: CdpApiClient) -> None:
         self._trades: TradesApi | None = None
         self._contract_invocations: ContractInvocationsApi | None = None
         self._smart_contracts: SmartContractsApi | None = None
+        self._balance_history: BalanceHistoryApi | None = None
+        self._transaction_history: TransactionHistoryApi | None = None
 
     @property
     def wallets(self) -> WalletsApi:
@@ -167,6 +171,21 @@ def contract_invocations(self) -> ContractInvocationsApi:
             self._contract_invocations = ContractInvocationsApi(api_client=self._cdp_client)
         return self._contract_invocations
 
+    @property
+    def balance_history(self) -> BalanceHistoryApi:
+        """Get the BalanceHistoryApi client instance.
+
+        Returns:
+            BalanceHistoryApi: The BalanceHistoryApi client instance.
+
+        Note:
+            This property lazily initializes the BalanceHistoryApi client on first access.
+
+        """
+        if self._balance_history is None:
+            self._balance_history = BalanceHistoryApi(api_client=self._cdp_client)
+        return self._balance_history
+
     @property
     def smart_contracts(self) -> SmartContractsApi:
         """Get the SmartContractsApi client instance.
@@ -181,3 +200,18 @@ def smart_contracts(self) -> SmartContractsApi:
         if self._smart_contracts is None:
             self._smart_contracts = SmartContractsApi(api_client=self._cdp_client)
         return self._smart_contracts
+
+    @property
+    def transaction_history(self) -> TransactionHistoryApi:
+        """Get the TransactionHistoryApi client instance.
+
+        Returns:
+            TransactionHistoryApi: The TransactionHistoryApi client instance.
+
+        Note:
+            This property lazily initializes the TransactionHistoryApi client on first access.
+
+        """
+        if self._transaction_history is None:
+            self._transaction_history = TransactionHistoryApi(api_client=self._cdp_client)
+        return self._transaction_history

cdp/historical_balance.py

@@ -0,0 +1,129 @@
+from collections.abc import Iterator
+from decimal import Decimal
+
+from cdp.asset import Asset
+from cdp.cdp import Cdp
+from cdp.client.models.address_historical_balance_list import AddressHistoricalBalanceList
+from cdp.client.models.historical_balance import HistoricalBalance as HistoricalBalanceModel
+
+
+class HistoricalBalance:
+    """A class representing a balance."""
+
+    def __init__(self, amount: Decimal, asset: Asset, block_height: str, block_hash: str):
+        """Initialize the Balance class.
+
+        Args:
+            amount (Decimal): The amount.
+            asset (Asset): The asset.
+            block_height (str): the block height where the balance is in.
+            block_hash (str): the block hash where the balance is in.
+
+        """
+        self._amount = amount
+        self._asset = asset
+        self._block_height = block_height
+        self._block_hash = block_hash
+
+    @classmethod
+    def from_model(cls, model: HistoricalBalanceModel) -> "HistoricalBalance":
+        """Create a Balance instance from a model.
+
+        Args:
+            model (BalanceModel): The model representing the balance.
+            asset_id (Optional[str]): The asset ID.
+
+        Returns:
+            Balance: The Balance instance.
+
+        """
+        asset = Asset.from_model(model.asset)
+
+        return cls(
+            amount=asset.from_atomic_amount(model.amount),
+            asset=asset,
+            block_height=model.block_height,
+            block_hash=model.block_hash
+        )
+
+    @classmethod
+    def list(cls, network_id: str, address_id: str, asset_id: str) -> Iterator["HistoricalBalance"]:
+        """List historical balances of an address of an asset.
+
+        Args:
+            network_id (str): The ID of the network to list historical balance for.
+            address_id (str): The ID of the address to list historical balance for.
+            asset_id(str): The asset ID to list historical balance.
+
+        Returns:
+            Iterator[Transaction]: An iterator of HistoricalBalance objects.
+
+        Raises:
+            Exception: If there's an error listing the historical_balances.
+
+        """
+        page = None
+        while True:
+            response: AddressHistoricalBalanceList = Cdp.api_clients.balance_history.list_address_historical_balance(
+                network_id=network_id,
+                address_id=address_id,
+                asset_id=Asset.primary_denomination(asset_id),
+                limit=100,
+                page=page,
+            )
+
+            for model in response.data:
+                yield cls.from_model(model)
+
+            if not response.has_more:
+                break
+
+            page = response.next_page
+
+    @property
+    def amount(self) -> Decimal:
+        """Get the amount.
+
+        Returns:
+            Decimal: The amount.
+
+        """
+        return self._amount
+
+    @property
+    def asset(self) -> Asset:
+        """Get the asset.
+
+        Returns:
+            Asset: The asset.
+
+        """
+        return self._asset
+
+    @property
+    def block_height(self) -> str:
+        """Get the block height.
+
+        Returns:
+            str: The block height.
+
+        """
+        return self._block_height
+
+    @property
+    def block_hash(self) -> str:
+        """Get the block hash.
+
+        Returns:
+            str: The block hash.
+
+        """
+        return self._block_hash
+
+    def __str__(self) -> str:
+        """Return a string representation of the Balance."""
+        return f"HistoricalBalance: (amount: {self.amount}, asset: {self.asset}, block_height: {self.block_height}, block_hash: {self.block_hash})"
+
+    def __repr__(self) -> str:
+        """Return a string representation of the Balance."""
+        return str(self)

cdp/transaction.py

@@ -1,11 +1,14 @@
 import json
+from collections.abc import Iterator
 from enum import Enum
 
 from eth_account.signers.local import LocalAccount
 from eth_account.typed_transactions import DynamicFeeTransaction
 from web3 import Web3
 
+from cdp.cdp import Cdp
 from cdp.client.models import Transaction as TransactionModel
+from cdp.client.models.address_transaction_list import AddressTransactionList
 
 
 class Transaction:
@@ -52,6 +55,38 @@ def __init__(self, model: TransactionModel):
         self._raw: DynamicFeeTransaction | None = None
         self._signature: str | None = model.signed_payload
 
+    @classmethod
+    def list(cls, network_id: str, address_id: str) -> Iterator["Transaction"]:
+        """List transactions of the address.
+
+        Args:
+            network_id (str): The ID of the network to list transaction for.
+            address_id (str): The ID of the address to list transaction for.
+
+        Returns:
+            Iterator[Transaction]: An iterator of Transaction objects.
+
+        Raises:
+            Exception: If there's an error listing the transactions.
+
+        """
+        page = None
+        while True:
+            response: AddressTransactionList = Cdp.api_clients.transaction_history.list_address_transactions(
+                network_id=network_id,
+                address_id=address_id,
+                limit=1,
+                page=page,
+            )
+
+            for model in response.data:
+                yield cls(model)
+
+            if not response.has_more:
+                break
+
+            page = response.next_page
+
     @property
     def unsigned_payload(self) -> str:
         """Get the unsigned payload."""

tests/factories/historical_balance_factory.py

@@ -0,0 +1,41 @@
+import pytest
+
+from cdp.client.models.historical_balance import HistoricalBalance as HistoricalBalanceModel
+from cdp.historical_balance import HistoricalBalance
+
+
+@pytest.fixture
+def historical_balance_model_factory(asset_model_factory):
+    """Create and return a factory for creating HistoricalBalance fixtures."""
+
+    def _create_historical_balance_model(
+        amount="1000000000000000000",
+        network_id="base-sepolia",
+        asset_id="eth",
+        decimals=18,
+        block_hash="0xblockhash",
+        block_height="12345"
+    ):
+        asset_model = asset_model_factory(network_id, asset_id, decimals)
+        return HistoricalBalanceModel(
+            amount=amount, block_hash=block_hash, block_height=block_height, asset=asset_model)
+
+    return _create_historical_balance_model
+
+
+@pytest.fixture
+def historical_balance_factory(asset_factory):
+    """Create and return a factory for creating HistoricalBalance fixtures."""
+
+    def _create_historical_balance(
+        amount="1000000000000000000",
+        network_id="base-sepolia",
+        asset_id="eth",
+        decimals=18,
+        block_hash="0xblockhash",
+        block_height="12345"
+    ):
+        asset = asset_factory(network_id=network_id, asset_id=asset_id, decimals=decimals)
+        return HistoricalBalance(amount, asset, block_height, block_hash)
+
+    return _create_historical_balance