Uniswap V3 API Python: Query Pools and Build Trading Tools

Uniswap V3 sits at the heart of on-chain finance — billions in daily volume, concentrated liquidity positions, and a fully public data layer any developer can tap into for free. Whether you're building an arbitrage scanner that compares DEX prices against Binance or OKX spot markets, monitoring pool health for an LP strategy, or wiring up a trading bot, Python gives you a clean, direct path to all of it. This guide covers the full stack: The Graph's GraphQL API, direct on-chain calls with web3.py, fee tier mechanics, and production-grade error handling.

What Is Uniswap V3?

Uniswap V3 is the third major version of Uniswap, an automated market maker (AMM) originally deployed on Ethereum mainnet and later expanded to Arbitrum, Polygon, Optimism, Base, and other EVM-compatible chains. The defining innovation over V2 was concentrated liquidity: rather than spreading capital across all possible prices on a bonding curve, LPs choose a specific price range. Capital sitting within the active range earns fees; capital outside the range earns nothing and sits idle. This makes each dollar of liquidity far more efficient — but demands active position management when prices move.

• Concentrated liquidity: LPs pick a price range, not a global curve
• Four fee tiers: 0.01%, 0.05%, 0.30%, and 1.00% per swap
• Non-fungible LP positions: each position is represented as an ERC-721 NFT
• Built-in TWAP oracle for time-weighted average prices directly on-chain
• Multi-chain deployment: Ethereum, Arbitrum, Polygon, Optimism, Base, and more

For traders who primarily live on centralized platforms like Binance, Bybit, or Coinbase, Uniswap V3 represents a fundamentally different execution model — no order book, no counterparty, no custodian. Understanding the protocol at the data level opens the door to MEV research, on-chain arbitrage, and LP analytics that most retail traders never touch.

Setting Up Your Python Environment

Two libraries cover the vast majority of Uniswap V3 use cases: requests for querying The Graph's GraphQL API, and web3 for direct Ethereum RPC calls. Set up a virtual environment first — installing globally will eventually cause version conflicts.

# Create and activate a virtual environment
python -m venv uniswap-env
source uniswap-env/bin/activate

# Install core dependencies
pip install requests web3 python-dotenv

# Store your Infura key in .env — never hardcode credentials
echo "INFURA_KEY=your_infura_key_here" > .env

You need a free Infura or Alchemy API key for any web3.py calls to the Ethereum node — both offer generous free tiers (Infura provides 100,000 requests/day on the free plan). For The Graph's hosted subgraph, no API key is required yet, though The Graph's decentralized network requires GRT tokens or a paid gateway key. Always load credentials from environment variables; never hardcode them in scripts you might accidentally push to a public repository.

Querying Uniswap V3 Pools with GraphQL

The Graph's Uniswap V3 subgraph indexes every pool, swap, mint, and burn event and exposes them via a GraphQL endpoint — no Ethereum node required. It handles historical queries, aggregate stats (TVL, volume, fee income), and tick-level data efficiently. The pool ID in every query is simply the pool contract address in lowercase.

import requests

SUBGRAPH = "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3"
POOL_ID = "0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640"  # ETH/USDC 0.05%

QUERY = """
{
  pool(id: "%s") {
    token0 { symbol decimals }
    token1 { symbol decimals }
    feeTier
    liquidity
    volumeUSD
    totalValueLockedUSD
    txCount
  }
}
""" % POOL_ID

response = requests.post(
    SUBGRAPH,
    json={"query": QUERY},
    timeout=10
)
response.raise_for_status()

pool = response.json()["data"]["pool"]
fee_pct = int(pool["feeTier"]) / 10000

print(f"Pair:   {pool['token0']['symbol']}/{pool['token1']['symbol']}")
print(f"Fee:    {fee_pct}%")
print(f"TVL:    ${float(pool['totalValueLockedUSD']):,.2f}")
print(f"Volume: ${float(pool['volumeUSD']):,.2f}")
print(f"Swaps:  {int(pool['txCount']):}")

The feeTier field returns raw integers: 500 means 0.05%, 3000 means 0.30%. volumeUSD accumulates over the entire lifetime of the pool — for 24h rolling volume you need to query poolDayDatas instead. You can also query multiple pools simultaneously using the pools(...) entity with where filters on token addresses or minimum TVL thresholds.

Reading Live Token Prices with web3.py

The Graph subgraph has a delay of roughly 15–45 seconds — several blocks. For true real-time prices useful when comparing against Bybit or Gate.io spot to detect arbitrage windows, you need to call the pool contract's slot0 function directly. Uniswap V3 stores the current price as sqrtPriceX96: the square root of the token1/token0 price ratio, scaled by 2^96. Decoding it requires a bit of math.

from web3 import Web3
from dotenv import load_dotenv
import os

load_dotenv()

w3 = Web3(Web3.HTTPProvider(
    f"https://mainnet.infura.io/v3/{os.getenv('INFURA_KEY')}"
))
assert w3.is_connected(), "Failed to connect to Ethereum node"

# Minimal ABI — only the slot0 view function we need
POOL_ABI = [{
    "inputs": [],
    "name": "slot0",
    "outputs": [
        {"name": "sqrtPriceX96",               "type": "uint160"},
        {"name": "tick",                         "type": "int24"},
        {"name": "observationIndex",             "type": "uint16"},
        {"name": "observationCardinality",       "type": "uint16"},
        {"name": "observationCardinalityNext",   "type": "uint16"},
        {"name": "feeProtocol",                  "type": "uint8"},
        {"name": "unlocked",                     "type": "bool"}
    ],
    "stateMutability": "view",
    "type": "function"
}]

def sqrt_price_to_human(sqrt_price_x96, token0_dec=6, token1_dec=18):
    """Decode sqrtPriceX96 to human-readable token1-per-token0 price."""
    price = (sqrt_price_x96 / (2 ** 96)) ** 2
    return price * (10 ** token0_dec) / (10 ** token1_dec)

# ETH/USDC 0.05% pool — token0 = USDC (6 dec), token1 = WETH (18 dec)
POOL_ADDR = "0x88e6A0c2ddd26FEEb64F039a2c41296FcB3f5640"
pool = w3.eth.contract(
    address=Web3.to_checksum_address(POOL_ADDR),
    abi=POOL_ABI
)

slot0 = pool.functions.slot0().call()
usdc_per_weth = sqrt_price_to_human(slot0[0])
eth_price = 1 / usdc_per_weth

print(f"ETH/USDC (live): ${eth_price:,.2f}")
print(f"Current tick:    {slot0[1]}")

Token order is critical: in this pool, token0 is USDC (6 decimals) and token1 is WETH (18 decimals). The decoded price gives USDC-per-WETH. If you flip the pool address you'll get a different token order and an inverted price. Always verify token0 and token1 before wiring prices into live trading logic.

Understanding Uniswap V3 Fees

Uniswap V3 fees work fundamentally differently from centralized exchanges. On Binance or OKX, you pay a flat maker/taker fee to the exchange operator. On Uniswap V3, the fee is baked into each pool at deployment and flows entirely to LPs whose position was active during the swap — not to Uniswap Labs. The protocol fee (a governance-controlled cut of LP earnings) is currently set to 0% on most pools, meaning LPs collect 100% of swap fees.

The feeTier values returned by the API are raw integers in units of 0.0001%: divide by 10,000 to display the percentage, or by 1,000,000 to get the decimal multiplier for fee calculations. For LP analytics, gross daily fee income is simply daily_volume_usd × (fee_tier / 1,000,000). A pool doing $100M in daily volume at the 0.05% tier generates $50,000 in gross LP fees per day.

Error Handling and Fetching Multiple Pools

The Graph's hosted service occasionally goes down, rate-limits heavy queries, or returns partial errors in the JSON body — GraphQL errors don't always map to non-200 HTTP responses. Any script running automated strategies or continuous monitoring needs to handle all of these cases cleanly without crashing. Here's a production-ready function that fetches the top pools by TVL with comprehensive error handling:

import requests
from requests.exceptions import Timeout, ConnectionError, HTTPError

SUBGRAPH = "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3"

def fetch_top_pools(limit=10, min_tvl_usd=1_000_000):
    """
    Fetch top Uniswap V3 pools by TVL.
    Returns a list of pool dicts, or [] on any failure.
    """
    query = """
    {
      pools(
        first: %d
        orderBy: totalValueLockedUSD
        orderDirection: desc
        where: { totalValueLockedUSD_gt: "%d" }
      ) {
        id
        token0 { symbol }
        token1 { symbol }
        feeTier
        totalValueLockedUSD
        volumeUSD
        txCount
      }
    }
    """ % (limit, min_tvl_usd)

    try:
        resp = requests.post(
            SUBGRAPH,
            json={"query": query},
            timeout=15
        )
        resp.raise_for_status()
        data = resp.json()

        if "errors" in data:
            print(f"GraphQL error: {data['errors']}")
            return []

        return data["data"]["pools"]

    except Timeout:
        print("Subgraph timed out — The Graph may be congested")
        return []
    except ConnectionError:
        print("Network error — check your connection")
        return []
    except HTTPError as e:
        print(f"HTTP {e.response.status_code} error from subgraph")
        return []
    except (KeyError, ValueError) as e:
        print(f"Response parse error: {e}")
        return []


if __name__ == "__main__":
    pools = fetch_top_pools(limit=5)
    for p in pools:
        pair = f"{p['token0']['symbol']}/{p['token1']['symbol']}"
        fee  = int(p["feeTier"]) / 10000
        tvl  = float(p["totalValueLockedUSD"])
        print(f"{pair:20s}  {fee}%   TVL: ${tvl:>12,.0f}")

If you're building a live monitoring or trading system on top of Uniswap V3 data, consider pairing it with VoiceOfChain — a real-time on-chain signal platform that adds market context to raw pool analytics. DEX data tells you what happened on-chain; signal layers help you interpret what it means for your next move.

Frequently Asked Questions

Conclusion

Uniswap V3 exposes one of the richest public data sets in all of DeFi — and Python gets you to it without paying for a proprietary API or running your own infrastructure. The Graph handles historical and aggregate queries; web3.py handles real-time on-chain state. Between the two, you can build pool monitors, fee calculators, price feeds that compare DEX liquidity against Bybit or KuCoin order books, and fully automated trading bots. The patterns in this guide are the foundation. Once pool queries are running cleanly, extending to 24h volume snapshots, tick-level data, or LP position tracking is just more GraphQL and more ABI calls. Start with one pool, get the numbers flowing, and build from there.