Python Twitter WebSocket Client: Real-Time Tweets to Trades

This tutorial builds a Python client for the Xanguard real-time Twitter/X WebSocket that watches a list of accounts, pulls contract addresses out of their tweets, and hands each new address to a trading function. The code survives disconnects and never buys the same token twice. Every snippet runs against the live contract at wss://api.xanguard.tech/v1/dt/realtime/ws, and steps 4 to 6 fit together into one file you can run.

1. What you'll build

A single-file Python bot, bot.py, that:

Connects to the Xanguard B2B WebSocket and completes the HELLO, LOGIN, READY handshake.
Parses twitter.post.new and twitter.post.update events.
Pulls Solana and EVM contract addresses from the tweet text, from expanded links (a pump.fun/coin/<mint> hidden behind a t.co link), from quoted tweets, and from image text if your key is on the Enterprise plan (OCR).
Sends heartbeats, notices a dead connection, and reconnects with jittered backoff.
Filters by account, tweet type and delay, then passes each new contract address to a execute_trade() stub that is safe to run in dry-run mode.

You need Python 3.10+ and one dependency: pip install "websockets>=14".

2. Get an API key from @B2B_Xanguard_bot

Open @B2B_Xanguard_bot on Telegram and send /start.
Pick a plan and pay in SOL. Plans start at $49/month for 50 handles (see B2B pricing).
Once the payment confirms, the bot sends a quick-start message with your key. It starts with dt_ and is shown once, so save it now. If you lose it, /apikey makes a new one and the old key stops working at its next login.
Add the accounts to watch, either in the bot (/add handle1 handle2) or over REST:

curl -X POST https://api.xanguard.tech/v1/dt/targets \
  -H "Authorization: Bearer $XANGUARD_KEY" \
  -H "Content-Type: application/json" \
  -d '{"handle": "elonmusk"}'

You don't need to reconnect after adding a handle. Open connections pick up the new handle list straight away.

export XANGUARD_KEY=dt_your_key_here

3. Minimal connect

The protocol uses opcodes, the same way TweetCatcher's does. The server speaks first:

op 10 HELLO arrives on connect with heartbeat_interval (30000 ms).
You send op 2 LOGIN within 15 seconds.
The server answers op 4 READY, or op 3 DISCONNECT with a reason.
After that, op 0 EVENT frames stream in. You send op 1 HEARTBEAT every interval, and the server replies op 11 HEARTBEAT_ACK.

import asyncio, json, os
import websockets

URL = "wss://api.xanguard.tech/v1/dt/realtime/ws"
API_KEY = os.environ["XANGUARD_KEY"]  # dt_...

async def main():
    async with websockets.connect(URL) as ws:
        hello = json.loads(await ws.recv())              # op 10 HELLO
        interval = hello["d"]["heartbeat_interval"] / 1000
        await ws.send(json.dumps({"op": 2, "d": API_KEY}))  # op 2 LOGIN

        reply = json.loads(await ws.recv())
        if reply["op"] == 3:                             # op 3 DISCONNECT
            raise SystemExit(f"login refused: {reply['d']['reason']}")
        print("READY:", reply["d"])                      # op 4 READY

        async def heartbeat():
            while True:
                await asyncio.sleep(interval)
                await ws.send(json.dumps({"op": 1}))     # op 1 HEARTBEAT

        hb = asyncio.create_task(heartbeat())
        async for raw in ws:
            msg = json.loads(raw)
            if msg["op"] == 0:                           # op 0 EVENT
                d = msg["d"]
                print(d["event"], "@" + d["task_info"]["handle"],
                      (d["data"].get("text") or "")[:100])

asyncio.run(main())

Run python minimal.py. With a wrong key you get login refused: Invalid or expired API key. With a good key you see READY, and then events as your accounts post. This version stops at the first disconnect. Steps 4 to 6 fix that.

4. Parse tweet events and extract the contract address

A tweet event looks like this (trimmed):

{"op": 0, "d": {
  "event": "twitter.post.new",
  "event_id": "evt_1234567890123456789",
  "task_info": {"handle": "somekol"},
  "data": {
    "id": "1234567890123456789",
    "created_at": 1758620000000,
    "type": "post",
    "text": "new one https://t.co/abc",
    "cashtags": [],
    "entities": {"urls": [{"expanded_url": "https://pump.fun/coin/..."}]},
    "quoted_tweet": null,
    "author": {"handle": "somekol", "name": "...", "stats": {"followers": 50000}}
  }}}

Things to know before you parse:

The tweet ID is data.id. created_at is epoch milliseconds.
data.type is one of post, reply, quote or repost.
A twitter.post.update can follow with the same event_id. It fills in fields the first frame was missing: reply or quote context, or the rest of a very long post. Merge it by event_id. Don't treat it as a new tweet.
Addresses often hide behind t.co links, so also scan entities.urls[].expanded_url. entities can be null.
extracted_cas (addresses read from images) is only there if OCR is on for your key, which needs the Enterprise plan.

Start bot.py with this:

import asyncio, json, os, random, re, time
import websockets

URL = "wss://api.xanguard.tech/v1/dt/realtime/ws"
API_KEY = os.environ["XANGUARD_KEY"]

SOL_RE = re.compile(r"\b[1-9A-HJ-NP-Za-km-z]{32,44}\b")  # Solana base58
EVM_RE = re.compile(r"\b0x[a-fA-F0-9]{40}\b")             # EVM hex

def log(*args):
    print(time.strftime("%H:%M:%S"), *args, flush=True)

def extract_cas(data):
    """Contract addresses from a tweet payload, first-seen order, no dupes."""
    sources = [data.get("text") or ""]
    # t.co links hide the address; entities.urls carries the expanded URL
    # (e.g. https://pump.fun/coin/<mint>). entities can be null.
    for u in (data.get("entities") or {}).get("urls") or []:
        sources.append(u.get("expanded_url") or "")
    # On quotes, quoted_tweet is the tweet being quoted. (On replies it holds
    # the parent tweet, someone else's words, so skip it.)
    if data.get("type") == "quote" and data.get("quoted_tweet"):
        sources.append(data["quoted_tweet"].get("text") or "")

    found, seen = [], set()
    def add(ca):
        key = ca.lower() if ca.startswith("0x") else ca
        if key not in seen:
            seen.add(key)
            found.append(ca)

    for ca in data.get("extracted_cas") or []:  # read from images (OCR, Enterprise plan)
        add(ca)
    for s in sources:
        for m in SOL_RE.finditer(s):
            add(m.group())
        for m in EVM_RE.finditer(s):
            add(m.group())
    return found

def handle_event(d):
    if d["event"] not in ("twitter.post.new", "twitter.post.update"):
        return  # follows, profile changes, etc.
    data = d["data"]
    lag_ms = int(time.time() * 1000) - (data.get("created_at") or 0)
    cas = extract_cas(data)
    log(f"{d['event']} @{d['task_info']['handle']} {data['type']} "
        f"id={data['id']} lag={lag_ms}ms cas={cas}")
    on_tweet(d["event_id"], d["task_info"]["handle"], data, cas, lag_ms)

The two regexes are the standard ones. Solana is base58 at 32 to 44 characters, and EVM is 0x plus 40 hex digits. EVM addresses are compared in lower case so a checksummed copy and a lower-case copy don't count as two tokens.

5. Reconnect with backoff, heartbeat

Here are the server rules this code follows:

No heartbeat for 90 seconds closes the connection, and no traffic of any kind for 180 seconds does too.
Backend deploys drop sockets for a few seconds with no DISCONNECT frame. That's normal, so reconnect right away.
Tweets are real-time only. Nothing older than 30 seconds is delivered, and missed tweets aren't replayed. (Follow and unfollow events from the last 15 minutes are replayed after READY.)
A key can hold 5 connections at once. A sixth gets Too many connections (max 5).
Several bad-key logins in a short burst get your IP temporarily blocked with HTTP 429. So never retry an invalid key in a loop.

Append this to bot.py:

FATAL = ("Invalid or expired API key", "Subscription expired",
         "Invalid login payload")

class Fatal(Exception): pass
class Refused(Exception): pass

async def heartbeat(ws, interval, st):
    while True:
        await asyncio.sleep(interval)
        if time.monotonic() - st["last_ack"] > 2 * interval + 5:
            log("two heartbeat ACKs missed, reconnecting")
            await ws.close()
            return
        await ws.send(json.dumps({"op": 1}))

async def session(st):
    async with websockets.connect(URL, open_timeout=10, close_timeout=2) as ws:
        hello = json.loads(await asyncio.wait_for(ws.recv(), 10))
        if hello.get("op") != 10:
            raise ConnectionError(f"expected HELLO, got {hello}")
        interval = hello["d"]["heartbeat_interval"] / 1000

        # d may be the bare key string, or an object with the key + filters,
        # e.g. {"api_key": KEY, "onlyCA": True, "types": ["tweet", "quote"]}
        await ws.send(json.dumps({"op": 2, "d": {"api_key": API_KEY}}))
        ready = json.loads(await asyncio.wait_for(ws.recv(), 15))
        if ready.get("op") == 3:
            reason = ready["d"]["reason"]
            raise (Fatal if reason in FATAL else Refused)(reason)
        if ready.get("op") != 4:
            raise ConnectionError(f"expected READY, got {ready}")
        log(f"READY: {ready['d']['handles']} handles, "
            f"modules={ready['d']['modules']}")
        st["ready"] = True
        st["last_ack"] = time.monotonic()

        hb = asyncio.create_task(heartbeat(ws, interval, st))
        try:
            async for raw in ws:
                msg = json.loads(raw)
                op = msg.get("op")
                if op == 0:
                    try:
                        handle_event(msg["d"])
                    except Exception as e:  # one bad event must not kill the feed
                        log("event handler error:", repr(e))
                elif op == 11:
                    st["last_ack"] = time.monotonic()
                elif op == 3:
                    reason = msg["d"]["reason"]
                    raise (Fatal if reason in FATAL else Refused)(reason)
        finally:
            hb.cancel()

async def run_forever():
    backoff = [1, 2, 5, 10, 30]
    attempt = 0
    while True:
        st = {"ready": False}
        try:
            await session(st)
            why = "socket closed"
        except Fatal as e:
            log(f"fatal: {e}. Fix the key or renew in @B2B_Xanguard_bot.")
            raise SystemExit(1)
        except Refused as e:  # e.g. "Too many connections (max 5)"
            why, attempt = f"refused: {e}", max(attempt, len(backoff) - 1)
        except websockets.InvalidStatus as e:
            code = e.response.status_code
            why = f"HTTP {code}"
            if code == 429:  # temporary block after repeated bad keys
                attempt = len(backoff) - 1
        except (OSError, asyncio.TimeoutError, ConnectionError,
                websockets.ConnectionClosed, websockets.InvalidHandshake) as e:
            why = repr(e)
        if st["ready"]:
            attempt = 0  # we were live; come back fast
        delay = backoff[min(attempt, len(backoff) - 1)] * random.uniform(0.8, 1.2)
        attempt += 1
        log(f"disconnected ({why}); reconnecting in {delay:.1f}s")
        await asyncio.sleep(delay)

Why it's shaped like this:

If two ACKs in a row go missing, the client closes and reconnects itself instead of waiting for the server to time out. A half-open TCP connection looks fine from your side and delivers nothing, so waiting is the costly mistake.
An invalid or expired key, or an expired subscription, stops the process. Retrying those can't succeed and only gets your IP blocked.
The backoff goes 1, 2, 5, 10, 30 seconds with ±20% jitter, so a fleet of bots doesn't reconnect in lockstep. It resets to 1 second after any session that reached READY.

6. Filter and trigger a trading action (stub)

The last block decides what gets traded. It keeps the accounts you trust, original posts and quotes only, and events that arrived fast enough to act on. It sends each contract address once, through a queue, so a slow order never stalls the socket reader:

WATCH = {h.strip().lower() for h in os.environ.get("WATCH", "").split(",") if h.strip()}
TYPES = {"post", "quote"}   # skip replies and reposts
MAX_LAG_MS = 5_000          # too late to trade? skip
DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"

acted = set()               # CAs already sent to the trader
trades = asyncio.Queue()

def on_tweet(event_id, handle, data, cas, lag_ms):
    if WATCH and handle.lower() not in WATCH:
        return
    if data["type"] not in TYPES or lag_ms > MAX_LAG_MS:
        return
    for ca in cas:
        key = ca.lower() if ca.startswith("0x") else ca
        if key in acted:
            continue        # already handled (post.update repeats, reposted CAs)
        acted.add(key)
        trades.put_nowait({"ca": ca, "handle": handle, "tweet_id": data["id"],
                           "event_id": event_id, "lag_ms": lag_ms})

async def execute_trade(sig):
    """STUB: replace with your swap / order code."""
    if DRY_RUN:
        log(f"DRY RUN buy {sig['ca']} (@{sig['handle']}, "
            f"https://x.com/{sig['handle']}/status/{sig['tweet_id']})")
        return
    raise NotImplementedError("wire up your exchange or DEX client here")

async def trader():
    while True:
        sig = await trades.get()
        try:
            await execute_trade(sig)
        except Exception as e:
            log("trade failed:", repr(e))

async def main():
    worker = asyncio.create_task(trader())
    try:
        await run_forever()
    finally:
        worker.cancel()

if __name__ == "__main__":
    asyncio.run(main())

Run it in dry-run mode first:

WATCH=somekol,otherkol python bot.py
# 12:01:07 READY: 42 handles, modules=['realtime']
# 12:03:55 twitter.post.new @somekol post id=19... lag=412ms cas=['9BB6...pump']
# 12:03:55 DRY RUN buy 9BB6...pump (@somekol, https://x.com/somekol/status/19...)

Before you set DRY_RUN=0, add the checks your strategy needs inside execute_trade: liquidity, token age, max position size. A regex can't tell a real token from a random base58 string or a DEX pair address. Treat every extracted address as a candidate until you've checked it on-chain.

Would you rather have the server do the first cut? LOGIN also accepts an object: {"api_key": KEY, "onlyCA": true, "types": ["tweet", "quote"]}. With onlyCA you only receive tweets that carry a contract address in their text (or, with OCR, in an image). Addresses that appear only inside a link won't pass that filter, and the client-side extractor above catches those.

7. Deploy tips

Run it under systemd with Restart=always and Environment=PYTHONUNBUFFERED=1 so logs show up in journalctl straight away. Keep the key in an EnvironmentFile with chmod 600, never in the repo.
Run two copies on two machines. Missed tweets aren't replayed, so a second connection covers deploys and network blips. You get 5 per key. Share the "already traded" set between them (for example Redis SET ca 1 NX EX 86400) so only one of them buys.
Keep your clock synced (chrony or systemd-timesyncd). The lag you log is your clock minus created_at, and a drifting clock makes a fast feed look slow, or the reverse.
Host close to where you trade: your RPC node, exchange or DEX API. That hop usually costs more than the feed.
Log every disconnect with a UTC timestamp. If you think you missed something, send support the exact window. Per-connection logs can show whether it was a deploy.
Watch the event rate. If you're connected and heartbeating but nothing arrives for longer than your accounts are ever quiet, check /v1/dt/status and your handle list before you blame the market.

8. Frequently Asked Questions

Does the WebSocket replay tweets I missed while disconnected?

No. The stream is real-time only, and tweets older than 30 seconds are dropped. The one exception is follow and unfollow events, which are replayed for the last 15 minutes after each READY. If you can't afford gaps, run two connections from different machines.

Why did I get the same tweet twice?

The second one was a twitter.post.update. It has the same event_id and fills in the reply or quote context or the rest of a long post. Merge it by event_id. The bot above does this through the acted set.

Can I use this WebSocket from a browser?

Technically yes, but your key would be exposed. Run the client on a server and forward what you need to your front end.

How many connections can one key open?

Five at once. A sixth receives op 3 DISCONNECT with Too many connections (max 5).

I'm migrating from TweetCatcher. What changes?

The opcodes are the same (HELLO 10, LOGIN 2, READY 4, EVENT 0, HEARTBEAT 1/11, DISCONNECT 3). Point your client at wss://api.xanguard.tech/v1/dt/realtime/ws, log in with your dt_ key, and check the field names in the event shape above.

How fast is it?

Tweet detection speed is measured on production traffic and published live at xanguard.tech/speed. Your own lag log line measures the same thing from your side.

Building in JavaScript instead? See the Node.js version of this tutorial. The full protocol reference is at docs.xanguard.tech.