Building an on-chain alerts bot in Python without any blockchain library

PolySignal — my Telegram bot that watches Polymarket — has a small subsystem called chain_watcher. It does one thing: cross-check every trade Polymarket's API reports against what actually happened on Polygon, and complain (via Sentry) when the API is wrong.

It's about 280 lines of Python with one dependency — httpx. No web3.py, no eth-account, no eth-abi. I'm going to walk through how to read public on-chain events from Python with nothing but the standard library and an HTTP client, because (a) I needed to and (b) most tutorials reach for the SDK as the first step when you genuinely don't need it for a read-only watcher.

This isn't an anti-SDK argument. If you're signing transactions, sending funds, or doing serious ABI work, use web3.py. The case here is narrower: for reading specific events from a known contract, you can do it cleanly with httpx + JSON-RPC.

What "watching the chain" actually means

A blockchain node exposes a JSON-RPC API. You can ask it questions:

"What block are you on?" → eth_blockNumber
"Give me all logs matching this filter from blocks N to M" → eth_getLogs
"What's the transaction at this hash?" → eth_getTransactionByHash

For an alerts bot you only need the first two. You repeatedly poll the head of the chain and ask: "any logs from contract X, with this event signature, since the last block I checked?" When new logs come back, decode them and act.

That's it. Forty lines for the polling loop, ten for the JSON-RPC client, twenty for the event decoder, the rest is glue.

The JSON-RPC client

A node listens for HTTP POSTs to its RPC URL. Each call is a single JSON body. You don't need an SDK; you need an HTTP client and a little patience.

class _ChainRPC:
    def __init__(self, url: str, timeout: float = 15.0) -> None:
        self._url = url
        self._client = httpx.AsyncClient(timeout=timeout)
        self._id = 0

    async def _call(self, method: str, params: list) -> object:
        for attempt in range(3):
            self._id += 1
            try:
                resp = await self._client.post(
                    self._url,
                    json={
                        "jsonrpc": "2.0",
                        "id": self._id,
                        "method": method,
                        "params": params,
                    },
                )
                resp.raise_for_status()
                body = resp.json()
                if body.get("error"):
                    raise RuntimeError(f"RPC {method} failed: {body['error']}")
                return body["result"]
            except (httpx.HTTPError, RuntimeError):
                if attempt == 2:
                    raise
                await asyncio.sleep(0.5 * (attempt + 1))
        raise RuntimeError("unreachable")

That's the whole client. Two convenience methods on top:

    async def block_number(self) -> int:
        result = await self._call("eth_blockNumber", [])
        return int(result, 16)  # the node returns hex strings

    async def get_logs(self, address, from_block, to_block, topics):
        return await self._call("eth_getLogs", [{
            "address": address,
            "fromBlock": hex(from_block),
            "toBlock": hex(to_block),
            "topics": topics,
        }])

Why the retries? Public Polygon RPCs (the polygon-rpc.com family and similar) return sporadic 5xx errors. A couple of half-second retries inside one tick smooths that out without stalling.

Filtering the events you actually want

A contract emits events with a known signature. For Polymarket's CTF Exchange V2, the event that matters is OrderFilled:

OrderFilled(
    bytes32 indexed orderHash,
    address indexed maker,
    address indexed taker,
    uint256 side,
    uint256 tokenId,
    uint256 makerAmountFilled,
    uint256 takerAmountFilled,
    uint256 fee
)

The topics field of an Ethereum log carries up to four things:

The Keccak hash of the event signature (topics[0]).
Each indexed parameter, padded to 32 bytes.

For OrderFilled, topics[0] is a known constant, and topics[2] / topics[3] are the maker and taker addresses (left-padded to 32 bytes). To find "any fill involving a watched wallet," I make two eth_getLogs calls — one filtering by the maker topic, one by the taker:

for filter_topics in (
    [ORDER_FILLED_TOPIC, None, watched_address_topics],         # as maker
    [ORDER_FILLED_TOPIC, None, None, watched_address_topics],   # as taker
):
    raw_logs += await rpc.get_logs(CONTRACT, from_block, to_block, filter_topics)

watched_address_topics is a list of 32-byte left-padded addresses. The node does the heavy lifting; you receive only matching logs.

Decoding a log

A log looks like {topics: [...], data: "0x...", transactionHash: "0x...", ...}. Topics are 32-byte hex; the data field is the concatenation of all non-indexed parameters, each padded to 32 bytes. Decoding doesn't need an ABI library for a known fixed-layout event — slice the hex.

def decode_order_filled(raw: dict) -> OrderFill:
    topics = raw["topics"]
    data = raw["data"].removeprefix("0x")
    words = [data[i : i + 64] for i in range(0, len(data), 64)]
    return OrderFill(
        tx_hash=raw["transactionHash"].lower(),
        order_hash=topics[1],
        maker="0x" + topics[2][-40:].lower(),
        taker="0x" + topics[3][-40:].lower(),
        side=int(words[0], 16),
        token_id=int(words[1], 16),
        maker_amount_filled=int(words[2], 16),
        taker_amount_filled=int(words[3], 16),
    )

That's it. A struct-shaped tuple per event.

Defensive bits worth keeping

Two production realities I built into the watcher:

1. Bounded block span. After an RPC outage, the "from last seen block to now" span can be thousands of blocks. Public nodes reject overlong eth_getLogs queries; the watcher gets stuck. So each tick caps the span at a fixed maximum (1000 blocks for me — about 30 minutes at Polygon's ~2s/block) and the watcher catches up in chunks across successive ticks.

2. Cross-check loop, not duplicate ingestion. I'm not using the chain as my primary data source; I'm using it to verify the API. The watcher records on-chain fills, waits a few minutes for the API to report the same trade, and complains via Sentry if the API never did. A single buggy API response throws a signal I'll see; if the chain were silent and the API were silent, neither layer would notice.

This is the part where SDK-less actually beats SDK: there's no library between me and the bytes. When something goes wrong, the trace is short.

When NOT to skip the SDK

To be honest about the trade:

Don't skip if you're sending transactions. Nonce management, gas estimation, signing — that's exactly what web3.py is for.
Don't skip if you need ABI introspection. Dynamic ABI parsing is genuinely a library-grade problem.
Don't skip if you want one client that handles ETH + several L2s with their quirks. A library normalises that.

For a passive watcher of a single event on a single chain, none of those apply. Eight imports, one HTTP client, ~280 lines.

The watcher in the wild

This runs inside PolySignal, my Telegram bot that alerts users when top wallets on Polymarket make a trade. The chain_watcher is the part that lets me sleep at night when Polymarket's API has its periodic moments.

If you want to read the full module, it's about 280 lines including comments and the cross-check loop. Happy to paste a Gist or answer questions in the comments.

推荐订阅源

DEV Community