Complete WebSocket example

Needs Node.js 22.18 or later, which runs TypeScript files directly. The program has no dependencies.

1. Copy the program

   tweetstream-example.tstypescriptCopy

   // 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. Save it as tweetstream-example.ts

3. Run it with your key

   TerminalbashCopy

   TWEETSTREAM_API_KEY=YOUR_API_KEY node tweetstream-example.ts

• Raw file: tweetstream-example.ts

Needs Python 3.10 or later. The run command creates a virtual environment and installs websockets into it.

1. Copy the program

   tweetstream_example.pypythonCopy

   """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. Save it as tweetstream_example.py

3. Run it with your key

   TerminalbashCopy

   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

• Raw file: tweetstream_example.py