X API v2 の Powerstream 低レイテンシーストリーミング API

Powerstream は、公開 X データにリアルタイムでアクセスするための最低レイテンシーのストリーミング API です。データのハイドレーションと配信を優先する他のストリーミングエンドポイント (P99 レイテンシー約 6-7 秒) とは異なり、Powerstream は速度に最適化されており、最小限の遅延でデータを配信します。レガシーの GNIP Powertrack API と同様に、キーワード、オペレーター、メタデータに基づいて Post をフィルタリングするためにルールを使用します。Powerstream エンドポイントへの永続的な HTTP 接続を確立すると、一致する Post をリアルタイムで受信できます。現在、Powerstream は最大 1,000 ルールをサポートし、各ルールは 2048 文字です。

主な機能

リアルタイムデータ配信: Post が公開された瞬間にストリーミングする最低レイテンシーオプション。
精密なフィルタリング: オペレーターを使ったブールクエリで、必要なデータだけを正確にフィルタリング。
配信: HTTP/1.1 の chunked transfer encoding を用いた JSON レスポンス。
ローカルデータセンターサポート: レプリケーションラグを回避してレイテンシーをさらに削減するために、ローカルデータセンターからのみ Post を取得。

Powerstream API は、選択された Enterprise プランで利用可能なプレミアムサービスです。Powerstream へのアクセスや Enterprise サービスについて詳しく知りたい方は、Enterprise リクエストフォームを送信して Sales チームまでお問い合わせください。 Powerstream がお客様のニーズをどのようにサポートできるかについて、喜んでご相談いたします。

クイックスタート

このセクションでは、requests ライブラリを使った Python で PowerStream エンドポイントをすぐに始める方法を紹介します。pip install requests でインストールできます。すべての例は OAuth 2.0 Bearer Token 認証を使用します。YOUR_BEARER_TOKEN を実際のトークンに置き換えてください (安全に保管してください。例: os.getenv('BEARER_TOKEN') 経由)。各エンドポイントをコードスニペット付きで説明します。以下のインポートを冒頭に含めているものとします:

import requests
import json
import time
import sys
import os  # For env vars

セットアップ

bearer_token = os.getenv('BEARER_TOKEN') or "YOUR_BEARER_TOKEN"  # Use env var for security
base_url = "https://api.x.com/2/powerstream"
rules_url = f"{base_url}/rules"  # For rule management
headers = {
   "Authorization": f"Bearer {bearer_token}",
   "Content-Type": "application/json"
}

1. ルールを作成 (POST /rules)

ストリームをフィルタリングするためのルールを追加します。

例

data = {
   "rules": [
       {
           "value": "(cat OR dog) lang:en -is:retweet",
           "tag": "pet-monitor"
       },
       # Add more rules as needed (up to 100)
   ]
}

response = requests.post(rules_url, headers=headers, json=data)
if response.status_code == 201:
   rules_added = response.json().get("data", {}).get("rules", [])
   print("Rules added:")
   for rule in rules_added:
       print(f"ID: {rule['id']}, Value: {rule['value']}, Tag: {rule.get('tag', 'N/A')}")
else:
   print(f"Error {response.status_code}: {response.text}")

2. ルールを削除 (POST /rules)

ルールを ID (推奨) または値で削除します。

例

data = {
   "rules": [
       {
           "value": "(cat OR dog) lang:en -is:retweet",
           "tag": "pet-monitor"
       },
       # Add more rules as needed (up to 100)
   ]
}

response = requests.delete(rules_url, headers=headers, json=data)
if response.status_code == 200:
   deleted = response.json().get("data", {})
   print(f"Deleted count: {deleted.get('deleted', 'N/A')}")
   if 'not_deleted' in deleted:
       print("Not deleted:", deleted['not_deleted'])
else:
   print(f"Error {response.status_code}: {response.text}")

ヒント: すべてのルールを削除するには、まず GET で取得して ID を抽出し、一括削除します。

3. ルールを取得 (GET /rules)

すべてのアクティブなルールを取得します。

response = requests.get(rules_url, headers=headers)
if response.status_code == 200:
   rules = response.json().get("data", {}).get("rules", [])
   if rules:
       print("Active rules:")
       for rule in rules:
           print(f"ID: {rule['id']}, Value: {rule['value']}, Tag: {rule.get('tag', 'N/A')}")
   else:
       print("No active rules.")
else:
   print(f"Error {response.status_code}: {response.text}")

4. PowerStream (GET /stream)

リアルタイムで低レイテンシーの Post 用にストリームに接続します。行ごとに読み取るために stream=True を使用します。堅牢性のために再接続ロジックを実装します。

例

stream_url = base_url

def main():
   while True:
       response = requests.request("GET", stream_url, headers=headers, stream=True)
       print(response.status_code)
       for response_line in response.iter_lines():
           if response_line:
               json_response = json.loads(response_line)
               print(json.dumps(json_response, indent=4, sort_keys=True))
               if response.status_code != 200:
                   print(response.headers)
                   raise Exception(
                       "Request returned an error: {} {}".format(
                           response.status_code, response.text
                       )
                   )

ローカルデータセンターサポート

レイテンシー最適化のために、Powerstream は接続が確立されたローカルデータセンターで発信または作成された Post のみを取得するオプションを提供します。これにより、レプリケーションラグを回避し、他のデータセンターからの Post に比べて高速に配信できます。これを有効にするには、ストリームエンドポイントにクエリパラメーター ?localDcOnly=true を付加します (例: /2/powerstream?localDcOnly=true)。接続しているデータセンターは、ストリームの最初のデータペイロードとレスポンスの HTTP ヘッダーの両方で示されます。コードでの使用例:

# For local datacenter only:
stream_url = "https://api.x.com/2/powerstream?localDcOnly=true"

localDcOnly パラメーターが有効になっている場合、ストリームが最初に接続すると、使用されているローカルデータセンターを示す次のレスポンスヘッダーが含まれます:

'x-powerstream-datacenter': 'atla',
'x-powerstream-localdconly': 'true'

これに加えて、データセンターを指定する初期ペイロードも送信されます:

{
    "type": "connection_metadata",
    "datacenter": "atla",
    "timestamp": 1762557264155
}

ヒント: レイテンシーを最適化するために、異なる地理的位置 (例えば米国東海岸のアトランタ付近と米国西海岸のポートランド付近) から接続をセットアップし、それぞれで localDcOnly=true を有効にします。これにより、それぞれのデータセンターからの Post に高速にアクセスできます。データセンターをまたがったデータを結合するために、ストリームをクライアント側で集約してください。

オペレーター

フィルタリング用のルールを設定するには、キーワードとオペレーターを使用できます。

Powerstream オペレーター

利用可能なオペレーターの完全なリスト

レスポンス

Powerstream API のペイロード形式は、レガシー GNIP Powertrack API と同じです。JSON レスポンスのサンプルは以下のとおりです:

レスポンス例

[
   {
       "created_at": "Tue Mar 21 20:50:14 +0000 2006",
       "id": 20,
       "id_str": "20",
       "text": "just setting up my twttr",
       "truncated": false,
       "entities": {
           "hashtags": [],
           "symbols": [],
           "user_mentions": [],
           "urls": []
       },
       "source": "<a href=\"http://x.com\" rel=\"nofollow\">X Web Client</a>",
       "in_reply_to_status_id": null,
       "in_reply_to_status_id_str": null,
       "in_reply_to_user_id": null,
       "in_reply_to_user_id_str": null,
       "in_reply_to_screen_name": null,
       "user": {
           "id": 12,
           "id_str": "12",
           "name": "jack",
           "screen_name": "jack",
           "location": "",
           "description": "no state is the best state",
           "url": "https://t.co/ZEpOg6rn5L",
           "entities": {
               "url": {
                   "urls": [
                       {
                           "url": "https://t.co/ZEpOg6rn5L",
                           "expanded_url": "http://primal.net/jack",
                           "display_url": "primal.net/jack",
                           "indices": [
                               0,
                               23
                           ]
                       }
                   ]
               },
               "description": {
                   "urls": []
               }
           },
           "protected": false,
           "followers_count": 6427829,
           "friends_count": 3,
           "listed_count": 32968,
           "created_at": "Tue Mar 21 20:50:14 +0000 2006",
           "favourites_count": 36306,
           "utc_offset": null,
           "time_zone": null,
           "geo_enabled": true,
           "verified": false,
           "statuses_count": 30134,
           "lang": null,
           "contributors_enabled": false,
           "is_translator": false,
           "is_translation_enabled": false,
           "profile_background_color": "EBEBEB",
           "profile_background_image_url": "http://abs.twimg.com/images/themes/theme7/bg.gif",
           "profile_background_image_url_https": "https://abs.twimg.com/images/themes/theme7/bg.gif",
           "profile_background_tile": false,
           "profile_image_url": "http://pbs.twimg.com/profile_images/1661201415899951105/azNjKOSH_normal.jpg",
           "profile_image_url_https": "https://pbs.twimg.com/profile_images/1661201415899951105/azNjKOSH_normal.jpg",
           "profile_banner_url": "https://pbs.twimg.com/profile_banners/12/1742427520",
           "profile_link_color": "990000",
           "profile_sidebar_border_color": "DFDFDF",
           "profile_sidebar_fill_color": "F3F3F3",
           "profile_text_color": "333333",
           "profile_use_background_image": true,
           "has_extended_profile": true,
           "default_profile": false,
           "default_profile_image": false,
           "following": null,
           "follow_request_sent": null,
           "notifications": null,
           "translator_type": "regular",
           "withheld_in_countries": []
       },
       "geo": null,
       "coordinates": null,
       "place": null,
       "contributors": null,
       "is_quote_status": false,
       "retweet_count": 122086,
       "favorite_count": 263321,
       "favorited": false,
       "retweeted": false,
       "lang": "en"
   }
]

制限とベストプラクティス

レート制限: ルール管理は 24 時間あたり 50 リクエスト。ストリームには制限はありません (ただし接続制限が適用されます)。
再接続: 切断時は指数バックオフを使用。
モニタリング: Connection: keep-alive ヘッダーを使用。

ストリーミングの基礎

ストリーミングデータの消費

ストリーミングクライアントのベストプラクティス

切断の処理

優雅に再接続する

大容量対応

高スループットを処理

リカバリと冗長性

レジリエントなアプリケーションを構築

Getting Started

Products

Fundamentals

Partners & Customers

Resources

X API v2 の Powerstream 低レイテンシーストリーミング API

主な機能

クイックスタート

セットアップ

1. ルールを作成 (POST /rules)

2. ルールを削除 (POST /rules)

3. ルールを取得 (GET /rules)

4. PowerStream (GET /stream)

ローカルデータセンターサポート

オペレーター

Powerstream オペレーター

レスポンス

制限とベストプラクティス

ストリーミングの基礎

ストリーミングデータの消費

切断の処理

大容量対応

リカバリと冗長性

​主な機能

​クイックスタート

​セットアップ

​1. ルールを作成 (POST /rules)

​2. ルールを削除 (POST /rules)

​3. ルールを取得 (GET /rules)

​4. PowerStream (GET /stream)

​ローカルデータセンターサポート

​オペレーター

Powerstream オペレーター

​レスポンス

​制限とベストプラクティス

​ストリーミングの基礎

ストリーミングデータの消費

切断の処理

大容量対応

リカバリと冗長性

主な機能

クイックスタート

セットアップ

1. ルールを作成 (POST /rules)

2. ルールを削除 (POST /rules)

3. ルールを取得 (GET /rules)

4. PowerStream (GET /stream)

ローカルデータセンターサポート

オペレーター

レスポンス

制限とベストプラクティス

ストリーミングの基礎