"""
This module defines the :class:`~eth_portfolio.address.PortfolioAddress` class, which represents an address managed by the `eth-portfolio` system.
The `eth-portfolio` acts as a manager of addresses, handling various lending protocols and external interactions.
The :class:`~eth_portfolio.address.PortfolioAddress` class is designed to manage different aspects of an Ethereum address within the portfolio,
such as transactions, transfers, balances, and interactions with both external and lending protocols.
Key components and functionalities provided by the :class:`~eth_portfolio.address.PortfolioAddress` class include:
- Handling Ethereum and token balances
- Managing debt and collateral from lending protocols
- Tracking transactions and transfers (both internal and token transfers)
- Providing comprehensive balance descriptions at specific block heights
The class leverages asynchronous operations using the `a_sync` library to efficiently gather and process data.
It also integrates with various submodules from `eth-portfolio` to load balances, manage ledgers, and interact
with external protocols.
"""
import asyncio
import logging
from typing import TYPE_CHECKING, Dict, Optional
import a_sync
from a_sync.exceptions import MappingIsEmptyError
from y import convert
from y._decorators import stuck_coro_debugger
from y.constants import EEE_ADDRESS
from y.datatypes import Address, Block
from eth_portfolio import protocols
from eth_portfolio._ledgers.address import (AddressInternalTransfersLedger,
AddressLedgerBase,
AddressTokenTransfersLedger,
AddressTransactionsLedger,
PandableLedgerEntryList)
from eth_portfolio._loaders import balances
from eth_portfolio._utils import _LedgeredBase
from eth_portfolio.typing import (Balance, RemoteTokenBalances, TokenBalances,
WalletBalances)
if TYPE_CHECKING:
from eth_portfolio.portfolio import Portfolio
logger = logging.getLogger(__name__)
[docs]
class PortfolioAddress(_LedgeredBase[AddressLedgerBase]):
"""
Represents a portfolio address within the eth-portfolio system.
"""
[docs]
def __init__(self, address: Address, portfolio: "Portfolio", asynchronous: bool = False) -> None: # type: ignore
"""
Initializes the PortfolioAddress instance.
Args:
address: The address to manage.
portfolio: The portfolio instance managing this address.
asynchronous (optional): Flag for asynchronous operation. Defaults to False.
Raises:
TypeError: If `asynchronous` is not a boolean.
Examples:
>>> portfolio = Portfolio()
>>> address = PortfolioAddress('0x1234...', portfolio)
"""
self.address = convert.to_address(address)
"""
The address being managed.
"""
if not isinstance(asynchronous, bool):
raise TypeError(f"`asynchronous` must be a boolean, you passed {type(asynchronous)}")
self.asynchronous = asynchronous
"""
Flag indicating if the operations are asynchronous.
"""
self.load_prices = portfolio.load_prices
"""
Indicates if price loading is enabled.
"""
super().__init__(portfolio._start_block)
self.transactions = AddressTransactionsLedger(self)
"""
Ledger for tracking transactions.
"""
self.internal_transfers = AddressInternalTransfersLedger(self)
"""
Ledger for tracking internal transfers.
"""
self.token_transfers = AddressTokenTransfersLedger(self)
"""
Ledger for tracking token transfers.
"""
def __str__(self) -> str:
"""
Returns the string representation of the address.
Returns:
The address as a string.
"""
return self.address
def __repr__(self) -> str:
"""
Returns the string representation of the PortfolioAddress instance.
Returns:
The string representation.
"""
return f"<{self.__class__.__name__} address={self.address} at {hex(id(self))}>"
def __eq__(self, other: object) -> bool:
"""
Checks equality with another object.
Args:
other: The object to compare with.
Returns:
True if equal, False otherwise.
"""
if isinstance(other, PortfolioAddress):
return self.address == other.address
elif isinstance(other, str):
return self.address == convert.to_address(other)
return False
def __hash__(self) -> int:
"""
Returns the hash of the address.
Returns:
The hash value.
"""
return hash(self.address)
# Primary functions
[docs]
@stuck_coro_debugger
async def describe(self, block: int) -> WalletBalances:
"""
Describes all of the wallet's balances at a given block.
Args:
block: The block number.
Returns:
:class:`~eth_portfolio.typing.WalletBalances`: The wallet balances.
Raises:
TypeError: If block is not an integer.
Examples:
>>> wallet_balances = await address.describe(12345678)
"""
if not isinstance(block, int):
raise TypeError(f"Block must be an integer. You passed {type(block)} {block}")
return WalletBalances(await a_sync.gather({
"assets": self.assets(block, sync=False),
"debt": self.debt(block, sync=False),
"external": self.external_balances(block, sync=False),
}))
[docs]
@stuck_coro_debugger
async def assets(self, block: Optional[Block] = None) -> TokenBalances:
"""
Retrieves the balances for every asset in the wallet at a given block.
Args:
block (optional): The block number to query. Defaults to None, which uses the latest block.
Returns:
:class:`~eth_portfolio.typing.TokenBalances`: The asset balances at `block`.
Examples:
>>> assets = await address.assets(12345678)
"""
return await self.balances(block=block, sync=False)
[docs]
@stuck_coro_debugger
async def debt(self, block: Optional[Block] = None) -> RemoteTokenBalances:
"""
Retrieves all debt balances for the wallet at a given block.
Args:
block (optional): The block number. Defaults to None, which uses the latest block.
Returns:
:class:`~eth_portfolio.typing.RemoteTokenBalances`: The debt balances at `block`.
Examples:
>>> debt = await address.debt(12345678)
"""
return await protocols.lending.debt(self.address, block=block)
[docs]
@stuck_coro_debugger
async def external_balances(self, block: Optional[Block] = None) -> RemoteTokenBalances:
"""
Retrieves the balances owned by the wallet, but not held *in* the wallet, at a given block.
Args:
block (optional): The block number. Defaults to None, which uses the latest block.
Returns:
:class:`~eth_portfolio.typing.RemoteTokenBalances`: The external balances.
Examples:
>>> external_balances = await address.external_balances(12345678)
"""
return sum(await asyncio.gather(self.staking(block, sync=False), self.collateral(block, sync=False)))
# Assets
[docs]
@stuck_coro_debugger
async def balances(self, block: Optional[Block]) -> TokenBalances:
"""
Retrieves balances for all assets in the wallet at a given block.
Args:
block: The block number.
Returns:
:class:`~eth_portfolio.typing.TokenBalances`: The balances.
Examples:
>>> balances = await address.balances(12345678)
"""
eth_balance, token_balances = await asyncio.gather(
self.eth_balance(block, sync=False),
self.token_balances(block, sync=False),
)
token_balances[EEE_ADDRESS] = eth_balance
return token_balances
[docs]
@stuck_coro_debugger
async def eth_balance(self, block: Optional[Block]) -> Balance:
"""
Retrieves the ETH balance for the wallet at a given block.
Args:
block: The block number.
Returns:
:class:`~eth_portfolio.typing.Balance`: The ETH balance at `block`.
Examples:
>>> eth_balance = await address.eth_balance(12345678)
"""
return await balances.load_eth_balance(self.address, block)
[docs]
@stuck_coro_debugger
async def token_balances(self, block) -> TokenBalances:
"""
Retrieves the balances for all tokens in the wallet at a given block.
Args:
block: The block number.
Returns:
:class:`~eth_portfolio.typing.TokenBalances`: The token balances at `block`.
Examples:
>>> token_balances = await address.token_balances(12345678)
"""
try:
data = a_sync.map(
balances.load_token_balance,
self.token_transfers._yield_tokens_at_block(block=block),
address=self.address,
block=block,
)
return TokenBalances(await data)
except MappingIsEmptyError:
return TokenBalances()
[docs]
@stuck_coro_debugger
async def collateral(self, block: Optional[Block] = None) -> RemoteTokenBalances:
"""
Retrieves all balances held by lending protocols on behalf of the wallet at a given block.
Args:
block (optional): The block number. Defaults to None, which uses the latest block.
Returns:
:class:`~eth_portfolio.typing.RemoteTokenBalances`: The collateral balances.
Examples:
>>> collateral = await address.collateral(12345678)
"""
return await protocols.lending.collateral(self.address, block=block)
[docs]
@stuck_coro_debugger
async def staking(self, block: Optional[Block] = None) -> RemoteTokenBalances:
"""
Retrieves all balances staked in protocols supported by eth_portfolio on behalf of the wallet at a given block.
Args:
block (optional): The block number. Defaults to None, which uses the latest block.
Returns:
:class:`~eth_portfolio.typing.RemoteTokenBalances`: The staked balances.
Examples:
>>> staking_balances = await address.staking(12345678)
"""
return await protocols.balances(self.address, block=block)
# Ledger Entries
[docs]
@stuck_coro_debugger
async def all(self, start_block: Block, end_block: Block) -> Dict[str, PandableLedgerEntryList]:
"""
Retrieves all ledger entries between two blocks.
Args:
start_block: The starting block number.
end_block: The ending block number.
Returns:
Dict[str, :class:`~eth_portfolio._ledgers.address.PandableLedgerEntryList`]: The ledger entries.
Examples:
>>> all_entries = await address.all(12000000, 12345678)
"""
return await a_sync.gather({
"transactions": self.transactions.get(start_block, end_block, sync=False),
"internal_transactions": self.internal_transfers.get(start_block, end_block, sync=False),
"token_transfers": self.token_transfers.get(start_block, end_block, sync=False),
})