Building a Real-Time DEX Price Aggregator: Compare Token Prices Across Uniswap, SushiSwap, and PancakeSwap with Python

Decentralized exchanges (DEXs) have become the backbone of DeFi trading, but prices for the same token pair often vary across different platforms. These discrepancies create arbitrage opportunities and make it difficult to know where to get the best execution price. In this tutorial, we'll build a Python-based DEX price aggregator that fetches real-time quotes across major DEXs so you can spot the best rates instantly.

Why Build a DEX Aggregator?

When you swap tokens on a DEX, the price depends on the liquidity pool's current state. A pool with deeper liquidity and less recent trading activity may offer a significantly better rate than a shallow pool on another exchange. An aggregator lets you:

• Save money by routing trades through the cheapest venue
• Spot arbitrage opportunities between pools
• Monitor liquidity depth across chains
• Build smarter bots that optimize execution

What We'll Cover

• Querying on-chain pair reserves using Web3.py
• Calculating token prices from reserve ratios
• Comparing quotes across Uniswap V2, SushiSwap, and PancakeSwap
• Supporting multiple EVM chains (Ethereum, BSC, Polygon)
• Formatting results into a clean dashboard

Prerequisites

pip install web3 requests

You'll also need an RPC endpoint. I recommend using a free tier from Infura, Alchemy, or a public RPC.

Step 1: Setting Up Multi-Chain Connections

from web3 import Web3

CHAINS = {
    "ethereum": "https://eth.llamarpc.com",
    "bsc": "https://bsc-dataseed.binance.org",
    "polygon": "https://polygon.llamarpc.com",
}

# Factory addresses for each DEX (Uniswap V2-style)
FACTORIES = {
    "uniswap_v2": {
        "ethereum": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f",
        "polygon": "0x5757371414417b8C6CAad45bAeF941aBc7d3Ab32",
    },
    "sushiswap": {
        "ethereum": "0xC0AEe478e3658e2610c5F7A4A2E1777cE9e4f2Ac",
        "bsc": "0xc35DADB65012eC5796536bD9864eD8773aBc74C4",
        "polygon": "0xc35DADB65012eC5796536bD9864eD8773aBc74C4",
    },
    "pancakeswap": {
        "bsc": "0xcA143Ce32Fe78f1f7019d7d551a6402fC5350c73",
    },
}

def get_w3(chain):
    return Web3(Web3.HTTPProvider(CHAINS[chain]))

Step 2: The Pair Contract ABI

Every Uniswap V2-style pair exposes getReserves() and token addresses:

PAIR_ABI = [
    {
        "constant": True,
        "inputs": [],
        "name": "getReserves",
        "outputs": [
            {"internalType": "uint112", "name": "reserve0", "type": "uint112"},
            {"internalType": "uint112", "name": "reserve1", "type": "uint112"},
            {"internalType": "uint32", "name": "blockTimestampLast", "type": "uint32"},
        ],
        "type": "function",
    },
    {
        "constant": True,
        "inputs": [],
        "name": "token0",
        "outputs": [{"internalType": "address", "name": "", "type": "address"}],
        "type": "function",
    },
    {
        "constant": True,
        "inputs": [],
        "name": "token1",
        "outputs": [{"internalType": "address", "name": "", "type": "address"}],
        "type": "function",
    },
]

Step 3: Fetching Pair Address from the Factory

FACTORY_ABI = [
    {
        "constant": True,
        "inputs": [
            {"internalType": "address", "name": "tokenA", "type": "address"},
            {"internalType": "address", "name": "tokenB", "type": "address"},
        ],
        "name": "getPair",
        "outputs": [{"internalType": "address", "name": "pair", "type": "address"}],
        "type": "function",
    },
]

def get_pair_address(w3, factory_addr, token_a, token_b):
    factory = w3.eth.contract(address=factory_addr, abi=FACTORY_ABI)
    return factory.functions.getPair(token_a, token_b).call()

Step 4: Calculating the Price

Given reserve0 and reserve1, the price of token0 in terms of token1 is simply reserve1 / reserve0. We also need to account for decimals:

ERC20_ABI = [
    {
        "constant": True,
        "inputs": [],
        "name": "decimals",
        "outputs": [{"internalType": "uint8", "name": "", "type": "uint8"}],
        "type": "function",
    },
    {
        "constant": True,
        "inputs": [],
        "name": "symbol",
        "outputs": [{"internalType": "string", "name": "", "type": "string"}],
        "type": "function",
    },
]

def get_price(w3, pair_address, token_in, token_out):
    pair = w3.eth.contract(address=pair_address, abi=PAIR_ABI)
    reserves = pair.functions.getReserves().call()
    token0 = pair.functions.token0().call()

    token_in_contract = w3.eth.contract(address=token_in, abi=ERC20_ABI)
    token_out_contract = w3.eth.contract(address=token_out, abi=ERC20_ABI)
    decimals_in = token_in_contract.functions.decimals().call()
    decimals_out = token_out_contract.functions.decimals().call()

    if token_in.lower() == token0.lower():
        price_raw = reserves[1] / reserves[0]
    else:
        price_raw = reserves[0] / reserves[1]

    # Adjust for decimals
    price = price_raw * (10 ** decimals_in) / (10 ** decimals_out)
    return price

Step 5: The Aggregator Loop

Now we tie it all together — iterate over chains and DEXs, fetch the pair, and compute the price:

def aggregate_quotes(token_in, token_out):
    results = []

    for chain_name, rpc in CHAINS.items():
        w3 = get_w3(chain_name)
        if not w3.is_connected():
            continue

        for dex_name, chain_factories in FACTORIES.items():
            factory_addr = chain_factories.get(chain_name)
            if not factory_addr:
                continue

            try:
                pair_addr = get_pair_address(w3, factory_addr, token_in, token_out)
                if pair_addr == "0x0000000000000000000000000000000000000000":
                    continue

                price = get_price(w3, pair_addr, token_in, token_out)
                symbol = w3.eth.contract(
                    address=token_out, abi=ERC20_ABI
                ).functions.symbol().call()

                results.append({
                    "chain": chain_name,
                    "dex": dex_name,
                    "price": price,
                    "pair": pair_addr,
                })
                print(f"  {chain_name}/{dex_name}: 1 token = {price:.6f} {symbol}")
            except Exception as e:
                print(f"  {chain_name}/{dex_name}: Error - {e}")

    return results

Step 6: Finding the Best Rate

# Example: WETH → USDC across all DEXs
WETH = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"
USDC = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"

print("🔍 Aggregating quotes for WETH → USDC...")
quotes = aggregate_quotes(WETH, USDC)

if quotes:
    best = max(quotes, key=lambda q: q["price"])
    worst = min(quotes, key=lambda q: q["price"])
    spread = ((best["price"] - worst["price"]) / worst["price"]) * 100

    print(f"Best:  {best['chain']}/{best['dex']} at {best['price']:.2f}")
    print(f"Worst: {worst['chain']}/{worst['dex']} at {worst['price']:.2f}")
    print(f"Spread: {spread:.2f}%")

Real-World Usage

You can extend this aggregator to:

• Run as a cron job that alerts you when spreads exceed a threshold
• Feed a trading bot that automatically executes on the best venue
• Build a web dashboard with Flask or FastAPI showing live rates

Try It Yourself

The complete code is available at github.com/Byaigo — clone the repo, grab an RPC endpoint, and start hunting for the best DEX prices.

---

Built with ❤️ for the open-source DeFi community. If you found this useful, consider supporting:

ETH: 0x18da907cb9d981bc798acb87ac27b03a2dc3cbb7

GitHub: github.com/Byaigo

Happy aggregating! 🚀