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

Represents a portfolio address within the eth-portfolio system.

This 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.

When awaited, a list of all T objects will be returned.

Example

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

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

Return type:

AsyncIterator[_T]

__getitem__(slice)
Parameters:

slice (slice)

Return type:

ASyncIterator[_T][_T]

__init__(address, start_block, load_prices, num_workers_transactions=1000, asynchronous=False)[source]

Initializes the PortfolioAddress instance.

Parameters:

  • address (str | HexBytes | AnyAddress | EthAddress) – The Ethereum address to manage.

  • start_block (int | BlockNumber) – The block number from which to start tracking.

  • load_prices (bool) – Flag indicating if price loading is enabled.

  • num_workers_transactions (optional) – Number of workers for transaction processing. Defaults to 1000.

  • asynchronous (optional) – Flag for asynchronous operation. Defaults to False.

Raises:

TypeError – If asynchronous is not a boolean.

Return type:

None

Examples

>>> address = PortfolioAddress('0x1234...', 0, True)
>>> print(address)

>>> address = PortfolioAddress('0x1234...', 0, False, num_workers_transactions=500, asynchronous=True)
>>> print(address)

See also

  • Portfolio

  • AddressTransactionsLedger

  • AddressInternalTransfersLedger

  • AddressTokenTransfersLedger

__iter__()

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.

filter(self, function: ViewFn[T])

Filters the T objects yielded by the ASyncIterator[T] based on a function.

Parameters:

function (ViewFn[T]) – 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 T objects from the ASyncIterator[T].

sort(self, key: SortKey[T] = None, reverse: bool = False)

Sort the T objects yielded by the ASyncIterator[T].

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 T objects yielded from this ASyncIterator[T], but sorted.

yield_forever()
Return type:

ASyncIterator[_T][_T]

property _ledgers: Tuple[_LT, _LT, _LT]

An iterator with the 3 ledgers (transactions, internal transfers, token transfers)

property _start_block: int | BlockNumber
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.

materialized

List[T]

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

Returns:

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

Type:

_AwaitableAsyncIterableMixin.materialized

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]

Categorize a token into a specific bucket based on its type.

This function attempts to categorize a given token into predefined buckets such as “Cash & cash equivalents”, “ETH”, “BTC”, “Other long term assets”, or “Other short term assets”. The categorization is based on the token’s characteristics and its presence in specific sets like ETH_LIKE, BTC_LIKE, and OTHER_LONG_TERM_ASSETS.

Parameters:

token (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the token to categorize.

Returns:

A string representing the bucket category of the token.

Raises:

  • ValueError – If the token’s source has not been verified and the error message

  • does not match the expected pattern. –

Return type:

str

Example

Categorize a stablecoin:

>>> await get_token_bucket("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48")
'Cash & cash equivalents'

Categorize an ETH-like token:

>>> await get_token_bucket("0xEeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE")
'ETH'

See also

  • _unwrap_token()

  • _is_stable()

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, num_workers_transactions=1000, asynchronous=False)[source]

Initialize a Portfolio instance.

Parameters:
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)
received(start_block=None, end_block=None)[source]
Parameters:

  • start_block (int | BlockNumber | None)

  • end_block (int | BlockNumber | None)

Return type:

AsyncIterator[Transaction | InternalTransfer | TokenTransfer]

sent(start_block=None, end_block=None)[source]
Parameters:

  • start_block (int | BlockNumber | None)

  • end_block (int | BlockNumber | None)

Return type:

AsyncIterator[Transaction | InternalTransfer | TokenTransfer]

_start_block = 0

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 = False

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[_T]

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: str = 'your portfolio'

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

ledger

A container for all of your fun taxable events.

Type:

PortfolioLedger

load_prices: bool = True

Whether to load prices. Defaults to True.

property token_transfers: PortfolioTokenTransfersLedger[_T]

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[_T]

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

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 T objects will be returned.

Example

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

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

Return type:

AsyncIterator[_T]

__getitem__(slice)
Parameters:

slice (slice)

Return type:

ASyncIterator[_T][_T]

__init__(portfolio)[source]

Initialize a PortfolioLedger instance.

Parameters:

portfolio (Portfolio) – The Portfolio instance to which this ledger belongs.

Return type:

None

__iter__()

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.

filter(self, function: ViewFn[T])

Filters the T objects yielded by the ASyncIterator[T] based on a function.

Parameters:

function (ViewFn[T]) – 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 T objects from the ASyncIterator[T].

received(start_block=None, end_block=None)[source]
Parameters:

  • start_block (int | BlockNumber | None)

  • end_block (int | BlockNumber | None)

Return type:

AsyncIterator[Transaction | InternalTransfer | TokenTransfer]

sent(start_block=None, end_block=None)[source]
Parameters:

  • start_block (int | BlockNumber | None)

  • end_block (int | BlockNumber | None)

Return type:

AsyncIterator[Transaction | InternalTransfer | TokenTransfer]

sort(self, key: SortKey[T] = None, reverse: bool = False)

Sort the T objects yielded by the ASyncIterator[T].

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 T objects yielded from this ASyncIterator[T], but sorted.

yield_forever()
Return type:

ASyncIterator[_T][_T]

property _ledgers: Tuple[_LT, _LT, _LT]

An iterator with the 3 ledgers (transactions, internal transfers, token transfers)

property _start_block: int | BlockNumber
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.

materialized

List[T]

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

Returns:

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

Type:

_AwaitableAsyncIterableMixin.materialized

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[eth_portfolio.typing._T]]

__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__(addresses, start_block, load_prices, num_workers_transactions)[source]

Initialize a PortfolioWallets instance.

Parameters:

  • 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.

  • start_block (int | BlockNumber)

  • load_prices (bool)

  • num_workers_transactions (int)

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[_T]]

A checksummed dictionary of PortfolioAddress objects.

Type:

ChecksumAddressDict[PortfolioAddress]

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, num_workers_transactions=1000, asynchronous=False)[source]

Initialize a Portfolio instance.

Parameters:
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)
received(start_block=None, end_block=None)[source]
Parameters:

  • start_block (int | BlockNumber | None)

  • end_block (int | BlockNumber | None)

Return type:

AsyncIterator[Transaction | InternalTransfer | TokenTransfer]

sent(start_block=None, end_block=None)[source]
Parameters:

  • start_block (int | BlockNumber | None)

  • end_block (int | BlockNumber | None)

Return type:

AsyncIterator[Transaction | InternalTransfer | TokenTransfer]

_start_block = 0

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 = False

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[_T]

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: str = 'your portfolio'

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

ledger

A container for all of your fun taxable events.

Type:

PortfolioLedger

load_prices: bool = True

Whether to load prices. Defaults to True.

property token_transfers: PortfolioTokenTransfersLedger[_T]

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[_T]

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