feed.database

Documentation for eth_defi.feed.database Python module.

DuckDB persistence for vault post tracking.

Functions

resolve_feed_database_path()

Resolve the vault post feed DuckDB database path.

Classes

`CollectedPost`	A single normalised post ready for database insertion.
`VaultPostDatabase`	DuckDB database for tracked sources and collected posts.

DEFAULT_VAULT_POST_DATABASE = PosixPath('/home/runner/.tradingstrategy/vaults/vault-post-database.duckdb'): Default DuckDB path for collected vault posts.

resolve_feed_database_path()

Resolve the vault post feed DuckDB database path.

Mirrors the resolution used by the post scanner (scripts/erc-4626/scan-vault-posts.py) so the JSON export reads the same database the feed collector writes. Keeping the resolver next to DEFAULT_VAULT_POST_DATABASE avoids each caller repeating the same environment lookup and path expansion logic.

The FEED_DB_PATH override takes precedence, falling back to the DB_PATH variable consumed by the post scanner, then the default path.

Returns: Path from FEED_DB_PATH, then DB_PATH, then the default vault post database path.
Return type: pathlib.Path

class CollectedPost

Bases: object

A single normalised post ready for database insertion.

external_post_id: str: Stable external identifier derived from the feed entry or a deterministic fallback.

title: Optional[str]: Entry title when the feed provides one.

post_url: Optional[str]: Canonical link to the source post when available.

published_at: Optional[datetime.datetime]: Original post publication timestamp in naive UTC.

fetched_at: datetime.datetime: Timestamp when the collector fetched this post in naive UTC.

short_description: str

Short preview text stored alongside the post.

Capped at 200 characters for compact listings; this is not the complete post body. See full_text for the full content.

full_text: str

Best available full text extracted from the feed entry.

For X/Twitter this is the complete note tweet body for tweets longer than 280 characters, populated via eth_defi.feed.twitter_api._extract_full_tweet_text(). See short_description for the truncated preview.

ai_summary: Optional[str]: Optional future AI-generated summary, null in the current version.

raw_payload: Optional[str]: JSON-serialised raw payload from the source API (e.g. full tweet object from X API).

__init__(external_post_id, title, post_url, published_at, fetched_at, short_description, full_text, ai_summary=None, raw_payload=None)

Parameters

external_post_id (str) –
title (Optional[str]) –
post_url (Optional[str]) –
published_at (Optional[datetime.datetime]) –
fetched_at (datetime.datetime) –
short_description (str) –
full_text (str) –
ai_summary (Optional[str]) –
raw_payload (Optional[str]) –

Return type

None

class VaultPostDatabase

Bases: object

DuckDB database for tracked sources and collected posts.

__init__(path)

Parameters: path (pathlib.Path) –

close()

Close the database connection.

Return type: None

save()

Force a checkpoint.

Return type: None

upsert_tracked_source(source)

Insert or update one tracked source and return its source ID.

Parameters: source (eth_defi.feed.sources.TrackedPostSource) –
Return type: int

upsert_tracked_sources(sources)

Insert or update tracked sources and return source IDs by logical key.

Parameters: sources (Iterable[eth_defi.feed.sources.TrackedPostSource]) –
Return type: dict[tuple[str, str, str, str], int]

mark_source_success(source_id, *, checked_at=None, last_post_published_at=None)

Update sync state for a successful source fetch.

Parameters

source_id (int) –
checked_at (Optional[datetime.datetime]) –
last_post_published_at (Optional[datetime.datetime]) –

Return type

None

mark_source_failure(source_id, error, *, checked_at=None)

Update sync state for a failed or skipped source fetch.

Parameters

source_id (int) –
error (str) –
checked_at (Optional[datetime.datetime]) –

Return type

None

insert_posts(source_id, posts)

Insert posts for a source and return the number of new rows.

Parameters

source_id (int) –
posts (Iterable[eth_defi.feed.database.CollectedPost]) –

Return type

int

prune_posts(max_post_age_days)

Delete posts older than the configured retention period.

Parameters: max_post_age_days (int) –
Return type: int

get_sync_state(key)

Read a value from the feed_sync_state table.

Parameters: key (str) –
Return type: Optional[str]

set_sync_state(key, value)

Write a value to the feed_sync_state table.

Parameters

key (str) –
value (str) –

Return type

None

get_known_post_ids(source_id=None)

Return all known external_post_id values, optionally filtered by source.

Parameters: source_id (Optional[int]) –
Return type: set[str]

get_source_last_post_timestamps(source_ids)

Return the stored last_post_published_at for the given source IDs.

Used to gate backfill fallbacks: a source whose stored timestamp is not None has already been seen before and does not need a fallback individual timeline read.

Parameters: source_ids (Iterable[int]) – Iterable of numeric source IDs to look up.
Returns: Mapping of source_id → last_post_published_at (None when the column has never been set for that row).
Return type: dict[int, datetime.datetime | None]

get_tracked_sources_df()

Return tracked source rows for diagnostics.

Return type: pandas.DataFrame

get_posts_df()

Return stored posts for diagnostics.

Return type: pandas.DataFrame

fetch_recent_posts_by_feeder(feeder_ids, max_per_feeder=10)

Fetch the most recent posts for each feeder across all source types.

Joins tracked_sources and posts on source_id, ranks posts per feeder by COALESCE(published_at, fetched_at) DESC, and returns the max_per_feeder newest posts per feeder.

Parameters

feeder_ids (Iterable[str]) – Iterable of feeder-id slugs to look up.
max_per_feeder (int) – Maximum number of posts to return per feeder.

Returns

Dict mapping feeder_id to a list of post dicts with keys title, short_description, full_text, post_url, source_type, published_at (always set via COALESCE fallback to fetched_at). Lists are ordered newest-first.

Return type

dict[str, list[dict]]