Build a KOL Tweet Alert Bot in 50 Lines of Python

What you need

• Python 3.10+ and pip install websockets aiohttp
• A 1322 API key + your WebSocket path (from the dashboard API access page after signup, plans from $250/mo, pricing)
• A Telegram bot token (free, talk to @BotFather) and your chat id

How the feed works under the hood: tweets from tracked accounts are detected in roughly 150-250ms and pushed to your open WebSocket as JSON events, in progressive stages (the earliest tweet.mini first, enriched stages after). Details: Twitter WebSocket API.

The bot

import asyncio, json, os, re
import aiohttp, websockets

API_KEY = os.environ["API_KEY"]            # 1322 api key
WS_URL = os.environ["WS_URL"]              # wss://1322.io/your-ws-path
TG_TOKEN = os.environ["TG_TOKEN"]          # telegram bot token
TG_CHAT = os.environ["TG_CHAT"]            # telegram chat id

# Solana base58 (32-44 chars) or EVM 0x address, catches contract drops
CA_PATTERN = re.compile(r"\b[1-9A-HJ-NP-Za-km-z]{32,44}\b|0x[a-fA-F0-9]{40}")

seen: set[str] = set()  # one alert per tweet across event stages


async def send_telegram(session: aiohttp.ClientSession, text: str) -> None:
    url = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage"
    await session.post(url, json={"chat_id": TG_CHAT, "text": text})


async def run() -> None:
    async with aiohttp.ClientSession() as session:
        while True:
            try:
                async with websockets.connect(
                    WS_URL, additional_headers={"X-Api-Key": API_KEY}
                ) as ws:
                    print("connected")
                    async for raw in ws:
                        e = json.loads(raw)
                        if e.get("platform") != "x":
                            continue
                        if not str(e.get("eventType", "")).startswith("tweet"):
                            continue
                        tid = e.get("id")
                        if not tid or tid in seen:
                            continue
                        seen.add(tid)

                        handle = e.get("handle", "?")
                        text = e.get("content", "")
                        ca = CA_PATTERN.search(text)
                        prefix = "🚨 CA DROP" if ca else "🐦"
                        msg = f"{prefix} @{handle}\n{text[:500]}"
                        await send_telegram(session, msg)
                        print("alerted:", handle, "ca!" if ca else "")
            except Exception as exc:
                print(f"reconnecting ({exc})")
                await asyncio.sleep(1)


asyncio.run(run())

Run it: API_KEY=... WS_URL=... TG_TOKEN=... TG_CHAT=... python bot.py

How it works

• Dedup by tweet id. Events arrive in stages (mini → update → expanded → full). The seen set fires one alert per tweet, on the earliest stage, which is exactly what you want for speed.
• Contract-address regex. Matches Solana base58 and EVM 0x addresses in the tweet text. A match upgrades the alert to a CA-drop notification, the highest-value signal for memecoin entries.
• Reconnect loop. The stream is real-time only (no replay), so the loop reconnects with a short backoff and keeps going.

Where to take it next

• More platforms, same socket: filter platform == "binance" for Binance Square coin-pair posts, or "truth" for political posts that move prediction markets.
• Ready-made minimal consumers for all 6 platforms (X, Instagram, Truth Social, YouTube, Binance Square, News) live in our open-source examples repo: github.com/SisoSol/social-monitor-examples.
• Zero-code route: the built-in Discord and Telegram delivery bots do this without a script, the custom code only earns its keep when you want your own filters (CA regex, ticker lists, follower thresholds).
• The full trading-bot picture: crypto KOL tracking.

FAQ