Skip to main content

​
Overview

The CoW Protocol Subgraph indexes on-chain trading data, making it easy to query historical trades, volume statistics, and protocol metrics. The SDK provides a type-safe SubgraphClient with built-in queries and support for custom GraphQL queries.

​
Setting Up the Subgraph Client

from cowdao_cowpy.subgraph.client import SubgraphClient
from cowdao_cowpy.subgraph.deployments import (
    build_subgraph_url,
    SubgraphEnvironment,
)
from cowdao_cowpy.common.chains import Chain

# Use default configuration (Mainnet, Production)
url = build_subgraph_url()
client = SubgraphClient(url=url)

# Or specify chain and environment
url = build_subgraph_url(
    chain=Chain.GNOSIS,
    env=SubgraphEnvironment.PRODUCTION
)
client = SubgraphClient(url=url)

​
Supported Networks

The Subgraph is available on:
  • Ethereum Mainnet - Chain.MAINNET
  • Gnosis Chain - Chain.GNOSIS
  • Arbitrum One - Chain.ARBITRUM_ONE
  • Base - Chain.BASE
  • Sepolia Testnet - Chain.SEPOLIA

​
Built-in Queries

The SDK includes pre-built queries for common use cases:

​
Total Protocol Statistics

import asyncio
from cowdao_cowpy.subgraph.client import SubgraphClient
from cowdao_cowpy.subgraph.deployments import build_subgraph_url

async def get_totals():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    # Fetch protocol totals
    totals = await client.totals()
    
    print(f"Total tokens: {totals.totals[0].tokens}")
    print(f"Total orders: {totals.totals[0].orders}")
    print(f"Total traders: {totals.totals[0].traders}")
    print(f"Total settlements: {totals.totals[0].settlements}")
    print(f"Volume USD: {totals.totals[0].volumeUsd}")
    print(f"Volume ETH: {totals.totals[0].volumeEth}")
    print(f"Fees USD: {totals.totals[0].feesUsd}")
    print(f"Fees ETH: {totals.totals[0].feesEth}")

asyncio.run(get_totals())

​
Daily Volume Statistics

async def get_daily_volume():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    # Get last 7 days of volume
    daily_volume = await client.last_days_volume(days=7)
    
    for day in daily_volume.dailyTotals:
        print(f"Date: {day.timestamp}")
        print(f"Volume: ${day.volumeUsd}")
        print()

asyncio.run(get_daily_volume())

​
Hourly Volume Statistics

async def get_hourly_volume():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    # Get last 24 hours of volume
    hourly_volume = await client.last_hours_volume(hours=24)
    
    for hour in hourly_volume.hourlyTotals:
        print(f"Hour: {hour.timestamp}")
        print(f"Volume: ${hour.volumeUsd}")

asyncio.run(get_hourly_volume())

​
Custom GraphQL Queries

You can execute custom GraphQL queries using the execute() method:

​
Basic Custom Query

from pprint import pprint

async def custom_query():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    # Custom query
    query = """
    query LastDaysVolume($days: Int!) {
        dailyTotals(orderBy: timestamp, orderDirection: desc, first: $days) {
            timestamp
            volumeUsd
            orders
            traders
        }
    }
    """
    
    # Execute with variables
    response = await client.execute(
        query=query,
        variables={"days": 30}
    )
    
    # Parse response
    data = client.get_data(response)
    pprint(data)

asyncio.run(custom_query())

​
Query Orders with Filters

async def query_orders():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    query = """
    query GetOrders($owner: String!, $first: Int!) {
        orders(
            where: { owner: $owner }
            first: $first
            orderBy: creationTimestamp
            orderDirection: desc
        ) {
            id
            owner
            sellToken
            buyToken
            sellAmount
            buyAmount
            creationTimestamp
            status
        }
    }
    """
    
    response = await client.execute(
        query=query,
        variables={
            "owner": "0x...",
            "first": 100
        }
    )
    
    data = client.get_data(response)
    
    for order in data["orders"]:
        print(f"Order {order['id']}:")
        print(f"  Status: {order['status']}")
        print(f"  Created: {order['creationTimestamp']}")
        print()

asyncio.run(query_orders())

​
Query Trades

async def query_trades():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    query = """
    query GetTrades($owner: String!, $first: Int!) {
        trades(
            where: { owner: $owner }
            first: $first
            orderBy: timestamp
            orderDirection: desc
        ) {
            id
            timestamp
            sellToken
            buyToken
            sellAmount
            buyAmount
            order {
                id
            }
        }
    }
    """
    
    response = await client.execute(
        query=query,
        variables={
            "owner": "0x...",
            "first": 50
        }
    )
    
    data = client.get_data(response)
    
    for trade in data["trades"]:
        print(f"Trade at {trade['timestamp']}:")
        print(f"  Sold: {trade['sellAmount']} {trade['sellToken']}")
        print(f"  Bought: {trade['buyAmount']} {trade['buyToken']}")
        print()

asyncio.run(query_trades())

​
Advanced Query Examples

​
Token Trading Statistics

async def token_stats():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    query = """
    query TokenStats($tokenAddress: String!) {
        token(id: $tokenAddress) {
            id
            address
            symbol
            name
            decimals
            totalVolume
            numberOfTrades
            totalVolumeUsd
        }
        tokenDailyTotals(
            where: { token: $tokenAddress }
            first: 30
            orderBy: timestamp
            orderDirection: desc
        ) {
            timestamp
            volumeUsd
            totalVolume
        }
    }
    """
    
    response = await client.execute(
        query=query,
        variables={
            "tokenAddress": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"  # USDC
        }
    )
    
    data = client.get_data(response)
    token = data["token"]
    
    print(f"Token: {token['symbol']}")
    print(f"Total volume: ${token['totalVolumeUsd']}")
    print(f"Number of trades: {token['numberOfTrades']}")
    print("\nLast 30 days:")
    
    for day in data["tokenDailyTotals"]:
        print(f"  {day['timestamp']}: ${day['volumeUsd']}")

asyncio.run(token_stats())

​
User Trading Activity

async def user_activity():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    query = """
    query UserActivity($userAddress: String!) {
        user(id: $userAddress) {
            id
            address
            numberOfTrades
            volumeUsd
            solvedAmountUsd
            orders(first: 10, orderBy: creationTimestamp, orderDirection: desc) {
                id
                status
                sellToken
                buyToken
                creationTimestamp
            }
            trades(first: 10, orderBy: timestamp, orderDirection: desc) {
                id
                timestamp
                sellAmount
                buyAmount
            }
        }
    }
    """
    
    response = await client.execute(
        query=query,
        variables={"userAddress": "0x..."}
    )
    
    data = client.get_data(response)
    user = data["user"]
    
    if user:
        print(f"User: {user['address']}")
        print(f"Total trades: {user['numberOfTrades']}")
        print(f"Total volume: ${user['volumeUsd']}")
        print(f"\nRecent orders: {len(user['orders'])}")
        print(f"Recent trades: {len(user['trades'])}")
    else:
        print("User not found")

asyncio.run(user_activity())

​
Trading Pairs Analysis

async def pair_analysis():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    query = """
    query TopPairs($first: Int!) {
        pairs(
            first: $first
            orderBy: volumeUsd
            orderDirection: desc
        ) {
            id
            token0
            token1
            volumeUsd
            numberOfTrades
        }
    }
    """
    
    response = await client.execute(
        query=query,
        variables={"first": 10}
    )
    
    data = client.get_data(response)
    
    print("Top Trading Pairs:")
    for i, pair in enumerate(data["pairs"], 1):
        print(f"{i}. {pair['token0']} / {pair['token1']}")
        print(f"   Volume: ${pair['volumeUsd']}")
        print(f"   Trades: {pair['numberOfTrades']}")
        print()

asyncio.run(pair_analysis())

​
Working with Context Manager

async def with_context_manager():
    url = build_subgraph_url()
    
    # Use async context manager for automatic cleanup
    async with SubgraphClient(url=url) as client:
        totals = await client.totals()
        print(f"Total traders: {totals.totals[0].traders}")
        
        daily_volume = await client.last_days_volume(days=7)
        print(f"Days of data: {len(daily_volume.dailyTotals)}")

asyncio.run(with_context_manager())

​
Error Handling

from cowdao_cowpy.subgraph.client.exceptions import (
    GraphQLClientError,
    GraphQLClientHttpError,
    GraphQLClientGraphQLMultiError,
)

async def safe_query():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)
    
    try:
        response = await client.execute(
            query="""query { invalidField }"""
        )
        data = client.get_data(response)
    except GraphQLClientGraphQLMultiError as e:
        print(f"GraphQL errors: {e.errors}")
    except GraphQLClientHttpError as e:
        print(f"HTTP error: {e.status_code}")
    except GraphQLClientError as e:
        print(f"Client error: {e}")

asyncio.run(safe_query())

​
Complete Example: Trading Dashboard

dashboard.py
import asyncio
from datetime import datetime
from cowdao_cowpy.subgraph.client import SubgraphClient
from cowdao_cowpy.subgraph.deployments import build_subgraph_url
from cowdao_cowpy.common.chains import Chain

async def create_dashboard():
    # Initialize client
    url = build_subgraph_url(chain=Chain.MAINNET)
    client = SubgraphClient(url=url)
    
    print("=" * 60)
    print("CoW Protocol Trading Dashboard")
    print("=" * 60)
    print()
    
    # Protocol totals
    totals = await client.totals()
    total_data = totals.totals[0]
    
    print("📊 PROTOCOL STATISTICS")
    print(f"  Total Traders: {total_data.traders:,}")
    print(f"  Total Orders: {total_data.orders:,}")
    print(f"  Total Settlements: {total_data.settlements:,}")
    print(f"  Total Volume (USD): ${float(total_data.volumeUsd):,.2f}")
    print(f"  Total Fees (USD): ${float(total_data.feesUsd):,.2f}")
    print()
    
    # Last 7 days volume
    daily_volume = await client.last_days_volume(days=7)
    
    print("📈 LAST 7 DAYS VOLUME")
    total_week_volume = 0
    for day in reversed(daily_volume.dailyTotals):
        date = datetime.fromtimestamp(int(day.timestamp)).strftime('%Y-%m-%d')
        volume = float(day.volumeUsd)
        total_week_volume += volume
        print(f"  {date}: ${volume:,.2f}")
    
    print(f"\n  Weekly Total: ${total_week_volume:,.2f}")
    print(f"  Daily Average: ${total_week_volume / 7:,.2f}")
    print()
    
    # Last 24 hours activity
    hourly_volume = await client.last_hours_volume(hours=24)
    
    print("⏰ LAST 24 HOURS")
    total_day_volume = sum(
        float(hour.volumeUsd) for hour in hourly_volume.hourlyTotals
    )
    print(f"  Total Volume: ${total_day_volume:,.2f}")
    print(f"  Hourly Average: ${total_day_volume / 24:,.2f}")
    print()
    
    print("=" * 60)

if __name__ == "__main__":
    asyncio.run(create_dashboard())

​
Available Schema Types

The Subgraph schema includes these main types:
  • User - Trader information and statistics
  • Order - Order details and status
  • Trade - Executed trade information
  • Token - Token metadata and statistics
  • Pair - Trading pair information
  • Settlement - Settlement transaction details
  • DailyTotal - Daily aggregated statistics
  • HourlyTotal - Hourly aggregated statistics
  • Total - Overall protocol statistics
Refer to the CoW Protocol Subgraph documentation for the complete schema and available fields.

​
Query Optimization Tips

  1. Use pagination: Limit results with first and skip parameters
  2. Filter early: Apply where clauses to reduce data transfer
  3. Select specific fields: Only query fields you need
  4. Use orderBy: Sort results at the subgraph level
  5. Cache results: Store frequently accessed data locally

​
Next Steps

Build docs developers (and LLMs) love

Get started for freeTalk to us