VaultHistoricalReadMulticaller

Documentation for eth_defi.vault.historical.VaultHistoricalReadMulticaller Python class.

class VaultHistoricalReadMulticaller

Bases: object

Read historical data from multiple vaults using multicall and archive node polling.

Parameters

supported_quote_tokens – Allows us to validate vaults against list of supported tokens

Methods summary

__init__(web3factory[, ...])	param supported_quote_tokens
generate_vault_historical_calls(readers[, ...])	Generate multicalls for each vault to read its state at any block.
prepare_readers(vaults[, stateful, saved_states])	Create readrs for vaults.
read_historical(vaults, start_block, ...[, ...])	Create an iterable that extracts vault record from RPC.
save_reader_state()	Save the state of all readers.
validate_vaults(vaults)	Check that we can read these vaults.

__init__(web3factory, supported_quote_tokens=set[eth_defi.token.TokenDetails] | None, max_workers=8, token_cache=None, require_multicall_result=False, hypersync_client=None, timestamp_cache_file=PosixPath('/home/runner/.tradingstrategy/block-timestamp'))

Parameters

• supported_quote_tokens – Allows us to validate vaults against list of supported tokens

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

• hypersync_client (hypersync.HypersyncClient | None) –

• timestamp_cache_file (pathlib.Path) –

validate_vaults(vaults)

Check that we can read these vaults.

• Validate that we know how to read vaults

Raises

VaultReadNotSupported – In the case we cannot read some of the vaults

Parameters

vaults (list[eth_defi.vault.base.VaultBase]) –

prepare_readers(vaults, stateful=False, saved_states=None)

Create readrs for vaults.

Parameters

• vaults (list[eth_defi.vault.base.VaultBase]) –

• saved_states (dict[eth_defi.erc_4626.vault.VaultReaderState, dict] | None) –

Return type

dict[eth_typing.evm.HexAddress, eth_defi.vault.base.VaultHistoricalReader]

generate_vault_historical_calls(readers, display_progress=True)

Generate multicalls for each vault to read its state at any block.

Parameters

• readers (dict[eth_typing.evm.HexAddress, eth_defi.vault.base.VaultHistoricalReader]) –

• display_progress (bool) –

Return type

Iterable[tuple[eth_defi.event_reader.multicall_batcher.EncodedCall, eth_defi.event_reader.multicall_batcher.BatchCallState]]

read_historical(vaults, start_block, end_block, step, reader_func=<function read_multicall_historical>, saved_states=None)

Create an iterable that extracts vault record from RPC.

Parameters

• start_block (int) –

  The first block to read from.

  Set to None to get from the saved state what we have not yet read.

• reader_func (Callable) – Either read_multicall_historical or read_multicall_historical_stateful

• vaults (list[eth_defi.vault.base.VaultBase]) –

• end_block (int) –

• step (int) –

• saved_states (dict[eth_defi.erc_4626.vault.VaultReaderState, dict] | None) –

Returns

Unordered results

Return type

Iterable[eth_defi.vault.base.VaultHistoricalRead]

save_reader_state()

Save the state of all readers.

Returns

Dictionary keyed by the vault spce

Return type

dict[eth_defi.vault.base.VaultSpec, dict]