divine-thegraph-token-api 0.2.2
A Python client for The Graph Token API with elegant EVM/SVM separation and excellent test coverage
License: MIT
Requires Python: >=3.13

# The Graph Token API Client

[![PyPI version](https://badge.fury.io/py/divine-thegraph-token-api.svg)](https://badge.fury.io/py/divine-thegraph-token-api)
[![Python versions](https://img.shields.io/pypi/pyversions/divine-thegraph-token-api.svg)](https://pypi.org/project/divine-thegraph-token-api/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Test Coverage](https://img.shields.io/badge/coverage-100%25-brightgreen.svg)](https://github.com/divinescreener/thegraph-token-api)
[![Code style: ruff](https://img.shields.io/badge/code%20style-ruff-000000.svg)](https://github.com/astral-sh/ruff)

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](https://github.com/pinax-network/token-api/releases/tag/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:

```python
# 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

```bash
# 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](https://thegraph.market) (free tier available)

## 🚀 Quick Start

### 1. Get Your API Key

Visit [thegraph.market](https://thegraph.market) and click "Get API Key" (free tier available).

### 2. Set Up Environment

Create a `.env` file:
```bash
THEGRAPH_API_KEY=your_api_key_here
```

### 3. Start Coding!

```python
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:

```python
# 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:

```python
# 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

```python
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

```python
# 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

```python
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)

```python
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

```python
# 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

```python
# 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

| Network | Chain ID | Type | Status |
|---------|----------|------|--------|
| Ethereum | `ethereum` | EVM | ✅ Supported |
| Polygon | `polygon` | EVM | ✅ Supported |
| BNB Chain | `bsc` | EVM | ✅ Supported |
| Arbitrum | `arbitrum` | EVM | ✅ Supported |
| Optimism | `optimism` | EVM | ✅ Supported |
| Avalanche | `avalanche` | EVM | ✅ Supported |
| Base | `base` | EVM | ✅ Supported |
| Unichain | `unichain` | EVM | ✅ Supported |
| Solana | `solana` | SVM | ✅ Supported |

### 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.

```python
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.

```python
# 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

```python
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

```python
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

```python
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

```python
# 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

```python
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

```bash
# 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](CONTRIBUTING.md).

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](LICENSE) for details.

## 🙏 Acknowledgments

- Built on [The Graph](https://thegraph.com/) infrastructure
- Type safety via [divine-type-enforcer](https://github.com/divinescreener/type-enforcer)
- HTTP handling by [divine-typed-requests](https://github.com/divinescreener/typed-requests)

## 🆘 Support

- 📖 **Documentation**: [API Reference](API_REFERENCE.md) | [Official Docs](https://thegraph.com/docs/en/token-api/quick-start/)
- 💬 **Discord**: [The Graph Discord](https://discord.gg/thegraph)
- 🐛 **Issues**: [GitHub Issues](https://github.com/divinescreener/thegraph-token-api/issues)
- 🔑 **API Keys**: [thegraph.market](https://thegraph.market)

---

<p align="center">
  Made with ❤️ by <a href="https://github.com/divinescreener">DIVINE</a>
</p>