"""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()