eth_portfolio package

Subpackages

Submodules

eth_portfolio.address module

This module defines the PortfolioAddress class, which represents an address managed by the eth-portfolio system. The 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 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.

class eth_portfolio.address.PortfolioAddress[source]

Bases: _LedgeredBase[AddressLedgerBase]

Represents a portfolio address within the eth-portfolio system.

When awaited, a list of all AddressLedgerBase will be returned.

Example

>>> my_object = PortfolioAddress(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], AddressLedgerBase)
True

__aiter__(self) → AsyncIterator[T]

Return an async iterator that yields {obj} from the {cls}.

Return type:: AsyncIterator[_T]

__getitem__(slice)

Parameters:: slice (slice)
Return type:: ASyncIterator[_T]

__init__(address, portfolio, asynchronous=False)[source]

Initializes the PortfolioAddress instance.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress) – The address to manage.
portfolio (Portfolio) – The portfolio instance managing this address.
asynchronous (optional) – Flag for asynchronous operation. Defaults to False.

Raises:

TypeError – If asynchronous is not a boolean.

Return type:

None

Examples

>>> portfolio = Portfolio()
>>> address = PortfolioAddress('0x1234...', portfolio)

__iter__(self) → Iterator[T]

Return an iterator that yields {obj} from the {cls}.

Note

Synchronous iteration leverages ASyncIterator, which uses asyncio.BaseEventLoop.run_until_complete() to fetch items. ASyncIterator.__next__() raises a SyncModeInAsyncContextError if the event loop is already running.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:: Iterator[T]

filter(self, function: ViewFn[T]) → 'ASyncFilter[T]'

Filters the {obj} yielded by the {cls} based on a function.

Parameters:: function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.
Returns:: An instance of ASyncFilter that yields the filtered {obj} from the {cls}.
Return type:: ASyncFilter[T]

sort(self, *, key: SortKey[T] = None, reverse: bool = False) → 'ASyncSorter[T]'

Sort the {obj} yielded by the {cls}.

Parameters:

key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.
reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the {obj} yielded from this {cls}, but sorted.

Return type:

ASyncSorter[T]

yield_forever()

Return type:: ASyncIterator[_T]

property _ledgers: Iterator[_LT]: An iterator with the 3 ledgers (transactions, internal transfers, token transfers)

property _start_block: int

address: The address being managed.

all[source]

Retrieves all ledger entries between two blocks.

Parameters:

start_block (int | BlockNumber) – The starting block number.
end_block (int | BlockNumber) – The ending block number.

Returns:

The ledger entries.

Return type:

Dict[str, PandableLedgerEntryList]

Examples

>>> all_entries = await address.all(12000000, 12345678)

assets[source]

Retrieves the balances for every asset in the wallet at a given block.

Parameters:: block (optional) – The block number to query. Defaults to None, which uses the latest block.
Returns:: The asset balances at block.
Return type:: TokenBalances

Examples

>>> assets = await address.assets(12345678)

asynchronous: Flag indicating if the operations are asynchronous.

balances[source]

Retrieves balances for all assets in the wallet at a given block.

Parameters:: block (int | BlockNumber | None) – The block number.
Returns:: The balances.
Return type:: TokenBalances

Examples

>>> balances = await address.balances(12345678)

collateral[source]

Retrieves all balances held by lending protocols on behalf of the wallet at a given block.

Parameters:: block (optional) – The block number. Defaults to None, which uses the latest block.
Returns:: The collateral balances.
Return type:: RemoteTokenBalances

Examples

>>> collateral = await address.collateral(12345678)

debt[source]

Retrieves all debt balances for the wallet at a given block.

Parameters:: block (optional) – The block number. Defaults to None, which uses the latest block.
Returns:: The debt balances at block.
Return type:: RemoteTokenBalances

Examples

>>> debt = await address.debt(12345678)

describe[source]

Describes all of the wallet’s balances at a given block.

Parameters:: block (int) – The block number.
Returns:: The wallet balances.
Return type:: WalletBalances
Raises:: TypeError – If block is not an integer.

Examples

>>> wallet_balances = await address.describe(12345678)

eth_balance[source]

Retrieves the ETH balance for the wallet at a given block.

Parameters:: block (int | BlockNumber | None) – The block number.
Returns:: The ETH balance at block.
Return type:: Balance

Examples

>>> eth_balance = await address.eth_balance(12345678)

external_balances[source]

Retrieves the balances owned by the wallet, but not held in the wallet, at a given block.

Parameters:: block (optional) – The block number. Defaults to None, which uses the latest block.
Returns:: The external balances.
Return type:: RemoteTokenBalances

Examples

>>> external_balances = await address.external_balances(12345678)

internal_transfers: _LT: Ledger for tracking internal transfers.

load_prices: Indicates if price loading is enabled.

property materialized: List[T]

Synchronously iterate through the {cls} and return all {obj}.

Returns:: A list of the {obj} yielded by the {cls}.

staking[source]

Retrieves all balances staked in protocols supported by eth_portfolio on behalf of the wallet at a given block.

Parameters:: block (optional) – The block number. Defaults to None, which uses the latest block.
Returns:: The staked balances.
Return type:: RemoteTokenBalances

Examples

>>> staking_balances = await address.staking(12345678)

token_balances[source]

Retrieves the balances for all tokens in the wallet at a given block.

Parameters:: block – The block number.
Returns:: The token balances at block.
Return type:: TokenBalances

Examples

>>> token_balances = await address.token_balances(12345678)

token_transfers: _LT: Ledger for tracking token transfers.

transactions: _LT: Ledger for tracking transactions.

eth_portfolio.buckets module

async eth_portfolio.buckets.get_token_bucket(token)[source]

Parameters:: token (str | HexBytes | AnyAddress | EthAddress | Contract | int)
Return type:: str

eth_portfolio.constants module

eth_portfolio.portfolio module

This module defines the Portfolio and PortfolioWallets classes, which are used to manage and interact with a collection of PortfolioAddress objects. It also includes the PortfolioLedger class for handling transactions, internal transfers, and token transfers associated with the portfolio. The Portfolio class leverages the a_sync library to support both synchronous and asynchronous operations. This allows it to efficiently gather data, perform portfolio-related tasks, and handle network calls without blocking, thus improving the overall responsiveness and performance of portfolio operations.

This file is part of a larger system that includes modules for handling portfolio addresses, ledger entries, and other related tasks.

class eth_portfolio.portfolio.Portfolio[source]

Bases: ASyncGenericBase

Used to export information about a group of PortfolioAddress objects.

Has all attributes of a PortfolioAddress.
All calls to function(*args, **kwargs) will return {address: PortfolioAddress(Address).function(*args, **kwargs)}

__getitem__(key)[source]

Get a PortfolioAddress by its key.

Parameters:: key (str | HexBytes | AnyAddress | EthAddress) – The address key.
Returns:: The PortfolioAddress object.
Return type:: PortfolioAddress

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> address = portfolio["0xAddress1"]
>>> print(address)

__init__(addresses, start_block=0, label='your portfolio', load_prices=True, asynchronous=False)[source]

Initialize a Portfolio instance.

Parameters:

addresses (str | HexBytes | AnyAddress | EthAddress | Iterable[str | HexBytes | AnyAddress | EthAddress]) – The addresses to include in the portfolio.
start_block (int)
label (str)
load_prices (bool)
asynchronous (bool)

Return type:

None

Examples

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> print(portfolio)

__iter__()[source]

Iterate over the PortfolioAddress objects.

Returns:: An iterator over PortfolioAddress objects.
Return type:: Iterator[PortfolioAddress]

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> for address in portfolio:
...     print(address)

_start_block: The starting block number. Defaults to 0.

addresses

A container for the Portfolio’s PortfolioAddress objects.

Type:: PortfolioWallets

Works like a Dict[Address, PortfolioAddress] except you get the values when you iterate instead of the keys.

asynchronous: bool: Whether to use asynchronous operations. Defaults to False.

describe[source]

Returns a full snapshot of your portfolio at a given block.

Parameters:: block (int) – The block number.
Returns:: A snapshot of the portfolio balances.
Return type:: PortfolioBalances

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> balances = await portfolio.describe(block=1200000)
>>> print(balances)

property internal_transfers: PortfolioInternalTransfersLedger

A container for all internal transfers to or from any of your PortfolioAddress.

Returns:: The PortfolioInternalTransfersLedger object.
Return type:: PortfolioInternalTransfersLedger

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> internal_transfers = portfolio.internal_transfers
>>> print(internal_transfers)

label: A label for the portfolio. Defaults to “your portfolio”

ledger

A container for all of your fun taxable events.

Type:: PortfolioLedger

load_prices: bool: Whether to load prices. Defaults to True.

property token_transfers: PortfolioTokenTransfersLedger

A container for all token transfers to or from any of your PortfolioAddress.

Returns:: The PortfolioTokenTransfersLedger object.
Return type:: PortfolioTokenTransfersLedger

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> token_transfers = portfolio.token_transfers
>>> print(token_transfers)

property transactions: PortfolioTransactionsLedger

A container for all transactions to or from any of your PortfolioAddress.

Returns:: The PortfolioTransactionsLedger object.
Return type:: PortfolioTransactionsLedger

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> transactions = portfolio.transactions
>>> print(transactions)

w3: Web3

The Web3 object which will be used to call your rpc for all read operations.

Type:: Web3

class eth_portfolio.portfolio.PortfolioLedger[source]

Bases: _LedgeredBase[PortfolioLedgerBase]

A container for all transactions, internal transfers, and token transfers to or from any of the wallets in your Portfolio.

When awaited, a list of all PortfolioLedgerBase will be returned.

Example

>>> my_object = PortfolioLedger(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], PortfolioLedgerBase)
True

__aiter__(self) → AsyncIterator[T]

Return an async iterator that yields {obj} from the {cls}.

Return type:: AsyncIterator[_T]

__getitem__(slice)

Parameters:: slice (slice)
Return type:: ASyncIterator[_T]

__init__(portfolio)[source]

Initialize a PortfolioLedger instance.

Parameters:: portfolio (Portfolio) – The Portfolio instance to which this ledger belongs.
Return type:: None

__iter__(self) → Iterator[T]

Return an iterator that yields {obj} from the {cls}.

Note

Synchronous iteration leverages ASyncIterator, which uses asyncio.BaseEventLoop.run_until_complete() to fetch items. ASyncIterator.__next__() raises a SyncModeInAsyncContextError if the event loop is already running.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:: Iterator[T]

filter(self, function: ViewFn[T]) → 'ASyncFilter[T]'

Filters the {obj} yielded by the {cls} based on a function.

Parameters:: function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.
Returns:: An instance of ASyncFilter that yields the filtered {obj} from the {cls}.
Return type:: ASyncFilter[T]

sort(self, *, key: SortKey[T] = None, reverse: bool = False) → 'ASyncSorter[T]'

Sort the {obj} yielded by the {cls}.

Parameters:

key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.
reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the {obj} yielded from this {cls}, but sorted.

Return type:

ASyncSorter[T]

yield_forever()

Return type:: ASyncIterator[_T]

property _ledgers: Iterator[_LT]: An iterator with the 3 ledgers (transactions, internal transfers, token transfers)

property _start_block: int

all_entries[source]

Returns a mapping containing all transactions, internal transfers, and token transfers to or from each wallet in your portfolio.

Parameters:

start_block (int | BlockNumber) – The starting block number.
end_block (int | BlockNumber) – The ending block number.

Returns:

A dictionary mapping PortfolioAddress to their ledger entries.

Return type:

Dict[PortfolioAddress, Dict[str, PandableLedgerEntryList]]

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> entries = await portfolio.ledger.all_entries(start_block=1000000, end_block=1100000)
>>> print(entries)

asynchronous: True if default mode is async, False if sync.

df[source]

Returns a DataFrame containing all transactions, internal transfers, and token transfers to or from any wallet in your portfolio.

Parameters:

start_block (int | BlockNumber) – The starting block number.
end_block (int | BlockNumber) – The ending block number.
full (optional) – Whether to include all columns or a subset. Defaults to False.

Returns:

A DataFrame with the ledger entries.

Return type:

DataFrame

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> df = await portfolio.ledger.df(start_block=1000000, end_block=1100000)
>>> print(df)

internal_transfers: _LT: A container for all internal transfers to or from any of your PortfolioAddress.

property materialized: List[T]

Synchronously iterate through the {cls} and return all {obj}.

Returns:: A list of the {obj} yielded by the {cls}.

portfolio: The Portfolio containing the wallets this ledger will pertain to.

token_transfers: _LT: A container for all token transfers to or from any of your PortfolioAddress.

transactions: _LT: A container for all transactions to or from any of your PortfolioAddress.

class eth_portfolio.portfolio.PortfolioWallets[source]

Bases: Iterable[PortfolioAddress], Dict[str | HexBytes | AnyAddress | EthAddress, PortfolioAddress]

A container that holds all PortfolioAddress objects for a specific Portfolio.

Works like a Dict[Address, PortfolioAddress] except when you iterate you get the values instead of the keys. You should not initialize these. They are created automatically during Portfolio initialization.

_wallets

A checksummed dictionary of PortfolioAddress objects.

Type:: checksum_dict.base.ChecksumAddressDict[eth_portfolio.address.PortfolioAddress]

__getitem__(address)[source]

Get the PortfolioAddress object for a given address.

Parameters:: address (str | HexBytes | AnyAddress | EthAddress) – The address to look up.
Returns:: The PortfolioAddress object.
Return type:: PortfolioAddress

__init__(portfolio, addresses)[source]

Initialize a PortfolioWallets instance.

Parameters:

portfolio (Portfolio) – The Portfolio instance to which this wallet belongs.
addresses (Iterable[str | HexBytes | AnyAddress | EthAddress]) – An iterable of addresses to be included in the portfolio.

Return type:

None

__iter__()[source]

Iterate over the PortfolioAddress objects in the wallets.

Returns:: An iterator over PortfolioAddress objects.
Return type:: Iterator[PortfolioAddress]

get(key, default=None, /): Return the value for key if key is in the dictionary, else default.

items()[source]

Get the items (address, PortfolioAddress pairs) of the wallets.

Returns:: An iterable of (address, PortfolioAddress) tuples.
Return type:: Iterable[Tuple[Address, PortfolioAddress]]

keys()[source]

Get the keys (addresses) of the wallets.

Returns:: An iterable of addresses.
Return type:: Iterable[Address]

pop(k[, d]) → v, remove specified key and return the corresponding value.: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

values()[source]

Get the values (PortfolioAddress objects) of the wallets.

Returns:: An iterable of PortfolioAddress objects.
Return type:: Iterable[PortfolioAddress]

_wallets: ChecksumAddressDict[PortfolioAddress]

A checksummed dictionary of PortfolioAddress objects.

Type:: ChecksumAddressDict[PortfolioAddress]

eth_portfolio.structs module

final class eth_portfolio.structs.InternalTransfer[source]

Bases: _LedgerEntryBase

The :class:`~structs.InternalTransfer`class represents an internal transfer or call within a blockchain transaction.

Captures operations that occur during transaction execution, such as contract-to-contract calls, contract creations, or self-destructs. These are not separate on-chain transactions but part of the execution of a single transaction.

Example

>>> internal_tx = InternalTransfer(trace=evmspec.FilterTrace(...), ...)
>>> internal_tx.type
'call'
>>> internal_tx.trace_address
'0.1'
>>> internal_tx.gas_used
21000

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:: attr (str) – The name of the attribute to access.
Raises:: KeyError – If the provided key is not a member of the struct.
Return type:: Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s['field1']
'value'
>>> s['field2']
Traceback (most recent call last):
    ...
KeyError: ('field2', MyStruct(field1='value'))

__iter__()

Iterate through the keys of the Struct.

Yields:: Struct key.
Return type:: Iterator[str]

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(iter(s))
['field1', 'field2']

async static from_trace(trace, load_prices)[source]

Asynchronously processes a raw internal transfer dictionary into an InternalTransfer object.

This function is the core of the internal transfer processing pipeline. It handles various types of transfers, including special cases like block and uncle rewards. It also filters out certain transfers (e.g., to Gnosis Safe Singleton) and verifies transaction success for non-reward transfers.

The function performs several data transformations: - Value and gas conversions - Optional USD price loading - Field standardization

Parameters:

transfer – A dictionary containing the raw internal transfer data. Expected to have keys such as ‘type’, ‘transactionHash’, ‘blockNumber’, ‘from’, ‘to’, ‘value’, ‘gas’, ‘gasUsed’, ‘traceAddress’.
load_prices (bool) – Flag to determine whether to load USD prices for the transfer value.
trace (Trace | Trace | Trace | Trace)

Returns:

A processed InternalTransfer object.

Return type:

InternalTransfer

Example

>>> transfer = {"type": "call", "transactionHash": "0x123...", "blockNumber": 15537393, "from": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", "to": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", "value": "0x10", "gas": "0x5208", "gasUsed": "0x5208", "traceAddress": [0]}
>>> internal_tx = await load_internal_transfer(transfer=transfer, load_prices=True); print(internal_tx)

Note

Transfers to the Gnosis Safe Singleton (0xd9Db270c1B5E3Bd161E8c8503c55cEABeE709552) are filtered out

as they typically don’t represent actual value transfers. - The traceAddress is converted to a string for consistent representation across different scenarios. - For block and uncle rewards, gas is set to 0 as these are not regular transactions. - When loading prices, the EEE_ADDRESS constant is used, which represents the native currency of the chain.

Integration with eth_portfolio ecosystem:

Uses the InternalTransfer struct from eth_portfolio.structs for standardized output.
Utilizes utility functions from eth_portfolio._loaders.utils and eth_portfolio._utils.
Interacts with the global ‘chain’ object from the brownie library for chain ID.

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

key (str) – The key to look up.
default (optional) – The value to return if the key is not present.

Return type:

Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s.get('field1')
'value'
>>> s.get('field2', 'default')
'default'

items()

Returns an iterator over the struct’s field name and value pairs.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.items())
[('field1', 'value'), ('field2', 42)]

Return type:: Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.keys())
['field1', 'field2']

Return type:: Iterator[str]

values()

Returns an iterator over the struct’s field values.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.values())
['value', 42]

Return type:: Iterator[Any]

property _evm_object: CallTrace | CreateTrace | RewardTrace | SuicideTrace: The EVM object associated with {cls_name}, exactly as it was received from the RPC.

property block_hash: HexBytes: The hash of the block containing the transaction that includes this InternalTransfer.

property block_number: int | BlockNumber: The block number where the {cls_name} was included.

property call_type: str | None: The type of call made in this InternalTransfer (e.g., “call”, “delegatecall”, “staticcall”).

property chainid: Network: The network ID where the {cls_name} occurred.

entry_type: ClassVar[Literal['internal_transfer']] = 'internal_transfer': Constant indicating this value transfer is an internal transfer or call entry.

property from_address: Address: The address from which the internal transfer was sent, if applicable.

property gas: int: The amount of gas allocated for this internal operation.

property gas_used: int: The amount of gas actually consumed by this internal operation, if known.

property hash: HexBytes | str: The unique hash of the transaction containing this internal transfer.

property input: HexBytes: The input data for this internal operation, if any.

property output: HexBytes: The output data from this internal operation, if any.

price: Decimal | None: The price of the cryptocurrency at the time of the {cls_name}, if known.

property reward_type: str | None: The type of the reward, for reward transactions.

property subtraces: int: The number of sub-operations spawned by this InternalTransfer.

property to_address: Address | None: The address to which the internal transfer was sent, if applicable.

trace: CallTrace | CreateTrace | RewardTrace | SuicideTrace: The raw trace object associated with this internal transfer.

property trace_address: List[int]

The path of sub-calls to reach this InternalTransfer within the transaction,

Example: - The following trace_address `python [0, 1, 2] ` represents the third sub-call of the second sub-call of the first top-level call.

property transaction_index: int | None: The index of the transaction within its block, if applicable.

property type: Literal['call', 'create', 'reward', 'suicide']

property value: Decimal: The value/amount of cryptocurrency transferred in the internal transfer, scaled to a human-readable decimal value.

value_usd: Decimal | None: The USD value of the cryptocurrency transferred in the {cls_name}, if price is known.

final class eth_portfolio.structs.TokenTransfer[source]

Bases: _LedgerEntryBase

The TokenTransfer class represents a token transfer event within a blockchain transaction.

Captures the movement of ERC20-like tokens between addresses. These are typically emitted as events by token contracts and are not separate transactions but part of the execution of a transaction interacting with the token contract.

Example

>>> token_transfer = TokenTransfer(
...     log=Log(...),
...     value=Decimal("1000000"),  # 1 USDC (assuming 6 decimals)
...     token="USDC",
... )
>>> token_transfer.token
'USDC'
>>> token_transfer.value
Decimal('1000000')
>>> token_transfer.log_index
3

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:: attr (str) – The name of the attribute to access.
Raises:: KeyError – If the provided key is not a member of the struct.
Return type:: Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s['field1']
'value'
>>> s['field2']
Traceback (most recent call last):
    ...
KeyError: ('field2', MyStruct(field1='value'))

__iter__()

Iterate through the keys of the Struct.

Yields:: Struct key.
Return type:: Iterator[str]

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(iter(s))
['field1', 'field2']

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

key (str) – The key to look up.
default (optional) – The value to return if the key is not present.

Return type:

Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s.get('field1')
'value'
>>> s.get('field2', 'default')
'default'

items()

Returns an iterator over the struct’s field name and value pairs.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.items())
[('field1', 'value'), ('field2', 42)]

Return type:: Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.keys())
['field1', 'field2']

Return type:: Iterator[str]

values()

Returns an iterator over the struct’s field values.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.values())
['value', 42]

Return type:: Iterator[Any]

property _evm_object: Log: The EVM object associated with {cls_name}, exactly as it was received from the RPC.

property block_number: int | BlockNumber: The block number where the {cls_name} was included.

property chainid: Network: The network ID where the {cls_name} occurred.

entry_type: ClassVar[Literal['token_transfer']] = 'token_transfer': Constant indicating this value transfer is a token transfer entry.

property from_address: Address

property hash: HexBytes: The unique hash of the transaction containing this token transfer.

log: Log: The log associated with this token transfer.

property log_index: int: The index of this transfer event within the transaction logs. Used to uniquely identify the Transfer event associated with this TokenTransfer within the transaction.

price: Decimal | None: The price of the cryptocurrency at the time of the {cls_name}, if known.

property to_address: Address

token: str | None: The symbol of the token being transferred, if known.

property token_address: Address: The contract address of the token being transferred.

transaction_index: int

value: Decimal: The amount of tokens transferred, scaled to a human-readable decimal value.

value_usd: Decimal | None: The USD value of the cryptocurrency transferred in the {cls_name}, if price is known.

final class eth_portfolio.structs.Transaction[source]

Bases: _TransactionBase

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:: attr (str) – The name of the attribute to access.
Raises:: KeyError – If the provided key is not a member of the struct.
Return type:: Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s['field1']
'value'
>>> s['field2']
Traceback (most recent call last):
    ...
KeyError: ('field2', MyStruct(field1='value'))

__iter__()

Iterate through the keys of the Struct.

Yields:: Struct key.
Return type:: Iterator[str]

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(iter(s))
['field1', 'field2']

classmethod from_rpc_response(transaction, *, price=None, value_usd=None)

Parameters:

transaction (TransactionLegacy | Transaction2930 | Transaction1559)
price (Decimal | None)
value_usd (Decimal | None)

Return type:

Transaction

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

key (str) – The key to look up.
default (optional) – The value to return if the key is not present.

Return type:

Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s.get('field1')
'value'
>>> s.get('field2', 'default')
'default'

items()

Returns an iterator over the struct’s field name and value pairs.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.items())
[('field1', 'value'), ('field2', 42)]

Return type:: Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.keys())
['field1', 'field2']

Return type:: Iterator[str]

values()

Returns an iterator over the struct’s field values.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.values())
['value', 42]

Return type:: Iterator[Any]

property _evm_object: TransactionLegacy | Transaction2930 | Transaction1559 | Trace | Trace | Trace | Trace | Log: The EVM object associated with {cls_name}, exactly as it was received from the RPC.

property access_list: Tuple[AccessListEntry, ...] | None: List of addresses and storage keys the transaction plans to access (for EIP-2930 and EIP-1559 transactions).

property block_hash: BlockHash: The hash of the block that includes this Transaction.

property block_number: int | BlockNumber: The block number where the {cls_name} was included.

property chainid: Network: The network ID where the {cls_name} occurred.

entry_type: ClassVar[Literal['transaction']] = 'transaction': Constant indicating this value transfer is an on-chain transaction entry.

property from_address: Address | None: The address from which the transaction was sent, if applicable.

property gas: int: The maximum amount of gas the sender is willing to use for the Transaction.

property gas_price: int: The price per unit of gas the sender is willing to pay (for legacy and EIP-2930 transactions).

property hash: TransactionHash: The unique transaction hash.

property input: HexBytes: The data payload sent with the Transaction, often used for contract interactions.

property max_fee_per_gas: int | None: The maximum total fee per gas the sender is willing to pay (for EIP-1559 transactions only).

property max_priority_fee_per_gas: int | None: The maximum priority fee per gas the sender is willing to pay (for EIP-1559 transactions only).

property nonce: int: The sender’s transaction count at the time of this Transaction.

price: Decimal | None: The price of the cryptocurrency at the time of the {cls_name}, if known.

property r: HexBytes: The R component of the Transaction’s ECDSA signature.

property s: HexBytes: The S component of the Transaction’s ECDSA signature.

property to_address: Address | None: The address to which the transaction was sent, if applicable.

transaction: TransactionLegacy | Transaction2930 | Transaction1559: The transaction object received by calling eth_getTransactionByHash.

property transaction_index: int | None: The index of the transaction within its block, if applicable.

property type: int | None: The transaction type (e.g., 0 for legacy, 1 for EIP-2930, 2 for EIP-1559). None for chains that don’t specify transaction types.

property v: int: The V component of the Transaction’s ECDSA signature, used for replay protection.

property value: Decimal: The value/amount of cryptocurrency transferred in the transaction, scaled to a human-readable decimal value.

value_usd: Decimal | None: The USD value of the cryptocurrency transferred in the {cls_name}, if price is known.

property y_parity: int | None: The Y parity of the transaction signature, used in EIP-2718 typed transactions.

final class eth_portfolio.structs.TransactionRLP[source]

Bases: _TransactionBase

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:: attr (str) – The name of the attribute to access.
Raises:: KeyError – If the provided key is not a member of the struct.
Return type:: Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s['field1']
'value'
>>> s['field2']
Traceback (most recent call last):
    ...
KeyError: ('field2', MyStruct(field1='value'))

__iter__()

Iterate through the keys of the Struct.

Yields:: Struct key.
Return type:: Iterator[str]

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(iter(s))
['field1', 'field2']

classmethod from_rpc_response(transaction, *, price=None, value_usd=None)

Parameters:

transaction (TransactionLegacy | Transaction2930 | Transaction1559)
price (Decimal | None)
value_usd (Decimal | None)

Return type:

Transaction

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

key (str) – The key to look up.
default (optional) – The value to return if the key is not present.

Return type:

Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s.get('field1')
'value'
>>> s.get('field2', 'default')
'default'

items()

Returns an iterator over the struct’s field name and value pairs.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.items())
[('field1', 'value'), ('field2', 42)]

Return type:: Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.keys())
['field1', 'field2']

Return type:: Iterator[str]

values()

Returns an iterator over the struct’s field values.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.values())
['value', 42]

Return type:: Iterator[Any]

property _evm_object: TransactionLegacy | Transaction2930 | Transaction1559 | Trace | Trace | Trace | Trace | Log: The EVM object associated with {cls_name}, exactly as it was received from the RPC.

property access_list: Tuple[AccessListEntry, ...] | None: List of addresses and storage keys the transaction plans to access (for EIP-2930 and EIP-1559 transactions).

property block_hash: BlockHash: The hash of the block that includes this Transaction.

property block_number: int | BlockNumber: The block number where the {cls_name} was included.

property chainid: Network: The network ID where the {cls_name} occurred.

entry_type: ClassVar[Literal['transaction']] = 'transaction': Constant indicating this value transfer is an on-chain transaction entry.

property from_address: Address | None: The address from which the transaction was sent, if applicable.

property gas: int: The maximum amount of gas the sender is willing to use for the Transaction.

property gas_price: int: The price per unit of gas the sender is willing to pay (for legacy and EIP-2930 transactions).

property hash: TransactionHash: The unique transaction hash.

property input: HexBytes: The data payload sent with the Transaction, often used for contract interactions.

property max_fee_per_gas: int | None: The maximum total fee per gas the sender is willing to pay (for EIP-1559 transactions only).

property max_priority_fee_per_gas: int | None: The maximum priority fee per gas the sender is willing to pay (for EIP-1559 transactions only).

property nonce: int: The sender’s transaction count at the time of this Transaction.

price: Decimal | None: The price of the cryptocurrency at the time of the {cls_name}, if known.

property r: HexBytes: The R component of the Transaction’s ECDSA signature.

property s: HexBytes: The S component of the Transaction’s ECDSA signature.

property to_address: Address | None: The address to which the transaction was sent, if applicable.

transaction: TransactionRLP: The transaction object received by calling eth_getTransactionByHash.

property transaction_index: int | None: The index of the transaction within its block, if applicable.

property type: int | None: The transaction type (e.g., 0 for legacy, 1 for EIP-2930, 2 for EIP-1559). None for chains that don’t specify transaction types.

property v: int: The V component of the Transaction’s ECDSA signature, used for replay protection.

property value: Decimal: The value/amount of cryptocurrency transferred in the transaction, scaled to a human-readable decimal value.

value_usd: Decimal | None: The USD value of the cryptocurrency transferred in the {cls_name}, if price is known.

property y_parity: int | None: The Y parity of the transaction signature, used in EIP-2718 typed transactions.

eth_portfolio.typing module

This module defines a set of classes to represent and manipulate various levels of balance structures within an Ethereum portfolio. The focus of these classes is on reading, aggregating, and summarizing balances, including the value in both tokens and their equivalent in USD.

The main classes and their purposes are as follows:

Balance: Represents the balance of a single token, including its token amount and equivalent USD value.
TokenBalances: Manages a collection of Balance objects for multiple tokens, providing operations such as summing balances across tokens.
RemoteTokenBalances: Extends TokenBalances to manage balances across different protocols, enabling aggregation and analysis of balances by protocol.
WalletBalances: Organizes token balances into categories such as assets, debts, and external balances for a single wallet. It combines TokenBalances and RemoteTokenBalances to provide a complete view of a wallet’s balances.
PortfolioBalances: Aggregates WalletBalances for multiple wallets, providing operations to sum balances across an entire portfolio.
WalletBalancesRaw: Similar to WalletBalances, but with a key structure optimized for accessing balances directly by wallet and token.
PortfolioBalancesByCategory: Provides an inverted view of PortfolioBalances, allowing access by category first, then by wallet and token.

These classes are designed for efficient parsing, manipulation, and summarization of portfolio data, without managing or altering any underlying assets.

final class eth_portfolio.typing.Balance[source]

Bases: DictStruct

Represents the balance of a single token, including its token amount and equivalent USD value.

Example

>>> balance1 = Balance(Decimal('100'), Decimal('2000'))
>>> balance2 = Balance(Decimal('50'), Decimal('1000'))
>>> combined_balance = balance1 + balance2
>>> combined_balance.balance
Decimal('150')
>>> combined_balance.usd_value
Decimal('3000')

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:: attr (str) – The name of the attribute to access.
Raises:: KeyError – If the provided key is not a member of the struct.
Return type:: Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s['field1']
'value'
>>> s['field2']
Traceback (most recent call last):
    ...
KeyError: ('field2', MyStruct(field1='value'))

__iter__()

Iterate through the keys of the Struct.

Yields:: Struct key.
Return type:: Iterator[str]

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(iter(s))
['field1', 'field2']

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

key (str) – The key to look up.
default (optional) – The value to return if the key is not present.

Return type:

Any

Example

>>> class MyStruct(DictStruct):
...     field1: str
>>> s = MyStruct(field1="value")
>>> s.get('field1')
'value'
>>> s.get('field2', 'default')
'default'

items()

Returns an iterator over the struct’s field name and value pairs.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.items())
[('field1', 'value'), ('field2', 42)]

Return type:: Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.keys())
['field1', 'field2']

Return type:: Iterator[str]

values()

Returns an iterator over the struct’s field values.

Example

>>> class MyStruct(DictStruct):
...     field1: str
...     field2: int
>>> s = MyStruct(field1="value", field2=42)
>>> list(s.values())
['value', 42]

Return type:: Iterator[Any]

balance: Decimal: The amount of the token.

block: int | BlockNumber: The block from which the balance was taken, if known.

token: str | HexBytes | AnyAddress | EthAddress: The token the balance is of, if known.

property usd: Decimal: An alias for usd_value. Returns the USD value of the token amount.

usd_value: Decimal: The USD equivalent value of the token amount.

class eth_portfolio.typing.PortfolioBalances[source]

Bases: DefaultChecksumDict[WalletBalances], _SummableNonNumericMixin

Aggregates WalletBalances for multiple wallets, providing operations to sum balances across an entire portfolio.

The class uses wallet addresses as keys and WalletBalances objects as values.

Parameters:: seed – An initial seed of portfolio balances, either as a dictionary or an iterable of tuples.

Example

>>> portfolio_balances = PortfolioBalances({'0x123': WalletBalances({'assets': TokenBalances({'0x456': Balance(Decimal('100'), Decimal('2000'))})})})
>>> portfolio_balances['0x123']['assets']['0x456'].balance
Decimal('100')

__getitem__(key)

x.__getitem__(y) <==> x[y]

Parameters:: key (AnyAddress)
Return type:: T

__init__(seed=None, *, block=None)[source]

Parameters:

seed (Dict[str | HexBytes | AnyAddress | EthAddress, WalletBalances] | Iterable[Tuple[str | HexBytes | AnyAddress | EthAddress, WalletBalances]] | None)
block (int | None)

Return type:

None

__iter__(): Implement iter(self).

get(key, default=None, /): Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

sum_usd()[source]

Sums the USD values of all wallet balances in the portfolio.

Returns:: The total USD value of all wallet balances in the portfolio.
Return type:: Decimal

Example

>>> portfolio_balances = PortfolioBalances({'0x123': WalletBalances({'assets': TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})})})
>>> total_usd = portfolio_balances.sum_usd()
>>> total_usd
Decimal('2000')

values() → an object providing a view on D's values

block

property dataframe: DataFrame

Converts the portfolio balances into a pandas DataFrame.

Returns:: A DataFrame representation of the portfolio balances.

property inverted: PortfolioBalancesByCategory

Returns an inverted view of the portfolio balances, grouped by category first.

Returns:: The inverted portfolio balances, grouped by category.
Return type:: PortfolioBalancesByCategory

Example

>>> portfolio_balances = PortfolioBalances({'0x123': WalletBalances({'assets': TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})})})
>>> inverted_pb = portfolio_balances.inverted
>>> inverted_pb['assets']['0x123'].balance
Decimal('100')

class eth_portfolio.typing.PortfolioBalancesByCategory[source]

Bases: DefaultDict[Literal[‘assets’, ‘debt’, ‘external’], WalletBalancesRaw], _SummableNonNumericMixin

Provides an inverted view of PortfolioBalances, allowing access by category first, then by wallet and token.

The class uses category labels as keys (assets, debt, external) and WalletBalancesRaw objects as values.

Parameters:: seed – An initial seed of portfolio balances by category, either as a dictionary or an iterable of tuples.

Example

>>> pb_by_category = PortfolioBalancesByCategory({'assets': WalletBalancesRaw({'0x123': TokenBalances({'0x456': Balance(Decimal('100'), Decimal('2000'))})})})
>>> pb_by_category['assets']['0x123']['0x456'].balance
Decimal('100')

__getitem__(): x.__getitem__(y) <==> x[y]

__init__(seed=None, *, block=None)[source]

Parameters:

seed (Dict[Literal['assets', 'debt', 'external'], ~eth_portfolio.typing.WalletBalancesRaw] | ~typing.Iterable[~typing.Tuple[~typing.Literal['assets', 'debt', 'external'], ~eth_portfolio.typing.WalletBalancesRaw]] | None)
block (int | None)

Return type:

None

__iter__(): Implement iter(self).

get(key, default=None, /): Return the value for key if key is in the dictionary, else default.

invert()[source]

Inverts the portfolio balances by category to group by wallet first.

Returns:: The inverted portfolio balances, grouped by wallet first.
Return type:: PortfolioBalances

Example

>>> pb_by_category = PortfolioBalancesByCategory({'assets': WalletBalancesRaw({'0x123': TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})})})
>>> inverted_pb = pb_by_category.invert()
>>> inverted_pb['0x123']['assets']['0x123'].balance
Decimal('100')

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

values() → an object providing a view on D's values

property assets: WalletBalancesRaw

Returns the asset balances across all wallets.

Returns:: The WalletBalancesRaw object representing the asset balances.
Return type:: WalletBalancesRaw

property debt: WalletBalancesRaw

Returns the debt balances across all wallets.

Returns:: The WalletBalancesRaw object representing the debt balances.
Return type:: WalletBalancesRaw

class eth_portfolio.typing.RemoteTokenBalances[source]

Bases: DefaultDict[str, TokenBalances], _SummableNonNumericMixin

Manages token balances across different protocols, extending the TokenBalances functionality to multiple protocols.

The class uses protocol labels as keys and TokenBalances objects as values.

Parameters:: seed – An initial seed of remote token balances, either as a dictionary or an iterable of tuples.

Example

>>> remote_balances = RemoteTokenBalances({'protocol1': TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})})
>>> remote_balances['protocol1']['0x123'].balance
Decimal('100')

__getitem__(): x.__getitem__(y) <==> x[y]

__init__(seed=None, *, block=None)[source]

Parameters:

seed (Dict[str, TokenBalances] | None)
block (int | None)

Return type:

None

__iter__(): Implement iter(self).

get(key, default=None, /): Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

sum_usd()[source]

Sums the USD values of all token balances across all protocols.

Returns:: The total USD value of all remote token balances.
Return type:: Decimal

Example

>>> remote_balances = RemoteTokenBalances({'protocol1': TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})})
>>> total_usd = remote_balances.sum_usd()
>>> total_usd
Decimal('2000')

values() → an object providing a view on D's values

property dataframe: DataFrame

Converts the remote token balances into a pandas DataFrame.

Returns:: A DataFrame representation of the remote token balances.

class eth_portfolio.typing.TokenBalances[source]

Bases: DefaultChecksumDict[Balance], _SummableNonNumericMixin

A specialized defaultdict subclass made for holding a mapping of token -> balance.

Manages a collection of Balance objects for multiple tokens, allowing for operations such as summing balances across tokens.

The class uses token addresses as keys and Balance objects as values. It supports arithmetic operations like addition and subtraction of token balances.

Token addresses are checksummed automatically when adding items to the dict, and the default value for a token not present is an empty Balance object.

Parameters:: seed – An initial seed of token balances, either as a dictionary or an iterable of tuples.

Example

>>> token_balances = TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})
>>> token_balances['0x123'].balance
Decimal('100')

__getitem__(key)[source]

x.__getitem__(y) <==> x[y]

Return type:: Balance

__init__(seed=None, *, block=None)[source]

Parameters:

seed (Dict[str | HexBytes | AnyAddress | EthAddress, Balance] | Iterable[Tuple[str | HexBytes | AnyAddress | EthAddress, Balance]] | None)
block (int | None)

Return type:

None

__iter__(): Implement iter(self).

get(key, default=None, /): Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

sum_usd()[source]

Sums the USD values of all token balances.

Returns:: The total USD value of all token balances.
Return type:: Decimal

Example

>>> token_balances = TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})
>>> total_usd = token_balances.sum_usd()
>>> total_usd
Decimal('2000')

values() → an object providing a view on D's values

block

property dataframe: DataFrame

Converts the token balances into a pandas DataFrame.

Returns:: A DataFrame representation of the token balances.

class eth_portfolio.typing.WalletBalances[source]

Bases: Dict[Literal[‘assets’, ‘debt’, ‘external’], TokenBalances | RemoteTokenBalances], _SummableNonNumericMixin

Organizes token balances into categories such as assets, debts, and external balances for a single wallet.

The class uses categories as keys (assets, debt, external) and TokenBalances or RemoteTokenBalances objects as values.

Parameters:: seed – An initial seed of wallet balances, either as a dictionary or an iterable of tuples.

Example

>>> wallet_balances = WalletBalances({'assets': TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})})
>>> wallet_balances['assets']['0x123'].balance
Decimal('100')

__getitem__(key)[source]

Retrieves the balance associated with the given category key.

Parameters:: key (Literal['assets', 'debt', 'external']) – The category label (assets, debt, or external).
Returns:: The balances associated with the category.
Raises:: KeyError – If the key is not a valid category.
Return type:: TokenBalances | RemoteTokenBalances

Example

>>> wallet_balances = WalletBalances({'assets': TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})})
>>> assets_balances = wallet_balances['assets']
>>> assets_balances['0x123'].balance
Decimal('100')

__init__(seed=None, *, block=None)[source]

Parameters:

seed (WalletBalances | Dict[Literal['assets', 'debt', 'external'], ~eth_portfolio.typing.TokenBalances] | ~typing.Iterable[~typing.Tuple[~typing.Literal['assets', 'debt', 'external'], ~eth_portfolio.typing.TokenBalances]] | None)
block (int | None)

Return type:

None

__iter__(): Implement iter(self).

get(key, default=None, /): Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

sum_usd()[source]

Sums the USD values of the wallet’s assets, debts, and external balances.

Returns:: The total USD value of the wallet’s net assets (assets - debt + external).
Return type:: Decimal

Example

>>> wallet_balances = WalletBalances({'assets': TokenBalances({'0x123': Balance(Decimal('100'), Decimal('2000'))})})
>>> total_usd = wallet_balances.sum_usd()
>>> total_usd
Decimal('2000')

values() → an object providing a view on D's values

property assets: TokenBalances

Returns the assets held by the wallet.

Returns:: The TokenBalances object representing the wallet’s assets.
Return type:: TokenBalances

property dataframe: DataFrame

Converts the wallet balances into a pandas DataFrame.

Returns:: A DataFrame representation of the wallet balances.

property debt: RemoteTokenBalances

Returns the debts associated with the wallet.

Returns:: The RemoteTokenBalances object representing the wallet’s debts.
Return type:: RemoteTokenBalances

property external: RemoteTokenBalances

Returns the external balances associated with the wallet.

Returns:: The RemoteTokenBalances object representing the wallet’s external balances.
Return type:: RemoteTokenBalances

class eth_portfolio.typing.WalletBalancesRaw[source]

Bases: DefaultChecksumDict[TokenBalances], _SummableNonNumericMixin

A structure for key pattern wallet -> token -> balance.

This class is similar to WalletBalances but optimized for key lookups by wallet and token directly. It manages TokenBalances objects for multiple wallets.

Parameters:: seed – An initial seed of wallet balances, either as a dictionary or an iterable of tuples.

Example

>>> raw_balances = WalletBalancesRaw({'0x123': TokenBalances({'0x456': Balance(Decimal('100'), Decimal('2000'))})})
>>> raw_balances['0x123']['0x456'].balance
Decimal('100')

__getitem__(key)

x.__getitem__(y) <==> x[y]

Parameters:: key (AnyAddress)
Return type:: T

__init__(seed=None, *, block=None)[source]

Parameters:

seed (Dict[str | HexBytes | AnyAddress | EthAddress, TokenBalances] | Iterable[Tuple[str | HexBytes | AnyAddress | EthAddress, TokenBalances]] | None)
block (int | None)

Return type:

None

__iter__(): Implement iter(self).

get(key, default=None, /): Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

values() → an object providing a view on D's values

block

class eth_portfolio.typing._SummableNonNumericMixin[source]

Bases: object

Mixin class for non-numeric summable objects.

This class provides an interface for objects that can be used with sum but are not necessarily numeric.

Module contents

class eth_portfolio.Portfolio[source]

Bases: ASyncGenericBase

Used to export information about a group of PortfolioAddress objects.

Has all attributes of a PortfolioAddress.
All calls to function(*args, **kwargs) will return {address: PortfolioAddress(Address).function(*args, **kwargs)}

__getitem__(key)[source]

Get a PortfolioAddress by its key.

Parameters:: key (str | HexBytes | AnyAddress | EthAddress) – The address key.
Returns:: The PortfolioAddress object.
Return type:: PortfolioAddress

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> address = portfolio["0xAddress1"]
>>> print(address)

__init__(addresses, start_block=0, label='your portfolio', load_prices=True, asynchronous=False)[source]

Initialize a Portfolio instance.

Parameters:

addresses (str | HexBytes | AnyAddress | EthAddress | Iterable[str | HexBytes | AnyAddress | EthAddress]) – The addresses to include in the portfolio.
start_block (int)
label (str)
load_prices (bool)
asynchronous (bool)

Return type:

None

Examples

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> print(portfolio)

__iter__()[source]

Iterate over the PortfolioAddress objects.

Returns:: An iterator over PortfolioAddress objects.
Return type:: Iterator[PortfolioAddress]

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> for address in portfolio:
...     print(address)

_start_block: The starting block number. Defaults to 0.

addresses

A container for the Portfolio’s PortfolioAddress objects.

Type:: PortfolioWallets

Works like a Dict[Address, PortfolioAddress] except you get the values when you iterate instead of the keys.

asynchronous: bool: Whether to use asynchronous operations. Defaults to False.

describe[source]

Returns a full snapshot of your portfolio at a given block.

Parameters:: block (int) – The block number.
Returns:: A snapshot of the portfolio balances.
Return type:: PortfolioBalances

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> balances = await portfolio.describe(block=1200000)
>>> print(balances)

property internal_transfers: PortfolioInternalTransfersLedger

A container for all internal transfers to or from any of your PortfolioAddress.

Returns:: The PortfolioInternalTransfersLedger object.
Return type:: PortfolioInternalTransfersLedger

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> internal_transfers = portfolio.internal_transfers
>>> print(internal_transfers)

label: A label for the portfolio. Defaults to “your portfolio”

ledger

A container for all of your fun taxable events.

Type:: PortfolioLedger

load_prices: bool: Whether to load prices. Defaults to True.

property token_transfers: PortfolioTokenTransfersLedger

A container for all token transfers to or from any of your PortfolioAddress.

Returns:: The PortfolioTokenTransfersLedger object.
Return type:: PortfolioTokenTransfersLedger

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> token_transfers = portfolio.token_transfers
>>> print(token_transfers)

property transactions: PortfolioTransactionsLedger

A container for all transactions to or from any of your PortfolioAddress.

Returns:: The PortfolioTransactionsLedger object.
Return type:: PortfolioTransactionsLedger

Example

>>> portfolio = Portfolio(addresses=["0xAddress1", "0xAddress2"])
>>> transactions = portfolio.transactions
>>> print(transactions)

w3: Web3

The Web3 object which will be used to call your rpc for all read operations.

Type:: Web3