The Graph Token API Client

A powerful Python client for The Graph Token API that brings blockchain data to your fingertips. Access token balances, NFT ownership, DeFi swaps, and price histories across Ethereum, Solana, and 8+ other chains with an elegant, type-safe interface.

Current Spec version: 4.0
Conforms to token-api v2.3.3

🌟 Why The Graph Token API Client?

The Challenge

Accessing blockchain data is complex:

• Multiple chains with different APIs and data formats 🔗
• Inconsistent interfaces between EVM and Solana ecosystems 🤔
• Raw responses requiring extensive parsing and validation 📊
• Rate limiting and error handling complexity 🚦

Our Solution

The Graph Token API Client provides a unified, elegant interface for all your blockchain data needs:

# Without this client - Complex and error-prone 😰
import requests
response = requests.get(
    "https://api.thegraph.com/v1/tokens/evm/ethereum/balances",
    headers={"Authorization": f"Bearer {api_key}"},
    params={"address": wallet_address}
)
if response.status_code == 200:
    data = response.json()
    # Parse nested structures, handle errors...

# With this client - Clean and simple ✨
from thegraph_token_api import TokenAPI, Currency

api = TokenAPI()
balances = await api.evm.balances(wallet_address)
# That's it! Fully typed, validated, and ready to use

# NEW: Unified Price API (convenience feature - not an API endpoint)
eth_price = await api.price.get(Currency.ETH)  # $3,873+ from Ethereum DEX swaps
sol_price = await api.price.get(Currency.SOL)  # $192+ from Solana DEX swaps
pol_price = await api.price.get(Currency.POL)  # $0.22 from Polygon network
bnb_price = await api.price.get(Currency.BNB)  # $845+ from BSC network
avax_price = await api.price.get(Currency.AVAX) # $26+ from Avalanche network

✨ Features

• 🏗️ Elegant Architecture: Intuitive separation between EVM and SVM (Solana) chains
• 🌐 Multi-Chain Support: 9 chains including Ethereum, Polygon, BSC, Arbitrum, and Solana
• 📊 Comprehensive Data: Token balances, NFTs, DeFi swaps, prices, transfers, and more
• ⚡ High Performance: Async/await support with connection pooling
• 🛡️ Type Safety: Full type hints with runtime validation
• 🔄 Smart Defaults: Auto-loads API keys, sensible limits, mainnet defaults
• 📈 Time-Series Data: Historical prices and time-filtered swap data
• 💰 Unified Price API: Real-time prices for 5 major cryptocurrencies (ETH, SOL, POL, BNB, AVAX)
• 🎯 Developer Friendly: Clean API, great docs, extensive examples

📦 Installation

# Using pip
pip install divine-thegraph-token-api

# Using uv
uv add divine-thegraph-token-api

# For development
git clone https://github.com/divine/thegraph-token-api
cd thegraph-token-api
uv sync

Requirements

• Python 3.13+
• API key from thegraph.market (free tier available)

🚀 Quick Start

1. Get Your API Key

Visit thegraph.market and click "Get API Key" (free tier available).

2. Set Up Environment

Create a .env file:

THEGRAPH_API_KEY=your_api_key_here

3. Start Coding!

import anyio
from thegraph_token_api import TokenAPI

async def main():
    # Initialize client (auto-loads from .env)
    api = TokenAPI()
    
    # Get Ethereum token balances
    vitalik_wallet = "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
    balances = await api.evm.balances(vitalik_wallet)
    
    for token in balances:
        if token["balance"] > 0:
            print(f"{token['symbol']}: {token['balance']} (${token['value_usd']:,.2f})")
    
    # Get current cryptocurrency prices (5 major currencies supported)
    eth_price = await api.price.get(Currency.ETH)   # ~$3,873
    sol_price = await api.price.get(Currency.SOL)   # ~$192
    pol_price = await api.price.get(Currency.POL)   # ~$0.22
    bnb_price = await api.price.get(Currency.BNB)   # ~$845
    avax_price = await api.price.get(Currency.AVAX) # ~$26
    
    print(f"\nCurrent Prices:")
    print(f"ETH: ${eth_price:.2f}") if eth_price else print("ETH: unavailable")
    print(f"SOL: ${sol_price:.2f}") if sol_price else print("SOL: unavailable")
    print(f"POL: ${pol_price:.2f}") if pol_price else print("POL: unavailable")
    print(f"BNB: ${bnb_price:.2f}") if bnb_price else print("BNB: unavailable")
    print(f"AVAX: ${avax_price:.2f}") if avax_price else print("AVAX: unavailable")
    
    # Get detailed price statistics
    eth_stats = await api.price.get(Currency.ETH, include_stats=True)
    if eth_stats:
        print(f"ETH confidence: {eth_stats['confidence']:.0%} (from {eth_stats['trades_analyzed']} trades)")

anyio.run(main)

📖 Core Concepts

Chain Separation

The API elegantly separates EVM and SVM operations:

# EVM chains (Ethereum, Polygon, BSC, etc.)
await api.evm.balances(address)        # Get token balances
await api.evm.nfts.ownerships(address) # Get NFT holdings
await api.evm.swaps(protocol=...)      # Get DEX swaps

# SVM (Solana)
await api.svm.balances(mint=...)       # Get SPL token holders
await api.svm.transfers(mint=...)      # Get token transfers
await api.svm.swaps(program_id=...)    # Get DEX swaps

# Unified Price API (convenience feature - uses API data internally)
await api.price.get(Currency.ETH)      # Get current ETH price from Ethereum DEX swaps
await api.price.get(Currency.SOL)      # Get current SOL price from Solana DEX swaps
await api.price.get(Currency.POL)      # Get current POL price from Polygon network
await api.price.get(Currency.BNB)      # Get current BNB price from BSC network
await api.price.get(Currency.AVAX)     # Get current AVAX price from Avalanche network

Smart Organization

Methods are logically grouped for discoverability:

# NFT operations grouped together
api.evm.nfts.ownerships(address)    # What NFTs does this address own?
api.evm.nfts.collection(contract)   # Get collection details
api.evm.nfts.activities(contract)   # Recent NFT activities

# Pool operations grouped together
api.evm.pools(token=address)        # Find liquidity pools
api.evm.pool_history(pool, "1h")    # Get pool metrics over time

# Price operations (5 major cryptocurrencies)
api.price.get(Currency.ETH)          # Current ETH price (~$3,873)
api.price.get(Currency.SOL)          # Current SOL price (~$192)
api.price.get(Currency.POL)          # Current POL price (~$0.22)
api.price.get(Currency.BNB)          # Current BNB price (~$845)
api.price.get(Currency.AVAX)         # Current AVAX price (~$26)
api.price.get_supported_currencies() # List all 5 supported currencies

🔥 Usage Examples

Token Balances Across Chains

from thegraph_token_api import TokenAPI, Chain

api = TokenAPI()

# Check balances on multiple chains
async def check_portfolio(address: str):
    chains = [Chain.ETHEREUM, Chain.POLYGON, Chain.ARBITRUM]
    
    total_value = 0
    for chain in chains:
        balances = await api.evm.balances(address, chain=chain)
        
        print(f"\n{chain.value.title()} Holdings:")
        for token in balances[:5]:  # Top 5 tokens
            value = token.get("value_usd", 0)
            total_value += value
            print(f"  {token['symbol']}: ${value:,.2f}")
    
    print(f"\nTotal Portfolio Value: ${total_value:,.2f}")

NFT Collection Analytics

# Analyze an NFT collection
async def analyze_nft_collection(contract: str):
    # Get collection info
    collection = await api.evm.nfts.collection(contract)
    print(f"Collection: {collection['name']}")
    print(f"Total Supply: {collection['total_supply']}")
    
    # Get recent activities
    activities = await api.evm.nfts.activities(
        contract, 
        chain=Chain.ETHEREUM,
        limit=10
    )
    
    for activity in activities:
        print(f"{activity['type']}: Token #{activity['token_id']} "
              f"for {activity['price']} ETH")

DeFi Swap Monitoring

from datetime import datetime, timedelta
from thegraph_token_api import Protocol, SwapPrograms

# Monitor recent Uniswap V3 swaps
async def monitor_uniswap_swaps():
    swaps = await api.evm.swaps(
        protocol=Protocol.UNISWAP_V3,
        chain=Chain.ETHEREUM,
        limit=20
    )
    
    for swap in swaps:
        print(f"Swap: {swap['amount_in']} {swap['token_in_symbol']} → "
              f"{swap['amount_out']} {swap['token_out_symbol']}")
        print(f"Value: ${swap['value_usd']:,.2f}")
        print(f"DEX: {swap['dex_name']}\n")

# Monitor Solana swaps with time filtering
async def monitor_solana_swaps():
    # Get swaps from last 30 minutes
    end_time = int(datetime.now().timestamp())
    start_time = int((datetime.now() - timedelta(minutes=30)).timestamp())
    
    swaps = await api.svm.swaps(
        program_id=SwapPrograms.RAYDIUM,
        start_time=start_time,
        end_time=end_time,
        limit=10
    )
    
    print(f"Recent Raydium swaps (last 30 min):")
    for swap in swaps:
        print(f"${swap['value_usd']:,.2f} swap at "
              f"{datetime.fromtimestamp(swap['timestamp'])}")

Unified Price API (Convenience Feature)

from thegraph_token_api import TokenAPI, Currency

# The Unified Price API is a convenience feature that uses DEX swap data
# internally to calculate current cryptocurrency prices for 5 major currencies
async def get_current_prices():
    api = TokenAPI()
    
    # Simple price queries using Currency enum (type-safe, enum-only interface)
    eth_price = await api.price.get(Currency.ETH)    # ~$3,873
    sol_price = await api.price.get(Currency.SOL)    # ~$192
    pol_price = await api.price.get(Currency.POL)    # ~$0.22
    bnb_price = await api.price.get(Currency.BNB)    # ~$845
    avax_price = await api.price.get(Currency.AVAX)  # ~$26
    
    print(f"ETH: ${eth_price:.2f}" if eth_price else "ETH: unavailable")
    print(f"SOL: ${sol_price:.2f}" if sol_price else "SOL: unavailable")
    print(f"POL: ${pol_price:.2f}" if pol_price else "POL: unavailable")
    print(f"BNB: ${bnb_price:.2f}" if bnb_price else "BNB: unavailable")
    print(f"AVAX: ${avax_price:.2f}" if avax_price else "AVAX: unavailable")
    
    # NOTE: String support has been removed - only Currency enum is supported
    # This ensures type safety and eliminates runtime errors
    
    # Get detailed statistics with confidence metrics
    eth_stats = await api.price.get(Currency.ETH, include_stats=True)
    if eth_stats:
        print(f"\nETH Detailed Analysis:")
        print(f"  Price: ${eth_stats['price']:.2f}")
        print(f"  Confidence: {eth_stats['confidence']:.0%}")
        print(f"  Trades analyzed: {eth_stats['trades_analyzed']}")
        print(f"  Volatility: ${eth_stats['std_deviation']:.2f}")
        print(f"  Range: ${eth_stats['min_price']:.2f} - ${eth_stats['max_price']:.2f}")
    
    # Check supported currencies
    supported = await api.price.get_supported_currencies()
    print(f"\nSupported currencies: {[c.value for c in supported]}")
    
    # Cache management
    await api.price.clear_cache(Currency.ETH)  # Clear specific currency
    await api.price.clear_cache()              # Clear all cache

Price History Analysis

# Get historical token prices
async def analyze_token_price(token_address: str):
    # Get 7-day price history with daily intervals
    prices = await api.evm.price_history(
        token=token_address,
        chain=Chain.ETHEREUM,
        interval="1d",
        days=7
    )
    
    # Calculate price change
    if len(prices) >= 2:
        start_price = prices[0]["price_usd"]
        end_price = prices[-1]["price_usd"]
        change = ((end_price - start_price) / start_price) * 100
        
        print(f"7-day price change: {change:+.2f}%")
        print(f"Current price: ${end_price:,.4f}")
        
        # Find min/max
        min_price = min(p["price_usd"] for p in prices)
        max_price = max(p["price_usd"] for p in prices)
        print(f"7-day range: ${min_price:,.4f} - ${max_price:,.4f}")

Token Holder Analysis

# Analyze token holders
async def analyze_token_holders(token_address: str):
    holders = await api.evm.token_holders(
        token_address,
        chain=Chain.ETHEREUM,
        limit=100
    )
    
    # Calculate concentration
    total_supply = sum(h["balance"] for h in holders)
    top_10_balance = sum(h["balance"] for h in holders[:10])
    concentration = (top_10_balance / total_supply) * 100
    
    print(f"Top 10 holders own {concentration:.1f}% of supply")
    
    # Show top holders
    for i, holder in enumerate(holders[:5], 1):
        pct = (holder["balance"] / total_supply) * 100
        print(f"{i}. {holder['address'][:10]}... - {pct:.2f}%")

🌐 Supported Networks & Protocols

Blockchains

DEX Protocols

EVM DEXs:

• Uniswap V2 & V3
• SushiSwap
• PancakeSwap
• And more...

Solana DEXs:

• Raydium
• Orca
• Jupiter
• Pump.fun

📚 API Reference

Core Classes

TokenAPI

The main entry point for all API operations.

api = TokenAPI(api_key: Optional[str] = None)
# If api_key is None, loads from THEGRAPH_API_KEY env var

# Access different interfaces
api.evm     # EVM chain operations
api.svm     # Solana operations  
api.price   # Unified Price API (convenience feature)

EVMInterface

Access via api.evm - handles all EVM chain operations.

# Token operations
await api.evm.balances(address: str, chain: Chain = Chain.ETHEREUM)
await api.evm.token_info(contract: str, chain: Chain = Chain.ETHEREUM)
await api.evm.token_holders(contract: str, chain: Chain = Chain.ETHEREUM)

# NFT operations  
await api.evm.nfts.ownerships(address: str, chain: Chain = Chain.ETHEREUM)
await api.evm.nfts.collection(contract: str, chain: Chain = Chain.ETHEREUM)
await api.evm.nfts.activities(contract: str, chain: Chain = Chain.ETHEREUM)

# DeFi operations
await api.evm.swaps(protocol: Protocol, chain: Chain = Chain.ETHEREUM)
await api.evm.pools(token: str, chain: Chain = Chain.ETHEREUM)

# Price data
await api.evm.price_history(token: str, interval: str, days: int)
await api.evm.pool_history(pool: str, interval: str)

# Transfers
await api.evm.transfers(from_address: str = None, to_address: str = None)

#### `UnifiedPriceAPI` (Convenience Feature)

Access via `api.price` - provides unified cryptocurrency prices.

**Note:** This is a convenience feature that uses the API's DEX swap data internally to calculate prices. It is not a separate API endpoint.

```python
# Simple price queries (enum-only interface)
await api.price.get(Currency.ETH)                    # Current ETH price (~$3,873)
await api.price.get(Currency.SOL)                    # Current SOL price (~$192)
await api.price.get(Currency.POL)                    # Current POL price (~$0.22)
await api.price.get(Currency.BNB)                    # Current BNB price (~$845)
await api.price.get(Currency.AVAX)                   # Current AVAX price (~$26)

# Detailed statistics
autostats = await api.price.get(Currency.ETH, include_stats=True)
# Returns: {"price": 3500.0, "confidence": 0.9, "trades_analyzed": 15, ...}

# Utility methods
await api.price.get_supported_currencies()          # List[Currency]
await api.price.is_supported(Currency.ETH)          # bool
await api.price.clear_cache(Currency.ETH)           # Clear specific cache
await api.price.clear_cache()                       # Clear all cache

# Force refresh (bypass cache)
await api.price.get(Currency.ETH, force_refresh=True)

#### `SVMInterface`

Access via `api.svm` - handles Solana operations.

```python
# Token operations
await api.svm.balances(mint: str, limit: int = 100)
await api.svm.transfers(mint: str, limit: int = 100)

# DEX operations with time filtering
await api.svm.swaps(
    program_id: str = None,
    start_time: int = None,
    end_time: int = None,
    limit: int = 100
)

Enums

from thegraph_token_api import Chain, Protocol, SwapPrograms, Currency

# EVM chains
Chain.ETHEREUM
Chain.POLYGON
Chain.BSC
# ... etc

# EVM DEX protocols
Protocol.UNISWAP_V2
Protocol.UNISWAP_V3
Protocol.SUSHISWAP
# ... etc

# Solana DEX programs
SwapPrograms.RAYDIUM
SwapPrograms.ORCA
SwapPrograms.JUPITER
SwapPrograms.PUMP_FUN

# Supported currencies for Unified Price API (5 major cryptocurrencies)
Currency.ETH    # Ethereum (~$3,873)
Currency.SOL    # Solana (~$192)
Currency.POL    # Polygon (~$0.22)
Currency.BNB    # BNB Chain (~$845)
Currency.AVAX   # Avalanche (~$26)

Error Handling

from thegraph_token_api import TokenAPIError
from type_enforcer import ValidationError

try:
    balances = await api.evm.balances(address)
except TokenAPIError as e:
    print(f"API error: {e}")
except ValidationError as e:
    print(f"Data validation error: {e}")

🛠️ Advanced Usage

Custom API Configuration

from thegraph_token_api import TokenAPI

# Custom initialization
api = TokenAPI(
    api_key="your_api_key",
    timeout=30.0,  # Request timeout in seconds
    max_retries=3  # Number of retries on failure
)

Batch Operations

# Fetch data for multiple addresses efficiently
async def batch_fetch_balances(addresses: list[str]):
    tasks = [
        api.evm.balances(addr, chain=Chain.ETHEREUM)
        for addr in addresses
    ]
    
    results = await anyio.gather(*tasks)
    
    for addr, balances in zip(addresses, results):
        total_value = sum(t.get("value_usd", 0) for t in balances)
        print(f"{addr}: ${total_value:,.2f}")

Caching Strategies

from functools import lru_cache
from datetime import datetime, timedelta

# Cache token info for 1 hour
@lru_cache(maxsize=1000)
async def get_token_info_cached(contract: str, chain: str):
    return await api.evm.token_info(contract, Chain(chain))

# Cache with TTL
class CachedTokenAPI:
    def __init__(self, api: TokenAPI, ttl: int = 3600):
        self.api = api
        self.cache = {}
        self.ttl = ttl
    
    async def get_token_info(self, contract: str):
        key = f"token:{contract}"
        now = datetime.now()
        
        if key in self.cache:
            data, timestamp = self.cache[key]
            if now - timestamp < timedelta(seconds=self.ttl):
                return data
        
        data = await self.api.evm.token_info(contract)
        self.cache[key] = (data, now)
        return data

🔧 Development

Setup

# Clone the repository
git clone https://github.com/divinescreener/thegraph-token-api
cd thegraph-token-api

# Install dependencies
uv sync

# Run tests
uv run pytest

# Run linting
uv run ruff check
uv run mypy src

# Run examples
uv run python examples/basic_usage.py

Project Structure

thegraph-token-api/
├── src/
│   └── thegraph_token_api/
│       ├── __init__.py
│       ├── core.py          # Main API client
│       ├── evm.py           # EVM chain interface
│       ├── svm.py           # Solana interface
│       ├── types.py         # Type definitions
│       └── utils.py         # Helper functions
├── tests/                   # Test suite
├── examples/                # Usage examples
│   ├── endpoints/          
│   │   ├── evm/            # EVM examples
│   │   └── svm/            # Solana examples
│   └── basic_usage.py      # Getting started
├── API_REFERENCE.md        # Detailed API docs
└── pyproject.toml          # Project configuration

Contributing

We welcome contributions! Please see our Contributing Guidelines.

Key requirements:

• Maintain 100% test coverage (382 tests currently passing)
• Follow the existing code style (ruff + mypy strict mode)
• Add tests for new features
• Update documentation

📊 Performance

The client is optimized for production use:

• Connection Pooling: Reuses HTTP connections
• Async Operations: Non-blocking I/O for high throughput
• Smart Caching: Unified Price API with volatility-based TTL
• Batch Support: Efficient multi-request handling
• Statistical Analysis: Median pricing with IQR outlier filtering
• Progressive Retry: Adaptive sampling for reliable price data
• 100% Test Coverage: 382 tests ensuring production reliability
• Multi-Chain Architecture: Unified EVM approach for scalability

Unified Price API Performance

The Unified Price API uses advanced techniques for reliable pricing:

• Volatility-Based Caching: Shorter cache (60s) during high volatility, longer (300s) during stable periods
• Multi-Source Aggregation: Uses recent DEX swap data from multiple sources
• Statistical Robustness: Median prices with outlier filtering prevent manipulation
• Confidence Scoring: Returns confidence metrics based on sample size and data quality
• Smart Retries: Progressive retry with increasing sample sizes for reliable data

🔒 Security

• API Key Safety: Never logged or exposed
• Input Validation: All parameters validated
• Type Safety: Runtime type checking
• Secure HTTP: TLS 1.2+ enforced

📄 License

MIT License - see LICENSE for details.

🙏 Acknowledgments

• Built on The Graph infrastructure
• Type safety via divine-type-enforcer
• HTTP handling by divine-typed-requests

🆘 Support

• 📖 Documentation: API Reference | Official Docs
• 💬 Discord: The Graph Discord
• 🐛 Issues: GitHub Issues
• 🔑 API Keys: thegraph.market

Made with ❤️ by DIVINE