如何监控 Serenity 在 X 上发布的新公开帖
如果 Serenity 发布了一条可能引发讨论的新公开帖，你可能希望在话题扩散前收到提醒。这个场景不需要长连接，也不需要持续抓全量时间线；更稳的做法是定时查询最新结果，把新出现的 tweet_id 和本地状态做对比。
TwexAPI 的 Advanced Twitter Search by Page 端点适合做这种轻量轮询。请求里使用 searchTerms 限定 Serenity 账号，sortBy 固定为 Latest，每次只看最新一页即可。
这不是流式连接。关键不是“每次抓多少”，而是保存状态：第一次运行建立基线，后续只对没见过的 tweet_id 发送告警。
监控策略
监控策略 指先决定告警范围，再写代码。默认建议只监控 Serenity 的原创新帖，避免回复和转推触发太多低价值通知。
推荐查询：
from:aleabitoreddit -filter:retweets -filter:replies这条查询会保留 Serenity 本人发布的新原创帖。如果你想监控他在别人帖子下的回复，去掉 -filter:replies 即可。
一个小型监控任务可以按这个顺序执行：
| 步骤 | 目的 |
|---|---|
| 拉取最新结果 | 使用 sortBy: Latest 获取最新一页 |
| 初始化基线 | 第一次运行只保存已有 tweet_id，不告警 |
| 对比新 ID | 后续运行只处理未见过的帖子 |
| 先落库再告警 | 避免告警发出后原始数据丢失 |
| 节流和重试 | 网络失败时重试，但同一 tweet_id 不重复通知 |
API 端点
API 端点 指轮询任务使用的搜索入口。请求体里的 searchTerms 是数组，sortBy 支持 Latest 或 Top；监控新帖时使用 Latest。
POST https://api.twexapi.io/twitter/advanced_search/page
Authorization: Bearer <your_token>
Content-Type: application/jsoncurl --request POST \
--url https://api.twexapi.io/twitter/advanced_search/page \
--header 'Authorization: Bearer <your_token>' \
--header 'Content-Type: application/json' \
--data '{
"searchTerms": ["from:aleabitoreddit -filter:retweets -filter:replies"],
"sortBy": "Latest"
}'Python 轮询示例
Python 轮询示例 指一个可以本地跑起来的最小监控任务。它用 JSON 文件保存 seen_ids，重启后不会忘记已经告警过的帖子。
1import json
2import time
3from pathlib import Path
4
5import requests
6
7TOKEN = "<your_token>"
8URL = "https://api.twexapi.io/twitter/advanced_search/page"
9QUERY = "from:aleabitoreddit -filter:retweets -filter:replies"
10HEADERS = {"Authorization": f"Bearer {TOKEN}"}
11STATE_FILE = Path("serenity-monitor-state.json")
12POLL_SECONDS = 300
13
14def load_seen_ids():
15 if not STATE_FILE.exists():
16 return set(), False
17 payload = json.loads(STATE_FILE.read_text(encoding="utf-8"))
18 return set(payload.get("seen_ids", [])), payload.get("initialized", False)
19
20def save_state(seen_ids, initialized=True):
21 STATE_FILE.write_text(
22 json.dumps(
23 {"initialized": initialized, "seen_ids": sorted(seen_ids)},
24 ensure_ascii=False,
25 indent=2,
26 ),
27 encoding="utf-8",
28 )
29
30def fetch_latest():
31 response = requests.post(
32 URL,
33 headers=HEADERS,
34 json={"searchTerms": [QUERY], "sortBy": "Latest"},
35 timeout=30,
36 )
37 response.raise_for_status()
38 return [tweet for tweet in response.json().get("data", []) if tweet]
39
40def notify(tweet):
41 print("NEW", tweet.get("tweet_id"), tweet.get("text", "")[:160])
42
43seen_ids, initialized = load_seen_ids()
44
45while True:
46 latest = fetch_latest()
47 latest_ids = {tweet["tweet_id"] for tweet in latest if tweet.get("tweet_id")}
48
49 if not initialized:
50 seen_ids.update(latest_ids)
51 initialized = True
52 save_state(seen_ids, initialized=True)
53 print(f"已初始化 {len(seen_ids)} 条已有帖子")
54 else:
55 new_tweets = []
56 for tweet in reversed(latest):
57 tweet_id = tweet.get("tweet_id")
58 if tweet_id and tweet_id not in seen_ids:
59 new_tweets.append(tweet)
60 seen_ids.add(tweet_id)
61
62 if new_tweets:
63 save_state(seen_ids, initialized=True)
64 for tweet in new_tweets:
65 notify(tweet)
66
67 time.sleep(POLL_SECONDS)这个示例先保存状态，再发送通知。这样即使 Slack、邮件或 webhook 临时失败，下一轮也不会把同一条帖子重复当成“新帖”。如果你更希望“通知成功后才标记已处理”，可以把通知结果写入单独的 delivery 表，而不是只靠 seen_ids。
生产环境检查清单
生产环境检查清单 指上线前要明确的几个状态和失败处理规则。监控类脚本最常见的问题不是抓不到数据，而是重复告警或丢失基线。
| 检查项 | 建议 |
|---|---|
| 状态存储 | 将 seen_ids 存入数据库、Redis 或可持久化文件 |
| 首次运行 | 明确是初始化基线，还是要对当前可见帖子全部告警 |
| 轮询间隔 | 根据实时性和 credits 预算，每 5 到 15 分钟轮询一次 |
| 告警顺序 | 发送 Slack、邮件或 webhook 前保存原始响应 |
| 重试策略 | 网络失败可以重试，但同一 tweet_id 不重复通知 |
| 查询范围 | 默认排除转推和回复；需要监控回复时再调整查询 |
监控只负责发现 Serenity 新发布的源帖。要观察源帖发布后的全站扩散，应继续使用追踪 Serenity 相关 Cashtag或回复分析流程。
相关资源
免责声明
免责声明 指本文只说明如何发现和保存公开 X 帖子。该监控不评估金融观点的准确性，也不构成投资建议。