How to Pull Crypto Prices From DEXs

build_pool_chart.py

def build_pool_chart(network: str, pool_address: str, pool_name: str, timeframe: str = "day"):

    """Build a price & volume chart for a liquidity pool from real OHLCV (no synthetic fallback)."""
    payload = get_pool_ohlcv(network, pool_address, timeframe)
    ohlcv = _extract_ohlcv(payload)
    
    if not ohlcv:
        # Helpful debug for users
        raise RuntimeError("No OHLCV data returned for the given pool/timeframe. Try a coarser timeframe (e.g., 'day') or set include_empty_intervals=true.")

    df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
    df["date"] = pd.to_datetime(df["timestamp"], unit="s")
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), sharex=True)
    
    # Price chart
    ax1.plot(df["date"], df["close"], linewidth=2, color="blue")
    ax1.set_ylabel("Price (USD)")
    ax1.set_title({pool_name} - Price & Volume")
    ax1.grid(True, alpha=0.3)

    # Volume chart
    ax2.bar(df["date"], df["volume"], alpha=0.7, color="orange")
    ax2.set_ylabel("Volume (USD)")
    ax2.set_xlabel("Date")
    ax2.grid(True, alpha=0.3)
    plt.xticks(rotation=45)
    plt.tight_layout()
    
    filename = {pool_name.replace('/', '_').replace(' ', '_').lower()}_{datetime.utcnow().date()}.png"
    plt.savefig(filename, dpi=300, bbox_inches="tight")
    plt.show()
    print(Pool chart saved as: {filename}; points: {len(df)}")
    return df, filename

if __name__ == "__main__":
    # Example: USDC/ETH 0.05% (Uniswap v3)
    network = "eth"
    pool_address = "0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640"
    print("Creating pool chart from real OHLCV data...")
    chart_data, filename = build_pool_chart(network, pool_address, "USDC/ETH Pool")
    print(Chart created: {filename}; rows: {chart_data.shape[0]}")

build_token_chart.py

def build_token_chart(network: str, token_address: str, token_name: str, timeframe: str = "day"):
    """Build a price chart from real OHLCV data (no synthetic fallback)."""
    payload = get_token_ohlcv(network, token_address, timeframe)
    ohlcv = _extract_ohlcv(payload)
    if not ohlcv:
        raise RuntimeError("No OHLCV data returned for the given token/timeframe. Try a coarser timeframe (e.g., 'day') or set include_empty_intervals=true.")

    df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
    df["date"] = pd.to_datetime(df["timestamp"], unit="s")

    plt.figure(figsize=(12, 6))
    plt.plot(df["date"], df["close"], linewidth=2, color="blue")
    plt.title({token_name} Price Chart")
    plt.xlabel("Date")
    plt.ylabel("Price (USD)")
    plt.grid(True, alpha=0.3)
    plt.xticks(rotation=45)
    plt.tight_layout()

    filename = {token_name.lower()}_price_chart_{datetime.utcnow().date()}.png"
    plt.savefig(filename, dpi=300, bbox_inches="tight")
    plt.show()

    print(Chart saved as: {filename}; points: {len(df)}")
    return df, filename

if __name__ == "__main__":
    # Example: USDC on Ethereum
    network = "eth"
    token_address = "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"  # USDC
    print("Creating token price chart from real OHLCV data...")
    chart_data, filename = build_token_chart(network, token_address, "USDC")
    print(Chart created: {filename}; rows: {chart_data.shape[0]}")

get_dex_pools.py

import requests
import os
from dotenv import load_dotenv

load_dotenv()

def get_dex_pools(dex_id, network='eth', page=1):
    """Get all pools for a specific DEX"""
    url = https://api.coingecko.com/api/v3/onchain/networks/{network}/dexes/{dex_id}/pools"
   
    headers = {
        'X-Cg-Demo-Api-Key': os.getenv('COINGECKO_DEMO_API_KEY')
    }
    params = {'page': page}
   
    response = requests.get(url, headers=headers, params=params)
    return response.json()

if __name__ == "__main__":
    # Example: Get Uniswap V3 pools
    result = get_dex_pools('uniswap_v3', 'eth')
    print("Uniswap V3 pools sample:")
    for pool in result['data'][:3]:  # Show first 3 results
        attrs = pool.get('attributes', {})
        print(Pool: {attrs.get('name', 'Unknown')}")
        print(Address: {attrs.get('address', 'Unknown')}")
        print(Fee: {attrs.get('pool_fee_percentage', 'Unknown')}%")
        print(TVL: ${float(attrs.get('reserve_in_usd', 0)):,.2f}")
        print("---")

get_pool_data.py

def get_pool_data(pool_address, network='eth', include_tokens=True):
    """Get detailed data for a specific pool"""
    url = https://api.coingecko.com/api/v3/onchain/networks/{network}/pools/{pool_address}"
    
    headers = {
        'X-Cg-Demo-Api-Key': os.getenv('COINGECKO_API_KEY')
    }
    
    params = {}
    if include_tokens:
        params['include'] = 'base_token,quote_token'
    
    response = requests.get(url, headers=headers, params=params)
    return response.json()

# Example: Get data for USDC/WETH pool on Uniswap V3
pool_address = "0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8"  # USDC/WETH 0.3%
result = get_pool_data(pool_address)

print("Pool details:")
data = result['data']['attributes']
dex_data = result['data']['relationships']['dex']['data']

print(Name: {data['name']}")
print(Address: {data['address']}")
print(DEX: {dex_data['id']}")
print(TVL: ${float(data['reserve_in_usd']):,.2f}")
print(Volume 24h: ${float(data['volume_usd']['h24']):,.2f}")
print(Base token price: ${float(data['base_token_price_usd']):.6f}")
print(Quote token price: ${float(data['quote_token_price_usd']):.2f}")

get_pool_ohlcv_data.py

import requests
import os
from dotenv import load_dotenv

load_dotenv()

def get_pool_ohlcv_data(pool_address, network='eth'):
    """Get historical OHLCV data for a specific pool address"""
    url = https://api.coingecko.com/api/v3/onchain/networks/{network}/pools/{pool_address}/ohlcv/day"
   
    headers = {
        'x-cg-demo-api-key': os.getenv('COINGECKO_DEMO_API_KEY')
    }
   
    params = {
        'include_empty_intervals': 'false'
    }
   
    try:
        response = requests.get(url, headers=headers, params=params, timeout=30)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(HTTP Error: {e}")
        return None
    except Exception as e:
        print(Error: {e}")
        return None

if __name__ == "__main__":
    # Example: Get historical OHLCV data for USDC/WETH pool on Uniswap V3
    pool_address = "0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8"  # USDC/WETH 0.3%
    result = get_pool_ohlcv_data(pool_address)
   
    if result and 'data' in result:
        print("Historical Pool OHLCV Data (Last 7 days):")
        data = result['data']['attributes']
        ohlcv_list = data['ohlcv_list']
       
        print(Pool: {pool_address}")
        print(Total data points: {len(ohlcv_list)}")
        print("\nDate\t\t\tOpen\t\tHigh\t\tLow\t\tClose\t\tVolume")
        print("-" * 80)
       
        for entry in ohlcv_list[-7:]:  # Show last 7 entries
            timestamp = entry[0]
            open_price = entry[1]
            high_price = entry[2]
            low_price = entry[3]
            close_price = entry[4]
            volume = entry[5]
           
            # Convert timestamp to readable date
            from datetime import datetime
            date = datetime.fromtimestamp(timestamp/1000).strftime('%Y-%m-%d %H:%M')          
            print({date}\t{open_price:.6f}\t{high_price:.6f}\t{low_price:.6f}\t{close_price:.6f}\t{volume:.2f}")
    else:
        print("Failed to retrieve pool OHLCV data - this endpoint requires Pro API access")

get_token_ohlcv.py

import requests
import os
from dotenv import load_dotenv

load_dotenv()

COINGECKO_API_KEY = os.getenv("COINGECKO_API_KEY")

def get_token_ohlcv(network: str, token_address: str, timeframe: str = "day") -> dict:
    """Fetch token OHLCV from CoinGecko Onchain API."""
    if not COINGECKO_API_KEY:
        raise RuntimeError("COINGECKO_API_KEY is not set in environment.")
    url = f"https://pro-api.coingecko.com/api/v3/onchain/networks/{network}/tokens/{token_address}/ohlcv/{timeframe}"
    headers = {"X-Cg-Pro-Api-Key": COINGECKO_PRO_API_KEY}
    params = {"aggregate": 1, "include_empty_intervals": "true"}
    resp = requests.get(url, headers=headers, params=params, timeout=30)
    resp.raise_for_status()
    return resp.json()

def _extract_ohlcv(payload: dict):
    """Return OHLCV list from possible response shapes."""
    if not isinstance(payload, dict):
        return []
    if isinstance(payload.get("ohlcv"), list):
        return payload["ohlcv"]
    if isinstance(payload.get("ohlcv_list"), list):
        return payload["ohlcv_list"]
    data = payload.get("data")
    if isinstance(data, dict):
        attrs = data.get("attributes", {})
        if isinstance(attrs.get("ohlcv"), list):
            return attrs["ohlcv"]
        if isinstance(attrs.get("ohlcv_list"), list):
            return attrs["ohlcv_list"]
    return []

if __name__ == "__main__":
    # Test with Polygon network and USDC token
    network = "polygon_pos"
    token_address = "0x2791bca1f2de4661ed88a30c99a7a9449aa84174"  # USDC on Polygon
   
    try:
        result = get_token_ohlcv(network, token_address)
        ohlcv_data = _extract_ohlcv(result)
       
        if ohlcv_data:
            print(f"USDC Token OHLCV Data ({len(ohlcv_data)} points):")
            for i, entry in enumerate(ohlcv_data[:5]):  # Show first 5
                print(f"  {i+1}: {entry}")
        else:
            print("No OHLCV data found")
            print(f"Response keys: {list(result.keys())}")
    except Exception as e:
        print(f"Error: {e}")

get_token_prices.py

import requests
import os
from dotenv import load_dotenv

load_dotenv()

def get_token_prices(token_addresses, vs_currencies='usd', include_24h_change=True):
    """Get current token prices from on-chain data"""
    if isinstance(token_addresses, list):
        addresses = ','.join(token_addresses)
    else:
        addresses = token_addresses
    url = https://api.coingecko.com/api/v3/onchain/simple/networks/eth/token_price/{addresses}"
   
    headers = {
        'X-Cg-Demo-Api-Key': os.getenv('COINGECKO_DEMO_API_KEY')
    }
   
    params = {
        'vs_currencies': vs_currencies,
        'include_24hr_change': str(include_24h_change).lower()
    }
   
    response = requests.get(url, headers=headers, params=params)
    return response.json()

if __name__ == "__main__":
    # Example: Get USDC and WETH prices
    token_addresses = [
    "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",  # USDC
    "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"   # WETH
    ]
    result = get_token_prices(token_addresses)

print("Token price data:")
token_prices = result['data']['attributes']['token_prices']
for address, price in token_prices.items():
        print(Token: {address}")
        print(Price: ${float(price):.6f}")
        print("---")

get_top_pools.py

import requests
import os
from dotenv import load_dotenv

load_dotenv()
def get_top_pools(network='eth', include_volume=True):
    """Get top DEX pools by TVL"""
    url = "https://api.coingecko.com/api/v3/onchain/networks/{network}/pools"
   
    headers = {
        'x-cg-demo-api-key': os.getenv('COINGECKO_DEMO_API_KEY')
    }
   
    params = {}
    if include_volume:
        params['include'] = 'base_token,quote_token'
   
    try:
        response = requests.get(url, headers=headers, params=params, timeout=30)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(HTTP Error: {e}")
        return None
    except Exception as e:
        print(Error: {e}")
        return None

if __name__ == "__main__":
    # Example: Get top pools on Ethereum by TVL
    result = get_top_pools('eth')
   
    if result and 'data' in result:
        print("Top DEX pools by TVL:")
        for pool in result['data'][:5]:  # Show first 5 results
            attrs = pool.get('attributes', {})
            dex_data = pool.get('relationships', {}).get('dex', {}).get('data', {})
           
            print(Pool: {attrs.get('name', 'Unknown')}")
            print(TVL: ${float(attrs.get('reserve_in_usd', 0)):,.2f}")
            print(DEX: {dex_data.get('id', 'Unknown')}")
            volume_24h = attrs.get('volume_usd', {}).get('h24', 0) if attrs.get('volume_usd') else 0
            print(Volume 24h: ${float(volume_24h):,.2f}")
            print("---")
    else:
        print("Failed to retrieve top pools data")

search_pools.py

import requests
import os
from dotenv import load_dotenv

load_dotenv()

def search_pools(query, network=None):
    """Search for DEX pools by query"""
    url = "https://api.coingecko.com/api/v3/onchain/search/pools"
   
    headers = {
        'X-Cg-Demo-Api-Key': os.getenv('COINGECKO_DEMO_API_KEY')
    }
   
    params = {'query': query}
    if network:
        params['network'] = network
   
    response = requests.get(url, headers=headers, params=params)
    return response.json()

if __name__ == "__main__":
    # Example: Search for USDC pools
    result = search_pools('USDC', 'eth')
    print("Sample output:")
   
    for i, pool in enumerate(result['data'][:3]):  # Show first 3 results
        print(Pool {i+1}:")
       
        # Extract from attributes
        attrs = pool.get('attributes', {})
        name = attrs.get('name', 'Unknown')
        address = attrs.get('address', 'Unknown')
       
        # Extract DEX from relationships
        dex_data = pool.get('relationships', {}).get('dex', {}).get('data', {})
        dex_id = dex_data.get('id', 'Unknown')
        # Extract additional useful info
        reserve_usd = attrs.get('reserve_in_usd', 0)
        volume_24h = attrs.get('volume_usd', {}).get('h24', 0)
        print(  Pool: {name}")
        print(  Address: {address}")
        print(  DEX: {dex_id}")
        print(  Network: eth")  # We searched for eth specifically
        print(  TVL: ${float(reserve_usd):,.2f}")
        print(  Volume 24h: ${float(volume_24h):,.2f}")