推特实时监控 API

#推特实时监控是怎么工作的

推特实时监控是一个循环。它向 X 请求某个信息源里最新的帖子，留下还没处理过的，几秒后再请求一次。这个循环叫轮询。信息源可以是一个账号、一个账号列表，或者对一组词的搜索：品牌名、对手的账号、cashtag、故障关键词，或者那些一发帖就能影响市场的分析师。

X 上的每条帖子都带着一个随时间递增的数字 ID，所以你处理过的最大 ID 就能当检查点：比它大的都是新的。两次轮询之间的间隔，决定了最长要等多久。刚轮询完就发布的帖子，要整整等一个间隔才被下一次轮询抓到；刚好在轮询之前发布的，则立刻就能抓到。

官方 X API 走的是相反的路，推送。它的 Filtered Stream 保持一个 HTTP 连接打开，把每条匹配你规则的帖子发给你，X 的 Filtered Stream 文档（在新标签页打开）称 99% 的帖子延迟约 4 到 5 秒。按量付费访问允许 1 个连接、1,000 条规则（每条最长 1,024 个字符）和每秒 250 条帖子，而且需要通过审批的开发者账号、一个项目、一个应用和一个 Bearer 令牌。

X 把数据流送来的每条帖子都按一次帖子读取计费：$0.005，也就是每 1,000 条 $5.00，每月最多 300 万次帖子读取，超过就要签 Enterprise 合同。根据 X 10 月 23 日的通知（在新标签页打开），自 2026 年 11 月 22 日起，随帖子返回的作者资料另加 $0.010，所以带作者的一条帖子是 $0.015。官方的其他费率，都拆解在 X API 价格指南里。

#用 Sorsa API 监控账号和关键词

要用代码监控 X，用 Sorsa API，一款 Twitter（X）API 与爬虫：一次请求就能读取最多 5,000 个账号的最新帖子，关键词信息源使用 X 自己的搜索语法，所有套餐都是每秒 20 次请求。每次轮询都返回新鲜数据，每条帖子都带着作者的完整资料，只需一个 ApiKey 请求头，而不是开发者账号加 OAuth。

按你要盯的对象选接口。每个接口返回一页约 20 条帖子，最新的在前，并附带一个指向更早一页的 next_cursor。

• 最多 5,000 个账号

  接口

  GET /list-tweets（在新标签页打开）

  一次请求返回什么

  一个公开 List 里每位成员的最新帖子

• 一个账号

  接口

  POST /user-tweets（在新标签页打开）

  一次请求返回什么

  它最新的帖子、回复或转发

• 词、话题标签、cashtag

  接口

  POST /search-tweets（在新标签页打开）

  一次请求返回什么

  用 "order": "latest" 得到最新的匹配项

• @ 了某个账号的帖子

  接口

  POST /mentions（在新标签页打开）

  一次请求返回什么

  最新的提及，支持点赞数和日期筛选

1. 获取密钥。注册后，在控制台复制 API 密钥。100 次免费请求已经在上面了。

2. **把账号放进一个公开 X List。**在 x.com 创建一个 List，添加最多 5,000 个账号，并保持公开：任何 API 密钥都读不了私有 List。List 的 ID 就是 x.com 上该 List 地址末尾的那串数字，也就是下面请求里发送的 list_id。

3. **轮询这个 List。**一个 GET 请求一次返回所有成员的最新帖子，每条都带着作者资料。

   curl -G "https://api.sorsa.io/v3/list-tweets" \
     -H "ApiKey: YOUR_API_KEY" \
     --data-urlencode "list_id=1849722563900214388"
   

   {
     "tweets": [
       {
         "id": "2117190277605314310",
         "full_text": "Status update: EU-West checkout errors are resolved.",
         "created_at": "2026-11-02T09:41:12Z",
         "lang": "en",
         "likes_count": 14,
         "retweet_count": 5,
         "reply_count": 3,
         "quote_count": 1,
         "view_count": 2140,
         "is_reply": false,
         "conversation_id_str": "2117190277605314310",
         "user": {
           "id": "1702291101836602498",
           "username": "harborline_status_hq",
           "display_name": "Harborline Status",
           "followers_count": 48210,
           "verified": false
         }
       },
       {
         "id": "2117190206607470143",
         "full_text": "Our Q3 freight index is out: spot rates down 4%.",
         "created_at": "2026-11-02T09:40:55Z",
         "lang": "en",
         "likes_count": 31,
         "retweet_count": 9,
         "reply_count": 4,
         "quote_count": 2,
         "view_count": 5320,
         "is_reply": false,
         "conversation_id_str": "2117190206607470143",
         "user": {
           "id": "1366689741639744314",
           "username": "quaywright_analytics",
           "display_name": "Quaywright Analytics",
           "followers_count": 12904,
           "verified": true
         }
       }
     ],
     "next_cursor": "DAABCgABG3kQ8x7_7ToKAAIbeRDz9yz3QAgAAwAAAAIAAA"
   }
   

   显示完整响应

4. **只保留新帖子。**把每个 ID 和你的检查点比较；如果自上次轮询以来到了不止一页，就沿着 next_cursor 继续读；已经处理过的 ID 直接跳过。下面的循环每 5 秒做完这三件事，并重新检查最近 2 分钟的 ID，所以迟到进入信息源的帖子也能抓住。

   import time
   import requests
   
   API = "https://api.sorsa.io/v3"
   HEADERS = {"ApiKey": "YOUR_API_KEY"}
   LIST_ID = "1849722563900214388"
   INTERVAL = 5              # seconds between polls
   OVERLAP = 120_000 << 22   # re-check the last 2 minutes of IDs
   
   
   def read_newer(floor):
       """Read pages until posts at or below floor show up."""
       posts, params = [], {"list_id": LIST_ID}
       for _ in range(5):  # at most 5 pages per poll
           r = requests.get(f"{API}/list-tweets", headers=HEADERS,
                            params=params, timeout=30)
           r.raise_for_status()
           page = r.json()
           tweets = page.get("tweets", [])
           newer = [t for t in tweets if int(t["id"]) > floor]
           posts += newer
           cursor = page.get("next_cursor")
           if not floor or len(newer) < len(tweets) or not cursor:
               break
           params["next_cursor"] = cursor
       return sorted(posts, key=lambda t: int(t["id"]))
   
   
   def handle(t):
       print(t["user"]["username"], t["created_at"], t["full_text"][:80])
   
   
   last_id, seen, first = 0, set(), True
   while True:
       try:
           posts = read_newer(max(last_id - OVERLAP, 0))
       except requests.RequestException as err:
           print("poll failed, retrying:", err)
           time.sleep(INTERVAL * 2)
           continue
       for t in posts:
           if t["id"] not in seen:
               seen.add(t["id"])
               if not first:  # the first poll only sets the checkpoint
                   handle(t)
           last_id = max(last_id, int(t["id"]))
       first = False
       time.sleep(INTERVAL)
   

   const API = "https://api.sorsa.io/v3";
   const headers = { ApiKey: "YOUR_API_KEY" };
   const LIST_ID = "1849722563900214388";
   const INTERVAL = 5_000;           // 5 seconds between polls
   const OVERLAP = 120_000n << 22n;  // re-check the last 2 minutes of IDs
   
   const sleep = (t) => new Promise((r) => setTimeout(r, t));
   
   async function readNewer(floor) {
     const posts = [];
     const params = new URLSearchParams({ list_id: LIST_ID });
     for (let i = 0; i < 5; i++) {
       const res = await fetch(`${API}/list-tweets?${params}`, { headers });
       if (!res.ok) throw new Error(`HTTP ${res.status}`);
       const page = await res.json();
       const tweets = page.tweets ?? [];
       const newer = tweets.filter((t) => BigInt(t.id) > floor);
       posts.push(...newer);
       if (!floor || newer.length < tweets.length || !page.next_cursor) break;
       params.set("next_cursor", page.next_cursor);
     }
     return posts.sort((a, b) => (BigInt(a.id) < BigInt(b.id) ? -1 : 1));
   }
   
   let lastId = 0n;
   let first = true;
   const seen = new Set();
   while (true) {
     try {
       const floor = lastId > OVERLAP ? lastId - OVERLAP : 0n;
       for (const t of await readNewer(floor)) {
         if (!seen.has(t.id)) {
           seen.add(t.id);
           if (!first) console.log(t.user.username, t.full_text.slice(0, 80));
         }
         if (BigInt(t.id) > lastId) lastId = BigInt(t.id);
       }
       first = false;
     } catch (err) {
       console.error("poll failed, retrying:", err.message);
       await sleep(INTERVAL);
     }
     await sleep(INTERVAL);
   }
   

5. **加上关键词信息源。**对品牌或 cashtag，轮询 POST /search-tweets，带上 "order": "latest"，并在查询里加入 since_id: 操作符，这样每次轮询只返回更新的匹配项。原生转发默认不在结果里，除非查询里加上 include:nativeretweets，其余操作符见推特搜索 API指南。

   floor = max(last_id - OVERLAP, 0)
   query = '("lumenforge" OR #lumenforge OR $LMFG) lang:en'
   body = {"query": f"{query} since_id:{floor}", "order": "latest"}
   r = requests.post(f"{API}/search-tweets", headers=HEADERS,
                     json=body, timeout=30)
   

6. 把每条新帖子送到需要它的地方。handle() 函数由你来替换：把帖子写进数据库，发到团队的 Slack 频道，或者交给 AI 智能体去分类和总结。

#一个 X List 覆盖 5,000 个账号

X List 把成千上万个账号变成一个信息源，这正是大规模轮询能保持低成本的原因。逐个盯 5,000 个账号，每次轮询要发 5,000 次时间线请求。把它们放进一个公开 List，每次轮询就只是对 GET /list-tweets 的一次请求，不管有多少成员发了帖。

这样，速率限制就不再是瓶颈。Sorsa API 在所有套餐上都允许每个密钥每秒 20 次请求，免费账号也一样。一个密钥可以每秒轮询 20 个 List，或者每 30 秒轮询 600 个 List，覆盖最多 300 万个账号。在官方 X API 上，List 时间线接口每个应用每 15 分钟允许 900 次请求，约每秒一次，而且对返回的每条帖子都计费。完整的对照表，见 X API 速率限制指南。

每次请求返回约 20 条帖子。当某个 List 在两次轮询之间发布的帖子超过这个数，上面的循环会沿着 next_cursor 回溯到检查点，每多一页多一次请求。缩短间隔也能起到同样的作用。读取 List 的成员和粉丝、批量创建 List，见 X 列表与社区 API指南。

#多久轮询一次，要花多少钱

先根据你的任务能接受的最长等待时间选定间隔，再数轮询次数。无论一页里有 20 条新帖子还是一条也没有，Sorsa API 每次轮询都只算一次请求，所以月成本取决于两个数：你轮询多少个信息源，以及多频繁。在 30 天的一个月里，一个信息源每 30 秒轮询一次，大约发出 8.6 万次请求，正好在 Pro 套餐的 10 万次以内。每个信息源，各间隔的成本如下：

• 5 秒

  每月请求数

  51.8 万

  最低

  $726

  按 Pro 费率

  $1,032

• 15 秒

  每月请求数

  17.3 万

  最低

  $242

  按 Pro 费率

  $344

• 30 秒

  每月请求数

  8.6 万

  最低

  $121

  按 Pro 费率

  $172

• 1 分钟

  每月请求数

  4.3 万

  最低

  $60

  按 Pro 费率

  $86

• 5 分钟

  每月请求数

  8,600

  最低

  $12

  按 Pro 费率

  $17

“最低”一列用的是 Sorsa API 的最低费率，每 1,000 次请求 $1.40；Pro 费率是 $1.99。这里的一个信息源，是一个含最多 5,000 个账号的 List，或一个搜索查询，所以每 30 秒轮询 5,000 个账号的成本，和轮询一个账号一样。下面的规划器可以加上更多信息源，并显示哪个套餐能覆盖它们。

按帖子算，繁忙的信息源差别最明显。一个 List 或一次搜索，如果每次轮询都填满 20 条的一页，在 Sorsa API 上采集帖子每 1,000 条 $0.07 起。官方 X API 对数据流送来的帖子每 1,000 条收费 $5.00，自 11 月 22 日起把每条帖子附带的作者也计费之后，就是每 1,000 条 $15.00。

失败的请求从不计费，而 GET /key-usage-info（在新标签页打开） 能报告一个密钥还剩多少请求，且不占用请求数。

#用 API 做 X 上的社交聆听

X 上的社交聆听，指收集人们发布的关于某个品牌、产品或话题的内容，再把它变成数字：声量、情感、触达，以及带动讨论的账号。聆听工具在自己的看板里完成这些。用 API，则是你自己采集帖子，在自己的代码、数据仓库或 AI 智能体里计算这些数字。

与多平台聆听工具的 X 连接器相比，专用的 X 数据 API 给你：

• 任意公开帖子的完整回复、引用和转发者，逐页读取；
• X 自己的搜索语法，从精确短语到 lang:、min_faves: 和 from:；
• 可回溯到 2006 年的任意日期窗口搜索，用来把一次发布和以往的发布做对比；
• 每条帖子里都有作者的完整资料，无需额外调用；
• 原始 JSON，没有席位限制，也没有导出上限，按请求固定计价。

品牌提及来自两个信息源：POST /mentions 抓到 @ 了你账号的帖子，名称搜索则抓到那些没有 @ 的帖子。推特提及 API指南按推文 ID 把两者合并。围绕任何一条火起来的帖子，POST /comments（在新标签页打开）、POST /quotes（在新标签页打开） 和 POST /retweeters（在新标签页打开） 分别读取回复、引用帖子和转发过它的账号。GET /followers（在新标签页打开） 每页读取 200 份受众资料，GET /info-batch（在新标签页打开） 一次读取 100 份资料，GET /trends（在新标签页打开） 则按 WOEID 返回某个国家或城市当前的热门趋势。

团队从信息源里算出来的指标

• **声量占比：**对你的品牌和每个竞争对手，在同一个时间窗口里运行同一个查询，然后比较匹配的帖子数，以及背后的浏览量和粉丝触达。推特竞品分析指南会搭出完整的对比。
• **情感：**用你自己的模型或 LLM，把每条帖子的文本分为正面、负面或中性，做法见我们的推特情感分析指南。
• **触达与扩散：**把一个窗口内帖子的 view_count、retweet_count 和 quote_count 加起来。
• **头部声音：**按 followers_count 和帖子获得的互动，对作者排名。

用代码做 X 上的社交聆听，Sorsa API 的搜索结果每 1,000 条 $0.07 起，历史可回溯到 2006 年，而同一个密钥也跑得动上面的实时监控。

#让监控保持可靠

一个运行几个月的监控，会遇到重启、突发流量和乱序到达的帖子。五个习惯能让它不漏帖，也不重复上报：

• **保存检查点。**把最后一个 ID 存在数据库的一行或一个文件里，这样重启后能接着上次停下的地方继续，而不是重放旧帖子或跳过新帖子。
• **重新检查一小段重叠区间。**一条帖子可能比更新的帖子晚几秒才进入 List 或搜索信息源。每次轮询都回读 2 分钟的 ID，并跳过已处理的 ID，就能抓住这些迟到的帖子。
• 比较 ID，不比较位置。POST /user-tweets 可能先返回账号的置顶帖子，所以要把这一页里最大的 ID 当作最新的。
• **出错时退避。**429 表示这个密钥超过了每秒 20 次请求：等一下再重试。403 表示密钥已没有剩余请求。两者都不花钱，因为失败的请求不计费。
• **稍后刷新计数。**发布几秒后就抓到的帖子，几乎还没有点赞或浏览量。一小时后用 POST /tweet-info-bulk（在新标签页打开） 重新读取你保留的帖子，每次请求 100 条，这样已知推文每 1,000 条 $0.014 起。

须知

轮询只读取公开数据。受保护账号、私有 List 和私信永远不会出现在任何信息源里。Sorsa API 在设计上是只读的：它从不发帖、点赞或关注，也从不索要你的 X 登录信息，所以你自己的 X 账号没有任何风险。

#开始你的第一个实时监控

推特实时监控归结起来就是一个间隔和一个检查点：对一个公开 List 每 5 秒循环一次，每次轮询只用一个请求，就能盯住 5,000 个账号。Sorsa API 用一个 API 密钥、所有套餐每秒 20 次请求跑这个循环，采集的帖子每 1,000 条 $0.07 起，而官方 X API 是 $5.00。把你的账号放进一个 List，用你的 100 次免费请求跑上面的循环，等第一批帖子进来之后，再加上关键词信息源。

#常见问题