---
title: "完整 WebSocket 示例"
canonical_url: "https://tweetstream.io/zh-cn/docs/examples"
last_updated: 2026-09-29
---

# 完整 WebSocket 示例

复制一个 TypeScript 或 Python 程序，用你的 API key 运行，就能看到它连接、重连、合并帖子更新、解析引用链，并为每条帖子计时。

Canonical URL: https://tweetstream.io/zh-cn/docs/examples

> 使用 AI 编程助手？
>
> 把本页的 Markdown 版本和协议规范交给它，并让它遵守以下规则：
>
> 1. 按 `(platform, tweetId)` 保存每条帖子。
> 2. 把 `tweet/update` 的字段合并到已保存的内容上。更新只携带发生变化的字段。
> 3. 更新带有 `ref` 时，替换已保存的 `ref`，并沿 `ref.subtweet` 读取后续引用链。
> 4. 每条 `tweet/meta` 都是完整快照，用它替换已保存的 `meta`。
> 5. 遇到未处理的 `t` 和 `op` 组合时跳过该消息，保持连接打开。
>
> - [本页 Markdown](https://tweetstream.io/zh-cn/docs/examples.md)
> - [AsyncAPI 规范](https://tweetstream.io/zh-cn/docs/asyncapi.json)
> - [OpenAPI 规范](https://tweetstream.io/zh-cn/docs/openapi.json)

## 运行程序

把 `YOUR_API_KEY` 替换为控制台中的 key，试用 key 也可以。监控账号发帖时才会收到帖子，所以请先在控制台添加至少一个账号，或设置 `TWEETSTREAM_ACCOUNTS`。

### TypeScript

需要 Node.js 22.18 或更高版本，它可以直接运行 TypeScript 文件。程序没有依赖。

1. 复制程序

```typescript
// TweetStream complete example: connect, reconnect, handle every event, and measure latency.
// Node.js 22.18 or later runs this TypeScript file directly. No dependencies.
//
//   TWEETSTREAM_API_KEY=YOUR_API_KEY node tweetstream-example.ts
//
// Optional settings:
//   TWEETSTREAM_ACCOUNTS=handle1,handle2               add these accounts before connecting
//   TWEETSTREAM_WS_URL=wss://ws-iad.tweetstream.io/ws  use the US endpoint

// The messages this program reads, limited to the fields it uses. Every field of every event:
// https://tweetstream.io/docs/asyncapi.json

// Every server message is a v1 envelope. Route on t (channel) and op (operation); d is the payload.
type Envelope = { v: 1; t: string; op: string; id?: string; ts: number; d: unknown };

// Posts name their author by handle. Some account events use username instead.
type Account = { id?: string; handle?: string; username?: string; platform?: string };

type PostKind = "post" | "quote" | "reply" | "retweet";

// A post that this one quotes, replies to, or reposts. It can reference another through subtweet.
type Reference = {
  type: Exclude<PostKind, "post">;
  tweetId: string;
  author?: Account;
  text?: string;
  subtweet?: Reference;
};

// tweet/content: a new post. author.platform is "twitter" when it is missing.
type PostContent = {
  tweetId: string;
  text: string;
  createdAt: number; // Unix milliseconds
  kind: PostKind;
  author: Account;
  ref?: Reference;
};

// tweet/update: only the fields that changed. textUpdateType "completion" delivers the full text
// of a post that arrived shortened; it is not an edit.
type PostUpdate = Partial<Omit<PostContent, "createdAt">> & {
  tweetId: string;
  platform?: string;
  textUpdateType?: "completion";
};

// tweet/meta: a complete snapshot of the tokens, markets, and image text found in a post.
type PostMeta = {
  tweetId: string;
  platform?: string;
  detected?: {
    tokens?: Array<{ symbol?: string; contract?: string; priceUsd?: number }>;
    prediction?: Array<{ title?: string; priceUsd?: number }>;
  };
  ocr?: { text: string };
};

// tweet/delete, tweet/pin, and tweet/unpin.
type PostEvent = { tweetId: string; platform?: string; author?: Account };

type Coin = { symbol?: string; address?: string };

// The payload of each t/op pair this program handles.
type Payloads = {
  "control/auth_pong": Record<string, never>;
  "tweet/content": PostContent;
  "tweet/update": PostUpdate;
  "tweet/meta": PostMeta;
  "tweet/delete": PostEvent;
  "tweet/pin": PostEvent;
  "tweet/unpin": PostEvent;
  "account/follow": { actor: Account; target: Account };
  "account/unfollow": { actor: Account; target: Account };
  "account/profile_update": { actor: Account; changes: Record<string, unknown> };
  "account/affiliate_update": { organization: Account; member: Account; action: string };
  "account/callout": { caller: Account; token: Coin };
  "account/callout_update": { updateOf: string; token?: Coin };
  "account/thesis": { author: Account; coin: Coin; thesis: { text: string } };
  "account/token_verified": { coin: Coin };
  "news/article": { source?: { name?: string }; title?: string };
  "charity/added": { name: string };
  "control/twitter_handles_result": {
    results: Array<{ input: string; state: string }>;
    error: string | null;
  };
};
type Pair = keyof Payloads;
type Handler<P> = (d: P, receivedAt: number) => void;

// What this program keeps for each post.
type StoredPost = {
  content: Partial<PostUpdate> | null;
  meta: PostMeta | null;
  deleted: boolean;
  pinned: boolean;
  hasContent: boolean;
};

const API_KEY = process.env.TWEETSTREAM_API_KEY?.trim();
const DASHBOARD = "https://tweetstream.io/dashboard?tab=api";
if (!API_KEY) {
  console.error(`Set TWEETSTREAM_API_KEY. Copy your key from ${DASHBOARD}`);
  process.exit(1);
}
const WS_URL = process.env.TWEETSTREAM_WS_URL || "wss://ws-global.tweetstream.io/ws";
const API_URL = "https://api.tweetstream.io";
const ACCOUNTS = (process.env.TWEETSTREAM_ACCOUNTS ?? "")
  .split(",")
  .map((handle) => handle.trim().replace(/^@/, ""))
  .filter(Boolean);
const TWITTER_EPOCH_MS = 1288834974657n;

const posts = new Map<string, StoredPost>(); // "platform:tweetId" -> stored post
const counts = new Map<string, number>(); // "t/op" -> number of events seen
const latencies: number[] = []; // publish-to-receipt ms for the most recent 10,000 new posts
let attempt = 0;
let hinted = false;

const log = (label: string, text: string) => console.log(`${label.padEnd(9)} ${text}`);
const at = (account: Account | undefined) =>
  `@${String(account?.handle ?? account?.username ?? "unknown").replace(/^@/, "")}`;
const clip = (text: unknown, max = 80) => {
  const flat = String(text ?? "").replace(/\s+/g, " ").trim();
  return JSON.stringify(flat.length > max ? `${flat.slice(0, max - 1)}…` : flat);
};

// Posts are keyed by (platform, tweetId). Content carries the platform on its author;
// later operations carry it on d.platform.
const platformOf = (d: PostEvent) => d.platform ?? d.author?.platform ?? "twitter";
function postFor(d: PostEvent, platform = platformOf(d)): StoredPost {
  const key = `${platform}:${d.tweetId}`;
  let post = posts.get(key);
  if (!post) {
    post = { content: null, deleted: false, hasContent: false, meta: null, pinned: false };
    posts.set(key, post);
    if (posts.size > 10_000) {
      const oldest = posts.keys().next().value;
      if (oldest !== undefined) posts.delete(oldest); // bound memory
    }
  }
  return post;
}

// Updates carry only the fields that changed, so merge them over what you stored.
const merge = (base: Partial<PostUpdate> | null, patch: Partial<PostUpdate> | null) => ({
  ...base,
  ...patch,
  author: { ...base?.author, ...patch?.author },
});

function publishedAtMs(platform: string, d: PostContent): number | null {
  // X post IDs encode their publish time: (id >> 22) + the Twitter epoch.
  if (platform === "twitter" && /^\d+$/.test(d.tweetId)) {
    return Number((BigInt(d.tweetId) >> 22n) + TWITTER_EPOCH_MS);
  }
  return typeof d.createdAt === "number" ? d.createdAt : null;
}

function referenceChain(ref: Reference | undefined): string {
  const links: string[] = [];
  // A referenced post can reference another one through ref.subtweet.
  for (let r = ref, depth = 0; r && depth < 8; r = r.subtweet, depth++) {
    links.push(`${r.type ?? "ref"} ${at(r.author)} ${r.tweetId} ${clip(r.text, 40)}`);
  }
  return links.join(" -> ");
}

function describeMeta(meta: PostMeta): string {
  const price = (value: number | undefined, prefix = "") => (value == null ? "" : ` ${prefix}${value}`);
  const tokens = meta.detected?.tokens ?? [];
  const markets = meta.detected?.prediction ?? [];
  const parts: string[] = [];
  if (tokens.length) {
    parts.push(`tokens ${tokens.map((t) => `${t.symbol ?? t.contract}${price(t.priceUsd, "$")}`).join(", ")}`);
  }
  if (markets.length) {
    parts.push(`markets ${markets.map((m) => `${clip(m.title, 40)}${price(m.priceUsd)}`).join(", ")}`);
  }
  if (meta.ocr?.text) parts.push(`ocr ${clip(meta.ocr.text, 40)}`);
  return parts.join(" | ") || "no detections";
}

// Delete, pin, and unpin change the stored post's state and keep its content.
function lifecycle(d: PostEvent, label: string, change: Partial<StoredPost>) {
  Object.assign(postFor(d), change);
  log(label, `${at(d.author)} ${d.tweetId}`);
}

const handlers: { [P in Pair]: Handler<Payloads[P]> } = {
  "control/auth_pong": () => log("auth", "ok, waiting for events"),
  "tweet/content": (d, receivedAt) => {
    const platform = d.author?.platform ?? "twitter";
    const post = postFor(d, platform);
    post.content = merge(d, post.content); // an update that arrived first still wins
    let latency = "";
    const publishedAt = publishedAtMs(platform, d);
    if (!post.hasContent && publishedAt !== null) {
      latencies.push(receivedAt - publishedAt);
      if (latencies.length > 10_000) latencies.shift();
      latency = `${receivedAt - publishedAt} ms  `;
    }
    post.hasContent = true;
    const where = platform === "twitter" ? "" : ` (${platform})`;
    log("post", `${at(d.author)}${where} ${d.tweetId}  ${latency}${clip(post.content.text)}`);
    if (d.ref) log("", `chain: ${referenceChain(d.ref)}`);
  },
  "tweet/update": (d) => {
    const post = postFor(d);
    post.content = merge(post.content, d); // a ref in the update replaces the stored one
    const who = `${at(post.content.author)} ${d.tweetId}`;
    const changed = Object.keys(d).filter((k) => !["platform", "receivedAt", "tweetId"].includes(k));
    if (d.ref) log("update", `${who} chain: ${referenceChain(d.ref)}`);
    else if (d.textUpdateType === "completion") log("update", `${who} full text ${clip(d.text)}`);
    else log("update", `${who} changed ${changed.join(", ")}`);
  },
  "tweet/meta": (d) => {
    const post = postFor(d);
    post.meta = d; // each meta is a complete snapshot: replace it, never merge
    log("meta", `${at(post.content?.author)} ${d.tweetId}  ${describeMeta(d)}`);
  },
  "tweet/delete": (d) => lifecycle(d, "delete", { deleted: true }),
  "tweet/pin": (d) => lifecycle(d, "pin", { pinned: true }),
  "tweet/unpin": (d) => lifecycle(d, "unpin", { pinned: false }),
  "account/follow": (d) => log("follow", `${at(d.actor)} followed ${at(d.target)}`),
  "account/unfollow": (d) => log("unfollow", `${at(d.actor)} unfollowed ${at(d.target)}`),
  "account/profile_update": (d) =>
    log("profile", `${at(d.actor)} changed ${Object.keys(d.changes ?? {}).join(", ")}`),
  "account/affiliate_update": (d) =>
    log("affiliate", `${at(d.organization)} ${d.action} ${at(d.member)}`),
  "account/callout": (d) =>
    log("callout", `${at(d.caller)} called ${d.token?.symbol} ${d.token?.address}`),
  "account/callout_update": (d) => log("callout", `update to ${d.updateOf}: ${d.token?.symbol}`),
  "account/thesis": (d) =>
    log("thesis", `${at(d.author)} on ${d.coin?.symbol}: ${clip(d.thesis?.text, 60)}`),
  "account/token_verified": (d) => log("verified", `${d.coin?.symbol ?? d.coin?.address}`),
  "news/article": (d) => log("news", `${d.source?.name}: ${clip(d.title)}`),
  "charity/added": (d) => log("charity", d.name),
  "control/twitter_handles_result": (d) =>
    log("handles", (d.results ?? []).map((r) => `${r.input} ${r.state}`).join(", ") || String(d.error)),
};
const isPair = (pair: string): pair is Pair => Object.hasOwn(handlers, pair);

function onMessage(data: unknown, receivedAt: number) {
  let event: Partial<Envelope> | null;
  try {
    event = JSON.parse(String(data));
  } catch {
    log("skip", "frame is not JSON");
    return;
  }
  if (event?.v !== 1 || typeof event.t !== "string" || typeof event.op !== "string") return;
  const pair = `${event.t}/${event.op}`;
  counts.set(pair, (counts.get(pair) ?? 0) + 1);
  if (!isPair(pair)) return; // unknown pairs are counted, then ignored
  // The payload is JSON from the server; the types above describe what each pair carries.
  const handle = handlers[pair] as Handler<unknown>;
  try {
    handle(event.d ?? {}, receivedAt);
  } catch (error) {
    log("error", `${pair}: ${error}`); // one malformed event must not stop the program
  }
}

function printSummary() {
  const sorted = [...latencies].sort((a, b) => a - b);
  const pct = (p: number) => sorted[Math.max(0, Math.ceil(p * sorted.length) - 1)];
  const timing = sorted.length
    ? `p50 ${pct(0.5)} ms  p95 ${pct(0.95)} ms  (n=${sorted.length})`
    : "no posts yet";
  const seen = [...counts].map(([pair, n]) => `${pair} ${n}`).join(", ") || "none";
  log("summary", `latency ${timing} | events ${seen}`);
}

// POST /api/add-account answers with one row per handle.
type AddAccountResponse = {
  results?: Array<{ input: string; state: string; message?: string }>;
  error?: string;
};

async function addAccounts() {
  const response = await fetch(`${API_URL}/api/add-account`, {
    body: JSON.stringify({ accounts: ACCOUNTS }),
    headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
    method: "POST",
    signal: AbortSignal.timeout(15_000),
  });
  const body = (await response.json().catch(() => ({}))) as AddAccountResponse;
  // One row per handle, even on 207 (some failed) or 400 (all failed). Each row says why.
  const rows = Array.isArray(body.results) ? body.results : [];
  for (const row of rows) {
    log("accounts", `${row.input} ${row.state}${row.message ? `: ${row.message}` : ""}`);
  }
  if (!rows.length && !response.ok) {
    log("accounts", `add failed (${response.status}): ${body.error ?? "unknown error"}`);
  }
}

// GET /api/me reports which pool the key connects to and how many connections each has open.
type ConnectionPool = { count: number; limit: number };
type MeResponse = {
  credentialScope?: "standard" | "ultra_speed";
  websocket?: ConnectionPool; // standard keys
  ultraSpeed?: { websocket?: ConnectionPool } | null; // Ultra Speed keys
};

async function explainRefusedConnection() {
  // Node's built-in WebSocket hides the HTTP status of a refused upgrade, so ask the REST API.
  const headers = { Authorization: `Bearer ${API_KEY}` };
  const response = await fetch(`${API_URL}/api/me`, {
    headers,
    signal: AbortSignal.timeout(10_000),
  }).catch(() => null);
  if (response?.status === 401) {
    console.error(`API key rejected (401). Copy a current key from ${DASHBOARD}`);
    process.exit(1);
  }
  if (response?.status === 403) {
    console.error("Streaming is not active for this key (403). Start a trial or renew your plan.");
    process.exit(1);
  }
  const me = (response?.ok ? await response.json().catch(() => null) : null) as MeResponse | null;
  const pool = me?.credentialScope === "ultra_speed" ? me.ultraSpeed?.websocket : me?.websocket;
  if (pool && pool.count >= pool.limit) {
    log("limit", `all ${pool.limit} connections are in use; close another client`);
  }
}

function connect() {
  const socket = new WebSocket(WS_URL, ["tweetstream.v1", `tweetstream.auth.token.${API_KEY}`]);
  let openedAt: number | null = null; // set while this socket is open
  socket.addEventListener("open", () => {
    openedAt = Date.now();
    log("connected", WS_URL);
    socket.send(JSON.stringify({ op: "auth_ping" })); // answered by control/auth_pong
    setTimeout(() => {
      if (openedAt !== null && !latencies.length && !hinted) {
        hinted = true;
        log("hint", "posts appear when a monitored account posts. Add accounts in the");
        log("hint", "dashboard, or run again with TWEETSTREAM_ACCOUNTS=handle1,handle2");
      }
    }, 120_000);
  });
  // Take the receipt time before any parsing so latency excludes your own processing.
  socket.addEventListener("message", (event) => onMessage(event.data, Date.now()));
  let settled = false; // some runtimes fire both error and close; reconnect only once
  const reconnect = async (reason: string) => {
    if (settled) return;
    settled = true;
    if (openedAt === null) await explainRefusedConnection();
    else if (Date.now() - openedAt >= 30_000) attempt = 0; // healthy session: reset backoff
    openedAt = null;
    const delay = Math.min(30_000, 1_000 * 2 ** attempt) + Math.floor(Math.random() * 250);
    attempt = Math.min(attempt + 1, 5);
    log("closed", `${reason}; reconnecting in ${(delay / 1000).toFixed(1)} s`);
    setTimeout(connect, delay);
  };
  // Node's built-in WebSocket reports a refused or failed handshake with error and no close.
  socket.addEventListener("error", () => openedAt === null && reconnect("connection failed"));
  socket.addEventListener("close", (event) => reconnect(`code ${event.code}`));
}

process.on("SIGINT", () => {
  printSummary();
  process.exit(0);
});
setInterval(printSummary, 60_000);
if (ACCOUNTS.length) {
  addAccounts()
    .catch((error: Error) => log("accounts", `add failed: ${error.message}`))
    .finally(connect);
} else {
  connect();
}
```

2. 保存为 `tweetstream-example.ts`

3. 用你的 key 运行

```bash
TWEETSTREAM_API_KEY=YOUR_API_KEY node tweetstream-example.ts
```

[原始文件：tweetstream-example.ts](https://tweetstream.io/docs/examples/tweetstream-example.ts)

### Python

需要 Python 3.10 或更高版本。运行命令会创建虚拟环境，并在其中安装 `websockets`。

1. 复制程序

```python
"""TweetStream complete example: connect, reconnect, handle every event, and measure latency.

Python 3.10 or later. One dependency, installed in a virtual environment:

    python3 -m venv tweetstream-venv
    tweetstream-venv/bin/pip install 'websockets>=14'
    TWEETSTREAM_API_KEY=YOUR_API_KEY tweetstream-venv/bin/python tweetstream_example.py

Optional settings:
    TWEETSTREAM_ACCOUNTS=handle1,handle2               add these accounts before connecting
    TWEETSTREAM_WS_URL=wss://ws-iad.tweetstream.io/ws  use the US endpoint
"""

import asyncio
import collections
import json
import math
import os
import random
import sys
import time
import urllib.error
import urllib.request

if sys.version_info < (3, 10):
    sys.exit(f"This example needs Python 3.10 or later; this is {sys.version.split()[0]}.")

import websockets

API_KEY = os.environ.get("TWEETSTREAM_API_KEY", "").strip()
DASHBOARD = "https://tweetstream.io/dashboard?tab=api"
if not API_KEY:
    sys.exit(f"Set TWEETSTREAM_API_KEY. Copy your key from {DASHBOARD}")
WS_URL = os.environ.get("TWEETSTREAM_WS_URL") or "wss://ws-global.tweetstream.io/ws"
API_URL = "https://api.tweetstream.io"
ACCOUNTS = [
    handle.strip().lstrip("@")
    for handle in os.environ.get("TWEETSTREAM_ACCOUNTS", "").split(",")
    if handle.strip()
]
TWITTER_EPOCH_MS = 1288834974657

posts = {}  # (platform, tweetId) -> {"content", "meta", "deleted", "pinned", "has_content"}
counts = {}  # "t/op" -> number of events seen
latencies = collections.deque(maxlen=10_000)  # publish-to-receipt ms, most recent new posts
hinted = False


def log(label, text):
    print(f"{label:<9} {text}", flush=True)


def at(account):
    account = account if isinstance(account, dict) else {}
    return "@" + str(account.get("handle") or account.get("username") or "unknown").lstrip("@")


def clip(text, limit=80):
    flat = " ".join(str(text or "").split())
    return json.dumps(flat if len(flat) <= limit else flat[: limit - 1] + "…", ensure_ascii=False)


def field(d, name, key):
    return (d.get(name) or {}).get(key)


# Posts are keyed by (platform, tweetId). Content carries the platform on its author;
# later operations carry it on d.platform.
def post_for(d, platform=None):
    platform = platform or d.get("platform") or field(d, "author", "platform") or "twitter"
    key = (platform, d["tweetId"])
    if key not in posts:
        posts[key] = {"content": None, "meta": None, "deleted": False, "pinned": False}
        posts[key]["has_content"] = False
        if len(posts) > 10_000:
            posts.pop(next(iter(posts)))  # bound memory
    return posts[key]


def merge(base, patch):
    # Updates carry only the fields that changed, so merge them over what you stored.
    base, patch = base or {}, patch or {}
    return {**base, **patch, "author": {**(base.get("author") or {}), **(patch.get("author") or {})}}


def published_at_ms(platform, d):
    tweet_id = str(d.get("tweetId", ""))
    if platform == "twitter" and tweet_id.isdigit():
        return (int(tweet_id) >> 22) + TWITTER_EPOCH_MS  # X post IDs encode their publish time
    created_at = d.get("createdAt")
    return created_at if isinstance(created_at, (int, float)) else None


def reference_chain(ref):
    links = []
    # A referenced post can reference another one through subtweet.
    while ref and len(links) < 8:
        text = clip(ref.get("text"), 40)
        links.append(f"{ref.get('type', 'ref')} {at(ref.get('author'))} {ref.get('tweetId')} {text}")
        ref = ref.get("subtweet")
    return " -> ".join(links)


def describe_meta(meta):
    detected = meta.get("detected") or {}
    tokens = [t.get("symbol") or t.get("contract") for t in detected.get("tokens") or []]
    prices = [t.get("priceUsd") for t in detected.get("tokens") or []]
    markets = [clip(m.get("title"), 40) for m in detected.get("prediction") or []]
    odds = [m.get("priceUsd") for m in detected.get("prediction") or []]
    parts = []
    if tokens:
        parts.append("tokens " + ", ".join(t if p is None else f"{t} ${p}" for t, p in zip(tokens, prices)))
    if markets:
        parts.append("markets " + ", ".join(m if p is None else f"{m} {p}" for m, p in zip(markets, odds)))
    if field(meta, "ocr", "text"):
        parts.append(f"ocr {clip(meta['ocr']['text'], 40)}")
    return " | ".join(parts) or "no detections"


def on_content(d, received_at):
    platform = field(d, "author", "platform") or "twitter"
    post = post_for(d, platform)
    post["content"] = merge(d, post["content"])  # an update that arrived first still wins
    latency = ""
    published_at = published_at_ms(platform, d)
    if not post["has_content"] and published_at is not None:
        latencies.append(round(received_at - published_at))
        latency = f"{latencies[-1]} ms  "
    post["has_content"] = True
    where = "" if platform == "twitter" else f" ({platform})"
    text = clip(post["content"].get("text"))
    log("post", f"{at(d.get('author'))}{where} {d['tweetId']}  {latency}{text}")
    if d.get("ref"):
        log("", f"chain: {reference_chain(d['ref'])}")


def on_update(d, _):
    post = post_for(d)
    post["content"] = merge(post["content"], d)  # a ref in the update replaces the stored one
    who = f"{at(post['content'].get('author'))} {d['tweetId']}"
    changed = ", ".join(k for k in d if k not in ("platform", "receivedAt", "tweetId"))
    if d.get("ref"):
        log("update", f"{who} chain: {reference_chain(d['ref'])}")
    elif d.get("textUpdateType") == "completion":
        log("update", f"{who} full text {clip(d.get('text'))}")
    else:
        log("update", f"{who} changed {changed}")


def on_meta(d, _):
    post = post_for(d)
    post["meta"] = d  # each meta is a complete snapshot: replace it, never merge
    log("meta", f"{at((post['content'] or {}).get('author'))} {d['tweetId']}  {describe_meta(d)}")


def lifecycle(label, **change):
    # Delete, pin, and unpin change the stored post's state and keep its content.
    def handle(d, _):
        post_for(d).update(change)
        log(label, f"{at(d.get('author'))} {d['tweetId']}")

    return handle


HANDLERS = {
    "tweet/content": on_content,
    "tweet/update": on_update,
    "tweet/meta": on_meta,
    "tweet/delete": lifecycle("delete", deleted=True),
    "tweet/pin": lifecycle("pin", pinned=True),
    "tweet/unpin": lifecycle("unpin", pinned=False),
}


def event_line(pair, d):
    # One readable line for account, news, charity, and control events.
    actor, target = at(d.get("actor")), at(d.get("target"))
    token, coin = d.get("token") or {}, d.get("coin") or {}
    results = ", ".join(f"{r.get('input')} {r.get('state')}" for r in d.get("results") or [])
    return {
        "control/auth_pong": ("auth", "ok, waiting for events"),
        "control/twitter_handles_result": ("handles", results or str(d.get("error"))),
        "account/follow": ("follow", f"{actor} followed {target}"),
        "account/unfollow": ("unfollow", f"{actor} unfollowed {target}"),
        "account/profile_update": ("profile", f"{actor} changed {', '.join(d.get('changes') or {})}"),
        "account/affiliate_update": (
            "affiliate",
            f"{at(d.get('organization'))} {d.get('action')} {at(d.get('member'))}",
        ),
        "account/callout": (
            "callout",
            f"{at(d.get('caller'))} called {token.get('symbol')} {token.get('address')}",
        ),
        "account/callout_update": ("callout", f"update to {d.get('updateOf')}: {token.get('symbol')}"),
        "account/thesis": (
            "thesis",
            f"{at(d.get('author'))} on {coin.get('symbol')}: {clip(field(d, 'thesis', 'text'), 60)}",
        ),
        "account/token_verified": ("verified", coin.get("symbol") or coin.get("address")),
        "news/article": ("news", f"{field(d, 'source', 'name')}: {clip(d.get('title'))}"),
        "charity/added": ("charity", d.get("name")),
    }.get(pair)


def on_message(raw, received_at):
    try:
        event = json.loads(raw)
    except ValueError:
        return log("skip", "frame is not JSON")
    # Envelope: {v, t, op, id?, ts, d}. Route by t, then op.
    if not isinstance(event, dict) or event.get("v") != 1:
        return
    pair = f"{event.get('t')}/{event.get('op')}"
    counts[pair] = counts.get(pair, 0) + 1
    d = event.get("d") or {}
    try:
        if pair in HANDLERS:
            HANDLERS[pair](d, received_at)
        elif event_line(pair, d):  # unknown pairs are counted, then ignored
            log(*event_line(pair, d))
    except Exception as error:  # one malformed event must not close the connection
        log("error", f"{pair}: {error!r}")


def print_summary():
    ordered = sorted(latencies)
    pct = lambda p: ordered[max(0, math.ceil(p * len(ordered)) - 1)]
    timing = f"p50 {pct(0.5)} ms  p95 {pct(0.95)} ms  (n={len(ordered)})" if ordered else "no posts yet"
    seen = ", ".join(f"{pair} {n}" for pair, n in counts.items()) or "none"
    log("summary", f"latency {timing} | events {seen}")


def add_accounts():
    request = urllib.request.Request(
        f"{API_URL}/api/add-account",
        data=json.dumps({"accounts": ACCOUNTS}).encode(),
        headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            status, raw = response.status, response.read()
    except urllib.error.HTTPError as error:  # 400, 429, and 503 can still carry results
        status, raw = error.code, error.read()
    except OSError as error:
        return log("accounts", f"add failed: {error}")
    try:
        body = json.loads(raw)
    except ValueError:
        body = {}
    body = body if isinstance(body, dict) else {}
    # One row per handle, even on 207 (some failed) or 400 (all failed). Each row says why.
    rows = body.get("results") if isinstance(body.get("results"), list) else []
    for row in rows:
        message = f": {row['message']}" if row.get("message") else ""
        log("accounts", f"{row.get('input')} {row.get('state')}{message}")
    if not rows and status >= 300:
        log("accounts", f"add failed ({status}): {body.get('error') or 'unknown error'}")


async def report_every_minute():
    while True:
        await asyncio.sleep(60)
        print_summary()


async def hint_when_quiet():
    global hinted
    await asyncio.sleep(120)
    if not latencies and not hinted:
        hinted = True
        log("hint", "posts appear when a monitored account posts. Add accounts in the")
        log("hint", "dashboard, or run again with TWEETSTREAM_ACCOUNTS=handle1,handle2")


async def main():
    asyncio.create_task(report_every_minute())
    protocols = ["tweetstream.v1", f"tweetstream.auth.token.{API_KEY}"]
    attempt = 0
    while True:
        opened_at, reason = None, "connection closed"
        try:
            async with websockets.connect(WS_URL, subprotocols=protocols) as ws:
                opened_at = time.monotonic()
                log("connected", WS_URL)
                await ws.send(json.dumps({"op": "auth_ping"}))  # answered by control/auth_pong
                hint = asyncio.create_task(hint_when_quiet())
                try:
                    async for raw in ws:
                        # Take the receipt time before parsing so latency excludes your own work.
                        on_message(raw, time.time() * 1000)
                finally:
                    hint.cancel()
        except Exception as error:
            # A refused upgrade carries its HTTP status on error.response. 429 means the pool
            # this key connects to (standard or Ultra Speed) has no free connection.
            status = getattr(getattr(error, "response", None), "status_code", None)
            if status == 401:
                sys.exit(f"API key rejected (401). Copy a current key from {DASHBOARD}")
            if status == 403:
                sys.exit("Streaming is not active for this key (403). Start a trial or renew your plan.")
            reason = "all connections are in use" if status == 429 else f"{type(error).__name__}: {error}"
        if opened_at is not None and time.monotonic() - opened_at >= 30:
            attempt = 0  # a healthy session resets backoff
        delay = min(30, 2**attempt) + random.random() / 4
        attempt = min(attempt + 1, 5)
        log("closed", f"{reason}; reconnecting in {delay:.1f} s")
        await asyncio.sleep(delay)


if __name__ == "__main__":
    if ACCOUNTS:
        add_accounts()
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print_summary()
```

2. 保存为 `tweetstream_example.py`

3. 用你的 key 运行

```bash
python3 -m venv tweetstream-venv && \
tweetstream-venv/bin/pip install 'websockets>=14' && \
TWEETSTREAM_API_KEY=YOUR_API_KEY tweetstream-venv/bin/python tweetstream_example.py
```

[原始文件：tweetstream_example.py](https://tweetstream.io/docs/examples/tweetstream_example.py)

### 预期输出

```text
connected wss://ws-global.tweetstream.io/ws
auth      ok, waiting for events
post      @marketdesk 2104934217937850514  183 ms  "SOL breaks out of the 4h range. Watching $98.5 as…"
update    @marketdesk 2104934217937850514 chain: quote @projectteam 2104932246614966349 "Mainnet is live." -> reply @devteam 2104929205744566281 "Ship date?"
update    @marketdesk 2104934217937850514 full text "SOL breaks out of the 4h range. Watching $98.5 as support. Chart below."
meta      @marketdesk 2104934217937850514  tokens SOL $98.5 | ocr "SOL/USD 4h"
delete    @marketdesk 2104934217937850514
follow    @marketdesk followed @projectteam
summary   latency p50 183 ms  p95 183 ms  (n=1) | events control/auth_pong 1, tweet/content 1, tweet/update 2, tweet/meta 1, tweet/delete 1, account/follow 1
```

## 事件顺序

上面的预期输出就来自这些消息，其中的时间戳仅作示意。每条帖子都从 `tweet/content` 开始。同一 `(platform, tweetId)` 的 `tweet/update` 会为它补充字段，其中的 `ref` 把帖子与它引用或回复的帖子连接起来。

1. 你发送 `auth_ping`. 连接打开后立即发送，用于检查 key。

```json
{
  "op": "auth_ping"
}
```

2. 服务端发送 `control/auth_pong`. key 有效。程序输出一行 `auth`，然后等待事件。

```json
{
  "v": 1,
  "t": "control",
  "op": "auth_pong",
  "ts": 1790690396000,
  "d": {}
}
```

3. 服务端发送 `tweet/content`. 一条新帖子，这里的正文是缩短版。程序按 `(platform, tweetId)` 保存它，并输出正文和延迟。

```json
{
  "v": 1,
  "t": "tweet",
  "op": "content",
  "id": "2104934217937850514",
  "ts": 1790690400160,
  "d": {
    "tweetId": "2104934217937850514",
    "text": "SOL breaks out of the 4h range. Watching $98.5 as…",
    "createdAt": 1790690400000,
    "receivedAt": 1790690400150,
    "kind": "quote",
    "author": {
      "id": "1450012345678901248",
      "handle": "@marketdesk",
      "name": "Market Desk",
      "platform": "twitter"
    }
  }
}
```

4. 服务端发送 `tweet/update`. 为同一条帖子补充引用。程序把字段合并到已保存的帖子上，替换其 `ref`，并沿 `ref.subtweet` 输出引用链。

```json
{
  "v": 1,
  "t": "tweet",
  "op": "update",
  "id": "2104934217937850514",
  "ts": 1790690402000,
  "d": {
    "tweetId": "2104934217937850514",
    "platform": "twitter",
    "receivedAt": 1790690401990,
    "kind": "quote",
    "ref": {
      "type": "quote",
      "tweetId": "2104932246614966349",
      "text": "Mainnet is live.",
      "author": {
        "id": "1380123456789012480",
        "handle": "@projectteam",
        "name": "Project Team"
      },
      "subtweet": {
        "type": "reply",
        "tweetId": "2104929205744566281",
        "text": "Ship date?",
        "author": {
          "id": "1290123456789012481",
          "handle": "@devteam",
          "name": "Dev Team"
        }
      }
    }
  }
}
```

5. 服务端发送 `tweet/update`. 提供完整正文，`textUpdateType` 为 `completion`。程序把它合并到已保存的帖子上，已有的 `ref` 保持不变。

```json
{
  "v": 1,
  "t": "tweet",
  "op": "update",
  "id": "2104934217937850514",
  "ts": 1790690403000,
  "d": {
    "tweetId": "2104934217937850514",
    "platform": "twitter",
    "receivedAt": 1790690402990,
    "text": "SOL breaks out of the 4h range. Watching $98.5 as support. Chart below.",
    "textUpdateType": "completion"
  }
}
```

6. 服务端发送 `tweet/meta`. 帖子的富化数据。程序用这份快照替换已保存的 `meta`，并输出代币、预测市场和 OCR 文本。

```json
{
  "v": 1,
  "t": "tweet",
  "op": "meta",
  "id": "2104934217937850514",
  "ts": 1790690404000,
  "d": {
    "tweetId": "2104934217937850514",
    "platform": "twitter",
    "receivedAt": 1790690403990,
    "detected": {
      "tokens": [
        {
          "chain": "solana",
          "contract": "So11111111111111111111111111111111111111112",
          "name": "Wrapped SOL",
          "networkId": 1399811149,
          "priceUsd": 98.5,
          "sources": [
            "text",
            "ocr"
          ],
          "symbol": "SOL"
        }
      ]
    },
    "ocr": {
      "text": "SOL/USD 4h"
    }
  }
}
```

7. 服务端发送 `tweet/delete`. 作者删除了帖子。程序把它标记为已删除，内容保留。

```json
{
  "v": 1,
  "t": "tweet",
  "op": "delete",
  "id": "2104934217937850514",
  "ts": 1790690430000,
  "d": {
    "eventId": "delete.264da58762fac7204da8",
    "tweetId": "2104934217937850514",
    "platform": "twitter",
    "author": {
      "id": "1450012345678901248",
      "handle": "@marketdesk"
    },
    "deletedAt": 1790690429990,
    "receivedAt": 1790690429990
  }
}
```

8. 服务端发送 `account/follow`. 一个监控账号关注了另一个账号。程序输出一行，不改动已保存的帖子。

```json
{
  "v": 1,
  "t": "account",
  "op": "follow",
  "id": "follow-1450012345678901248-1380123456789012480",
  "ts": 1790690438000,
  "d": {
    "actor": {
      "id": "1450012345678901248",
      "handle": "@marketdesk"
    },
    "target": {
      "id": "1380123456789012480",
      "handle": "@projectteam"
    },
    "kind": "FOLLOW",
    "eventId": "follow-1450012345678901248-1380123456789012480",
    "observedAt": 1790690437990,
    "receivedAt": 1790690437990
  }
}
```

## 连接与重连

`connect()` 使用两个子协议 `tweetstream.v1` 和 `tweetstream.auth.token.<API_KEY>` 打开连接，然后发送 `auth_ping`。收到 `control/auth_pong` 即表示 key 有效。

断线后，程序依次等待 1、2、4、8、16 秒，之后每次等待 30 秒，并加入少量随机抖动。连接保持 30 秒后，等待时间会重置。

遇到 `401` 或 `403` 时，程序会输出说明并退出，因为重试无法修复 key 或套餐问题。Node.js 不提供被拒连接的 HTTP 状态码，所以 TypeScript 程序会调用 `GET /api/me` 查询原因。Python 程序直接从 `websockets` 的错误中读取状态码。

## 添加账号

`TWEETSTREAM_ACCOUNTS` 中有账号时，`addAccounts()` 会在连接前把它们发送到 `POST https://api.tweetstream.io/api/add-account`。`207` 响应表示部分账号添加成功、部分失败，因此程序为每个账号输出一行，并显示其 `state`。

程序退出后，已添加的账号仍保留在监控列表中。

## 帖子与更新链

程序按 `(platform, tweetId)` 保存每条帖子。`tweet/content` 创建帖子，`tweet/update` 只合并它携带的字段，已有的正文和作者信息保持不变。

更新包含 `ref` 时，程序会替换已保存的引用，并沿 `ref.subtweet` 逐层输出引用链。更新可能先于内容到达，合并时会保留更新中的字段。`tweet/delete`、`tweet/pin` 和 `tweet/unpin` 只改变已保存帖子的状态，不改动内容。

### 合并更新并输出引用链 (TypeScript)

```typescript
// Updates carry only the fields that changed, so merge them over what you stored.
const merge = (base: Partial<PostUpdate> | null, patch: Partial<PostUpdate> | null) => ({
  ...base,
  ...patch,
  author: { ...base?.author, ...patch?.author },
});

function referenceChain(ref: Reference | undefined): string {
  const links: string[] = [];
  // A referenced post can reference another one through ref.subtweet.
  for (let r = ref, depth = 0; r && depth < 8; r = r.subtweet, depth++) {
    links.push(`${r.type ?? "ref"} ${at(r.author)} ${r.tweetId} ${clip(r.text, 40)}`);
  }
  return links.join(" -> ");
}

  "tweet/update": (d) => {
    const post = postFor(d);
    post.content = merge(post.content, d); // a ref in the update replaces the stored one
    const who = `${at(post.content.author)} ${d.tweetId}`;
    const changed = Object.keys(d).filter((k) => !["platform", "receivedAt", "tweetId"].includes(k));
    if (d.ref) log("update", `${who} chain: ${referenceChain(d.ref)}`);
    else if (d.textUpdateType === "completion") log("update", `${who} full text ${clip(d.text)}`);
    else log("update", `${who} changed ${changed.join(", ")}`);
  },
```

### 合并更新并输出引用链 (Python)

```python
def merge(base, patch):
    # Updates carry only the fields that changed, so merge them over what you stored.
    base, patch = base or {}, patch or {}
    return {**base, **patch, "author": {**(base.get("author") or {}), **(patch.get("author") or {})}}


def reference_chain(ref):
    links = []
    # A referenced post can reference another one through subtweet.
    while ref and len(links) < 8:
        text = clip(ref.get("text"), 40)
        links.append(f"{ref.get('type', 'ref')} {at(ref.get('author'))} {ref.get('tweetId')} {text}")
        ref = ref.get("subtweet")
    return " -> ".join(links)


def on_update(d, _):
    post = post_for(d)
    post["content"] = merge(post["content"], d)  # a ref in the update replaces the stored one
    who = f"{at(post['content'].get('author'))} {d['tweetId']}"
    changed = ", ".join(k for k in d if k not in ("platform", "receivedAt", "tweetId"))
    if d.get("ref"):
        log("update", f"{who} chain: {reference_chain(d['ref'])}")
    elif d.get("textUpdateType") == "completion":
        log("update", f"{who} full text {clip(d.get('text'))}")
    else:
        log("update", f"{who} changed {changed}")
```

## 富化数据

`tweet/meta` 可能在帖子之后到达。每条都是完整快照，所以程序会替换已保存的元数据，而不是合并。

`meta` 行会列出检测到的代币及价格、预测市场及价格，以及从图片中识别的 OCR 文本（如有）。

## 其他事件

关注、取消关注、资料变更、关联账号变更、喊单、FOMO 观点与代币认证、新闻文章和慈善项目通知都会各输出一行。喊单、新闻、慈善和 FOMO 事件需要先在控制台开启才会收到。

遇到未知的 `t` 和 `op` 组合时，程序只计数并跳过，连接保持打开。某个处理函数出错时，程序会输出一行 `error` 并继续读取。

## 测量延迟

消息处理函数在解析任何内容之前先读取时钟。对于 X 帖子，发布时间来自帖子 ID：将 ID 右移 22 位，再加上 Twitter 纪元 `1288834974657`。其他平台使用 `createdAt`。

两者之差就是帖子从发布到在你的机器上被接收的时间，其中包括到你这里的网络传输。重放的帖子不会再次计入样本。程序每 60 秒以及你按下 Ctrl+C 时，都会输出最近 10,000 条帖子的 p50 和 p95，以及每种事件的数量。

### 读取发布时间 (TypeScript)

```typescript
function publishedAtMs(platform: string, d: PostContent): number | null {
  // X post IDs encode their publish time: (id >> 22) + the Twitter epoch.
  if (platform === "twitter" && /^\d+$/.test(d.tweetId)) {
    return Number((BigInt(d.tweetId) >> 22n) + TWITTER_EPOCH_MS);
  }
  return typeof d.createdAt === "number" ? d.createdAt : null;
}
```

### 读取发布时间 (Python)

```python
def published_at_ms(platform, d):
    tweet_id = str(d.get("tweetId", ""))
    if platform == "twitter" and tweet_id.isdigit():
        return (int(tweet_id) >> 22) + TWITTER_EPOCH_MS  # X post IDs encode their publish time
    created_at = d.get("createdAt")
    return created_at if isinstance(created_at, (int, float)) else None
```

## 可选设置

在运行命令中，把这些环境变量和 `TWEETSTREAM_API_KEY` 放在一起。

| 变量 | 默认值 | 用途 |
| --- | --- | --- |
| `TWEETSTREAM_API_KEY` | 必填 | 你的 API key。程序不会输出它。 |
| `TWEETSTREAM_ACCOUNTS` | 无 | 连接前要添加的账号，用逗号分隔，例如 `marketdesk,projectteam` |
| `TWEETSTREAM_WS_URL` | `wss://ws-global.tweetstream.io/ws` | 在美国运行时使用 `wss://ws-iad.tweetstream.io/ws` |

## 常见错误

- 把 `tweet/update` 当作新帖子。它属于你按同一 `(platform, tweetId)` 保存的帖子。
- 用更新替换已保存的帖子。更新只携带发生变化的字段，应当合并。
- 丢弃先于内容到达的更新。先保留它，内容到达后再合并进来，不覆盖更新中的字段。
- 把 `tweet/meta` 合并到旧的元数据中。每条都是完整快照，应替换已保存的那份。
- 用信封中的 `ts` 或 `id` 给帖子排序或去重。帖子身份请使用 `(platform, tweetId)`。
- 遇到未知的 `t` 和 `op` 组合就关闭连接。跳过这条消息并继续读取。
- 对 `401` 或 `403` 反复重试。新连接无法修复 key 或套餐问题。

## 故障排查

- 只看到 `auth` 和 `summary` 行。监控账号发帖时才会收到帖子。请添加一个活跃账号，或用 `GET /api/me` 检查监控列表。连续两分钟没有帖子时，程序会输出提示。
- 程序因 `API key rejected (401)` 退出。请从控制台重新复制 key。轮换后的旧 key 会失效。
- 程序因 `Streaming is not active for this key (403)` 退出。请开始试用或续订套餐，然后重新运行。
- 程序提示所有连接都在使用中。你的套餐限制了同时打开的 WebSocket 连接数。请关闭另一个客户端，或等待旧连接关闭。
- 延迟为负数或远高于预期。这个数值比较的是你的时钟和帖子的发布时间，所以请用 NTP 同步时钟，并在运行交易系统的机器上测量。
- Node.js 报错 `Unknown file extension ".ts"`。说明你的 Node.js 低于 22.18。请升级 Node.js 后重新运行命令。
- Python 报错 `No module named 'websockets'`。请按照运行命令，用 `tweetstream-venv/bin/python` 运行程序，这个环境里已经安装了 `websockets`。
- 创建虚拟环境时提示 `ensurepip is not available`。在 Debian 或 Ubuntu 上，请先用 `sudo apt install python3-venv` 安装 venv 模块，然后重新运行 Python 命令。
- pip 提示 `No matching distribution found for websockets>=14`。说明你的 `python3` 低于 3.10。请安装 Python 3.10 或更高版本，并用它创建虚拟环境。

## 在此基础上构建

要把这个程序改成长期运行的消费者，可以先看类型化分发器和参考消费者。

- [分发所有操作](https://tweetstream.io/zh-cn/docs/production-consumer#complete-router): 覆盖全部 18 种服务端操作的类型化处理程序。
- [构建参考消费者](https://tweetstream.io/zh-cn/docs/production-consumer#reference-consumer): 重连、持久恢复和账号管理，每个主题一页。
- [开源 Solana 狙击机器人](https://github.com/tweetstream-io/solana-sniper-bot): 基于这个数据流的 TypeScript 和 Rust 机器人，在发帖、简介变更、关注和喊单时买入。