著者：Sorsa Editorial

2026年7月4日更新：100回分の無料リクエストの開始オプションを追加し、公式 X API の従量課金のコスト比較を刷新しました。

目次

要点 {#key-takeaway}

リアルタイムTwitterモニタリングは、REST エンドポイントを5〜30秒ごとにポーリングし、返されたツイート ID を最後に見た ID と比較し、新しいツイートをアラートのパイプラインにプッシュすることで機能します。多くのアカウントを一度に追跡するには、最大5,000件を1つの X リストにまとめ、単一のリストエンドポイントをポーリングします。

なぜ2023年以降、リアルタイムTwitterモニタリングは難しくなったか {#why-real-time-twitter-monitoring-got-harder-after-2023}

リアルタイムTwitterモニタリングは、かつて1つのことを意味しました。フィルタードストリームへの永続的な接続を開き、いくつかのルールを定義し、投稿されたそばからツイートをアプリケーションに流し込む、ということです。2023年の料金の大改定がそれを月額$5,000の Proプランの機能に変え、2026年初頭に X はさらに踏み込みました。配信されるすべての投稿が課金され、継続的な監視ワークロードが200万件の読み取り上限に達する前に月に数千ドルに達する従量課金モデルに切り替えたのです。ほとんどのチーム（インディー開発者、ブランド監視チーム、トレーディングボット、ニュースデスク、リード獲得の代理店）にとって、その料金がプロジェクトを殺しました。実用的な代替は、プレーンな REST API から自分でデータを取得することです。

本ガイドは、その取得ベースのパターンを Sorsa API の上に構築します。Sorsa は、すべてのリクエストで新鮮なツイートデータを返す Twitter/X の代替 API プロバイダーで、OAuth フローも審査待ちの行列もない単一の API キーで動き、すべてのプランで一律の毎秒20リクエストを許可し、最大5,000アカウントを1回の /list-tweets 呼び出しに折り込めます。読み取りアクセスは、バッチエンドポイントで1,000ツイートあたり$0.02（1,000プロフィールあたり$0.01）から始まり、40個のエンドポイントすべてをカバーする、1回限り・クレジットカード不要の100回分の無料リクエストで全体をテストできます。約300ms のレスポンスタイムと正しいエンドポイントの選択で、投稿されてから数秒以内に新しいツイートを検出でき、読み取りの多い監視には、公式 API が課金する額のごく一部で済みます。プロバイダーのより広い比較については、Twitter API の代替 の概要をご覧ください。

用語について1つ。本記事は「ポーリング」を意図的に使います。それは「遅延」の婉曲表現ではありません。正しく行えば、ポーリングループは、選んだ間隔に API のレスポンスタイムを加えた時間内にツイートを返します。ループが300ms のエンドポイントに対して5秒ごとに走るなら、最悪ケースのレイテンシはおよそ5.3秒です。それはほとんどの消費者向けストリーミング製品に匹敵するか上回り、そして決定的に、フラット料金の REST 予算で機能します。

ポーリング対ストリーミング：どちらを選ぶべきか？ {#polling-vs-streaming-which-approach-fits-your-case}

ソーシャルプラットフォームからデータを得る方法は2つあります。プッシュベース（ストリーミング、Webhook）か、プルベース（ポーリング）です。ほとんどのエンジニアは、より速そうに聞こえるという理由でストリーミングを既定にします。実際には、選択は3つのことによります。何をいくつ監視しているか、レイテンシの予算、そして接続の状態にどれだけ耐えられるか、です。

要因ポーリング（REST）ストリーミング（WebSocket / フィルタードストリーム）
最初のツイートまでのレイテンシ間隔 ＋ API のレスポンスタイム接続レイテンシ、通常1〜3秒
セットアップの複雑さループの中の API 呼び出し1つ永続的な接続、再接続ロジック、バックプレッシャー処理
認証ヘッダーの中の API キーOAuth またはトークンローテーション
クラッシュ時の復旧保存したカーソルから再開再接続、再生バッファ、重複排除の窓
コストモデルリクエスト単位配信されたリソース単位、またはエンタープライズ契約
最適な用途1〜5,000の監視対象、1〜30秒の許容度のアラートサブ秒のレイテンシ、フルファイアホースの取り込み、ミリ秒スケールのトレーディング

ストリーミングが勝つのは、大きなキーワード空間でミリ秒レベルのアラートが必要で、再接続、ギャップの復旧、ルール管理を扱うエンジニアリング能力があるときです。それ以外のすべて（ブランドリスニング、ニュース監視、リード獲得、コンプライアンスアラート、中頻度のトレーディングシグナル、コンテンツモデレーション）には、ポーリングのほうが単純で、安く、十分です。

ポーリングのもう1つの過小評価された利点：スクリプトがクラッシュしても、次のサイクルで中断したところから再開します。ストリーミングのパイプラインは、データ損失なく障害を扱うために別の再生バッファを必要とします。Sorsa は意図的にマネージドな Webhook 製品を出していません。プルのパターンは制御プレーンを自分の手に保ち、一律の毎秒20リクエストのレート制限 が本番規模のループに十分な余裕を与えます。

監視対象に合ったエンドポイントを選ぶ {#pick-the-right-endpoint-for-what-youre-monitoring}

最初の設計上の決定は、対象の形に合ったエンドポイントを選ぶことです。4つのエンドポイントが、事実上あらゆるリアルタイム監視のニーズをカバーします。

監視対象エンドポイントメソッド理由
単一アカウント/user-tweetsPOST1ユーザーのタイムラインから最新のツイートを返す
一度に最大5,000アカウント/list-tweetsGET1回のリクエストが X リストのすべてのメンバーをカバー
キーワード、ハッシュタグ、検索クエリ/search-tweetsPOST完全なコマンド構文、時系列の結果に order: latest をサポート
特定ハンドルへの @メンション/mentionsPOSTエンゲージメントフィルター付きのメンション追跡用に専用設計

ほとんどのチームを驚かせるエンドポイントは /list-tweets です。50アカウントを1つのリストに入れてリストエンドポイントをポーリングすると、各アカウントを個別にポーリングするのと比べてリクエスト量がおよそ50分の1になります。同じパターンが、リクエスト量を変えずに500や5,000アカウントにスケールし、リスト自体は x.com で作成します。

レベル1：単一アカウントを追跡する {#level-1-track-a-single-account}

最も単純なケースです。1つのアカウント、競合、CEO、規制当局、インフルエンサーが投稿した瞬間を知りたいとします。低ボリュームの監視や、スケールアップ前のパイプラインのテストに役立ちます。/user-tweets エンドポイント を使います。

Python

python
import requests
import time

API_KEY = "YOUR_API_KEY"
USERNAME = "elonmusk"
POLL_INTERVAL = 5  # seconds

URL = "https://api.sorsa.io/v3/user-tweets"
HEADERS = {"ApiKey": API_KEY, "Content-Type": "application/json"}

last_seen_id = None

print(f"Monitoring @{USERNAME}...")

while True:
    try:
        resp = requests.post(URL, headers=HEADERS, json={"username": USERNAME})
        resp.raise_for_status()
        tweets = resp.json().get("tweets", [])

        if tweets:
            # Tweet IDs are Snowflake strings. Cast to int for safe comparison
            # since lexicographic order can break across ID length boundaries.
            top_id = int(tweets[0]["id"])

            if last_seen_id is None:
                last_seen_id = top_id
                print(f"Baseline set: {last_seen_id}")
            else:
                new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                # Print in chronological order (oldest first).
                for tweet in reversed(new_tweets):
                    print(f"[NEW] @{USERNAME}: {tweet['full_text'][:140]}")
                if new_tweets:
                    last_seen_id = top_id

    except requests.exceptions.RequestException as e:
        print(f"Request error: {e}")
        time.sleep(POLL_INTERVAL * 2)
        continue

    time.sleep(POLL_INTERVAL)

JavaScript

javascript
const API_KEY = "YOUR_API_KEY";
const USERNAME = "elonmusk";
const POLL_INTERVAL = 5000;

let lastSeenId = null;
console.log(`Monitoring @${USERNAME}...`);

while (true) {
  try {
    const resp = await fetch("https://api.sorsa.io/v3/user-tweets", {
      method: "POST",
      headers: { "ApiKey": API_KEY, "Content-Type": "application/json" },
      body: JSON.stringify({ username: USERNAME }),
    });
    if (!resp.ok) throw new Error(`HTTP ${resp.status}`);

    const tweets = (await resp.json()).tweets || [];

    if (tweets.length > 0) {
      // BigInt comparison avoids precision loss on 64-bit Snowflake IDs.
      const topId = BigInt(tweets[0].id);

      if (lastSeenId === null) {
        lastSeenId = topId;
        console.log(`Baseline set: ${lastSeenId}`);
      } else {
        const newTweets = tweets.filter((t) => BigInt(t.id) > lastSeenId);
        for (const t of [...newTweets].reverse()) {
          console.log(`[NEW] @${USERNAME}: ${t.full_text.slice(0, 140)}`);
        }
        if (newTweets.length) lastSeenId = topId;
      }
    }
  } catch (err) {
    console.error(`Error: ${err.message}`);
    await new Promise((r) => setTimeout(r, POLL_INTERVAL * 2));
    continue;
  }
  await new Promise((r) => setTimeout(r, POLL_INTERVAL));
}

上のコードで2つの細部が重要です。第一に、ツイート ID は Snowflake の値で、文字列として届きます。文字列として比較すると単一の時間窓の内側では機能しますが、ID の桁数の境界をまたぐと脆弱です。Python では int、JavaScript では BigInt にキャストします。第二に、ループはタイムライン全体をダンプするのではなく、最初の成功した呼び出しでベースラインを確立します。それが起動時のスパムの噴出を避けます。

これは機能しますが、スケールが悪いです。50アカウントの監視は、50個の別々のポーリングループと50倍の API リクエストを意味します。そこで X リストの出番です。

レベル2：1回のリクエストで最大5,000アカウントを追跡する {#level-2-track-up-to-5000-accounts-in-one-request}

X リストは、複数アカウントの監視に最も有用で、最も活用されていないツールです。リストはアカウントの公開グループ（最大5,000件）で、/list-tweets エンドポイント は、すべてのメンバーにわたる統合された最新のツイートを1回のリクエストで返します。リストを一度作り、ポーラーをそれに向ければ、公式のものに支払うことなく、実質的に独自のカスタムファイアホースを構築したことになります。

ステップ1：公開の X リストを作成する

  1. X リスト に行き、新しいリストを作成します。
  2. 監視したいアカウント（最大5,000件）を追加します。
  3. リストを 公開 に設定します。非公開のリストは API 経由でアクセスできません。
  4. URL から リスト ID をコピーします。https://x.com/i/lists/1234567890 なら ID は 1234567890 です。

ステップ2：リストをポーリングする

python
import requests
import time

API_KEY = "YOUR_API_KEY"
LIST_ID = "YOUR_LIST_ID"
POLL_INTERVAL = 5

URL = f"https://api.sorsa.io/v3/list-tweets?list_id={LIST_ID}"
HEADERS = {"ApiKey": API_KEY, "Accept": "application/json"}


def monitor_list(callback, interval=POLL_INTERVAL):
    """Poll an X List and call `callback` for each new tweet detected."""
    last_seen_id = None
    print(f"Monitoring List {LIST_ID} (interval: {interval}s)")

    while True:
        try:
            resp = requests.get(URL, headers=HEADERS, timeout=10)
            resp.raise_for_status()
            tweets = resp.json().get("tweets", [])

            if not tweets:
                time.sleep(interval)
                continue

            top_id = int(tweets[0]["id"])

            if last_seen_id is None:
                last_seen_id = top_id
                print(f"Baseline set: {last_seen_id}")
            else:
                new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                if new_tweets:
                    for tweet in reversed(new_tweets):
                        callback(tweet)
                    last_seen_id = top_id

        except requests.exceptions.RequestException as e:
            print(f"Request error: {e}. Retrying in {interval * 2}s")
            time.sleep(interval * 2)
            continue

        time.sleep(interval)


def on_new_tweet(tweet):
    user = tweet["user"]
    print(f"[NEW] @{user['username']}: {tweet['full_text'][:120]}")
    print(
        f"       Likes: {tweet.get('likes_count', 0)} | "
        f"RTs: {tweet.get('retweet_count', 0)} | "
        f"Views: {tweet.get('view_count', 'N/A')}\n"
    )


if __name__ == "__main__":
    monitor_list(on_new_tweet)

効率の向上は劇的です。 50アカウントを個別に10秒間隔で監視すると、1日432,000リクエスト（50ループ × 各8,640リクエスト）かかります。その同じ50アカウントを1つの X リストに入れて /list-tweets をポーリングすると、1日8,640リクエストです。それはカバレッジを失うことなく50分の1の削減です。

1つの落とし穴：/list-tweets は1ページあたり最大20件のツイートを返します。リストのメンバーが頻繁にツイートし、1回のポーリング間隔内に20件を超える新しいツイートが届くと、いくつかを取りこぼすことがあります。2つの対処：間隔を2〜3秒に下げるか、以前に見た ID に到達するまで next_cursor でページネーションで取得します。ほとんどのユースケース（ブランド監視、ニュースデスク、オーディエンス調査）では、5〜10秒あたり20件のツイートは十分すぎる余裕です。

レベル3：キーワード、ハッシュタグ、検索クエリを追跡する {#level-3-track-keywords-hashtags-and-search-queries}

アカウントベースの監視は、既知の情報源が何を言うかを捉えます。キーワードベースの監視は、監視対象のトピックについて誰かが何を言うかを捉えます。時系列の結果を得るには、order: "latest" とともに /search-tweets エンドポイント を使います。

python
import requests
import time

API_KEY = "YOUR_API_KEY"
QUERY = '"your brand" OR @yourbrand lang:en'
POLL_INTERVAL = 10

URL = "https://api.sorsa.io/v3/search-tweets"
HEADERS = {"ApiKey": API_KEY, "Content-Type": "application/json"}


def monitor_keyword(query, callback, interval=10):
    last_seen_id = None
    print(f"Monitoring: {query} (interval: {interval}s)")

    while True:
        try:
            resp = requests.post(
                URL,
                headers=HEADERS,
                json={"query": query, "order": "latest"},
                timeout=10,
            )
            resp.raise_for_status()
            tweets = resp.json().get("tweets", [])

            if tweets:
                top_id = int(tweets[0]["id"])
                if last_seen_id is None:
                    last_seen_id = top_id
                    print(f"Baseline set: {last_seen_id}")
                else:
                    new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                    for tweet in reversed(new_tweets):
                        callback(tweet)
                    if new_tweets:
                        last_seen_id = top_id

        except requests.exceptions.RequestException as e:
            print(f"Error: {e}")
            time.sleep(interval * 2)
            continue

        time.sleep(interval)

本当の力はクエリ文字列に宿ります。API は Twitter の高度な検索コマンドのセット全体をサポートします（非公式のリファレンスは igorbrigadir/twitter-advanced-search にあります）。たとえば、自社ブランドのエンゲージメントの高い英語のメンションを追跡し、リツイートを飛ばすには：

python
monitor_keyword('"your brand" min_faves:10 lang:en -filter:retweets', on_new_tweet)

リアルタイム監視で価値を発揮するいくつかのコマンド：

  • min_faves:N、min_retweets:N で、すでにトレンドになっているコンテンツをフィルターする。
  • -filter:retweets、-filter:replies でノイズを落とす。
  • from:user1 OR from:user2 で、リストなしで少数のアカウントを監視する。
  • (keyword1 OR keyword2) (problem OR issue OR broken) で、感情を帯びたメンションを捉える。
  • near:"san francisco" within:25mi で、地理で区切った監視をする。

2026年のキーワードストリームは、かつてよりノイズが多いです。自動化された返信、詐欺アカウント、AI 生成のスパムが、あらゆる人気の用語に積み重なります。エンゲージメントのしきい値が、ソースにおける最も安いフィルターです。min_faves:5 や min_replies:2 のような下限が、コールバックに届く前に使い捨てのノイズの大半を取り除くため、少なくともいくらかの勢いのある投稿が残ります。オープンなキーワードではなく特にハンドルのメンションを見張っているなら、Twitter メンション API が、まさにこの種のクリーンアップのための最も豊富なフィルターのセット（min_likes、min_replies、min_retweets、日付の範囲）を公開します。

クエリ文字列が扱いにくく感じ始めたら、検索ビルダーの playground が、それを視覚的に構築し、進めながらコマンドのセット全体を見せてくれます。

新しいツイートを Slack、Discord、または任意の HTTP エンドポイントにプッシュする {#push-new-tweets-to-slack-discord-or-any-http-endpoint}

ポーリングループはプロデューサーです。コールバックは、各新しいツイートに何が起きるかを決める場所です。コールバックはただの関数なので、同じモニターが HTTP を話すあらゆるものにルーティングできます。最も一般的な宛先なので Slack を最初に、それからいくつかの手早いバリエーションを。

Incoming Webhook 経由の Slack

python
import requests

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"


def send_to_slack(tweet):
    user = tweet["user"]
    text = (
        f"*New tweet from @{user['username']}*\n"
        f"{tweet['full_text']}\n"
        f"Likes: {tweet.get('likes_count', 0)} | "
        f"RTs: {tweet.get('retweet_count', 0)} | "
        f"Views: {tweet.get('view_count', 'N/A')}\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(SLACK_WEBHOOK_URL, json={"text": text})


# Plug into any monitor:
monitor_list(send_to_slack)
# or: monitor_keyword("bitcoin lang:en min_faves:50", send_to_slack)

Slack アプリの設定で Slack Incoming Webhook の URL をセットアップします（Slack 公式ドキュメント）。Discord、Telegram、または任意の内部エンドポイントも同じパターンです。

Discord

python
DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/YOUR/WEBHOOK"


def send_to_discord(tweet):
    user = tweet["user"]
    content = (
        f"**@{user['username']}** just tweeted:\n"
        f"{tweet['full_text']}\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(DISCORD_WEBHOOK_URL, json={"content": content})

Discord の Webhook のセットアップは discord.com/developers/docs/resources/webhook に文書化されています。

Telegram

python
TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"


def send_to_telegram(tweet):
    user = tweet["user"]
    text = (
        f"New tweet from @{user['username']}\n\n"
        f"{tweet['full_text']}\n\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(
        f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
        json={"chat_id": TELEGRAM_CHAT_ID, "text": text},
    )

任意のカスタム HTTP エンドポイント

python
def send_to_internal_api(tweet):
    requests.post(
        "https://internal.example.com/events/twitter",
        json={
            "tweet_id": tweet["id"],
            "username": tweet["user"]["username"],
            "text": tweet["full_text"],
            "metrics": {
                "likes": tweet.get("likes_count", 0),
                "retweets": tweet.get("retweet_count", 0),
                "views": tweet.get("view_count", 0),
            },
            "url": f"https://x.com/{tweet['user']['username']}/status/{tweet['id']}",
        },
        headers={"Authorization": "Bearer YOUR_INTERNAL_TOKEN"},
        timeout=5,
    )

構築したものは、実質的に独自の Webhook リレーです。API がデータを供給し、自分のコールバックが、各新しいツイートについて誰に知らせるかを決めます。そのリレーを自分で所有する利点は、ベンダーのダッシュボードの中ではなく、すべてのルーティングルール（フィルタリング、レート制限、複数チャンネルへのファンアウト、リトライ方針）を自分のコードに保てることです。

どのくらいの頻度でポーリングすべきか、そしていくらかかるか？ {#how-often-should-you-poll-and-what-does-it-cost}

ループの各サイクルは1リクエストかかり、1回の Sorsa リクエストは、どのエンドポイントに当たるかにかかわらず、プランから1単位です。選んだ間隔が月間の使用量を直接動かし、フラットなリクエスト単位の料金では、それがきれいにプランに対応します。

間隔リクエスト/時リクエスト/日リクエスト/30日1ループのプラン
1秒3,60086,4002,592,000カスタム（Enterprise 超）
5秒72017,280518,400カスタム（Enterprise のすぐ上）
10秒3608,640259,200Enterprise（$899/月）
30秒1202,88086,400Pro（$199/月）
1分601,44043,200Pro（$199/月）

数値は単一の連続ループのものです。複数のループを並行して実行すると、それぞれのリクエスト数が加わります。プランのクォータは、Starter 1万、Pro 10万、Enterprise 月50万リクエストで、それ以上はカスタムクォータです。

これらのループを本番で運用してきた、いくつかの実務的なガイドライン：

  • ブランドリスニング、ニュース監視、リード獲得： 10〜30秒で十分です。投稿から30秒以内にどの新しいツイートも捉え、月間のフットプリントは小さいです。
  • 金融シグナルの検出、速報ボット、トレーディングのワークフロー： 1〜5秒。より多くのリクエストを使いますが、レイテンシの予算がそれを正当化します。
  • コンプライアンス、監査、ゆっくり動く研究： 1〜5分。行動の窓が時間単位で測られるユースケースには、リアルタイムは過剰です。

単一のリストで30秒〜1分の頻度は、月額$199の Proプランに収まります。10秒のループ（約259,000リクエスト）に締めると$899の Enterprise に移り、1つの優先度の高い対象への1秒のループでさえ、公式 X API が同等のリアルタイムアクセスに課金する額をはるかに下回ります。

本番向けの堅牢化：公開前に直すべき5つのこと {#production-hardening-five-things-to-fix-before-going-live}

上の例は意図的に最小限です。これらのいずれかを本番のトラフィックに向ける前に、次の5つの懸念に対処してください。

1. 再起動をまたいで last_seen_id を永続化する

スクリプトがクラッシュして、チェックポイントを覚えずに再起動すると、2つのことがうまくいきません。古いツイートを再処理する（Slack チャンネルへの重複アラート）か、新しいベースラインを設定して静かにギャップを取りこぼすか、です。チェックポイントをファイル、Redis、またはデータベースに保存します。

python
import json
import os

STATE_FILE = "monitor_state.json"


def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f).get("last_seen_id")
    return None


def save_state(last_seen_id):
    with open(STATE_FILE, "w") as f:
        json.dump({"last_seen_id": last_seen_id}, f)

起動時に読み込み、カーソルを更新する成功したポーリングのたびに保存します。

2. エラー時の指数バックオフ

ネットワークの問題、一時的な5xx レスポンス、レート制限のヒット（HTTP 429）は起きます。即座に再試行して事態を悪化させる代わりに、上限を設けて段階的にバックオフします。

python
retry_delay = POLL_INTERVAL
MAX_DELAY = 60

while True:
    try:
        resp = requests.get(URL, headers=HEADERS, timeout=10)
        if resp.status_code == 429:
            print(f"Rate limited. Backing off {retry_delay}s")
            time.sleep(retry_delay)
            retry_delay = min(retry_delay * 2, MAX_DELAY)
            continue
        resp.raise_for_status()
        retry_delay = POLL_INTERVAL  # reset on success
        # process tweets
    except requests.exceptions.RequestException as e:
        print(f"Error: {e}")
        time.sleep(retry_delay)
        retry_delay = min(retry_delay * 2, MAX_DELAY)
        continue

    time.sleep(POLL_INTERVAL)

段階的にバックオフし、遅延に上限を設け、最初の成功で通常の間隔にリセットして、短い429 がモニターを遅い頻度に固定しないようにします。

3. ポーリングと処理を分離する

高価な操作（感情スコアリング、データベースへの書き込み、外部 API の呼び出し、AI 分類）をポーリングループの中で同期的に実行しないでください。下流のシステムが遅くなると、ループがスケジュールから遅れ、レイテンシが急上昇します。新しいツイートをキューにプッシュし、別のワーカーでそれらを処理します。

python
from collections import deque
import threading

tweet_queue = deque()


def polling_loop():
    """Fast loop: poll and enqueue. No heavy work here."""
    # standard polling code, but instead of calling the callback directly:
    # tweet_queue.append(tweet)
    pass


def processing_worker():
    """Separate thread: dequeue and dispatch."""
    while True:
        if tweet_queue:
            tweet = tweet_queue.popleft()
            send_to_slack(tweet)
            save_to_database(tweet)
        else:
            time.sleep(0.1)


threading.Thread(target=processing_worker, daemon=True).start()
polling_loop()

より重いワークロードには、インメモリの deque を Redis、RabbitMQ、SQS、またはスタックがすでに動かしている任意のメッセージブローカーに差し替えます。

4. ヘルスチェックとモニターの監視

すべてのポーリングサイクルをログします。タイムスタンプ、新しいツイートの数、レスポンスタイム、エラーです。モニターが直近 N 分間、成功したポーリングを完了していなければアラートします。静かな失敗は最も高くつく種類で、特にアラートのパイプラインでは、アラートの不在が、誰かがデータのギャップに気づくまで「何も起きなかった」を意味します。自分のコードをデバッグする前にプラットフォームの問題を除外するために、Sorsa ステータスページ で API の稼働状況を確認できます。

5. 本番で噛みつくエッジケースを扱う

  • 削除されたツイート： 取得したときとコールバックが発火するときの間にツイートが削除されると、URL は404 になります。これはエラーではなく、想定されたものとして扱います。
  • 保護されたアカウント： 追跡中のユーザーが非公開になると、/user-tweets は空のリストを返します。ログして続行します。
  • 固定ツイート： /user-tweets レスポンスの最初のツイートは、最新のものではなく固定ツイートであることがしばしばです。厳密な時系列の順序が重要なら created_at で並べます。
  • リツイート対元のツイート： リツイートには tweet["retweeted_status"] が入ります。両方が欲しいのか、元のツイートだけが欲しいのかを決めます。
  • 返信が制限されたツイート： is_replies_limited は、作者が返信を制限したことを示します。一部の監視のユースケースに有用なシグナルです。

実例：ブランド監視の移行 {#in-practice-a-brand-monitoring-migration}

当社が協力したある SaaS 企業は、2018年以降、公式のフィルタードストリームでブランド監視を運用していました。同社の構成はおよそ200のキーワードルールと60の優先アカウントを追跡し、2024年初頭までに Proプランで月額$5,000かかっていました。社内での提案はあけすけでした。ストリームを切り、お金を節約し、何も壊れないことを祈る、と。

移行は2週間かかりました。キーワードのルールを、30秒ごとにポーリングする2つの複合的な /search-tweets ワーカーにまとめ（ルールは OR コマンドのブールクエリに畳み込まれました）、60アカウントのフォローを、15秒ごとにポーリングする1つの X リストに置き換えました。合わせたフットプリントは月およそ350,000リクエストになり、同社が支払っていた$5,000に対して、月額$899の Enterpriseプランに余裕をもって収まりました。エンドツーエンドのアラートのレイテンシは、フィルタードストリームのおよそ2秒から、90パーセンタイルでおよそ15秒になりました。Slack でブランドメンションに対応する PR チームにとって、レイテンシの変化は見えないものでした。請求額の変化はそうではありませんでした。

このアプローチが価値を発揮するユースケース {#use-cases-where-this-approach-earns-its-keep}

最もよく見る5つのパターンです。

ブランドリスニングとソーシャル CRM。 自社ブランドのハンドルに製品名のキーワードを加えて /search-tweets を15〜30秒ごとにポーリングします。メッセージに感情のヒントを焼き込んで Slack にルーティングすれば、PR チームが数分以内に対応します。これはあらゆる ブランド・ソーシャルリスニング の構成の中核です。

ニュースとシグナルの検出。 速報ニュースのハンドル（Reuters、AP、Bloomberg、地域の媒体、担当記者）のリストを作り、5秒ごとにポーリングします。Discord サーバーやトレーディングダッシュボードにファンアウトします。これは2026年に構築できる「ニュースファイアホース」の最も安い版です。

競合インテリジェンス。 競合アカウントに、その CEO とプロダクトリードを加えたリストを、30秒ごとにポーリングします。新しいツイートが共有チャンネルに着地し、誰も40のプロフィールをタブで行き来せずに、PMM チームが無料のインテルフィードを得ます。これは継続的な 競合トラッキング の構成のライブ層です。

リード獲得。 問題提起のクエリで /search-tweets をポーリングします。"any recommendations for" (CRM OR analytics OR transcription)、"looking for an alternative to"、"we just churned from" です。営業がレビューする Slack チャンネルにルーティングします。これを運用するほとんどのチームは、クエリのまとまりごとに週5〜15件の適格なリードを捉えます。これは大規模な Twitter リード獲得 の背後にあるリアルタイムのエンジンです。

暗号資産の KOL シグナル。 暗号資産のインフルエンサーとプロジェクトアカウントのリストを作り、2〜5秒でポーリングし、任意で Sorsa スコアのエンドポイントを使ってオーディエンスの質でシグナルを重み付けします。

リアルタイムモニタリングにおける公式 X API とのコスト比較 {#cost-vs-the-official-x-api-for-real-time-monitoring}

開示：Sorsa は当社の製品であるため、これはあくまで提供元の見方として扱い、自分のワークロードで各選択肢を実際にテストしてください。両側の数字は本物で、2026年7月時点で最新です。

2つのプロバイダーは、まったく異なる単位で課金します。公式 X API は取得したリソース単位で課金します。2026年初頭以降有効な従量課金モデルの下では、各投稿の読み取りが$0.005で、ツイートに付く作者プロフィールは別の$0.010のユーザー読み取りです。Sorsa はリクエスト単位で課金し、1回のリクエストが、作者データを含めておよそ20件のツイート（またはフォロワーエンドポイントでは最大200件のプロフィール）を返します。その構造的なギャップが、監視におけるコストの差を生みます。

公式 X API（従量課金、2026年）Sorsa
料金モデル取得したリソース単位リクエスト単位のフラット（1呼び出し = 1リクエスト）
投稿の読み取り投稿の読み取り1件あたり$0.005リクエストに含まれ、投稿ごとの課金なし
ツイート内の作者プロフィール別途課金、ユーザー読み取り1件あたり$0.010ツイートレスポンスに無料で込み
24/7 の監視（月約170万投稿の読み取り）約$8,600/月Enterpriseプラン、$899/月
月間の読み取り上限200万投稿の読み取り、その後 Enterprise が必要プランベースのクォータ、投稿ごとの上限なし
上限を超えるとEnterprise 契約、歴史的に約$42,000+/月クォータを引き上げたカスタムプラン
認証OAuth 2.0 + ベアラートークンヘッダーの中の単一の API キー
レート制限エンドポイントにより変動すべてのプランで一律の毎秒20リクエスト

トレードオフはレイテンシです。フィルタードストリームは1〜2秒以内にツイートを着地させ、一方ポーリングは、間隔に約300ms のレスポンスタイムを加えた時間内に着地させます。ほとんどの監視のユースケースでは、そのギャップは見えません。サブ秒のトレーディングボットには見え、その場合はベンダーにかかわらず真のストリームが正しいツールです。しかしブランド、ニュース、または競合の規模での読み取りの多い監視には、配信された投稿ごとに支払うのは速く積み重なり、200万件の読み取りの壁にぶつかります。フラットなリクエスト単位のプランはそうなりません。ユースケース別の完全な料金の分解については、2026年の Twitter API 料金 をご覧ください。

よくある質問 {#faq}

REST のポーリングは本当に「リアルタイム」？

REST のポーリングはニアリアルタイムです。レイテンシの予算は、ポーリング間隔に API のレスポンスタイム（速いエンドポイントで約300ms）を加えたものです。5秒間隔では、最悪ケースはツイートが投稿されてからコールバックが発火するまでおよそ5.3秒です。監視のユースケースの大多数では、それがリアルタイムの実務的な定義を満たします。真のストリームを必要とするのは、ミリ秒スケールのトレーディングとライブイベントのオークションだけです。

1つの API キーで何アカウント監視できる？

Sorsa では、1つの API キーで、X リストを通じて事実上無制限の数のアカウントを監視できます。1つのリストは最大5,000アカウントを保持し、ポーリングごとに1回の /list-tweets リクエストとして数えられます。複数のリストが一律の毎秒20リクエストの制限内で並行して走り、それが単一のキーで数百の同時監視ジョブの余地を残します。

レート制限に達したらどうなる？

Sorsa は、一律の毎秒20リクエストの制限を超えると HTTP 429 レスポンスを返します。1秒バックオフして再試行すれば、ループは続きます。ペナルティボックスもロックアウトもありません。ほとんどのポーリング頻度は毎秒20リクエストを十分に下回るため、本番のモニターが429 を見ることはめったになく、より高い制限もリクエストに応じて利用できます。

スクリプトが再起動したときに重複アラートをどう避ける？

重複アラートを避けるには、各成功したポーリングの後、最後に見たツイート ID を永続的なストレージ（ファイル、Redis、またはデータベース）に保存します。起動時にその ID を読み込み、それをベースラインとして使うことで、ループはチェックポイントより新しいツイートだけをディスパッチします。永続化がなければ、再起動は古いツイートを再生するか、静かにギャップをスキップします。

非公開・保護されたアカウントを監視できる？

どのツールも非公開・保護された Twitter/X アカウントにはアクセスできず、Sorsa は公開データだけを浮かび上がらせます。追跡中のアカウントが監視の途中で非公開になると、エンドポイントは空のリストを返し、ポーリングループはエラーなく続きます。これはプラットフォームレベルのプライバシールールであって、特定のプロバイダーに固有の制限ではありません。

マネージドな Webhook はサポートされている？

Sorsa はマネージドな Webhook 製品を出していません。サポートされるリアルタイムの経路は本ガイドのポーリングパターンで、そこでは自分のコールバックが各新しいツイートをルーティングします。利点は、フィルタリング、ファンアウト、リトライロジックを自分のコードで完全に制御できることです。特にベンダーがホストするプッシュ配信が欲しいチームは、エンタープライズプランでの公式 X API の Account Activity Webhook を見るでしょう。

返されるデータはどれだけ新鮮？

データはすべてのリクエストで新鮮で、自分の呼び出しとプラットフォームの間にキャッシュ層はありません。ツイートが半秒前に投稿されたなら、次のポーリングがそれを拾います。約300ms のレスポンスタイムと組み合わさって、その新鮮さが、事後の分析だけでなく監視にポーリングを使えるものにしています。

リアルタイム監視を過去分のバックフィルと組み合わせられる？

はい、そしてほとんどの本番パイプラインは両方を行います。監視に使う同じエンドポイント（/user-tweets、/search-tweets、/list-tweets）は、1回限りのバックフィルのために履歴を後方にたどる next_cursor パラメータを受け付け、それから新しいデータには前向きにポーリングループに切り替えます。バックフィル側については、過去のTwitterデータ ガイドをご覧ください。

始め方 {#getting-started}

独自のリアルタイム監視パイプラインを構築するには：

  1. API キーを取得する。Sorsa ダッシュボード から。1つのキーがすべてのエンドポイントで機能し、すべてのアカウントは100回分の無料リクエスト（1回限り、クレジットカード不要）で始まるため、プランを選ぶ前にループを組めます。
  2. 対話的にテストする。API playground で、既知のユーザー名やリスト ID で /user-tweets や /list-tweets を叩き、ライブの結果が見えることを確認します。
  3. ループの1つをコピーする（単一アカウント、リスト、またはキーワード）。この記事のものから、API キーを差し替えます。
  4. コールバックを追加する。アラートを送りたいところ（Slack、Discord、内部 API、キュー）にルーティングします。
  5. 本番向けの堅牢化を加える（状態の永続化、バックオフ、分離した処理）。基本のループが安定したら。

間隔の表から月間の使用量を見積もり、プランを選び、出荷します。クイックスタートガイド が、最初の呼び出しを端から端まで案内します。標準のプランを超えるより高いレート制限や量が必要なら、営業に相談 してください。カスタムクォータを一緒に詰めます。

監修：Keksich（Sorsa創業者、マーケター兼X APIリサーチャー）

本ガイドは、2022年以降、累計50億件を超えるリクエストを処理してきた Twitter/X の代替 API である Sorsa を構築・運用するチームによって書かれ、維持されています。すべてのコードサンプルは、稼働中の /user-tweets、/list-tweets、/search-tweets のエンドポイントに対して動き、ポーリング・バックオフ・永続化のパターンは、本番で運用する監視ループから取られています。コスト比較は、2026年4月の更新後に有効な公式 X API の従量課金モデルを反映しており、料金とレート制限の詳細は2026年7月に確認しました。チームの詳細は about ページ にあります。