Skip to content

Query API Reference

The Query API provides synchronous and asynchronous methods for fetching data from Dexscreener. All methods have both sync and async versions.

Overview

Query methods are one-time data fetches that return immediately with the current data. They are ideal for:

  • Getting current prices
  • Searching for tokens
  • Fetching pair information
  • Retrieving token profiles

Rate Limits

  • Pair queries: 300 requests/minute
  • Token profiles/orders: 60 requests/minute

The SDK automatically handles rate limiting with retry logic.

Pair Queries

get_pair / get_pair_async

def get_pair(address: str) -> Optional[TokenPair]
async def get_pair_async(address: str) -> Optional[TokenPair]

Get information for a single trading pair (implemented via search, no need to specify chain).

Example:

# Sync
pair = client.get_pair("JUPyiwrYJFskUPiHa7hkeR8VUtAeFoSYbKedZNsDvCN")
if pair:
    print(f"{pair.base_token.symbol}: ${pair.price_usd}")

# Async
pair = await client.get_pair_async("JUPyiwrYJFskUPiHa7hkeR8VUtAeFoSYbKedZNsDvCN")

get_pair_by_pair_address / get_pair_by_pair_address_async

def get_pair_by_pair_address(chain_id: str, pair_address: str) -> Optional[TokenPair]
async def get_pair_by_pair_address_async(chain_id: str, pair_address: str) -> Optional[TokenPair]

Get trading pair information on a specified blockchain.

Parameters:

  • chain_id: Blockchain identifier (e.g., "ethereum", "solana", "bsc")
  • pair_address: The pair contract address

Example:

pair = client.get_pair_by_pair_address(
    "ethereum",
    "0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640"  # USDC/WETH
)

get_pairs_by_pairs_addresses / get_pairs_by_pairs_addresses_async

def get_pairs_by_pairs_addresses(chain_id: str, pair_addresses: List[str]) -> List[TokenPair]
async def get_pairs_by_pairs_addresses_async(chain_id: str, pair_addresses: List[str]) -> List[TokenPair]

Batch fetch trading pair information (same chain). Supports up to 30 addresses, exceeding this will raise ValueError.

Example:

pairs = client.get_pairs_by_pairs_addresses(
    "ethereum",
    [
        "0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640",
        "0x11b815efb8f581194ae79006d24e0d814b7697f6",
        "0x4e68ccd3e89f51c3074ca5072bbac773960dfa36"
    ]
)

search_pairs / search_pairs_async

def search_pairs(query: str) -> List[TokenPair]
async def search_pairs_async(query: str) -> List[TokenPair]

Search for trading pairs (by name, symbol, or address).

Example:

# Search by symbol
results = client.search_pairs("PEPE")

# Search by name
results = client.search_pairs("Shiba Inu")

# Search by address
results = client.search_pairs("0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640")

get_pairs_by_token_address / get_pairs_by_token_address_async

def get_pairs_by_token_address(chain_id: str, token_address: str) -> List[TokenPair]
async def get_pairs_by_token_address_async(chain_id: str, token_address: str) -> List[TokenPair]

Get all trading pairs for a single token on a specified chain. Returns all trading pairs for that token across all DEXs on that chain.

Example:

# Get all USDC pairs on Ethereum
usdc_pairs = client.get_pairs_by_token_address(
    "ethereum",
    "A0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"  # USDC
)

for pair in usdc_pairs:
    other_token = pair.quote_token if pair.base_token.address.lower() == usdc_address.lower() else pair.base_token
    print(f"USDC/{other_token.symbol} on {pair.dex_id}: ${pair.price_usd}")

get_pairs_by_token_addresses / get_pairs_by_token_addresses_async

def get_pairs_by_token_addresses(chain_id: str, token_addresses: List[str]) -> List[TokenPair]
async def get_pairs_by_token_addresses_async(chain_id: str, token_addresses: List[str]) -> List[TokenPair]

Batch fetch all trading pairs for multiple tokens on a specified chain. Returns all trading pairs containing any of the specified tokens (deduplicated).

Important Notes:

  • Chain ID must be specified (e.g., "solana", "ethereum", etc.)
  • Supports up to 30 token addresses max, exceeding this will raise ValueError
  • API returns maximum 30 trading pairs (most relevant/active)
  • Returns a collection of all trading pairs containing any of these tokens. If a trading pair contains multiple specified tokens (e.g., USDC/SOL pair contains both USDC and SOL), it will only appear once

Example:

# Get pairs for multiple tokens
token_addresses = [
    "A0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",  # USDC
    "dac17f958d2ee523a2206206994597c13d831ec7",  # USDT
    "6B175474E89094C44Da98b954EedeAC495271d0F"   # DAI
]

stablecoin_pairs = client.get_pairs_by_token_addresses("ethereum", token_addresses)

Token Information Queries

get_latest_token_profiles / get_latest_token_profiles_async

def get_latest_token_profiles() -> List[TokenInfo]
async def get_latest_token_profiles_async() -> List[TokenInfo]

Get latest token profiles. Rate limit: 60 requests/minute.

Example:

profiles = client.get_latest_token_profiles()
for token in profiles[:10]:
    print(f"{token.token_address}: {token.description}")

get_latest_boosted_tokens / get_latest_boosted_tokens_async

def get_latest_boosted_tokens() -> List[TokenInfo]
async def get_latest_boosted_tokens_async() -> List[TokenInfo]

Get latest boosted tokens. Rate limit: 60 requests/minute.

Example:

boosted = client.get_latest_boosted_tokens()
for token in boosted:
    print(f"{token.chain_id}: {token.token_address} - Boost: {token.amount}")

get_tokens_most_active / get_tokens_most_active_async

def get_tokens_most_active() -> List[TokenInfo]
async def get_tokens_most_active_async() -> List[TokenInfo]

Get most active boosted tokens. Rate limit: 60 requests/minute.

Pool Information Queries

get_pools_by_token_address / get_pools_by_token_address_async

def get_pools_by_token_address(chain_id: str, token_address: str) -> List[TokenPair]
async def get_pools_by_token_address_async(chain_id: str, token_address: str) -> List[TokenPair]

Get pool information using the token-pairs/v1 endpoint. Similar to get_pairs_by_pairs_addresses but uses a different API endpoint. Rate limit: 300 requests/minute.

Order Queries

get_orders_paid_of_token / get_orders_paid_of_token_async

def get_orders_paid_of_token(chain_id: str, token_address: str) -> List[OrderInfo]
async def get_orders_paid_of_token_async(chain_id: str, token_address: str) -> List[OrderInfo]

Get paid orders for a token. Rate limit: 60 requests/minute.

Example:

orders = client.get_orders_paid_of_token("ethereum", token_address)
for order in orders:
    print(f"Order {order.type}: {order.status} at {order.payment_timestamp}")

Best Practices

1. Use Async for Multiple Queries

When fetching data for multiple tokens/pairs, use async methods:

import asyncio

async def fetch_multiple_tokens():
    client = DexscreenerClient()

    tokens = ["address1", "address2", "address3"]

    # Concurrent fetching
    tasks = [
        client.get_pairs_by_token_address_async("ethereum", addr)
        for addr in tokens
    ]

    results = await asyncio.gather(*tasks)
    return results

2. Handle Rate Limits

The SDK handles rate limits automatically, but you can also be proactive:

# Space out requests for token profiles (60/min limit)
import time

for token_address in large_token_list:
    orders = client.get_orders_paid_of_token("ethereum", token_address)
    process_orders(orders)
    time.sleep(1.1)  # ~54 requests per minute to be safe

3. Error Handling

Always handle potential None returns:

pair = client.get_pair_by_pair_address("ethereum", pair_address)
if pair:
    # Process pair data
    print(f"Price: ${pair.price_usd}")
else:
    print(f"Pair {pair_address} not found")

4. Batch Operations

For multiple addresses on the same chain, use batch methods:

# Instead of this:
pairs = []
for address in addresses:
    pair = client.get_pair_by_pair_address("ethereum", address)
    if pair:
        pairs.append(pair)

# Do this:
pairs = client.get_pairs_by_pairs_addresses("ethereum", addresses)

Common Patterns

Finding Arbitrage Opportunities

async def find_arbitrage(token_address: str):
    client = DexscreenerClient()

    # Get pairs across multiple chains
    chains = ["ethereum", "bsc", "polygon", "arbitrum"]

    tasks = [
        client.get_pairs_by_token_address_async(chain, token_address)
        for chain in chains
    ]

    all_pairs = await asyncio.gather(*tasks)

    # Flatten and analyze
    prices_by_chain = {}
    for chain, pairs in zip(chains, all_pairs):
        if pairs:
            # Get the most liquid pair
            best_pair = max(pairs, key=lambda p: p.liquidity.usd or 0)
            prices_by_chain[chain] = best_pair.price_usd

    # Find arbitrage
    if len(prices_by_chain) > 1:
        min_price = min(prices_by_chain.values())
        max_price = max(prices_by_chain.values())
        spread = ((max_price - min_price) / min_price) * 100

        if spread > 1:  # 1% threshold
            print(f"Arbitrage opportunity: {spread:.2f}% spread")
            return prices_by_chain

Token Discovery

def discover_new_tokens():
    client = DexscreenerClient()

    # Get latest profiles
    profiles = client.get_latest_token_profiles()

    # Filter interesting tokens
    interesting = []
    for token in profiles:
        # Get trading data
        pairs = client.get_pairs_by_token_address(token.chain_id, token.token_address)

        if pairs:
            total_volume = sum(p.volume.h24 for p in pairs if p.volume.h24)
            total_liquidity = sum(p.liquidity.usd for p in pairs if p.liquidity and p.liquidity.usd)

            if total_volume > 100_000 and total_liquidity > 50_000:
                interesting.append({
                    'token': token,
                    'pairs': pairs,
                    'volume': total_volume,
                    'liquidity': total_liquidity
                })

    return interesting