ForgeYieldsHistoricalReader

Documentation for eth_defi.erc_4626.vault_protocol.forgeyields.vault.ForgeYieldsHistoricalReader Python class.

class ForgeYieldsHistoricalReader

Bases: eth_defi.erc_4626.vault.ERC4626HistoricalReader

Read ForgeYields vault data — share price from on-chain, TVL from offchain API.

The on-chain totalAssets() reverts on the TokenGateway and convertToAssets(totalSupply()) returns only the gateway residual.

For rows near the chain head (within 24 hours of now), this reader writes the current denomination-token TVL from the ForgeYields API into total_assets. Older rows get total_assets=None — the backfill script fills those from the API’s 30-day historyReports.

The current TVL is also fed into the reader state so adaptive polling does not degrade to faded/tiny cadence.

Attributes summary

NEAR_HEAD_WINDOW

address

first_block

one_raw_share

Methods summary

__init__(vault, stateful)

construct_core_erc_4626_multicall()

Polling endpoints defined in ERC-4626 spec.

construct_multicalls()

Get the onchain calls that are needed to read the share price.

dictify_multicall_results(block_number, ...)

Convert batch of multicalls made for this vault to more digestible dict.

get_warmup_calls()

Yield (function_name, callable, contract_call) tuples for warmup testing.

process_core_erc_4626_result(call_by_name)

Decode common ERC-4626 calls.

process_result(block_number, timestamp, ...)

Process the result of mult

should_skip_call(function_name)

Check if a specific function call should be skipped.
__init__(vault, stateful)
Parameters
construct_core_erc_4626_multicall()

Polling endpoints defined in ERC-4626 spec.

  • Does not include fee calls which do not have standard

Return type

Iterable[eth_defi.event_reader.multicall_batcher.EncodedCall]

construct_multicalls()

Get the onchain calls that are needed to read the share price.

Return type

Iterable[eth_defi.event_reader.multicall_batcher.EncodedCall]

dictify_multicall_results(block_number, call_results, allow_failure=True)

Convert batch of multicalls made for this vault to more digestible dict.

  • Assert that all multicalls succeed

Returns

Dictionary where each multicall is keyed by its EncodedCall.extra_data["function"]

Parameters
Return type

dict[str, eth_defi.event_reader.multicall_batcher.EncodedCallResult]

get_warmup_calls()

Yield (function_name, callable, contract_call) tuples for warmup testing.

Each callable should execute a single contract call. If it raises, the function is marked as broken.

The optional contract_call is used for gas estimation to detect expensive calls before executing them. If provided, calls using excessive gas (>1M gas) will be marked as broken without execution.

Override in subclasses to add protocol-specific calls.

Returns

Iterable of (function_name, test_callable, contract_call) tuples. contract_call may be None if gas estimation is not needed.

Return type

Iterable[tuple[str, callable, Any]]

process_core_erc_4626_result(call_by_name)

Decode common ERC-4626 calls.

Parameters

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

Return type

tuple

process_result(block_number, timestamp, call_results)

Process the result of mult

  • Calls are created in construct_multicalls()

  • This method combines result of this calls to a easy to manage historical record VaultHistoricalRead

Parameters
Return type

eth_defi.vault.base.VaultHistoricalRead

should_skip_call(function_name)

Check if a specific function call should be skipped.

Uses the reader state’s call_status map if available.

Parameters

function_name (str) – The function name to check

Returns

True if the call should be skipped

Return type

bool