token

Documentation for eth_defi.token Python module.

ERC-20 token deployment and manipulation.

Deploy ERC-20 tokens to be used within your test suite.

Read also unit test suite for tokens to see how ERC-20 can be manipulated in pytest.

Functions

create_token(web3, deployer, name, symbol, ...)	Deploys a new ERC-20 token on local dev, testnet or mainnet.
fetch_erc20_details(web3, token_address[, ...])	Read token details from on-chain data.
get_chain_known_quote_tokens(chain_id)	Get all good quote tokens on chain.
get_chain_stablecoins(chain_id)	Get all good known stablecoins on a chain.
get_erc20_contract(web3, address[, ...])	Wrap address as ERC-20 standard interface.
get_weth_contract(web3[, name])	Get WETH9 contract for the chain
get_wrapped_native_token_address(chain_id)
reset_default_token_cache()	Purge the cached token data.

Classes

DummyPickledContract	Contract placeholder making contract references pickable
TokenCacheWarmupResult
TokenDetails	ERC-20 token Python presentation.
TokenDiskCache	Token cache that stores tokens in disk.

Exceptions

TokenDetailError

Cannot extract token details for an ERC-20 token for some reason.

class DummyPickledContract

Bases: object

Contract placeholder making contract references pickable

__init__(address)

Parameters

address (str) –

Return type

None

class TokenCacheWarmupResult

Bases: TypedDict

__init__(*args, **kwargs)

__new__(**kwargs)

clear()

Remove all items from the dict.

copy()

Return a shallow copy of the dict.

fromkeys(value=None, /)

Create a new dictionary with keys from iterable and values set to value.

get(key, default=None, /)

Return the value for key if key is in the dictionary, else default.

items()

Return a set-like object providing a view on the dict’s items.

keys()

Return a set-like object providing a view on the dict’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.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None.  Update D from mapping/iterable E and F.

If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values()

Return an object providing a view on the dict’s values.

exception TokenDetailError

Bases: Exception

Cannot extract token details for an ERC-20 token for some reason.

__init__(*args, **kwargs)

__new__(**kwargs)

add_note(note, /)

Add a note to the exception

with_traceback(tb, /)

Set self.__traceback__ to tb and return self.

class TokenDetails

Bases: object

ERC-20 token Python presentation.

• A helper class to work with ERC-20 tokens.

• Read on-chain data, deal with token value decimal conversions.

• Any field can be None for non-well-formed tokens.

• Supports one-way pickling

• Don’t construct directly, use fetch_erc20_details() to read data onchain, with optional caching

Example how to get USDC details on Polygon:

usdc = fetch_erc20_details(web3, "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174")  # USDC on Polygon
formatted = f"Token {usdc.name} ({usdc.symbol}) at {usdc.address} on chain {usdc.chain_id}"
assert formatted == "Token USD Coin (PoS) (USDC) at 0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174 on chain 137"

__init__(contract, name=None, symbol=None, total_supply=None, decimals=None, extra_data=<factory>)

Parameters

• contract (web3.contract.contract.Contract) –

• name (Optional[str]) –

• symbol (Optional[str]) –

• total_supply (Optional[int]) –

• decimals (Optional[int]) –

• extra_data (dict[str, Any]) –

Return type

None

property address: eth_typing.evm.HexAddress

The address of this token.

See also address_lower().

property address_lower: eth_typing.evm.HexAddress

The address of this token.

Always lowercase.

approve(to, amount)

Prepare a ERC20.approve() transaction with human-readable amount.

Example:

usdc_amount = Decimal(9.00)
tx_hash = usdc.approve(vault.address, usdc_amount).transact({"from": depositor})
assert_transaction_success_with_explanation(web3, tx_hash)

Returns

Bound contract function you need to turn to a tx

Parameters

• to (Union[eth_typing.evm.HexAddress, str]) –

• amount (decimal.Decimal) –

Return type

web3.contract.contract.ContractFunction

property chain_id: int

The EVM chain id where this token lives.

convert_to_decimals(raw_amount)

Convert raw token units to decimals.

Example:

details = fetch_erc20_details(web3, token_address)
# Convert 1 wei units to edcimals
assert details.convert_to_decimals(1) == Decimal("0.0000000000000001")

Parameters

raw_amount (int) –

Return type

decimal.Decimal

convert_to_raw(decimal_amount)

Convert decimalised token amount to raw uint256.

Example:

details = fetch_erc20_details(web3, token_address)
# Convert 1.0 USDC to raw unit with 6 decimals
assert details.convert_to_raw(1) == 1_000_000

Parameters

decimal_amount (decimal.Decimal) –

Return type

int

export()

Create a serialisable entry of this class.

Removes web3 connection and such unserialisable data.

Returns

Python dict of exported data.

Return type

dict

fetch_balance_of(address, block_identifier='latest')

Get an address token balance.

Parameters

• block_identifier – A specific block to query if doing archive node historical queries

• address (Union[eth_typing.evm.HexAddress, str]) –

Returns

Converted to decimal using convert_to_decimal()

Return type

decimal.Decimal

fetch_raw_balance_of(address, block_identifier='latest')

Get an address token balance.

Parameters

• block_identifier – A specific block to query if doing archive node historical queries

• address (Union[eth_typing.evm.HexAddress, str]) –

Returns

Raw token amount.

Return type

decimal.Decimal

property functions: web3.contract.contract.ContractFunctions

Alias for underlying Web3 contract method

static generate_cache_key(chain_id, address)

Generate a cache key for this token.

• Cached by (chain, address) as a string

• Validate the inputs before generating the key

• Address is always lowercase

Returns

Human reaadable {chain_id}-{address}

Parameters

• chain_id (int) –

• address (str) –

Return type

str

is_stablecoin_like()

Smell test for stablecoins.

• Symbol check for common stablecoins

• Not immune to scams

• For the list see is_stablecoin_like()

Returns

True if we think this could be a stablecoin.

Return type

bool

transfer(to, amount)

Prepare a ERC20.transfer() transaction with human-readable amount.

Example:

another_new_depositor = web3.eth.accounts[6]
tx_hash = base_usdc.transfer(another_new_depositor, Decimal(500)).transact({"from": usdc_holder, "gas": 100_000})
assert_transaction_success_with_explanation(web3, tx_hash)

Returns

Bound contract function you need to turn to a tx

Parameters

• to (Union[eth_typing.evm.HexAddress, str]) –

• amount (decimal.Decimal) –

Return type

web3.contract.contract.ContractFunction

class TokenDiskCache

Bases: eth_defi.sqlite_cache.PersistentKeyValueStore

Token cache that stores tokens in disk.

• Use with fetch_erc20_details()

• For loading hundreds of tokens once

• Shared across chains

• Enable fast cache warmup with load_token_details_with_multicall()

• Persistent: Make sure subsequent batch jobs do not refetch token data over RPC as it is expensive

• Store as a SQLite database

Example:

addresses = [
    "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",  # USDC
    "0x4200000000000000000000000000000000000006",  # WETH
    "0x50c5725949A6F0c72E6C4a641F24049A917DB0Cb",  # DAI
    "0x554a1283cecca5a46bc31c2b82d6702785fc72d9",  # UNI
]

cache = TokenDiskCache(tmp_path / "disk_cache.sqlite")
web3factory = MultiProviderWeb3Factory(JSON_RPC_BASE)
web3 = web3factory()

#
# Do single token lookups against cache
#
token = fetch_erc20_details(
    web3,
    token_address="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",
    chain_id=web3.eth.chain_id,
    cache=cache,
)
assert token.extra_data["cached"] == False
assert len(cache) == 1
# After one look up, we should have it cached
token = fetch_erc20_details(
    web3,
    token_address="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",
    chain_id=web3.eth.chain_id,
    cache=cache,
)
assert token.extra_data["cached"] == True
cache.purge()

#
# Warm up multiple on dry cache
#
result = cache.load_token_details_with_multicall(
    chain_id=web3.eth.chain_id,
    web3factory=web3factory,
    addresses=addresses,
    max_workers=max_workers,
    display_progress=False,
)
assert result["tokens_read"] == 4
assert "8453-0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913".lower() in cache
assert "8453-0x4200000000000000000000000000000000000006".lower() in cache

cache_data = cache["8453-0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913".lower()]
assert cache_data["name"] == "USD Coin"
assert cache_data["symbol"] == "USDC"
assert cache_data["decimals"] == 6
assert cache_data["supply"] > 1_000_000

token = fetch_erc20_details(
    web3,
    token_address="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",
    chain_id=web3.eth.chain_id,
    cache=cache,
)
assert token.extra_data["cached"] == True

Parameters

• filename – Path to the sqlite database

• autocommit – Whether to autocommit every time new entry is added to the database

__init__(filename=PosixPath('/home/runner/.tradingstrategy/cache/eth-defi-tokens.sqlite'), max_str_length=256)

Parameters

• filename – Path to the sqlite database

• autocommit – Whether to autocommit every time new entry is added to the database

• max_str_length (int) –

__new__(**kwargs)

clear()

Remove all items from the dict.

property conn: sqlite3.Connection

One connection per thread

copy()

Return a shallow copy of the dict.

create_cache_entry(call_results)

Map multicall results to token details data for one address

Parameters

call_results (dict[str, eth_defi.event_reader.multicall_batcher.EncodedCallResult]) –

Return type

dict

decode_value(value)

Hook to convert SQLite values to Python objects

Parameters

value (str) –

Return type

Any

encode_multicalls(address)

Generate multicalls for each token address

Parameters

address (eth_typing.evm.HexAddress) –

Return type

eth_defi.event_reader.multicall_batcher.EncodedCall

encode_value(value)

Hook to convert Python objects to cache format

Parameters

value (dict) –

Return type

Any

fromkeys(value=None, /)

Create a new dictionary with keys from iterable and values set to value.

get(key, default=None)

Return the value for key if key is in the dictionary, else default.

items()

Return a set-like object providing a view on the dict’s items.

keys()

Return a set-like object providing a view on the dict’s keys.

load_token_details_with_multicall(chain_id, web3factory, addresses, display_progress=False, max_workers=8, block_identifier='latest', checkpoint=32)

Warm up cache and load token details for multiple

Parameters

• chain_id (int) –

• web3factory (eth_defi.event_reader.web3factory.Web3Factory) –

• addresses (list[eth_typing.evm.HexAddress]) –

• display_progress (Union[str, bool]) –

• checkpoint (int) –

Return type

eth_defi.token.TokenCacheWarmupResult

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.

purge()

Delete all keys and save.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None.  Update D from mapping/iterable E and F.

If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values()

Return an object providing a view on the dict’s values.

create_token(web3, deployer, name, symbol, supply, decimals=18)

Deploys a new ERC-20 token on local dev, testnet or mainnet.

• Uses ERC20Mock contract for the deployment.

• Waits until the transaction has completed

Example:

# Deploys an ERC-20 token where 100,000 tokens are allocated ato the deployer address
token = create_token(web3, deployer, "Hentai books token", "HENTAI", 100_000 * 10**18)
print(f"Deployed token contract address is {token.address}")
print(f"Deployer account {deployer} has {token.functions.balanceOf(user_1).call() / 10**18} tokens")

Find more examples in Tutorials and unit testing source code.

Parameters

• web3 (web3.main.Web3) – Web3 instance

• deployer (str) –

  Deployer account as 0x address.

  Make sure this account has enough ETH or native token to cover the gas cost.

• name (str) – Token name

• symbol (str) – Token symbol

• supply (int) –

  Token starting supply as raw units.

  E.g. 500 * 10**18 to have 500 tokens minted to the deployer at the start.

• decimals (int) – How many decimals ERC-20 token values have

Returns

Instance to a deployed Web3 contract.

Return type

web3.contract.contract.Contract

fetch_erc20_details(web3, token_address, max_str_length=256, raise_on_error=True, contract_name='ERC20MockDecimals.json', cache=LRUCache({}, maxsize=1024, currsize=0), chain_id=None, cause_diagnostics_message=None)

Read token details from on-chain data.

Connect to Web3 node and do RPC calls to extract the token info. We apply some sanitazation for incoming data, like length checks and removal of null bytes.

The function should not raise an exception as long as the underlying node connection does not fail.

Note

Always give chain_id when possible. Otherwise the caching of data is inefficient.

Example:

details = fetch_erc20_details(web3, token_address)
assert details.name == "Hentai books token"
assert details.decimals == 6

Parameters

• web3 (web3.main.Web3) – Web3 instance

• token_address (Union[eth_typing.evm.HexAddress, str]) – ERC-20 contract address:

• max_str_length (int) – For input sanitisation

• raise_on_error – If set, raise TokenDetailError on any error instead of silently ignoring in and setting details to None.

• contract_name –

  Contract ABI file to use.

  The default is ERC20MockDecimals.json. For USDC use centre/FiatToken.json.

• cache (Optional[dict]) –

  Use this cache for cache token detail calls.

  The main purpose is to easily reduce JSON-RPC API call count.

  By default, we use LRU cache of 1024 entries.

  Set to None to disable the cache.

  Instance of cachetools.Cache. See cachetools documentation for details.

• chain_id (int) –

  Chain id hint for the cache.

  If not given do eth_chainId RPC call to figure out.

• cause_diagnostics_message (Optional[str]) – Log in Python logging subsystem why this fetch was done to debug RPC overuse.

Returns

Sanitised token info

Return type

eth_defi.token.TokenDetails

get_chain_known_quote_tokens(chain_id)

Get all good quote tokens on chain.

Parameters

chain_id (int) –

Return type

set[eth_defi.token.TokenDetails]

get_chain_stablecoins(chain_id)

Get all good known stablecoins on a chain.

Raises

AssertionError – Chain has zero stablecoins configured

Parameters

chain_id (int) –

Return type

set[eth_defi.token.TokenDetails]

get_erc20_contract(web3, address, contract_name='ERC20MockDecimals.json')

Wrap address as ERC-20 standard interface.

Parameters

• web3 (web3.main.Web3) –

• address (eth_typing.evm.HexAddress) –

Return type

web3.contract.contract.Contract

get_weth_contract(web3, name='1delta/IWETH9.json')

Get WETH9 contract for the chain

• See WETH9 contract

• WETH9 is different contract with different functions on different chain

Parameters

• web3 (web3.main.Web3) – Web3 instance

• name (str) – Alternative implementation.

Returns

WETH token details

Return type

web3.contract.contract.Contract

reset_default_token_cache()

Purge the cached token data.

See DEFAULT_TOKEN_CACHE