vault.historical

Documentation for eth_defi.vault.historical Python module.

Read historical state of vaults.

  • Use multicall to get data points for multiple vaults once

  • Include

    • Share price

    • TVL

    • Fees

See VaultHistoricalReadMulticaller for usage.

Module Attributes

DEFAULT_BLACK_LIST

List of contracts we cannot scan.

Functions

pformat_scan_result(self)

Format the result as a string.

scan_historical_prices_to_parquet(...[, ...])

Scan all historical vault share prices of vaults and save them in to Parquet file.

Classes

ParquetScanResult

Result of generating historical prices Parquet file.

VaultHistoricalReadMulticaller

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

Exceptions

VaultReadNotSupported

Vault cannot be read due to misconfiguration somewhere.
DEFAULT_BLACK_LIST = []

List of contracts we cannot scan. These will bomb out with out of gas. See Mantle issues.

class ParquetScanResult

Bases: TypedDict

Result of generating historical prices Parquet file.

__init__(*args, **kwargs)
__new__(**kwargs)
clear() → None.  Remove all items from D.
copy() → a shallow copy of D
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() → 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.

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() → an object providing a view on D's values
pformat_scan_result(self)

Format the result as a string.

Return type

str

exception VaultReadNotSupported

Bases: Exception

Vault cannot be read due to misconfiguration somewhere.

__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 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

__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
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
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
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
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]

scan_historical_prices_to_parquet(output_fname, web3, web3factory, vaults, token_cache, start_block=None, end_block=None, step=None, chunk_size=1024, compression='zstd', max_workers=8, require_multicall_result=False, frequency='1d', reader_states=None, hypersync_client=None, timestamp_cache_file=PosixPath('/home/runner/.tradingstrategy/block-timestamp'))

Scan all historical vault share prices of vaults and save them in to Parquet file.

  • Write historical prices to a Parquet file

  • Multiprocess-boosted

  • The same Parquet file can contain data from multiple chains

Parameters

  • output_fname (pathlib.Path) –

    Path to a destination Parquet file.

    If the file exists, all entries for the current chain are deleted and rewritten.

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

  • web3factory (eth_defi.event_reader.web3factory.Web3Factory) – Creation of connections in subprocess

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

    Vaults of which historical price we scan.

    All vaults must have their first_seen_at_block attribute set to increase scan performance.

  • start_block –

    First block to scan.

    Leave empty to autodetect

  • end_block –

    Last block to scan.

    Leave empty to autodetect.

  • step_duration –

    What is the historical step size (1 day).

    Will be automatically attmpeted to map to a block time.

  • step – What is the step is in number of blocks.

  • chunk_size – How many rows to write to the Parquet file in one buffer.

  • max_workers – Number of subprocesses to use for multicall

  • hypersync_client – Speed up the discovery of timestamps

  • token_cache (eth_defi.token.TokenDiskCache) –

  • frequency (Literal['1d', '1h']) –

  • reader_states (dict[eth_defi.vault.base.VaultSpec, dict] | None) –

Returns

Scan report.

Return type

eth_defi.vault.historical.ParquetScanResult