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 )
Start simple. Monitor one pair. Then expand to one DEX. Then to every DEX on three chains. dex explorer script
pip install web3 requests python-dotenv Now, connect to Ethereum mainnet (or BSC, Polygon, etc.): to_block): contract = w3.eth.contract(address=pair_address