balances

Documentation for eth_defi.balances Python module.

Token holding and portfolio for addresses.

Functions

`convert_balances_to_decimal`(web3, raw_balances)	Convert mapping of ERC-20 holdings to decimals.
`create_erc20_balance_call`(web3, ...[, debug])	Create a multicall wrapper for ERC-20 balanceOf call.
`fetch_erc20_balances_by_token_list`(web3, ...)	Get all current holdings of an account for a limited set of ERC-20 tokens.
`fetch_erc20_balances_by_transfer_event`(web3, ...)	Get all current holdings of an account.
`fetch_erc20_balances_fallback`(web3, address, ...)	Get all onchain balances of the token.
`fetch_erc20_balances_multicall`(web3, ...[, ...])	Read balance of multiple ERC-20 tokens on an address using internal multicall implementation.
`fetch_erc20_balances_multicall_v6`(web3, ...)	Read balance of multiple ERC-20 tokens on an address once using multicall.
`fetch_erc20_balances_multicall_v7`(web3, ...)	Read balance of multiple ERC-20 tokens on an address using internal multicall implementation.

Classes

`DecimalisedHolding`	A helper class to represent token holdings.
`ERC20BalanceCall`	Multicall wrapper for ERC-20 balanceOf calls.

Exceptions

BalanceFetchFailed

Could not read balances for an address.

class DecimalisedHolding

Bases: object

A helper class to represent token holdings.

Exposes the underlying decimals the ERC-20 wants to express.

__init__(value, decimals, contract)

Parameters

value (decimal.Decimal) –
decimals (int) –
contract (web3.contract.contract.Contract) –

Return type

None

exception BalanceFetchFailed

Bases: Exception

Could not read balances for an address.

Usually this means that you tried to read balances for an address with too many transactions and the underlying GoEthereun node craps out.

__init__(*args, **kwargs)

__new__(**kwargs)

add_note(): Exception.add_note(note) – add a note to the exception

with_traceback(): Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class ERC20BalanceCall

Bases: eth_defi.event_reader.multicall_batcher.MulticallWrapper

Multicall wrapper for ERC-20 balanceOf calls.

get_key()

Get key that will identify this call in the result dictionary

Return type: Hashable

handle(succeed, raw_return_value)

Parse the call result.

Parameters

succeed (bool) – Did we revert or not
raw_return_value (bytes) – Undecoded bytes from the Solidity function call

Returns

The value placed in the return dict

Return type

int | None

__init__(call, debug=False, token_address=None, holder_address=None)

Parameters

call (object) –
debug (bool) –
token_address (eth_typing.evm.HexAddress) –
holder_address (eth_typing.evm.HexAddress) –

Return type

None

get_human_args()

Get Solidity args as human readable string for debugging.

Return type: str

multicall_callback(succeed, raw_return_value)

Convert the raw Solidity function call result to a denominated token amount.

Multicall library callback

Returns

The token amount in the reserve currency we get on the market sell.

None if this path was not supported (Solidity reverted).

Parameters

succeed (bool) –
raw_return_value (Any) –

Return type

Any

fetch_erc20_balances_by_transfer_event(web3, owner, from_block=1, last_block_num=None)

Get all current holdings of an account.

We attempt to build a list of token holdings by analysing incoming ERC-20 Transfer events to a wallet.

The blockchain native currency like ETH or MATIC is not included in the analysis, because native currency transfers do not generate events.

We are not doing any throttling: If you ask for too many events once this function and your Ethereum node are likely to blow up.

Note

Because the limitations of GoEthereum, this method is likely to fail on public JSON-RPC nodes for blockchains like Binance Smart Chain, Polygon and others. E.g. BSC nodes will fail with {‘code’: -32000, ‘message’: ‘exceed maximum block range: 5000’}. Even if the nodes don’t directly fail, their JSON-RPC APIs are likely to timeout.

Example:

# Load up the user with some tokens
usdc.functions.transfer(user_1, 500).transact({"from": deployer})
aave.functions.transfer(user_1, 200).transact({"from": deployer})
balances = fetch_erc20_balances(web3, user_1)
assert balances[usdc.address] == 500
assert balances[aave.address] == 200

Parameters

web3 (web3.main.Web3) – Web3 instance
owner (eth_typing.evm.HexAddress) – The address we are analysis
from_block (Optional[int]) – As passed to eth_getLogs
last_block_num (Optional[eth_typing.evm.BlockNumber]) – Set to the last block, inclusive, if you want to have an analysis of in a point of history.

Returns

Map of (token address, amount)

Return type

Dict[eth_typing.evm.HexAddress, int]

fetch_erc20_balances_by_token_list(web3, owner, tokens, block_identifier=None, decimalise=False)

Get all current holdings of an account for a limited set of ERC-20 tokens.

If you know what tokens you are interested in, this method is much more efficient way than fetch_erc20_balances_by_transfer_event() to query token balances.

Example:

def test_portfolio_token_list(web3: Web3, deployer: str, user_1: str, usdc: Contract, aave: Contract):
    # Create a set of tokens
    tokens = {aave.address, usdc.address}
    # Load up the user with some tokens
    usdc.functions.transfer(user_1, 500).transact({"from": deployer})
    aave.functions.transfer(user_1, 200).transact({"from": deployer})
    balances = fetch_erc20_balances_by_token_list(web3, user_1, tokens)
    assert balances[usdc.address] == 500
    assert balances[aave.address] == 200

Parameters

tokens (Collection[Union[eth_typing.evm.HexAddress, str]]) – ERC-20 list
block_identifier (Union[Literal['latest', 'earliest', 'pending', 'safe', 'finalized'], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, hexbytes.main.HexBytes, int]) – Fetch at specific height
decimalise –
If True, convert output amounts to humanised format in Python Decimal.

Use cached TokenDetails data.
web3 (web3.main.Web3) –
owner (Union[eth_typing.evm.HexAddress, str]) –

Raises

BalanceFetchFailed – When you give a non-ERC-20 contract as a token.

Return type

Dict[Union[eth_typing.evm.HexAddress, str], int | decimal.Decimal]

convert_balances_to_decimal(web3, raw_balances, require_decimals=True)

Convert mapping of ERC-20 holdings to decimals.

Issues a JSON-RPC call to fetch token data for each ERC-20 in the input dictionary.

Example:

raw_balances = fetch_erc20_balances_by_token_list(web3, address, tokens)
return convert_balances_to_decimal(web3, raw_balances)

Parameters

raw_balances (Dict[Union[eth_typing.evm.HexAddress, str], int]) – Token address -> uint256 mappings
require_decimals – Safety check to ensure ERC-20 tokens have valid decimals set. Prevents some wrong addresses and broken tokens.

Returns

Token address -> DecimalisedHolding mappings

Return type

Dict[eth_typing.evm.HexAddress, eth_defi.balances.DecimalisedHolding]

fetch_erc20_balances_multicall_v6(web3, address, tokens, block_identifier, decimalise=True, chunk_size=50, token_cache=LRUCache({}, maxsize=1024, currsize=0), gas_limit=10000000, raise_on_error=True, max_workers=1)

Read balance of multiple ERC-20 tokens on an address once using multicall.

Fast, batches multiple calls on one JSON-RPC request
Using internal multicall module eth_defi.event_reader.multicall_batcher

Example:

def test_fetch_erc20_balances_multicall(web3):
    tokens = {
        "0x6921B130D297cc43754afba22e5EAc0FBf8Db75b",  # DogInMe
        "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913",  # USDC on Base
    }

    # Velvet vault
    address = "0x9d247fbc63e4d50b257be939a264d68758b43d04"

    block_number = get_almost_latest_block_number(web3)

    balances = fetch_erc20_balances_multicall(
        web3,
        address,
        tokens,
        block_identifier=block_number,
    )

    existing_dogmein_balance = balances["0x6921B130D297cc43754afba22e5EAc0FBf8Db75b"]
    assert existing_dogmein_balance > 0

    existing_usdc_balance = balances["0x833589fcd6edb6e08f4c7c32d4f71b54bda02913"]
    assert existing_usdc_balance > Decimal(1.0)

Parameters

address (Union[eth_typing.evm.HexAddress, str]) – Our wallet address of which balances we query
tokens (list[Union[eth_typing.evm.HexAddress, str]] | set[Union[eth_typing.evm.HexAddress, str]]) – List of ERC-20 addresses.
block_identifier –
Fetch at specific height.

Must be given for a multicall.
chunk_size – How many ERC-20 addresses feed to multicall once
gas_limit – Gas limit of the multicall request
decimalise –
If True, convert output amounts to humanised format in Python Decimal.

Use cached TokenDetails data.
token_cache (cachetools.Cache | None) – Cache ERC-20 decimal data.
max_workers – Use this many worker processes
raise_on_error – See BalanceFetchFailed.
web3 (web3.main.Web3) –

Raises

BalanceFetchFailed –

balanceOf() call failed.

When you give a non-ERC-20 contract as a token.

Returns

Map of token address -> balance.

If ERC-20 call failed, balance is set to None if raise_on_error is False.

Return type

dict[Union[eth_typing.evm.HexAddress, str], decimal.Decimal]

create_erc20_balance_call(web3, token_address, holder_address, debug=False)

Create a multicall wrapper for ERC-20 balanceOf call.

Parameters

web3 (web3.main.Web3) –
token_address (Union[eth_typing.evm.HexAddress, str]) –
holder_address (Union[eth_typing.evm.HexAddress, str]) –
debug (bool) –

Return type

eth_defi.balances.ERC20BalanceCall

fetch_erc20_balances_multicall_v7(web3, address, tokens, block_identifier, decimalise=True, chunk_size=50, token_cache=LRUCache({}, maxsize=1024, currsize=0), gas_limit=10000000, raise_on_error=True)

Read balance of multiple ERC-20 tokens on an address using internal multicall implementation.

Fast, batches multiple calls on one JSON-RPC request
Uses internal Multicall3 implementation without external dependencies

Example:

def test_fetch_erc20_balances_multicall(web3):
    tokens = {
        "0x6921B130D297cc43754afba22e5EAc0FBf8Db75b",  # DogInMe
        "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913",  # USDC on Base
    }

    # Velvet vault
    address = "0x9d247fbc63e4d50b257be939a264d68758b43d04"

    block_number = get_almost_latest_block_number(web3)

    balances = fetch_erc20_balances_multicall(
        web3,
        address,
        tokens,
        block_identifier=block_number,
    )

    existing_dogmein_balance = balances["0x6921B130D297cc43754afba22e5EAc0FBf8Db75b"]
    assert existing_dogmein_balance > 0

    existing_usdc_balance = balances["0x833589fcd6edb6e08f4c7c32d4f71b54bda02913"]
    assert existing_usdc_balance > Decimal(1.0)

Parameters

address (Union[eth_typing.evm.HexAddress, str]) – Address of which balances we query
tokens (list[Union[eth_typing.evm.HexAddress, str]] | set[Union[eth_typing.evm.HexAddress, str]]) – List of ERC-20 addresses.
block_identifier (Union[Literal['latest', 'earliest', 'pending', 'safe', 'finalized'], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, hexbytes.main.HexBytes, int]) –
Fetch at specific height.

Must be given for a multicall.
chunk_size – How many ERC-20 addresses feed to multicall once
gas_limit – Gas limit of the multicall request
decimalise –
If True, convert output amounts to humanised format in Python Decimal.

Use cached TokenDetails data.
token_cache (cachetools.Cache | None) – Cache ERC-20 decimal data.
raise_on_error – See BalanceFetchFailed.
web3 (web3.main.Web3) –

Raises

BalanceFetchFailed –

balanceOf() call failed.

When you give a non-ERC-20 contract as a token.

Returns

Map of token address -> balance.

If ERC-20 call failed, balance is set to None if raise_on_error is False.

Return type

dict[Union[eth_typing.evm.HexAddress, str], decimal.Decimal]

fetch_erc20_balances_multicall(web3, address, tokens, block_identifier, decimalise=True, chunk_size=50, token_cache=LRUCache({}, maxsize=1024, currsize=0), gas_limit=10000000, raise_on_error=True)

Read balance of multiple ERC-20 tokens on an address using internal multicall implementation.

Fast, batches multiple calls on one JSON-RPC request
Uses internal Multicall3 implementation without external dependencies

Example:

def test_fetch_erc20_balances_multicall(web3):
    tokens = {
        "0x6921B130D297cc43754afba22e5EAc0FBf8Db75b",  # DogInMe
        "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913",  # USDC on Base
    }

    # Velvet vault
    address = "0x9d247fbc63e4d50b257be939a264d68758b43d04"

    block_number = get_almost_latest_block_number(web3)

    balances = fetch_erc20_balances_multicall(
        web3,
        address,
        tokens,
        block_identifier=block_number,
    )

    existing_dogmein_balance = balances["0x6921B130D297cc43754afba22e5EAc0FBf8Db75b"]
    assert existing_dogmein_balance > 0

    existing_usdc_balance = balances["0x833589fcd6edb6e08f4c7c32d4f71b54bda02913"]
    assert existing_usdc_balance > Decimal(1.0)

Parameters

address (Union[eth_typing.evm.HexAddress, str]) – Address of which balances we query
tokens (list[Union[eth_typing.evm.HexAddress, str]] | set[Union[eth_typing.evm.HexAddress, str]]) – List of ERC-20 addresses.
block_identifier (Union[Literal['latest', 'earliest', 'pending', 'safe', 'finalized'], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, hexbytes.main.HexBytes, int]) –
Fetch at specific height.

Must be given for a multicall.
chunk_size – How many ERC-20 addresses feed to multicall once
gas_limit – Gas limit of the multicall request
decimalise –
If True, convert output amounts to humanised format in Python Decimal.

Use cached TokenDetails data.
token_cache (cachetools.Cache | None) – Cache ERC-20 decimal data.
raise_on_error – See BalanceFetchFailed.
web3 (web3.main.Web3) –

Raises

BalanceFetchFailed –

balanceOf() call failed.

When you give a non-ERC-20 contract as a token.

Returns

Map of token address -> balance.

If ERC-20 call failed, balance is set to None if raise_on_error is False.

Return type

dict[Union[eth_typing.evm.HexAddress, str], decimal.Decimal]

fetch_erc20_balances_fallback(web3, address, tokens, block_identifier, decimalise=True, chunk_size=50, token_cache=LRUCache({}, maxsize=1024, currsize=0), gas_limit=10000000, raise_on_error=True, disable_multicall=None)

Get all onchain balances of the token.

Safe variant
Try multicall approach first
If it fails for some reason, fall to individual JSON-RPC API approach
A reason for the failure would be crappy RPC providers

See fetch_erc20_balances_multicall() for usage and argument descriptions.

Parameters

disable_multicall (bool) –
Disable multicall behaviour.

If set to None autodetect local dev/test chain and disable based on the presence of Anvil: assume no multicall contract deployed there. On the mainnet fork, assume the presence of multicall contract.
web3 (web3.main.Web3) –
address (Union[eth_typing.evm.HexAddress, str]) –
tokens (list[Union[eth_typing.evm.HexAddress, str]] | set[Union[eth_typing.evm.HexAddress, str]]) –
block_identifier (Union[Literal['latest', 'earliest', 'pending', 'safe', 'finalized'], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, hexbytes.main.HexBytes, int]) –
token_cache (cachetools.Cache | None) –

Return type

dict[Union[eth_typing.evm.HexAddress, str], decimal.Decimal]