Querying the Subgraph

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

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)

    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)

    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}")

asyncio.run(get_daily_volume())

Hourly Volume Statistics

async def get_hourly_volume():
    url = build_subgraph_url()
    client = SubgraphClient(url=url)

    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

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)

    query = """
    query LastDaysVolume($days: Int!) {
        dailyTotals(orderBy: timestamp, orderDirection: desc, first: $days) {
            timestamp
            volumeUsd
            orders
            traders
        }
    }
    """

    response = await client.execute(
        query=query,
        variables={"days": 30}
    )

    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']}")

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']}")

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']}")

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']}")
    else:
        print("User not found")

asyncio.run(user_activity())

Working with Context Manager

async def with_context_manager():
    url = build_subgraph_url()

    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

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():
    url = build_subgraph_url(chain=Chain.MAINNET)
    client = SubgraphClient(url=url)

    print("=" * 60)
    print("CoW Protocol Trading Dashboard")
    print("=" * 60)

    # Protocol totals
    totals = await client.totals()
    total_data = totals.totals[0]

    print("\nPROTOCOL 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}")

    # Last 7 days volume
    daily_volume = await client.last_days_volume(days=7)

    print("\nLAST 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}")

    # Last 24 hours activity
    hourly_volume = await client.last_hours_volume(hours=24)

    total_day_volume = sum(
        float(hour.volumeUsd) for hour in hourly_volume.hourlyTotals
    )
    print(f"\nLAST 24 HOURS")
    print(f"  Total Volume: ${total_day_volume:,.2f}")
    print(f"  Hourly Average: ${total_day_volume / 24:,.2f}")

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

Query Optimization Tips

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

Next Steps

Learn about Managing Orders to create and track orders
Explore Swapping Tokens for executing trades
Check App Data for order metadata

Getting Started

Concepts

Guides

Advanced

API Reference

Overview

Setting Up the Subgraph Client

Supported Networks

Built-in Queries

Total Protocol Statistics

Daily Volume Statistics

Hourly Volume Statistics

Custom GraphQL Queries

Basic Custom Query

Query Orders with Filters

Query Trades

Advanced Query Examples

Token Trading Statistics

User Trading Activity

Working with Context Manager

Error Handling

Complete Example: Trading Dashboard

Available Schema Types

Query Optimization Tips

Next Steps

​Overview

​Setting Up the Subgraph Client

​Supported Networks

​Built-in Queries

​Total Protocol Statistics

​Daily Volume Statistics

​Hourly Volume Statistics

​Custom GraphQL Queries

​Basic Custom Query

​Query Orders with Filters

​Query Trades

​Advanced Query Examples

​Token Trading Statistics

​User Trading Activity

​Working with Context Manager

​Error Handling

​Complete Example: Trading Dashboard

​Available Schema Types

​Query Optimization Tips

​Next Steps

Overview

Setting Up the Subgraph Client

Supported Networks

Built-in Queries

Total Protocol Statistics

Daily Volume Statistics

Hourly Volume Statistics

Custom GraphQL Queries

Basic Custom Query

Query Orders with Filters

Query Trades

Advanced Query Examples

Token Trading Statistics

User Trading Activity

Working with Context Manager

Error Handling

Complete Example: Trading Dashboard

Available Schema Types

Query Optimization Tips

Next Steps