feed.stablecoin_rate

Documentation for eth_defi.feed.stablecoin_rate Python module.

Stablecoin rate refresh and depeg lookups.

This module maintains mutable rate metadata in eth_defi/data/stablecoins/*.yaml. It fetches stablecoin prices from the CoinGecko simple price API, stores the latest USD rate and fetch timestamp, and marks a sticky depegged_at timestamp when a token trades below the configured threshold in its inferred peg currency.

Vault metric calculation should use StablecoinRateFeeder instead of reading YAML files directly. The feeder owns the in-process lookup caches used to resolve a vault denomination token to stablecoin rate metadata.

See CoinGecko simple price documentation.

Functions

apply_coingecko_mapping_file(data_dir, ...)	Apply explicit CoinGecko id mappings to stablecoin YAML files.
build_depegged_stablecoin_lookups([data_dir])	Build contract and unambiguous symbol lookups for depegged stablecoins.
build_stablecoin_rate_lookups([data_dir])	Build contract and unambiguous symbol lookups for all known rate data.
extract_coingecko_id(url)	Extract a CoinGecko coin id from a human-readable CoinGecko URL.
fetch_stablecoin_rates(targets[, timeout, ...])	Fetch CoinGecko prices for due stablecoin targets.
iter_stablecoin_rate_targets([data_dir])	Iterate stablecoin YAML entries that can participate in rate refreshes.
refresh_stablecoin_rates([data_dir, now_, ...])	Refresh stablecoin rates and persist YAML metadata updates.
resolve_coingecko_metadata(entry)	Resolve the CoinGecko id, page link, and id source for one YAML entry.

Classes

DenominationTokenRate	Rate data section exported for a vault denomination token.
StablecoinRateFeeder	Cached stablecoin rate/depeg lookup helper for vault metrics.
StablecoinRateRefreshSummary	Counters for one stablecoin rate refresh run.
StablecoinRateTarget	One stablecoin YAML entry that may be refreshed from CoinGecko.

class DenominationTokenRate

Bases: object

Rate data section exported for a vault denomination token.

__init__(coingecko_id, usd_rate, usd_rate_fetched_at, usd_rate_source)

Parameters

• coingecko_id (Optional[str]) –

• usd_rate (Optional[float]) –

• usd_rate_fetched_at (Optional[datetime.datetime]) –

• usd_rate_source (Optional[str]) –

Return type

None

class StablecoinRateFeeder

Bases: object

Cached stablecoin rate/depeg lookup helper for vault metrics.

__init__(data_dir=PosixPath('/home/runner/work/web3-ethereum-defi/web3-ethereum-defi/eth_defi/data/stablecoins'))

Parameters

data_dir (pathlib.Path) –

Return type

None

property depegged_contracts: set[tuple[int, str]]

Set of (chain_id, lowercased_address) keys for depegged tokens (built lazily, cached).

property depegged_symbols: set[str]

Set of unambiguously depegged normalised symbols (built lazily, cached).

get_denomination_token_rate_section(chain_id, address, symbol)

Resolve a vault denomination token to exported rate metadata.

Parameters

• chain_id (Optional[int]) –

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

• symbol (Optional[str]) –

Return type

eth_defi.feed.stablecoin_rate.DenominationTokenRate

is_depegged_stablecoin_token(chain_id, address, symbol)

Return True when a denomination token is marked depegged.

Parameters

• chain_id (Optional[int]) –

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

• symbol (Optional[str]) –

Return type

bool

class StablecoinRateRefreshSummary

Bases: object

Counters for one stablecoin rate refresh run.

__init__(files_scanned=0, entries_seen=0, due_count=0, skipped_attempted_today_count=0, skipped_failed_today_count=0, skipped_succeeded_today_count=0, rates_fetched=0, files_updated=0, depegged_count=0, unactionable_depegged_count=0, skipped_missing_coingecko=0, skipped_unknown_peg=0, failed_count=0)

Parameters

• files_scanned (int) –

• entries_seen (int) –

• due_count (int) –

• skipped_attempted_today_count (int) –

• skipped_failed_today_count (int) –

• skipped_succeeded_today_count (int) –

• rates_fetched (int) –

• files_updated (int) –

• depegged_count (int) –

• unactionable_depegged_count (int) –

• skipped_missing_coingecko (int) –

• skipped_unknown_peg (int) –

• failed_count (int) –

Return type

None

class StablecoinRateTarget

Bases: object

One stablecoin YAML entry that may be refreshed from CoinGecko.

__init__(yaml_path, entry_index, slug, symbol, category, name, coingecko_id, coingecko_link, coingecko_id_source, coingecko_id_verified_at, peg_currency, usd_rate, usd_rate_fetched_at, usd_rate_updated_at, peg_rate, peg_rate_currency, rate_fetch_failed_at, rate_fetch_failed_reason, depegged_at, contract_addresses, ambiguous_symbol=False)

Parameters

• yaml_path (pathlib.Path) –

• entry_index (Optional[int]) –

• slug (str) –

• symbol (str) –

• category (str) –

• name (str) –

• coingecko_id (Optional[str]) –

• coingecko_link (Optional[str]) –

• coingecko_id_source (Optional[str]) –

• coingecko_id_verified_at (Optional[datetime.datetime]) –

• peg_currency (Optional[str]) –

• usd_rate (Optional[float]) –

• usd_rate_fetched_at (Optional[datetime.datetime]) –

• usd_rate_updated_at (Optional[datetime.datetime]) –

• peg_rate (Optional[float]) –

• peg_rate_currency (Optional[str]) –

• rate_fetch_failed_at (Optional[datetime.datetime]) –

• rate_fetch_failed_reason (Optional[str]) –

• depegged_at (Optional[datetime.datetime]) –

• contract_addresses (list[tuple[int, str]]) –

• ambiguous_symbol (bool) –

Return type

None

apply_coingecko_mapping_file(data_dir, mapping_path, progress_bar=False)

Apply explicit CoinGecko id mappings to stablecoin YAML files.

The mapping JSON shape is intentionally simple:

{
  "usdx": {
    "Kava USDX": {
      "coingecko_id": "usdx",
      "coingecko_link": "https://www.coingecko.com/en/coins/usdx"
    }
  }
}

Top-level keys are stablecoin slugs. For multi-entry files, the nested key can be the entry name or a numeric entry index encoded as a string. Standard files may use "default".

Parameters

• data_dir (pathlib.Path) –

• mapping_path (pathlib.Path) –

• progress_bar (bool) –

Return type

int

build_depegged_stablecoin_lookups(data_dir=PosixPath('/home/runner/work/web3-ethereum-defi/web3-ethereum-defi/eth_defi/data/stablecoins'))

Build contract and unambiguous symbol lookups for depegged stablecoins.

Determine which denomination tokens should be treated as depegged for vault blacklisting. Two independent lookups are returned:

• A set of (chain_id, lowercased_address) contract keys for every entry that carries a depegged_at marker, regardless of whether the entry lives in a single- or multi-entry YAML file. This is the precise path: it pins the dead token by address and never blacklists an unrelated token that merely reuses the same ticker.

• A set of normalised symbols that are unambiguously depegged. A symbol is only eligible here when a single flat (single-entry) YAML file owns it and that file is marked depegged_at.

Symbol matching is deliberately not used for multi-entry tokens. Tickers such as USDX are reused by several unrelated tokens in the wild — e.g. the dead Stables Labs USDX (0xf3527ef8…) versus the live Axis USD (USDx) on Plasma — so matching by ticker would blacklist healthy vaults. Multi-entry depegged tokens must therefore be pinned by contract_addresses in their YAML entry. A warning is logged for any depegged entry that has no contract address and is not covered by an unambiguous symbol, because such a depeg silently fails to blacklist anything (this is the gap that previously let USDX-denominated vaults stay listed).

Parameters

data_dir (pathlib.Path) – Directory of stablecoin metadata YAML files.

Returns

Tuple of (depegged_contract_keys, depegged_symbols).

Return type

tuple[set[tuple[int, str]], set[str]]

build_stablecoin_rate_lookups(data_dir=PosixPath('/home/runner/work/web3-ethereum-defi/web3-ethereum-defi/eth_defi/data/stablecoins'))

Build contract and unambiguous symbol lookups for all known rate data.

Parameters

data_dir (pathlib.Path) –

Return type

tuple[dict[tuple[int, str], eth_defi.feed.stablecoin_rate.DenominationTokenRate], dict[str, eth_defi.feed.stablecoin_rate.DenominationTokenRate]]

extract_coingecko_id(url)

Extract a CoinGecko coin id from a human-readable CoinGecko URL.

Parameters

url (Optional[str]) –

Return type

Optional[str]

fetch_stablecoin_rates(targets, timeout=20.0, progress_bar=False)

Fetch CoinGecko prices for due stablecoin targets.

Parameters

• targets (Sequence[eth_defi.feed.stablecoin_rate.StablecoinRateTarget]) – Due targets with resolved CoinGecko ids.

• timeout (float) – HTTP request timeout in seconds.

• progress_bar (bool) – Show a tqdm progress bar for CoinGecko request batches.

Returns

CoinGecko response keyed by coin id.

Return type

dict[str, dict[str, Any]]

iter_stablecoin_rate_targets(data_dir=PosixPath('/home/runner/work/web3-ethereum-defi/web3-ethereum-defi/eth_defi/data/stablecoins'))

Iterate stablecoin YAML entries that can participate in rate refreshes.

Parameters

data_dir (pathlib.Path) –

Return type

Iterator[eth_defi.feed.stablecoin_rate.StablecoinRateTarget]

refresh_stablecoin_rates(data_dir=PosixPath('/home/runner/work/web3-ethereum-defi/web3-ethereum-defi/eth_defi/data/stablecoins'), now_=None, force=False, timeout=20.0, progress_bar=False)

Refresh stablecoin rates and persist YAML metadata updates.

This function is intentionally entry tolerant: missing CoinGecko ids and missing prices are written to per-entry failure fields instead of aborting the whole run. HTTP-level failures are recorded for all due targets in the batch, then returned in the summary. Depeg decisions are made only when the token’s peg currency can be inferred and CoinGecko returns that currency.

Parameters

• data_dir (pathlib.Path) –

• now_ (Optional[datetime.datetime]) –

• force (bool) –

• timeout (float) –

• progress_bar (bool) –

Return type

eth_defi.feed.stablecoin_rate.StablecoinRateRefreshSummary

resolve_coingecko_metadata(entry)

Resolve the CoinGecko id, page link, and id source for one YAML entry.

Parameters

entry (dict[str, Any]) –

Return type

tuple[str | None, str | None, str | None]