Aggregation Queries

Timeseries and aggregation features for analytics and metrics in Graph Node

Graph Node provides built-in aggregation capabilities for timeseries data, enabling efficient storage and querying of metrics, statistics, and time-bucketed analytics. This feature is available from spec version 1.1.0 onwards.

Overview

Aggregations are declared in your subgraph schema through two complementary types:

1. Timeseries type: Stores raw data points with timestamps
2. Aggregation type: Defines how raw data is aggregated over time intervals

Graph Node automatically computes and maintains aggregations as new data arrives, rolling up statistics for hourly and daily intervals.

Defining Timeseries

A timeseries entity stores individual data points with automatic timestamps:

type Data @entity(timeseries: true) {
  id: Int8!
  timestamp: Timestamp!
  price: BigDecimal!
  volume: BigDecimal!
}

@entity(timeseries: true) · required

Marks the type as a timeseries. These entities are immutable and append-only.

id · Int8! · required

Automatically set by Graph Node in insertion order. Must be of type Int8.

timestamp · Timestamp! · required

Automatically set by Graph Node to the current block timestamp. User-provided values are silently overridden.

Timeseries Characteristics

• Immutable: Once created, timeseries entities cannot be updated or deleted
• Automatic IDs: The id field is set automatically in ascending order
• Automatic timestamps: The timestamp field is always set to block timestamp
• Efficient storage: Optimized for append-only operations

Defining Aggregations

Aggregations compute statistics over timeseries data at specified time intervals:

type Stats @aggregation(intervals: ["hour", "day"], source: "Data") {
  id: Int8!
  timestamp: Timestamp!
  sum: BigDecimal! @aggregate(fn: "sum", arg: "price")
  avg: BigDecimal! @aggregate(fn: "avg", arg: "price")
  max: BigDecimal! @aggregate(fn: "max", arg: "price")
  count: Int8! @aggregate(fn: "count")
}

@aggregation · required

Marks the type as an aggregation with two required arguments:

• intervals: Array of time intervals ("hour" or "day")
• source: Name of the timeseries type to aggregate

id · Int8! · required

Automatically set by Graph Node. Set to the id of one of the source data points (unspecified which).

timestamp · Timestamp! · required

Automatically set to the beginning of the aggregation time interval.

Aggregation Functions

The @aggregate directive supports these functions:

Sum

sum: BigDecimal! @aggregate(fn: "sum", arg: "price")

count: Int8! @aggregate(fn: "count")

minPrice: BigDecimal! @aggregate(fn: "min", arg: "price")
maxPrice: BigDecimal! @aggregate(fn: "max", arg: "price")

firstPrice: BigDecimal! @aggregate(fn: "first", arg: "price")
lastPrice: BigDecimal! @aggregate(fn: "last", arg: "price")

sum · Aggregation Function

Sum of all values in the interval.

count · Aggregation Function

Number of data points in the interval.

min · Aggregation Function

Minimum value in the interval.

max · Aggregation Function

Maximum value in the interval.

first · Aggregation Function

First value in the interval (by timeseries id ordering).

last · Aggregation Function

Last value in the interval (by timeseries id ordering).

Dimensions

Dimensions are non-aggregated fields used to group data. They enable multi-dimensional aggregations:

type Token @entity {
  id: ID!
  symbol: String!
  decimals: Int!
}

type TokenData @entity(timeseries: true) {
  id: Int8!
  timestamp: Timestamp!
  token: Token!  # Dimension
  price: BigDecimal!
  volume: BigDecimal!
}

type TokenStats @aggregation(intervals: ["hour", "day"], source: "TokenData") {
  id: Int8!
  timestamp: Timestamp!
  token: Token!  # Dimension - groups by token
  totalVolume: BigDecimal! @aggregate(fn: "sum", arg: "volume")
  avgPrice: BigDecimal! @aggregate(fn: "avg", arg: "price")
  count: Int8! @aggregate(fn: "count")
}

How dimensions work:

• Fields without @aggregate are dimensions
• Each unique combination of dimension values creates a separate aggregation series
• In the example above, each token gets its own hourly and daily statistics

Cumulative Aggregations

By default, aggregations compute values for each time interval independently. Cumulative aggregations compute running totals across all intervals:

type TokenStats @aggregation(intervals: ["hour", "day"], source: "TokenData") {
  id: Int8!
  timestamp: Timestamp!
  token: Token!
  
  # Non-cumulative: volume for this hour/day only
  intervalVolume: BigDecimal! @aggregate(fn: "sum", arg: "volume")
  
  # Cumulative: total volume from beginning of time
  cumulativeVolume: BigDecimal! @aggregate(fn: "sum", arg: "volume", cumulative: true)
  
  # Cumulative count
  totalTransactions: Int8! @aggregate(fn: "count", cumulative: true)
}

cumulative · Boolean · default: "false"

When true, aggregates over the entire timeseries up to the end of the current interval.

Aggregation Expressions

The arg parameter can be a field name or a SQL expression:

type SwapData @entity(timeseries: true) {
  id: Int8!
  timestamp: Timestamp!
  amount0: BigDecimal!
  amount1: BigDecimal!
  priceUSD: BigDecimal!
}

type SwapStats @aggregation(intervals: ["hour", "day"], source: "SwapData") {
  id: Int8!
  timestamp: Timestamp!
  
  # Calculate total USD value
  totalValueUSD: BigDecimal! @aggregate(
    fn: "sum", 
    arg: "amount0 * priceUSD"
  )
  
  # Maximum of two amounts
  maxAmount: BigDecimal! @aggregate(
    fn: "max",
    arg: "greatest(amount0, amount1)"
  )
  
  # Conditional aggregation
  largeSwapsCount: Int8! @aggregate(
    fn: "count",
    arg: "case when amount0 > 1000 then 1 else 0 end"
  )
}

Supported Expression Syntax

Expressions use SQL syntax with these supported features:

Operators

• Arithmetic: +, -, *, /, %, ^
• Comparison: =, !=, <, <=, >, >=
• Logical: and, or, not
• Other: is [not] {null|true|false}, is [not] distinct from

Math Functions

• abs(x): Absolute value
• ceil(x), ceiling(x): Round up
• floor(x): Round down
• div(x, y): Integer division
• mod(x, y): Modulo
• power(x, y): Exponentiation
• gcd(x, y): Greatest common divisor
• lcm(x, y): Least common multiple
• sign(x): Sign (-1, 0, or 1)

Conditional Functions

• coalesce(x, y, ...): First non-null value
• nullif(x, y): NULL if x equals y
• greatest(x, y, ...): Maximum value
• least(x, y, ...): Minimum value
• case when ... then ... else ... end: Conditional expression

Querying Aggregations

Graph Node creates top-level query fields for each aggregation:

Basic Aggregation Query

query {
  tokenStats(
    interval: "hour"
    first: 24
  ) {
    id
    timestamp
    totalVolume
    avgPrice
    count
  }
}

interval · Enum · required

Time interval to query: "hour" or "day" (must match intervals defined in schema).

current · Enum · default: "exclude"

Whether to include the current, partially filled bucket:

• exclude: Only return completed, rolled-up buckets (default)
• include: Also return in-progress bucket computed on-the-fly

Filtering Aggregations

Filter by dimensions and timestamp ranges:

query {
  tokenStats(
    interval: "hour"
    where: {
      token: "0x1234567890123456789012345678901234567890"
      timestamp_gte: "1704067200000000"
      timestamp_lt: "1704153600000000"
    }
    first: 24
  ) {
    timestamp
    token {
      symbol
    }
    totalVolume
    avgPrice
  }
}

Available Timestamp Filters

timestamp_eq · String

Exact timestamp match (microseconds since epoch).

timestamp_gte · String

Greater than or equal to timestamp.

timestamp_gt · String

Greater than timestamp.

timestamp_lte · String

Less than or equal to timestamp.

timestamp_lt · String

Less than timestamp.

timestamp_in · [String]

Timestamp in list.

Timestamps are strings containing microseconds since Unix epoch.

Example: "1704164640000000" = 2024-01-02T03:04:00Z

Convert from seconds: seconds * 1000000

Current Bucket

By default, aggregation queries return only completed, rolled-up buckets. Setting current: include adds the in-progress bucket:

Completed Only (Default)

query {
  tokenStats(
    interval: "hour"
    current: exclude
  ) {
    timestamp
    totalVolume
  }
}

query {
  tokenStats(
    interval: "hour"
    current: include
  ) {
    timestamp
    totalVolume
  }
}

How it works:

• exclude: Only returns buckets whose time interval has ended and been rolled up
• include: Adds one additional bucket computed on-the-fly from unrolled source data
• The current bucket covers from the last completed bucket to the most recent data point

Nested Aggregation Queries

The current argument works on nested aggregation fields:

query {
  tokens {
    id
    symbol
    stats(interval: "hour", current: include) {
      timestamp
      totalVolume
      avgPrice
    }
  }
}

Current bucket support for nested fields is only available when the field references a single aggregation type. It’s not supported for aggregations accessed through interfaces with multiple implementations.

Complete Example

Here’s a comprehensive example showing timeseries and aggregations for DEX trading data:

schema.graphql

type Token @entity {
  id: ID!
  symbol: String!
  decimals: Int!
  stats: [TokenStats!]! @derivedFrom(field: "token")
}

type Pair @entity {
  id: ID!
  token0: Token!
  token1: Token!
  stats: [PairStats!]! @derivedFrom(field: "pair")
}

# Raw swap events
type SwapData @entity(timeseries: true) {
  id: Int8!
  timestamp: Timestamp!
  pair: Pair!
  amount0: BigDecimal!
  amount1: BigDecimal!
  amountUSD: BigDecimal!
}

# Aggregated pair statistics
type PairStats @aggregation(intervals: ["hour", "day"], source: "SwapData") {
  id: Int8!
  timestamp: Timestamp!
  pair: Pair!
  
  # Interval statistics
  volumeUSD: BigDecimal! @aggregate(fn: "sum", arg: "amountUSD")
  swapCount: Int8! @aggregate(fn: "count")
  
  # Price tracking
  openPrice: BigDecimal! @aggregate(fn: "first", arg: "amount1 / amount0")
  closePrice: BigDecimal! @aggregate(fn: "last", arg: "amount1 / amount0")
  highPrice: BigDecimal! @aggregate(fn: "max", arg: "amount1 / amount0")
  lowPrice: BigDecimal! @aggregate(fn: "min", arg: "amount1 / amount0")
  
  # Cumulative metrics
  cumulativeVolumeUSD: BigDecimal! @aggregate(
    fn: "sum", 
    arg: "amountUSD", 
    cumulative: true
  )
  totalSwaps: Int8! @aggregate(fn: "count", cumulative: true)
}

Query Examples

Recent Hourly Stats

query {
  pairStats(
    interval: "hour"
    first: 24
    orderBy: timestamp
    orderDirection: desc
  ) {
    timestamp
    pair {
      token0 { symbol }
      token1 { symbol }
    }
    volumeUSD
    swapCount
    closePrice
  }
}

query {
  pairStats(
    interval: "day"
    where: {
      pair: "0xpairaddress"
      timestamp_gte: "1704067200000000"
    }
    first: 30
  ) {
    timestamp
    volumeUSD
    openPrice
    closePrice
    highPrice
    lowPrice
    cumulativeVolumeUSD
  }
}

query {
  pairStats(
    interval: "day"
    where: {
      timestamp_gte: "1704067200000000"
      timestamp_lt: "1704153600000000"
    }
    orderBy: volumeUSD
    orderDirection: desc
    first: 10
  ) {
    pair {
      token0 { symbol }
      token1 { symbol }
    }
    volumeUSD
    swapCount
  }
}

Sorting Behavior

Aggregations have a default sort order:

• Sorted by timestamp and id in descending order by default
• Returns most recent buckets first
• Can be overridden with orderBy and orderDirection

# Default: Most recent first
query {
  tokenStats(interval: "hour") {
    timestamp
    totalVolume
  }
}

# Override: Oldest first
query {
  tokenStats(
    interval: "hour"
    orderBy: timestamp
    orderDirection: asc
  ) {
    timestamp
    totalVolume
  }
}

Best Practices

1. Choose appropriate intervals: Use hourly for recent data, daily for historical trends
2. Use dimensions wisely: Each dimension combination creates separate aggregation series
3. Leverage cumulative aggregations: Useful for running totals and lifetime metrics
4. Filter by timestamp: Always specify timestamp ranges to limit data scanned
5. Monitor aggregation lag: Current buckets may not reflect all recent data until rollup completes
6. Use expressions for computed metrics: Calculate derived values at aggregation time

Next Steps

• Learn about time-travel queries for historical state
• Review query syntax for advanced filtering
• Explore GraphQL API overview for basic concepts