Uniswap DeFi Testing

Simple load testing for Uniswap decentralized exchange with price quotes and pool data

LoadForge can record your browser, graphically build tests, scan your site with a wizard and more. Sign up now to run your first test.

Sign up now


This guide shows how to load test Uniswap DeFi protocol endpoints. Perfect for testing DEX price data, pool information, and DeFi protocol reliability.

Use Cases

  • Test Uniswap API response times
  • Monitor DEX price quote accuracy
  • Validate pool liquidity data
  • Check DeFi protocol availability

Simple Implementation

from locust import task, HttpUser
import json
import random

class UniswapUser(HttpUser):
    def on_start(self):
        # Uniswap V3 Subgraph endpoint
        self.subgraph_url = "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3"
        
        # Popular trading pairs for testing
        self.token_pairs = [
            {"token0": "WETH", "token1": "USDC", "address": "0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8"},
            {"token0": "WETH", "token1": "USDT", "address": "0x4e68ccd3e89f51c3074ca5072bbac773960dfa36"},
            {"token0": "WETH", "token1": "DAI", "address": "0xc2e9f25be6257c210d7adf0d4cd6e3e881ba25f8"},
            {"token0": "USDC", "token1": "USDT", "address": "0x3416cf6c708da44db2624d63ea0aaef7113527c6"}
        ]
        
        # Common token addresses
        self.tokens = {
            "WETH": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
            "USDC": "0xa0b86a33e6441b8435b662303c0f479c7e1b5b8e", 
            "USDT": "0xdac17f958d2ee523a2206206994597c13d831ec7",
            "DAI": "0x6b175474e89094c44da98b954eedeac495271d0f"
        }

    @task(3)
    def get_pool_data(self):
        """Get pool information from Uniswap"""
        pair = random.choice(self.token_pairs)
        
        query = """
        {
          pool(id: "%s") {
            id
            token0 {
              symbol
              decimals
            }
            token1 {
              symbol
              decimals
            }
            feeTier
            liquidity
            sqrtPrice
            tick
            token0Price
            token1Price
            volumeUSD
            txCount
          }
        }
        """ % pair["address"].lower()
        
        payload = {"query": query}
        
        with self.client.post(
            self.subgraph_url,
            json=payload,
            name="Pool Data"
        ) as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    if "data" in data and data["data"]["pool"]:
                        pool = data["data"]["pool"]
                        token0_symbol = pool["token0"]["symbol"]
                        token1_symbol = pool["token1"]["symbol"]
                        price = float(pool["token0Price"])
                        volume = float(pool["volumeUSD"])
                        
                        print(f"{token0_symbol}/{token1_symbol}: Price {price:.6f}, Volume ${volume:,.0f}")
                    else:
                        response.failure("No pool data returned")
                        
                except (json.JSONDecodeError, KeyError, ValueError) as e:
                    response.failure(f"Failed to parse pool data: {e}")
            else:
                response.failure(f"Subgraph error: {response.status_code}")

    @task(2)
    def get_token_prices(self):
        """Get current token prices"""
        query = """
        {
          tokens(first: 5, orderBy: volumeUSD, orderDirection: desc) {
            id
            symbol
            name
            decimals
            derivedETH
            volumeUSD
            txCount
          }
        }
        """
        
        payload = {"query": query}
        
        with self.client.post(
            self.subgraph_url,
            json=payload,
            name="Token Prices"
        ) as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    if "data" in data and "tokens" in data["data"]:
                        tokens = data["data"]["tokens"]
                        
                        for token in tokens[:3]:  # Show top 3
                            symbol = token["symbol"]
                            price_eth = float(token["derivedETH"])
                            volume = float(token["volumeUSD"])
                            print(f"{symbol}: {price_eth:.6f} ETH, Volume ${volume:,.0f}")
                    else:
                        response.failure("No token data returned")
                        
                except (json.JSONDecodeError, KeyError, ValueError) as e:
                    response.failure(f"Failed to parse token data: {e}")
            else:
                response.failure(f"Subgraph error: {response.status_code}")

    @task(2)
    def get_recent_swaps(self):
        """Get recent swap transactions"""
        query = """
        {
          swaps(first: 10, orderBy: timestamp, orderDirection: desc) {
            id
            timestamp
            pool {
              token0 {
                symbol
              }
              token1 {
                symbol
              }
            }
            amount0
            amount1
            amountUSD
            sender
          }
        }
        """
        
        payload = {"query": query}
        
        with self.client.post(
            self.subgraph_url,
            json=payload,
            name="Recent Swaps"
        ) as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    if "data" in data and "swaps" in data["data"]:
                        swaps = data["data"]["swaps"]
                        print(f"Retrieved {len(swaps)} recent swaps")
                        
                        if swaps:
                            swap = swaps[0]  # Show most recent
                            token0 = swap["pool"]["token0"]["symbol"]
                            token1 = swap["pool"]["token1"]["symbol"]
                            amount_usd = float(swap["amountUSD"])
                            print(f"Latest swap: {token0}/{token1} - ${amount_usd:,.2f}")
                    else:
                        response.failure("No swap data returned")
                        
                except (json.JSONDecodeError, KeyError, ValueError) as e:
                    response.failure(f"Failed to parse swap data: {e}")
            else:
                response.failure(f"Subgraph error: {response.status_code}")

    @task(1)
    def get_pool_stats(self):
        """Get overall Uniswap statistics"""
        query = """
        {
          uniswapDayDatas(first: 1, orderBy: date, orderDirection: desc) {
            date
            volumeUSD
            tvlUSD
            txCount
          }
        }
        """
        
        payload = {"query": query}
        
        with self.client.post(
            self.subgraph_url,
            json=payload,
            name="Pool Stats"
        ) as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    if "data" in data and "uniswapDayDatas" in data["data"]:
                        day_data = data["data"]["uniswapDayDatas"]
                        
                        if day_data:
                            stats = day_data[0]
                            volume = float(stats["volumeUSD"])
                            tvl = float(stats["tvlUSD"])
                            tx_count = int(stats["txCount"])
                            
                            print(f"Daily Volume: ${volume:,.0f}, TVL: ${tvl:,.0f}, Txs: {tx_count:,}")
                    else:
                        response.failure("No stats data returned")
                        
                except (json.JSONDecodeError, KeyError, ValueError) as e:
                    response.failure(f"Failed to parse stats data: {e}")
            else:
                response.failure(f"Subgraph error: {response.status_code}")

Setup Instructions

  1. No API key required for The Graph public endpoints
  2. Uniswap V3 subgraph provides real-time DEX data
  3. Test with popular trading pairs for reliable data
  4. Monitor for rate limits on The Graph

What This Tests

  • Subgraph Performance: Tests The Graph indexing service
  • DeFi Data Accuracy: Validates pool and price data
  • Real-time Updates: Checks how current the data is
  • Query Complexity: Tests different GraphQL query types

Key Metrics

  • Pool Liquidity: Total value locked in trading pairs
  • Trading Volume: 24h volume for pools and tokens
  • Price Data: Current exchange rates between tokens
  • Transaction Count: Number of swaps and interactions

Common Issues

  • Subgraph Delays: Data may be slightly behind blockchain
  • Rate Limits: The Graph has query limits
  • Pool Addresses: Ensure correct pool contract addresses
  • Price Volatility: DeFi prices change rapidly

DeFi Concepts

  • TVL: Total Value Locked in the protocol
  • Liquidity: Available tokens for trading
  • Slippage: Price impact of large trades
  • Fees: Trading fees collected by liquidity providers

Ready to run your test?
Launch your locust test at scale.