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.
This guide shows how to load test Uniswap DeFi protocol endpoints. Perfect for testing DEX price data, pool information, and DeFi protocol reliability.
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}")