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)}")

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)}")

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)