Chain Sniper

A high-performance EVM blockchain listener and transaction monitor.

Install

git clone https://github.com/rakibhossain72/chain-sniper.git
cd chain-sniper
uv sync

Or with pip:

pip install chain-sniper

Quick Start

from chain_sniper import ChainSniper

ERC20_ABI = [...]  # Transfer event ABI
USDT = "0x55d398326f99059fF775485246999027B3197955"

sniper = ChainSniper("wss://bsc-ws-node.nariox.org:443")

@sniper.event(contract=USDT, abi=ERC20_ABI, name="Transfer")
async def handle_transfer(event):
    print(f"Transfer: {event['args']['value'] / 10**18} USDT")

await sniper.start()

Use a WebSocket URL (wss:// or ws://) — that's the only supported transport.

RPC Pool (fault-tolerant)

from chain_sniper.rpc_pool import RPCPool

pool = await RPCPool.create(
    rpcs=[
        "https://bsc-dataseed1.binance.org",
        "wss://bsc-ws-node.nariox.org:443",
    ],
    expected_chain_id=56,
)
sniper = ChainSniper(pool)

Transaction Monitoring

sniper.block_detail("full_block")

@sniper.on_transaction
async def handle_tx(tx):
    print(f"{tx['hash']} | {tx['from']} -> {tx['to']}")

Filtering

MongoDB-style rules with operators: $eq, $ne, $gt, $gte, $lt, $lte, $in, $nin, $regex, $exists.

from chain_sniper import Filter

f = Filter()
f.add_tx_rule({"value": {"_op": "$gte", "_value": 10**18}})
f.add_log_rule({"args.value": {"_op": "$gte", "_value": 1000 * 10**18}})

sniper.filter(f)

Dynamic Rules via Redis

Inject or remove filter rules at runtime without restarting:

from chain_sniper.listener import RedisRuleListener

redis_listener = RedisRuleListener(
    dynamic_filter=f,
    redis_url="redis://localhost",
    channel="sniper_rules",
)
await redis_listener.start()

Push rules from anywhere:

import redis, json

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
r.publish("sniper_rules", json.dumps({"action": "add_tx", "rule": {"to": "0x..."}}))

Key API

See examples/ for complete usage patterns.

License

MIT