Uniswap DeFi Testing

---

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