SubgraphClient

Client for querying CoW Protocol subgraph data

The SubgraphClient class provides methods to query historical data from the CoW Protocol subgraph, including trading volumes, totals, and other aggregated metrics.

Initialization

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

# Default mainnet
client = SubgraphClient(url=build_subgraph_url())

# Gnosis Chain
from cowdao_cowpy.common.chains import Chain
from cowdao_cowpy.subgraph.deployments import SubgraphEnvironment

url = build_subgraph_url(chain=Chain.GNOSIS, env=SubgraphEnvironment.PRODUCTION)
client = SubgraphClient(url=url)

Methods

last_days_volume

Query the trading volume for the last N days.

async def last_days_volume(self, days: int, **kwargs: Any) -> LastDaysVolume

Parameter	Type	Description
days	int	Number of days of volume data

Returns: LastDaysVolume with daily_totals list containing:

• timestamp (int): Unix timestamp for the day
• volume_usd (Optional[Any]): Trading volume in USD

volume_data = await client.last_days_volume(days=7)

for daily in volume_data.daily_totals:
    print(f"Date: {daily.timestamp}, Volume: ${daily.volume_usd}")

GraphQL Query:

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

---

last_hours_volume

Query the trading volume for the last N hours.

async def last_hours_volume(self, hours: int, **kwargs: Any) -> LastHoursVolume

Parameter	Type	Description
hours	int	Number of hours of volume data

Returns: LastHoursVolume with hourly_totals list containing:

• timestamp (int): Unix timestamp for the hour
• volume_usd (Optional[Any]): Trading volume in USD

volume_data = await client.last_hours_volume(hours=24)

for hourly in volume_data.hourly_totals:
    print(f"Hour: {hourly.timestamp}, Volume: ${hourly.volume_usd}")

GraphQL Query:

query LastHoursVolume($hours: Int!) {
  hourlyTotals(orderBy: timestamp, orderDirection: desc, first: $hours) {
    timestamp
    volumeUsd
  }
}

---

totals

Query aggregate protocol statistics.

async def totals(self, **kwargs: Any) -> Totals

Returns: Totals with a totals list containing:

Field	Type	Description
tokens	Any	Total unique tokens traded
orders	Any	Total orders created
traders	Any	Total unique traders
settlements	Any	Total settlements executed
volume_usd	Optional[Any]	Total volume in USD
volume_eth	Optional[Any]	Total volume in ETH
fees_usd	Optional[Any]	Total fees in USD
fees_eth	Optional[Any]	Total fees in ETH

totals_data = await client.totals()
stats = totals_data.totals[0]

print(f"Traders: {stats.traders}")
print(f"Orders: {stats.orders}")
print(f"Volume: ${stats.volume_usd}")

GraphQL Query:

query Totals {
  totals {
    tokens
    orders
    traders
    settlements
    volumeUsd
    volumeEth
    feesUsd
    feesEth
  }
}

---

execute

Execute a custom GraphQL query.

async def execute(self, query: str, variables: dict = None, **kwargs) -> Any

query = """
query GetOrders($owner: String!, $first: Int!) {
    orders(where: { owner: $owner }, first: $first) {
        id
        status
        sellAmount
        buyAmount
    }
}
"""

response = await client.execute(query=query, variables={"owner": "0x...", "first": 10})
data = client.get_data(response)

Complete Example

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

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

    totals = await client.totals()
    daily_volume = await client.last_days_volume(days=30)
    hourly_volume = await client.last_hours_volume(hours=24)

    if totals.totals:
        stats = totals.totals[0]
        print(f"Traders: {stats.traders}")
        print(f"Orders: {stats.orders}")
        print(f"Volume: ${stats.volume_usd}")

    volume_30d = sum(
        float(day.volume_usd)
        for day in daily_volume.daily_totals
        if day.volume_usd
    )
    print(f"30-Day Volume: ${volume_30d:,.2f}")

    volume_24h = sum(
        float(hour.volume_usd)
        for hour in hourly_volume.hourly_totals
        if hour.volume_usd
    )
    print(f"24-Hour Volume: ${volume_24h:,.2f}")

asyncio.run(analyze_protocol())