跳到文档正文
文档

WebSocket 快速开始

打开一个服务端 WebSocket 连接，传入 v1 协议和 API-key 协议，然后根据 envelope 字段 `t` 和 `op` 分发处理。

5 分钟 smoke test

先在 dashboard 添加一个监控账号，然后从服务端脚本打开 socket。安静的 socket 通常表示监控列表为空，或该账号在连接后还没有发布新内容。

TypeScript smoke testtypescript
import WebSocket from "ws";
 
const apiKey = process.env.TWEETSTREAM_API_KEY;
if (!apiKey) {
  throw new Error("Missing TWEETSTREAM_API_KEY");
}
 
const ws = new WebSocket("wss://ws.tweetstream.io/ws", [
  "tweetstream.v1",
  `tweetstream.auth.token.${apiKey}`,
]);
 
ws.on("open", () => {
  console.log("TweetStream connected");
});
 
ws.on("message", (raw) => {
  const event = JSON.parse(raw.toString());
  console.log(event.t, event.op, event.d);
});
 
ws.on("close", (code, reason) => {
  console.log("TweetStream closed", code, reason.toString());
});
 

连接

标准实时端点是 `wss://ws.tweetstream.io/ws`。付费 Ultra Speed 客户端应使用 dashboard 中的 Ultra API key 连接 `wss://ws-iad.tweetstream.io/ws`。服务端会选择 `tweetstream.v1` 作为应用协议，并在接受连接前移除认证 token 协议。这个示例使用指数退避重连，因为实时 socket 应作为长期运行的基础设施处理。

Node.js consumertypescript
import WebSocket from "ws";
 
type StreamEvent = {
  t?: string;
  op?: string;
  d?: {
    author?: { handle?: string };
    detected?: unknown;
    text?: string;
  };
};
 
const apiKey = process.env.TWEETSTREAM_API_KEY;
if (!apiKey) {
  throw new Error("Missing TWEETSTREAM_API_KEY");
}
 
let retry = 0;
 
function connect() {
  const ws = new WebSocket("wss://ws.tweetstream.io/ws", [
    "tweetstream.v1",
    `tweetstream.auth.token.${apiKey}`,
  ], {
    headers: {
      "fly-prefer-region": "iad",
    },
  });
 
  ws.on("upgrade", (response) => {
    console.log("connected region", response.headers["x-fly-region"]);
  });
 
  ws.on("open", () => {
    retry = 0;
    console.log("TweetStream connected");
  });
 
  ws.on("message", (raw) => {
    const event = JSON.parse(raw.toString()) as StreamEvent;
 
    if (event.t === "tweet" && event.op === "content") {
      const tweet = event.d;
      console.log(tweet?.author?.handle, tweet?.text);
    }
 
    if (event.t === "tweet" && event.op === "meta") {
      console.log("enrichment", event.d.detected);
    }
  });
 
  ws.on("close", (code, reason) => {
    console.warn("TweetStream closed", code, reason.toString());
    const delayMs = Math.min(30_000, 1_000 * 2 ** retry) + Math.floor(Math.random() * 500);
    retry += 1;
    setTimeout(connect, delayMs);
  });
}
 
connect();
 

区域路由

TweetStream 接受可选的 `fly-prefer-region` header，并在 `x-fly-region` 中返回实际选择的区域。公开区域为 `iad`、`nrt` 和 `ams`；除非你的基础设施实测其他路径更好，否则优先使用 `iad`。

区域位置默认建议
iad美国弗吉尼亚 Ashburn优先使用；通常最快
nrt日本东京当交易系统更靠近亚洲时使用
ams荷兰阿姆斯特丹当交易系统更靠近欧洲时使用
指定并读取区域typescript
import WebSocket from "ws";
 
const ws = new WebSocket("wss://ws.tweetstream.io/ws", [
  "tweetstream.v1",
  `tweetstream.auth.token.${process.env.TWEETSTREAM_API_KEY}`,
], {
  headers: {
    "fly-prefer-region": "iad",
  },
});
 
ws.on("upgrade", (response) => {
  console.log("connected region", response.headers["x-fly-region"]);
});
 

Python consumer

任何能传入两个 subprotocol 的 WebSocket 运行时都可以使用。Python 客户端也应该在传输错误后重连，原因与 Node 客户端相同。

Python websocketspython
import asyncio
import json
import os
import websockets
 
API_KEY = os.environ["TWEETSTREAM_API_KEY"]
URI = "wss://ws.tweetstream.io/ws"
PROTOCOLS = ["tweetstream.v1", f"tweetstream.auth.token.{API_KEY}"]
 
async def main():
    retry = 0
    while True:
        try:
            async with websockets.connect(URI, subprotocols=PROTOCOLS) as ws:
                retry = 0
                async for raw in ws:
                    event = json.loads(raw)
                    if event["t"] == "tweet" and event["op"] == "content":
                        tweet = event["d"]
                        print(tweet.get("author", {}).get("handle"), tweet.get("text"))
        except Exception as error:
            wait = min(30, 2 ** retry)
            retry += 1
            print(f"reconnecting in {wait}s after {error}")
            await asyncio.sleep(wait)
 
asyncio.run(main())
 

Envelope 合约

每条实时消息都是一个 envelope。先按 `t` 路由，再按 `op` 路由，并保持 handler 幂等，因为历史回放可能返回你已经实时处理过的事件。

Envelope 类型typescript
type VerifiedType = 'blue' | 'business' | 'government' | 'none';
type TweetVerifiedLabel = {
  badge: string | null;
  description: string;
  url: string | null;
};
 
type TweetAuthor = {
  banner?: string;
  bio?: string;
  followersCount?: number;
  followingCount?: number;
  id?: string;
  joinedAt?: number;
  location?: string;
  metrics?: {
    likes?: number;
    tweets?: number;
  };
  // Includes a leading @ when present, for example "@elonmusk".
  handle?: string;
  name?: string;
  platform?: 'twitter' | 'truth_social';
  profileImage?: string;
  url?: string;
  verifiedLabel?: TweetVerifiedLabel;
  verifiedType?: VerifiedType;
};
 
type Media = {
  url: string;
  type?: 'image' | 'video' | 'gif';
  thumbnail?: string;
};
 
type TweetUrl = {
  url: string;
  name?: string;
  tco?: string;
};
 
type TweetMention = {
  handle?: string;
  id?: string;
  name?: string;
};
 
type TweetContentKind = 'post' | 'reply' | 'quote' | 'retweet';
 
type TweetReference = {
  type: 'reply' | 'quote' | 'retweet';
  tweetId?: string;
  text?: string;
  translatedText?: string;
  author?: TweetAuthor;
  media?: Media[];
  // True when the upstream only supplied truncated text.
  partial?: boolean;
  quoted?: TweetReference;
};
 
type TweetContent = {
  tweetId: string;
  kind: TweetContentKind;
  // Original tweet text when the event includes both original and translated text.
  text: string;
  // Translation, present only when available.
  translatedText?: string;
  createdAt: number;
  author: TweetAuthor;
  link?: string;
  media?: Media[];
  mentions?: TweetMention[];
  // True when text is known to be truncated.
  partial?: boolean;
  receivedAt?: number;
  urls?: TweetUrl[];
  ref?: TweetReference;
};
 
type TweetMeta = {
  tweetId: string;
  ocr?: {
    text: string;
  };
  detected?: {
    tokens?: Array<{
      symbol?: string;
      name?: string;
      contract?: string;
      chain?: string;
      networkId?: number;
      priceUsd?: number;
      sources: Array<'text' | 'ocr'>;
    }>;
    cex?: Array<{
      exchange: 'bybit' | 'binance' | 'hyperliquid';
      symbol?: string;
      priceUsd?: number;
      url?: string;
      baseAsset?: string;
      quoteAsset?: string;
      sources: Array<'text' | 'ocr'>;
    }>;
    prediction?: Array<{
      exchange: 'polymarket' | 'kalshi';
      marketId?: string;
      title?: string;
      priceUsd?: number;
      url?: string;
      sources: Array<'text' | 'ocr'>;
    }>;
  };
};
 
type TweetUpdate = {
  tweetId: string;
  kind?: TweetContentKind;
  text?: string;
  translatedText?: string;
  author?: TweetAuthor;
  media?: Media[];
  mentions?: TweetMention[];
  partial?: boolean;
  receivedAt?: number;
  urls?: TweetUrl[];
  ref?: TweetContent['ref'];
};
 
type TweetDeleteEvent = {
  tweetId: string;
  eventId: string;
  deletedAt?: number;
  receivedAt?: number;
  author?: TweetAuthor;
  text?: string;
};
 
type TweetPinEvent = {
  tweetId: string;
  eventId: string;
  observedAt: number;
  receivedAt?: number;
  action: 'pin' | 'unpin';
  author: TweetAuthor;
  text?: string;
  tweet?: TweetContent;
};
 
type AccountEventActor = TweetAuthor & {
  websiteUrl?: string;
};
 
type ProfileUpdateEvent = {
  kind: 'PROFILE';
  eventId: string;
  observedAt: number;
  receivedAt?: number;
  actor: AccountEventActor;
  changes: {
    avatar?: string;
    banner?: string;
    bio?: string;
    handle?: string;
    location?: string;
    name?: string;
    verifiedLabel?: TweetVerifiedLabel | null;
    websiteUrl?: string | null;
  };
  previous?: {
    avatar?: string;
    banner?: string;
    bio?: string;
    handle?: string;
    location?: string;
    name?: string;
    verifiedLabel?: TweetVerifiedLabel | null;
    websiteUrl?: string | null;
  };
};
 
type FollowEvent = {
  kind: 'FOLLOW' | 'UNFOLLOW';
  eventId: string;
  observedAt: number;
  receivedAt?: number;
  actor: AccountEventActor;
  target: AccountEventActor;
};
 
type TwitterHandlesResult = {
  action: 'follow' | 'unfollow';
  requestId: string | null;
  results: Array<{
    input: string;
    handle?: string;
    name?: string;
    normalizedHandle?: string;
    profileImage?: string;
    twitterId?: string;
    state:
      | 'added'
      | 'already_following'
      | 'invalid_input'
      | 'duplicate'
      | 'not_found'
      | 'failed'
      | 'removed'
      | 'not_following';
    message?: string;
  }>;
  error: string | null;
};
 
type EnvelopeBase<
  TFamily extends 'tweet' | 'account' | 'control',
  TOp extends string,
  TPayload extends object,
> = {
  v: 1;
  t: TFamily;
  op: TOp;
  id?: string;
  ts: number;
  d: TPayload;
};
 
type TweetContentEnvelope = EnvelopeBase<'tweet', 'content', TweetContent>;
type TweetMetaEnvelope = EnvelopeBase<'tweet', 'meta', TweetMeta>;
type TweetUpdateEnvelope = EnvelopeBase<'tweet', 'update', TweetUpdate>;
type TweetLifecycleEnvelope = EnvelopeBase<
  'tweet',
  'delete' | 'pin' | 'unpin',
  TweetDeleteEvent | TweetPinEvent
>;
type AccountEnvelope = EnvelopeBase<
  'account',
  'profile_update' | 'follow' | 'unfollow',
  ProfileUpdateEvent | FollowEvent
>;
type ControlEnvelope = EnvelopeBase<
  'control',
  'auth_ping' | 'auth_pong' | 'twitter_handles_result',
  TwitterHandlesResult
>;
 
type TweetStreamEnvelope =
  | TweetContentEnvelope
  | TweetMetaEnvelope
  | TweetUpdateEnvelope
  | TweetLifecycleEnvelope
  | AccountEnvelope
  | ControlEnvelope;
 
function route(event: TweetStreamEnvelope) {
  if (event.t === 'tweet' && event.op === 'content') {
    console.log(event.d.tweetId, event.d.text);
  }
  if (event.t === 'tweet' && event.op === 'meta') {
    console.log(event.d.tweetId, event.d.detected);
  }
}
 

操作类型

FamilyOperation含义
tweetcontent原始发帖、回复、引用、转推，或支持的 Truth Social 帖子内容
tweetmeta已有 tweetId 的富化信息：OCR、检测到的代币、CEX、预测市场
tweetupdate已知 tweetId 的渐进式内容更新
tweetdelete, pin, unpin观察到的推文生命周期事件
accountprofile_update, follow, unfollow被监控账号的账号状态变化
controltwitter_handles_resultWebSocket handle-management 命令的结果

心跳和断开

TweetStream 每 30 秒发送原生 WebSocket ping frame。标准 Node 和 Python WebSocket 客户端会自动回复 pong frame。如果客户端停止响应，服务端会终止 socket；重连后可用 History API 回补已存内容、资料和关注事件。

重连后回补

记录你处理过的最后一个 content 时间戳或 tweetId。重连后用有界的 `startDate` 调用 History API，并以幂等方式回放，避免下游机器人重复执行。delete、pin、unpin 等生命周期事件属于实时流事件；下游状态应能在这些事件实时到达时继续处理。

历史回放窗口typescript
const lastSeen = new Date(Date.now() - 60_000).toISOString();
const url = new URL("https://api.tweetstream.io/api/history");
 
url.searchParams.set("handles", "marketdesk");
url.searchParams.set("startDate", lastSeen);
url.searchParams.set("limit", "1000");
url.searchParams.set("type", "TWEET");
 
const response = await fetch(url, {
  headers: {
    Authorization: `Bearer ${process.env.TWEETSTREAM_API_KEY}`,
  },
});
 
const replay = await response.json();
 

握手错误

WebSocket upgrade 失败会在 socket 被接受前返回 JSON body。这些错误可被客户端观察，也可以安全记录。

状态可能原因修复方式
400WebSocket 协议 header 缺失或无效发送 tweetstream.v1 和有效认证 token 协议
401API key 缺失、为空或无效在 dashboard 重新生成或轮换 API key
403订阅状态不允许实时流开始试用、续费或升级账号
429活跃 WebSocket 连接数达到限制关闭旧 socket 或升级套餐
503连接限制器暂时不可用使用退避策略重试
Upgrade 拒绝响应json
{
  "error": "WebSocket connection limit reached (3/3 active). Close an existing connection and retry.",
  "status": 429
}
 

限制和重试

套餐限制会控制活跃 WebSocket 连接、监控账号数量和 History API 权限。如果端点返回 `429`，暂停对应流程，并按响应窗口或你的常规退避间隔重试。

接口限制信号推荐处理
WebSocketupgrade 时返回 429关闭不用的 socket，或升级到更多连接数的套餐
History API限流时返回 retryAfterSeconds等待后再回放下一个窗口
监控账号/api/me 返回套餐用量批量添加前检查 count 和 limit

测量 snowflake 延迟

对于 X/Twitter 事件，解码 tweet snowflake 时间戳，并与消息到达瞬间记录的本地接收时间比较。测量主机必须保持 NTP 时间同步，否则 wall-clock 延迟没有意义。

Snowflake 延迟typescript
type ContentEvent = {
  t?: string;
  op?: string;
  d?: {
    tweetId?: string;
  };
};
 
const TWITTER_EPOCH_MS = 1_288_834_974_657n;
 
function tweetIdToTimestampMs(tweetId: string) {
  const id = BigInt(tweetId);
  return Number((id >> 22n) + TWITTER_EPOCH_MS);
}
 
function measureSnowflakeLatency(tweetId: string) {
  const arrivedAtMs = Date.now();
  const tweetedAtMs = tweetIdToTimestampMs(tweetId);
  return arrivedAtMs - tweetedAtMs;
}
 
ws.on("message", (raw) => {
  const event = JSON.parse(raw.toString()) as ContentEvent;
  const tweetId = event.d?.tweetId;
 
  if (event.t === "tweet" && event.op === "content" && tweetId) {
    console.log("snowflake latency ms", measureSnowflakeLatency(tweetId));
  }
});
 

生产建议

  • close 或网络错误后使用退避策略重连。
  • 优先按 envelope id 去重；没有 id 时按 `d.tweetId` 加 operation 去重。
  • 把 `meta` 视为某条已路由推文的迟到富化信息。
  • 跟踪活跃 WebSocket 连接数，避免超过套餐限制。