如何用 TwexAPI 构建实时 X 情感分析流水线

情感分析最怕两件事：数据已经过期，或者分数没有证据可查。一个看板如果只显示“市场情绪 -0.42”，但说不清用了哪些帖子、哪个查询、什么模型版本、什么时候抓到的，运营和研究人员很难信任它。

更稳的做法是把它当成数据流水线：用 TwexAPI Advanced Search by Page 轮询近期帖子，用 tweet_id 去重，保存原始 JSON，再用轻量模型先跑一遍分数。后续你可以把 VADER 换成 transformer 或 LLM，但采集、去重、评分、聚合和复核这几层不应该变。

流水线结构

实时情感分析流水线 指把 X 帖子从查询结果变成可复核情感指标的固定流程。它通常分成五部分：

查询设计：把品牌、事件、语言、排除条件和时间窗口写进查询策略。
采集：调用 POST /twitter/advanced_search/page，用 sortBy: "Latest" 获取最新帖子。
存储：保存原始 JSON，并在评分前用 tweet_id 去重。
评分：用轻量模型处理大部分内容，把边界样本或高影响帖子交给更重的模型。
聚合与复核：按时间窗口输出均值、数量、异常样本和模型版本，方便分析人员解释结果。

步骤 1：轮询采集

轮询采集 指按固定间隔请求最新搜索结果，而不是假装自己有流式连接。TwexAPI 的 POST /twitter/advanced_search/page 每页最多返回 20 条推文；第一页不传 next_cursor，后续页面传上一次响应里的 next_cursor。

下面示例用 sortBy: "Latest" 获取近期帖子，并用 seen_ids 避免重复评分。生产环境里，seen_ids 应该放进数据库或缓存，而不是只存在内存里。

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

步骤 2：高速情感评分

高速情感评分 指先用可解释、成本低的模型给所有新帖子打基础分，再把边界样本交给更重的模型。这样看板可以保持更新速度，也不会把所有文本都送进昂贵的 LLM。

VADER 适合做英文社交文本的初筛。多语言场景建议先按 lang 分流：英文走 VADER，其他语言走对应模型或翻译后的统一模型。

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

步骤 3：窗口聚合

窗口聚合 指把单条帖子分数合并成可以看趋势的时间窗口。不要只看平均分；同时保存样本量、负面占比、正面占比和高互动异常样本，否则小样本很容易误导团队。

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)

步骤 4：生产环境扩展

生产环境扩展 指把采集、评分、聚合和告警拆成可恢复任务。可以通过多个查询 worker 扩展采集层，但写入路径要保持严格：先去重，再评分，并把模型输出和产生它的查询绑定在一起。

生产环境至少要处理这些问题：

查询配置化：把查询、语言、排除条件、刷新间隔和最大页数放进配置表。
游标和去重分开：next_cursor 用于翻页；tweet_id 才是去重主键。
空窗口保护：没有新帖子时不要计算假平均分，直接标记样本量为 0。
模型版本管理：模型或阈值变化后，旧分数不能和新分数混在一起比较。
情感衰减：对活跃主题按计划刷新，避免旧情绪长期占据看板。
人工复核队列：负面分数高、互动量高或来自关键账号的帖子，应进入人工复核。

应保存哪些字段

应保存的字段 指能让每个分数回到原始帖子、原始查询和模型版本的最小证据集合。保存足够上下文，后续才能解释每个分数：

字段	用途
`tweet_id`	去重和追溯
`query`	说明帖子为什么进入数据集
`sortBy` 和 `fetched_at`	说明采集方式和进入系统的时间
`text`、`full_text` 和原始 JSON	模型变化后可重新处理
`lang`、`created_at_datetime`	支持语言过滤和时间窗口
`favorite_count`、`retweet_count`、`reply_count`	识别高影响样本
`sentiment_score` 和 `sentiment_label`	保存模型输出和阈值分类
`model_version`	让分数变化可审计
`review_status`	标记是否已人工确认

告警规则

告警规则 指把情感分数转成团队可执行的信号，而不是每次负面波动都报警。建议同时看分数、样本量和互动权重。

可先从三类告警开始：

负面占比突增：例如最近 15 分钟负面占比高于上一窗口 2 倍，且样本量超过最低阈值。
高互动负面帖：单条负面帖的回复、转发或浏览明显高于同窗口中位数。
关键查询命中：品牌名、产品故障、监管、价格、停服等高风险查询出现新负面样本。

告警内容应包含查询、时间窗口、样本量、平均分、负面占比、代表性 tweet_id 和模型版本。这样接收者能马上判断要不要升级处理。

小结

小结指先把实时情感分析做成可复核的数据链路，再追求更复杂的模型：用 /twitter/advanced_search/page 拉取最新帖子，用 tweet_id 去重，保存原始响应和查询上下文，用轻量模型打分，再按时间窗口聚合和告警。等流水线稳定后，再扩展查询集，并只在确实改善决策的地方加入更重的模型。