feed.twitter_api

Documentation for eth_defi.feed.twitter_api Python module.

X/Twitter API v2 integration for tweet collection.

Uses tweepy to read tweets from X lists and individual user timelines. Provides a user metadata cache to avoid repeated handle-to-ID lookups.

Reading requires only a bearer token (TWITTER_BEARER_TOKEN). List membership writes require full OAuth 1.0a credentials.

Note tweets and full text

The X API v2 caps the text field of every tweet at 280 characters. Tweets longer than this — so called note tweets, available to premium and verified accounts — are returned with a truncated text value (ending in an ellipsis) and the complete body, up to ~25,000 characters, in a separate note_tweet object.

The note_tweet object is only present when the note_tweet field is explicitly requested via the tweet_fields request parameter. Both fetch_tweets_from_x_list() and fetch_user_tweets() request it, and _extract_full_tweet_text() prefers note_tweet.text over the truncated text so the full body is preserved.

The full body lands in full_text, while short_description keeps a 200-character preview for compact listings. Both are carried through to the vault JSON bundle as full_text and snippet respectively.

See the X API v2 note tweet documentation: https://docs.x.com/x-api/fundamentals/note-tweets

Functions

compute_handles_hash(handles)	Compute a deterministic hash of a sorted set of Twitter handles.
fetch_tweets_from_x_list(list_id, ...[, ...])	Fetch tweets from an X list timeline, grouped by author user_id.
fetch_user_tweets(user_id, bearer_token, ...)	Fetch recent tweets from a single user timeline.
resolve_twitter_handles(handles, ...[, ...])	Resolve Twitter handles to user IDs, using the cache for known entries.
resolve_x_list_id_by_name(list_name, ...)	Resolve an X list ID by exact list name for the authenticated user.
sync_x_list_members(list_id, ...[, ...])	Sync X list membership with the provided Twitter handles.

Classes

CachedTwitterUser	Cached user metadata from a handle-to-ID lookup.
TwitterUserCache	File-backed cache of Twitter handle-to-user-ID mappings.

Exceptions

XApiError	Raised when the X API returns an unrecoverable error.
XRateLimitError	Raised when the X API rate-limits a list sync operation.

class CachedTwitterUser

Bases: object

Cached user metadata from a handle-to-ID lookup.

__init__(user_id, name, handle, fetched_at)

Parameters

• user_id (str) –

• name (str) –

• handle (str) –

• fetched_at (str) –

Return type

None

class TwitterUserCache

Bases: object

File-backed cache of Twitter handle-to-user-ID mappings.

Stored at ~/.tradingstrategy/vaults/feeds/twitter-users.json.

__init__(path=None)

Parameters

path (Optional[pathlib.Path]) –

get(handle)

Look up a cached user entry by handle (case-insensitive).

Parameters

handle (str) –

Return type

Optional[eth_defi.feed.twitter_api.CachedTwitterUser]

get_all_user_ids()

Return a mapping of handle → user_id for all cached entries.

Return type

dict[str, str]

get_by_user_id(user_id)

Look up a cached user entry by numeric user ID.

Parameters

user_id (str) –

Return type

Optional[eth_defi.feed.twitter_api.CachedTwitterUser]

is_stale(handle, max_age_days=30)

Check whether a cache entry is missing or older than max_age_days.

Parameters

• handle (str) –

• max_age_days (int) –

Return type

bool

put(handle, user_id, name)

Store or update a cache entry.

Parameters

• handle (str) –

• user_id (str) –

• name (str) –

Return type

None

save()

Persist the cache to disk.

Return type

None

exception XApiError

Bases: RuntimeError

Raised when the X API returns an unrecoverable error.

__init__(*args, **kwargs)

__new__(**kwargs)

add_note(note, /)

Add a note to the exception

with_traceback(tb, /)

Set self.__traceback__ to tb and return self.

exception XRateLimitError

Bases: eth_defi.feed.twitter_api.XApiError

Raised when the X API rate-limits a list sync operation.

__init__(*args, **kwargs)

__new__(**kwargs)

add_note(note, /)

Add a note to the exception

with_traceback(tb, /)

Set self.__traceback__ to tb and return self.

compute_handles_hash(handles)

Compute a deterministic hash of a sorted set of Twitter handles.

Parameters

handles (list[str]) –

Return type

str

fetch_tweets_from_x_list(list_id, bearer_token, user_cache, *, max_tweets=100, known_post_ids=None)

Fetch tweets from an X list timeline, grouped by author user_id.

Uses cursor-based pagination. Stops when encountering tweets already in the database (checked against known_post_ids) or when max_tweets is reached.

Returns

Mapping of user_id → list of CollectedPost.

Parameters

• list_id (str) –

• bearer_token (str) –

• user_cache (eth_defi.feed.twitter_api.TwitterUserCache) –

• max_tweets (int) –

• known_post_ids (Optional[set[str]]) –

Return type

dict[str, list[eth_defi.feed.database.CollectedPost]]

fetch_user_tweets(user_id, bearer_token, author_handle, *, max_tweets=10, since=None)

Fetch recent tweets from a single user timeline.

GET /2/users/:id/tweets supports start_time for incremental reads. Used when LIMIT is set or for backfilling newly added members.

Parameters

• user_id (str) –

• bearer_token (str) –

• author_handle (str) –

• max_tweets (int) –

• since (Optional[datetime.datetime]) –

Return type

list[eth_defi.feed.database.CollectedPost]

resolve_twitter_handles(handles, bearer_token, cache, *, max_age_days=30)

Resolve Twitter handles to user IDs, using the cache for known entries.

Only looks up handles that are missing from the cache or stale. Returns a mapping of handle → user_id.

The X API get_users response may include an errors list describing why specific handles could not be resolved (suspended, not found, renamed, etc.). These reasons are logged so operators can take corrective action on the corresponding YAML files.

Parameters

• handles (list[str]) –

• bearer_token (str) –

• cache (eth_defi.feed.twitter_api.TwitterUserCache) –

• max_age_days (int) –

Return type

dict[str, str]

resolve_x_list_id_by_name(list_name, consumer_key, consumer_secret, access_token, access_token_secret)

Resolve an X list ID by exact list name for the authenticated user.

Uses OAuth 1.0a user context to read the current X user and enumerate the lists owned by that account. This is intended for operator scripts where the production list owner is also the OAuth user.

Parameters

• list_name (str) – Exact X list name to find, e.g. Best builders in DeFi.

• consumer_key (str) – OAuth 1.0a consumer key.

• consumer_secret (str) – OAuth 1.0a consumer secret.

• access_token (str) – OAuth 1.0a user access token.

• access_token_secret (str) – OAuth 1.0a user access token secret.

Returns

Numeric X list ID as a string.

Raises

XApiError – If the current user cannot be read, or if zero or multiple owned lists match the requested name.

Return type

str

See the X API v2 list lookup endpoints: https://docs.x.com/x-api/lists/list-lookup

sync_x_list_members(list_id, twitter_handles, consumer_key, consumer_secret, access_token, access_token_secret, user_cache, bearer_token, db, *, add_delay_seconds=1.0, rate_limit_sleep_max_seconds=1200.0)

Sync X list membership with the provided Twitter handles.

Only runs when the set of handles has changed (hash-based detection using feed_sync_state table). Returns the number of members added.

Requires full OAuth 1.0a credentials for list write operations.

Parameters

• list_id (str) –

• twitter_handles (list[str]) –

• consumer_key (str) –

• consumer_secret (str) –

• access_token (str) –

• access_token_secret (str) –

• user_cache (eth_defi.feed.twitter_api.TwitterUserCache) –

• bearer_token (str) –

• db (eth_defi.feed.database.VaultPostDatabase) –

• add_delay_seconds (float) –

• rate_limit_sleep_max_seconds (float) –

Return type

int