research.sparkline_export

Documentation for eth_defi.research.sparkline_export Python module.

Bounded sparkline preparation, rendering and publication orchestration.

Module Attributes

`SPARKLINE_STATE_SCHEMA_VERSION`	Version of the persisted state envelope.
`SPARKLINE_LOW_TVL_INTERVAL`	Maximum age of a low-TVL successful completion.
`SPARKLINE_STATE_RETENTION_DAYS`	Retain state entries for recently observed vaults.
`SPARKLINE_BATCH_SIZE`	Bound image bytes and worker result queues.
`SHA256_HEX_LENGTH`	Number of hexadecimal characters in a SHA-256 digest.
`SPARKLINE_REQUIRED_ASSETS`	Required assets, identified by file extension and optional variant.
`SPARKLINE_FORMAT_COUNT`	Number of output assets required for a complete vault publication.
`SUPPORTED_SPARKLINE_FAMILIES`	Supported image families.

Functions

`calculate_sparkline_input_digest`(sparkline_data)	Calculate a canonical renderer/input digest before rendering.
`get_included_vault_ids`(vault_db, prices_df)	Select supported vaults with finite TVL and two valid price observations.
`latest_total_assets_by_id`(prices_df, prepared)	Resolve the latest finite TVL through each chart's final UTC day.
`load_sparkline_price_data`(prices_path)	Load and validate the canonical daily crypto sparkline Parquet.
`load_sparkline_state`(path[, now])	Load and strictly validate the persistent sparkline state.
`make_empty_sparkline_state`(now)	Create an empty versioned sparkline state envelope.
`prepare_vault_sparklines`(prices_df, included_ids)	Prepare daily sparse observations for all eligible vaults.
`render_sparklines`(vault_data, max_workers[, ...])	Render one bounded batch using the selected Joblib backend.
`render_vault_sparklines`(vault_id, sparkline_data)	Render one vault's deterministic SVG, square PNG and table PNG.
`run_sparkline_export`(*[, data_dir, ...])	Run bounded deterministic sparkline rendering and publication.
`save_sparkline_state`(state, path[, now])	Prune and atomically persist one sparkline state envelope.
`upload_sparkline`(s3_client, bucket_name, ...)	Gzip and publish one image, skipping an unchanged R2 object.
`upload_sparklines`(s3_client, bucket_name, ...)	Upload a bounded image list using the configured thread count.

Classes

`RenderData`	Rendered image payload for one vault and one output format.
`SparklineExportResult`	Counters and terminal status from one sparkline export run.
`SparklineVaultWorkResult`	One worker's preparation, cadence and optional render result.

SPARKLINE_STATE_SCHEMA_VERSION = 1: Version of the persisted state envelope.

SPARKLINE_LOW_TVL_INTERVAL = datetime.timedelta(days=3): Maximum age of a low-TVL successful completion.

SPARKLINE_STATE_RETENTION_DAYS = 90: Retain state entries for recently observed vaults.

SPARKLINE_BATCH_SIZE = 100: Bound image bytes and worker result queues.

SHA256_HEX_LENGTH = 64: Number of hexadecimal characters in a SHA-256 digest.

SPARKLINE_REQUIRED_ASSETS = frozenset({('png', None), ('png', 'table'), ('svg', None)}): Required assets, identified by file extension and optional variant.

SPARKLINE_FORMAT_COUNT = 3: Number of output assets required for a complete vault publication.

SUPPORTED_SPARKLINE_FAMILIES = frozenset({DenominationFamily.btc, DenominationFamily.eth, DenominationFamily.stablecoin}): Supported image families.

class RenderData

Bases: eth_defi.research.sparkline_export._RenderVariant

Rendered image payload for one vault and one output format.

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

class SparklineExportResult

Bases: object

Counters and terminal status from one sparkline export run.

__init__(success, counters)

Parameters

success (bool) –
counters (dict[str, int]) –

Return type

None

class SparklineVaultWorkResult

Bases: object

One worker’s preparation, cadence and optional render result.

vault_id: str: Canonical vault identifier.

status: str: Processing outcome for the parent coordinator.

latest_total_assets: Optional[float]: Latest finite native-unit TVL through the chart’s final day.

classification: Optional[tuple[eth_defi.vault.denomination.DenominationFamily, decimal.Decimal, bool]]: Currency family, native threshold and low-TVL flag when TVL is valid.

digest: Optional[str]: Canonical input digest when rendering was due.

images: Optional[list[eth_defi.research.sparkline_export.RenderData]]: SVG, square PNG and table PNG payloads when rendering succeeded.

error: Optional[str]: Render error when one vault failed.

__init__(vault_id, status, latest_total_assets=None, classification=None, digest=None, images=None, error=None)

Parameters

vault_id (str) –
status (str) –
latest_total_assets (Optional[float]) –
classification (Optional[tuple[eth_defi.vault.denomination.DenominationFamily, decimal.Decimal, bool]]) –
digest (Optional[str]) –
images (Optional[list[eth_defi.research.sparkline_export.RenderData]]) –
error (Optional[str]) –

Return type

None

make_empty_sparkline_state(now)

Create an empty versioned sparkline state envelope.

Parameters: now (datetime.datetime) –
Return type: dict[str, Any]

load_sparkline_state(path, now=None)

Load and strictly validate the persistent sparkline state.

Parameters

path (pathlib.Path) –
now (Optional[datetime.datetime]) –

Return type

dict[str, Any]

save_sparkline_state(state, path, now=None)

Prune and atomically persist one sparkline state envelope.

Parameters

state (dict[str, Any]) –
path (pathlib.Path) –
now (Optional[datetime.datetime]) –

Return type

None

get_included_vault_ids(vault_db, prices_df)

Select supported vaults with finite TVL and two valid price observations.

Eligibility requires distinct valid observation timestamps, even within one day. The metadata supplies the supported denomination family.

Parameters

vault_db (eth_defi.vault.vaultdb.VaultDatabase) – Vault metadata rows used for denomination-family classification.
prices_df (pandas.DataFrame) – Naive-UTC price rows containing id, share_price and total_assets columns.

Returns

Canonical IDs with supported denominations, two valid timestamps and at least one finite TVL observation.

Return type

set[str]

prepare_vault_sparklines(prices_df, included_ids)

Prepare daily sparse observations for all eligible vaults.

Parameters

prices_df (pandas.DataFrame) – Naive-UTC price rows with id, share_price and total_assets.
included_ids (set[str]) – IDs admitted by get_included_vault_ids().

Returns

Prepared chart data and the number of IDs skipped for insufficient finite history.

Return type

tuple[list[tuple[str, eth_defi.research.sparkline.SparklineData]], int]

latest_total_assets_by_id(prices_df, prepared)

Resolve the latest finite TVL through each chart’s final UTC day.

Parameters

prices_df (pandas.DataFrame) – Price rows indexed by naive UTC timestamps with id and total_assets columns.
prepared (list[tuple[str, eth_defi.research.sparkline.SparklineData]]) – Prepared charts whose end dates bound the TVL lookup.

Returns

Mapping from vault ID to the latest finite native-unit TVL, or None when no usable value exists in the chart’s final day.

Return type

dict[str, float | None]

calculate_sparkline_input_digest(sparkline_data)

Calculate a canonical renderer/input digest before rendering.

Parameters: sparkline_data (eth_defi.research.sparkline.SparklineData) –
Return type: str

render_vault_sparklines(vault_id, sparkline_data)

Render one vault’s deterministic SVG, square PNG and table PNG.

Parameters

vault_id (str) – Canonical chain-address vault identifier used in object keys.
sparkline_data (eth_defi.research.sparkline.SparklineData) – Prepared daily observations and fixed chart bounds.

Returns

Exactly one SVG, one legacy square PNG and one native 4:1 table PNG.

Raises

ValueError – If the prepared data cannot be rendered.

Return type

list[eth_defi.research.sparkline_export.RenderData]

render_sparklines(vault_data, max_workers, backend='processes')

Render one bounded batch using the selected Joblib backend.

Rendering stays batch-bounded so only the current batch’s image payloads need to be returned to the coordinator.

Parameters

vault_data (list[tuple[str, eth_defi.research.sparkline.SparklineData]]) – Vault IDs paired with prepared daily chart data.
max_workers (int) – Maximum number of concurrent renderer workers.
backend (str) – Joblib preference, either processes or threads.

Returns

Flattened SVG and PNG payloads for successfully rendered vaults.

Return type

list[eth_defi.research.sparkline_export.RenderData]

upload_sparkline(s3_client, bucket_name, render_data)

Gzip and publish one image, skipping an unchanged R2 object.

Parameters

s3_client (Any) – Thread-safe S3-compatible client.
bucket_name (str) – Destination R2 bucket.
render_data (eth_defi.research.sparkline_export.RenderData) – One deterministic SVG or PNG payload.

Returns

True when an object was uploaded, False when the remote object already matched the source digest.

Return type

bool

upload_sparklines(s3_client, bucket_name, render_data, max_workers)

Upload a bounded image list using the configured thread count.

Each task publishes one vault’s SVG and PNG while retaining the existing conditional upload and partial-failure behaviour.

This compatibility helper retains the standalone script’s previous public API. The production coordinator uploads per-vault asset sets directly so one partial set cannot be mistaken for a complete publication.

Parameters

s3_client (Any) – Authenticated S3-compatible client.
bucket_name (str) – Destination R2 bucket.
render_data (list[eth_defi.research.sparkline_export.RenderData]) – Rendered image payloads.
max_workers (int) – Upload thread count.

Returns

Number of uploaded and unchanged objects.

Return type

tuple[int, int]

load_sparkline_price_data(prices_path)

Load and validate the canonical daily crypto sparkline Parquet.

Parameters: prices_path (pathlib.Path) – Daily crypto-vault Parquet path.
Returns: Stable-sorted price rows with numeric share_price and total_assets columns.
Raises: ValueError – If the file schema or timestamp index is incompatible with export.
Return type: pandas.DataFrame

run_sparkline_export(*, data_dir=None, vault_db_path=None, prices_path=None, state_path=None, max_workers=None, render_workers=None, upload_workers=None, render_backend=None, batch_size=None, force=None, dry_run=False)

Run bounded deterministic sparkline rendering and publication.

Parameters

data_dir (Optional[pathlib.Path]) – Pipeline data directory used to resolve omitted paths.
vault_db_path (Optional[pathlib.Path]) – Vault metadata database path.
prices_path (Optional[pathlib.Path]) – Canonical crypto daily price Parquet path.
state_path (Optional[pathlib.Path]) – Persistent sparkline state path.
max_workers (Optional[int]) – Deprecated compatibility setting for both render and upload workers.
render_workers (Optional[int]) – Process or thread count for bounded rendering.
upload_workers (Optional[int]) – Thread count for R2 publication.
render_backend (Optional[str]) – processes or threads. Defaults to the environment setting or processes.
batch_size (Optional[int]) – Maximum number of vaults whose image payloads coexist in memory. When omitted, SPARKLINE_BATCH_SIZE is read from the environment.
force (Optional[bool]) – Bypass cadence, retry backoff and local digest skips.
dry_run (bool) – Skip R2 client creation and uploads while exercising rendering, batching and state updates. Requires force=True and a new scratch state path; an existing or default production state file is rejected.

Returns

Run counters and terminal status.

Return type

eth_defi.research.sparkline_export.SparklineExportResult