Why Python works well with TweetStream

TweetStream sends JSON envelopes over WebSocket, so any language with a WebSocket client will work. Python is a natural fit when your trading logic already uses NumPy, pandas, CCXT, or an LLM pipeline.

This page uses the websockets package (not websocket-client), an async client that handles subprotocol negotiation.

1. Install

bash

pip install websockets

2. Minimal example

Use this small working stream to print tweet content as it arrives. It is a quick connection check before you add your downstream system.

Content events are scoped to the accounts tracked in your dashboard. Payload author.handle values include a leading @ when present, so strip it before comparing against a bare username watchlist.

python

import asyncio
import json
import os
import websockets
 
API_KEY = os.environ["TWEETSTREAM_API_KEY"]
WS_URL = "wss://ws.tweetstream.io/ws"
SUBPROTOCOLS = ["tweetstream.v1", f"tweetstream.auth.token.{API_KEY}"]
 
 
def normalize_handle(value):
    return (value or "").removeprefix("@").lower()
 
 
async def stream():
    async with websockets.connect(WS_URL, subprotocols=SUBPROTOCOLS) as ws:
        async for raw in ws:
            envelope = json.loads(raw)
            if envelope["t"] == "tweet" and envelope["op"] == "content":
                tweet = envelope["d"]
                handle = normalize_handle(tweet["author"].get("handle"))
                print(f"[{tweet['kind']}] [@{handle or 'unknown'}] {tweet['text']}")
 
 
asyncio.run(stream())

3. Production pattern

Adds structured logging, envelope dispatch, and an exponential-backoff reconnect loop. Use it as a starting point for a trading bot.

python

import asyncio
import json
import logging
import os
from typing import Any
 
import websockets
from websockets.exceptions import ConnectionClosed
 
API_KEY = os.environ["TWEETSTREAM_API_KEY"]
WS_URL = "wss://ws.tweetstream.io/ws"
SUBPROTOCOLS = ["tweetstream.v1", f"tweetstream.auth.token.{API_KEY}"]
 
RECONNECT_DELAY_SECONDS = 5
MAX_BACKOFF_SECONDS = 60
 
log = logging.getLogger("tweetstream")
 
 
def normalize_handle(value: str | None) -> str:
    return (value or "").removeprefix("@").lower()
 
 
async def handle_envelope(envelope: dict[str, Any]) -> None:
    t, op, data = envelope["t"], envelope["op"], envelope["d"]
 
    if t == "tweet" and op == "content":
        handle = normalize_handle(data["author"].get("handle"))
        log.info(
            "tweet %s kind=%s @%s: %s",
            data["tweetId"],
            data["kind"],
            handle or "unknown",
            data["text"],
        )
        # Fan-out to your trading bot / queue / database here.
        return
 
    if t == "tweet" and op == "meta":
        tokens = (data.get("detected") or {}).get("tokens") or []
        for token in tokens:
            log.info(
                "token %s priceUsd=%s chain=%s",
                token.get("symbol"),
                token.get("priceUsd"),
                token.get("chain"),
            )
        return
 
    if t == "account" and op == "profile_update":
        log.info("profile update %s", data)
        return
 
    if t == "account" and op == "affiliate_update":
        log.info(
            "affiliate %s organization=%s member=%s event=%s",
            data["action"],
            data["organization"]["id"],
            data["member"]["id"],
            data["eventId"],
        )
        return
 
    log.debug("ignoring unsupported envelope t=%s op=%s", t, op)
 
 
async def stream_forever() -> None:
    backoff = RECONNECT_DELAY_SECONDS
    while True:
        try:
            async with websockets.connect(WS_URL, subprotocols=SUBPROTOCOLS) as ws:
                log.info("connected to TweetStream")
                backoff = RECONNECT_DELAY_SECONDS
                async for raw in ws:
                    try:
                        await handle_envelope(json.loads(raw))
                    except Exception:
                        log.exception("error handling envelope")
        except ConnectionClosed as exc:
            log.warning("connection closed: %s", exc)
        except Exception:
            log.exception("stream error")
 
        log.info("reconnecting in %ss", backoff)
        await asyncio.sleep(backoff)
        backoff = min(backoff * 2, MAX_BACKOFF_SECONDS)
 
 
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(stream_forever())

Envelope fields you will use

t: "tweet", op: "content": tweet text, author, and kind (post, reply, quote, or retweet)
t: "tweet", op: "meta": detected tokens, contract addresses, live prices, and OCR text
t: "tweet", op: "delete" | "pin" | "unpin": lifecycle changes observed for tracked tweets
t: "account", op: "profile_update": avatar, banner, bio, handle, name, location, website, and affiliation changes on tracked accounts
t: "account", op: "follow" | "unfollow": follow graph changes from tracked accounts
t: "account", op: "affiliate_update": affiliate list additions and removals for enabled tracked businesses. Replay stored changes with type=AFFILIATE.

See the payloads docs for the complete envelope reference.

Deployment tips

Run the worker as a long-lived process (systemd, Docker, Kubernetes Deployment). WebSocket sessions are stateless on our side, which keeps reconnect handling simple.
Put a queue (Redis Streams, SQS, or NATS) between the stream worker and your trading logic. Handlers should return in under a second; longer work goes on the queue.
Store the last-seen tweet ID if you need gap detection. Use the history REST API on Pro plans to backfill if your worker is offline.

Stream TweetStream tweets with Python + websockets