How to Retrieve Transactions on Uniswap Using Web3Py

Decentralized finance (DeFi) has transformed how users interact with digital assets, and Uniswap stands at the forefront of this movement as one of the most widely used decentralized exchanges. Every swap, liquidity addition, and token interaction on Uniswap is recorded on-chain, creating a rich source of transparent and valuable data.

For developers, being able to access and process this transaction data opens the door to building powerful tools such as trading bots, analytics dashboards, and real-time monitoring systems. In this guide, we’ll walk through how to retrieve transaction data from Uniswap using Web3.py, and how to leverage blockchain events to extract the information you need efficiently.

First, you need to prepare your Python environment and install the Web3.py library using.

pip install web3==6.*

Save this Python file as main.py.

from web3 import Web3
from hexbytes import HexBytes
import time

RPC_URL = "https://eth.llamarpc.com"
w3 = Web3(Web3.HTTPProvider(RPC_URL))

# topic0 = keccak of event signature
topic0 = w3.keccak(
    text="Swap(address,address,int256,int256,uint160,uint128,int24)"
)

# minimal ABI for pool calls
POOL_ABI = [
    {"name": "token0", "outputs": [{"type": "address"}], "inputs": [], "stateMutability": "view", "type": "function"},
    {"name": "token1", "outputs": [{"type": "address"}], "inputs": [], "stateMutability": "view", "type": "function"},
    {"name": "factory", "outputs": [{"type": "address"}], "inputs": [], "stateMutability": "view", "type": "function"},
]

# cache to reduce RPC calls
pool_cache = {}

# get current block
current_block = w3.eth.block_number
from_block = current_block
to_block = current_block + 1

# wait until next block is mined
while w3.eth.block_number < to_block:
    time.sleep(1)

# fetch logs without filtering by address
logs = w3.eth.get_logs({
    "fromBlock": from_block,
    "toBlock": to_block,
    "topics": [topic0]
})

for log in logs:
    try:
        # decode indexed parameters from topics
        sender = Web3.to_checksum_address("0x" + log["topics"][1].hex()[-40:])
        recipient = Web3.to_checksum_address("0x" + log["topics"][2].hex()[-40:])

        # decode non-indexed data
        data_field = log["data"]

        if isinstance(data_field, HexBytes):
            data_bytes = bytes(data_field)
        elif isinstance(data_field, str):
            data_bytes = bytes.fromhex(data_field[2:])
        else:
            continue

        # validate data length (5 params * 32 bytes)
        if len(data_bytes) != 32 * 5:
            continue

        amount0, amount1, sqrtPriceX96, liquidity, tick = w3.codec.decode(
            ["int256", "int256", "uint160", "uint128", "int24"],
            data_bytes
        )

        pool_address = Web3.to_checksum_address(log["address"])

        # get from cache or fetch from chain
        if pool_address in pool_cache:
            token0, token1, factory = pool_cache[pool_address]
        else:
            try:
                contract = w3.eth.contract(address=pool_address, abi=POOL_ABI)
                token0 = contract.functions.token0().call()
                token1 = contract.functions.token1().call()
                factory = contract.functions.factory().call()

                pool_cache[pool_address] = (token0, token1, factory)
            except Exception:
                continue  # skip if not a valid pool

        print("==== SWAP GLOBAL ====")
        print("Block:", log["blockNumber"])
        print("Pool:", pool_address)
        print("Factory:", factory)
        print("Token0:", token0)
        print("Token1:", token1)
        print("Sender:", sender)
        print("Recipient:", recipient)
        print("amount0:", amount0)
        print("amount1:", amount1)
        print("---------------------")

    except Exception as e:
        print("Skip log:", e)

If you run the script, the output will appear as shown below:

==== SWAP GLOBAL ====
Block: 24723221
Pool: 0x6546055f46e866a4B9a4A13e81273e3152BAE5dA
Factory: 0x1F98431c8aD98523631AE4a59f267346ea31F984
Token0: 0x68749665FF8D2d112Fa859AA293F07A622782F38
Token1: 0xdAC17F958D2ee523a2206206994597C13D831ec7
Sender: 0x0BB7a4Fdea32910038DEc59c20ccAe3a6E66b09f
Recipient: 0x111111125421cA6dc452d289314280a0f8842A65
amount0: 105679
amount1: -465823017
---------------------

You can see that token0 is being swapped to token1. To verify whether this transaction comes from Uniswap, you can check the factory address. A list of factory addresses can be found here: Uniswap Addresses

PRO TIP!

To save time, you can use Salishook at https://salishook.xyz — Salishook is a platform designed to simplify blockchain data extraction and automation, helping you build faster without handling complex low-level interactions.

With Salishook, you only need to input your webhook URL. The platform will handle parsing the required blockchain data, and you simply receive it ready to use.