Add a simple loop with a sleep interval to monitor swaps every 12 seconds (Ethereum block time):

if current_block > last_block: print(f"\n--- New blocks: last_block+1 to current_block ---") for pair in monitored_pairs: # List of pair addresses swaps = get_recent_swaps(pair, last_block+1, current_block) for swap in swaps: if swap["amount0_out"] > 100: # Filter for large trades print(f"🐋 LARGE SWAP: swap['amount0_out'] tokens") last_block = current_block

def get_recent_swaps(pair_address, from_block, to_block): contract = w3.eth.contract(address=pair_address, abi=DEX_ABI) # Create event filter swap_filter = contract.events.Swap.create_filter( from_block=from_block, to_block=to_block )

while True: current_block = w3.eth.block_number

pip install web3 requests python-dotenv Now, connect to Ethereum mainnet (or BSC, Polygon, etc.):

for s in swaps[:5]: print(f"Swap by s['sender']: s['amount0_out'] tokens out") Hardcoding a single pair address is fine for testing. A real explorer would query a factory contract to discover all pairs dynamically. Step 4: Making It Useful – Real-Time Monitoring A script that runs once is a toy. A script that runs forever is a tool.

if w3.is_connected(): print(f"Connected to chain: w3.eth.chain_id") A DEX like Uniswap V2 emits two critical events: Swap and Sync . We only need the ABI fragments for those events:

If you’ve ever wanted to peek under the hood of a decentralized exchange (DEX) like Uniswap or PancakeSwap, you’ve probably realized that while the data is public , it’s not exactly easy to read.