Build a Real-Time X Sentiment Pipeline with TwexAPI

Sentiment analysis fails when the data is stale or the score cannot be audited. If a dashboard only says "market sentiment: -0.42" but cannot show which posts, which query, which model version, and when the data was collected, analysts will not trust it.

A better approach is to treat sentiment as a data pipeline: poll recent posts with TwexAPI Advanced Search by Page, deduplicate by tweet_id, store raw JSON, then run a lightweight model first. You can later replace VADER with a transformer or LLM, but the collection, dedupe, scoring, aggregation, and review layers should stay stable.

Pipeline Shape

A real-time sentiment pipeline means turning X posts from query results into auditable sentiment metrics. It usually has five parts:

Query design: Store brand terms, events, language filters, exclusions, and refresh windows in query configuration.
Collection: Call POST /twitter/advanced_search/page with sortBy: "Latest" to collect recent posts.
Storage: Save raw JSON and deduplicate by tweet_id before scoring.
Scoring: Use a lightweight model for most posts and send borderline or high-impact samples to heavier models.
Aggregation and review: Output averages, sample counts, outliers, and model versions by time window.

Step 1: Polling Collection

Polling collection means requesting recent search results on a schedule instead of pretending you have a streaming connection. TwexAPI's POST /twitter/advanced_search/page returns up to 20 tweets per page. The first request omits next_cursor; later requests pass the cursor from the previous response.

The example below uses sortBy: "Latest" and seen_ids to avoid rescoring the same tweet. In production, store seen_ids in a database or cache rather than in memory.

1import requests
2from datetime import datetime, timezone
3
4API_URL = "https://api.twexapi.io/twitter/advanced_search/page"
5TOKEN = "YOUR_TWEXAPI_BEARER_TOKEN"
6
7def search_page(query, next_cursor=None):
8    payload = {
9        "searchTerms": [query],
10        "sortBy": "Latest"
11    }
12
13    if next_cursor:
14        payload["next_cursor"] = next_cursor
15
16    response = requests.post(
17        API_URL,
18        headers={
19            "Authorization": f"Bearer {TOKEN}",
20            "Content-Type": "application/json",
21            "Accept": "application/json",
22        },
23        json=payload,
24        timeout=30,
25    )
26    response.raise_for_status()
27    data = response.json()
28
29    if data.get("code", 200) >= 400:
30        raise RuntimeError(data.get("msg", "TwexAPI returned an error"))
31
32    return data
33
34def collect_recent_tweets(query, seen_ids, max_pages=3):
35    next_cursor = None
36    collected = []
37
38    for page_number in range(max_pages):
39        page = search_page(query, next_cursor)
40
41        for tweet in page.get("data", []):
42            if not tweet:
43                continue
44
45            tweet_id = tweet.get("tweet_id") or tweet.get("id")
46            if not tweet_id or tweet_id in seen_ids:
47                continue
48
49            seen_ids.add(tweet_id)
50            collected.append({
51                "query": query,
52                "fetched_at": datetime.now(timezone.utc).isoformat(),
53                "raw": tweet,
54            })
55
56        if not page.get("has_next_page"):
57            break
58
59        next_cursor = page.get("next_cursor")
60        if not next_cursor:
61            break
62
63    return collected
64
65seen_ids = set()
66tweets = collect_recent_tweets(
67    query='"crypto regulation" lang:en -filter:retweets',
68    seen_ids=seen_ids,
69    max_pages=2,
70)
71
72print(f"New tweets ready for scoring: {len(tweets)}")

Step 2: Fast Sentiment Scoring

Fast sentiment scoring means using an explainable, low-cost model to score all new posts first, then sending only selected samples to a heavier model. The dashboard stays fresh without sending every text snippet to an expensive LLM.

VADER is useful as a first pass for English social text. For multilingual pipelines, route by lang: English can go to VADER, while other languages can use language-specific models or a translated common model.

1from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
2
3analyzer = SentimentIntensityAnalyzer()
4MODEL_VERSION = "vader-3.3.2-en-baseline"
5
6def sentiment_label(score):
7    if score >= 0.2:
8        return "positive"
9    if score <= -0.2:
10        return "negative"
11    return "neutral"
12
13def text_for_model(tweet):
14    return tweet.get("full_text") or tweet.get("text") or ""
15
16def score_tweets(records):
17    scored = []
18
19    for record in records:
20        tweet = record["raw"]
21        text = text_for_model(tweet).strip()
22        if not text:
23            continue
24
25        compound = analyzer.polarity_scores(text)["compound"]
26        scored.append({
27            "tweet_id": tweet.get("tweet_id") or tweet.get("id"),
28            "query": record["query"],
29            "fetched_at": record["fetched_at"],
30            "created_at_datetime": tweet.get("created_at_datetime"),
31            "lang": tweet.get("lang"),
32            "text": text,
33            "sentiment_score": compound,
34            "sentiment_label": sentiment_label(compound),
35            "model_version": MODEL_VERSION,
36            "favorite_count": tweet.get("favorite_count"),
37            "retweet_count": tweet.get("retweet_count"),
38            "reply_count": tweet.get("reply_count"),
39            "raw": tweet,
40        })
41
42    return scored
43
44scored_tweets = score_tweets(tweets)
45print(f"Scored tweets: {len(scored_tweets)}")

Step 3: Window Aggregation

Window aggregation means combining individual tweet scores into time windows that show trends. Do not look only at the average score; also store sample size, positive ratio, negative ratio, and high-engagement examples.

1def summarize_window(scored_rows):
2    if not scored_rows:
3        return {
4            "tweet_count": 0,
5            "avg_sentiment": None,
6            "positive_ratio": 0,
7            "negative_ratio": 0,
8            "top_negative_examples": [],
9        }
10
11    tweet_count = len(scored_rows)
12    avg_sentiment = sum(row["sentiment_score"] for row in scored_rows) / tweet_count
13    positive_count = sum(1 for row in scored_rows if row["sentiment_label"] == "positive")
14    negative_count = sum(1 for row in scored_rows if row["sentiment_label"] == "negative")
15
16    top_negative_examples = sorted(
17        scored_rows,
18        key=lambda row: (row["sentiment_score"], -int(row.get("reply_count") or 0)),
19    )[:5]
20
21    return {
22        "tweet_count": tweet_count,
23        "avg_sentiment": round(avg_sentiment, 4),
24        "positive_ratio": round(positive_count / tweet_count, 4),
25        "negative_ratio": round(negative_count / tweet_count, 4),
26        "top_negative_examples": [
27            {
28                "tweet_id": row["tweet_id"],
29                "score": row["sentiment_score"],
30                "text": row["text"][:180],
31            }
32            for row in top_negative_examples
33        ],
34    }
35
36summary = summarize_window(scored_tweets)
37print(summary)

Step 4: Production Expansion

Production expansion means splitting collection, scoring, aggregation, and alerts into recoverable jobs. You can scale collection with multiple query workers, but the write path should stay strict: dedupe first, score second, and keep every model output tied to the query that produced it.

At minimum, handle these points:

Configurable queries: Store query, language, exclusions, refresh interval, and max page count in configuration.
Separate cursor and dedupe: next_cursor is for pagination; tweet_id is the dedupe key.
Empty-window protection: When there are no new posts, do not calculate fake averages; mark sample count as 0.
Model versioning: Scores from changed models or thresholds should not be compared as if they came from the same model.
Sentiment decay: Refresh active topics on a schedule so old sentiment does not dominate the dashboard.
Human review queue: Posts with high negative scores, high engagement, or key accounts should go to review.

Fields To Store

Fields to store means the minimum evidence needed to trace every score back to the source post, query, and model version.

Field	Use
`tweet_id`	Deduplication and traceability
`query` and `sortBy`	Explains why the post entered the dataset
`fetched_at` and `created_at_datetime`	Separates collection time from publish time
`lang`	Routes text to the right model
`text` and raw JSON	Allows reprocessing when parsing or scoring changes
`sentiment_score` and `sentiment_label`	Stores model output
`model_version`	Makes score changes auditable
Engagement counts	Helps prioritize outliers for review

Summary

Summary means building sentiment analysis as a traceable data pipeline: poll /twitter/advanced_search/page with sortBy: "Latest", page with next_cursor, deduplicate by tweet_id, score new posts, aggregate by time window, and save the raw evidence behind every number.