API Pagination Response

Cursor Pagination

The API will automatically generate the URL for the next page of results when applicable. If the full payload returns in the first payload, the "next" response will be null.

The URL for the next page is found under the property for a 200 - OK HTTP response like so:

  "status": 200,
  "title": "OK",
  "description": "Successful request",
  "payload": {
    "metadata": {
      "next": ""

As an API user, to access the next page of results, you should:

  1. Retrieve the URL from
  2. Copy your request headers from your initial API call e.g. x-api-key, x-amberdata-blockchain-id etc.
  3. Make a HTTP GET request with the next page URL and the copied headers

Querying Long Timeframes

API endpoints have a maximum supported range for the query parameters endDate and startDate.


Increasing the range for endDate and startDate

Amberdata reserves the right to increase the supported range for endDate and startDate for any endpoint. An increase in the supported range is fully backwards compatible.

There are ways to call endpoints and get data for timeframes longer than the maximum range of endDate and startDate. Here below is an example of how to get 1 year of data for the DEX - Trades endpoint which has a maximum range of 30 days.



The code below has been verified and tested for demonstration purposes only.

import os
from datetime import datetime
from dataclasses import dataclass, field

from endpoint_timerange_handler import EndpointTimeRangeHandler, Endpoint
from endpoint_caller_v2 import EndpointCaller

def http_ok_next_page_url_extractor(page):
    Function that extracts the next page url from the current page of data

    page: dict

    next_page_url: str
    if 'payload' in page and page['payload'] is not None:
        payload = page['payload']
        if 'metadata' in payload and payload['metadata'] is not None:
            metadata = payload['metadata']
            if 'next' in metadata and metadata['next'] is not None:
                next_page_url = metadata['next']
                return next_page_url
    return ""

class DEXTradesHistorical(Endpoint):
    poolAddress: str
    path_template: str = field(default='/market/defi/trades/{}/historical')
    max_interval_in_seconds: int = field(default=2592000) #30 days * 24 hrs * 60 min * 60 seconds
    def format_path(self) -> str:
        return self.path_template.format(self.poolAddress)

You can implement similar dataclasses for other endpoints as needed.

`max_interval_in_seconds` is the maximum supported range for endDate and startDate.

The maximum range for each endpoint can be found in the documentation.

def get_api_responses(start_date: datetime, end_date: datetime, endpoint: DEXTradesHistorical) -> None:
    Get all pages of data between `start_date` and `end_date`.
    endpoint_caller = EndpointCaller(os.getenv('PRODUCTION_API_KEY'))
    endpoint_timerange_handler = EndpointTimeRangeHandler(endpoint_caller)
    for page in endpoint_timerange_handler.get_data_for_timerange(
        response =
        print(f"Timestamp of first entry in the page: {response['payload']['data'][0][1]}")
        # print('uncomment here below as needed')
        # print(
        # print(page.duration_seconds)

def call_dex_trades_historical() -> None:
    Example configuration of an endpoint to be called.

    This example demonstrates a way to query a time range larger than the endpoint's max supported range (endDate - startDate).
    start_date = "2022-01-01T00:00:00"
    end_date = "2023-01-01T00:00:00"
    start_date_as_dt = datetime.fromisoformat(start_date).replace(microsecond=0)
    end_date_as_dt = datetime.fromisoformat(end_date).replace(microsecond=0)
    poolAddress = '0xcbcdf9626bc03e24f779434178a73a0b4bad62ed' # WBTC/ETH 0.3%
    exchange = 'uniswapv3'

    dex_trades_historical = DEXTradesHistorical(poolAddress=poolAddress)
    dex_trades_historical.add_query_parameter('exchange', exchange)
    get_api_responses(start_date_as_dt, end_date_as_dt, dex_trades_historical)

if __name__ == "__main__":

from datetime import datetime
from datetime import timedelta
from dataclasses import dataclass, field

from endpoint_caller_v2 import EndpointCaller

class Endpoint:
    This is a parent class that should be inherited and implemented for specific endpoints.
    query: dict = field(default_factory=dict)
    headers: dict = field(default_factory=dict)

    def add_query_parameter(self, parameter_name: str, parameter_value) -> None:
        if parameter_name is not None and len(parameter_name) > 0:
            self.query[parameter_name] = parameter_value
    def format_path(self) -> str:
        Child classes must implement this function.
        return ""
    def add_header(self, header_name: str, header_value: str) -> None:
        if header_name is not None and len(header_name) > 0:
            self.headers[header_name] = header_value

class EndpointTimeRangeHandler:
    def __init__(self, endpoint_caller: EndpointCaller) -> None:
        self.endpoint_caller = endpoint_caller

    def get_data(self,
                 start_date: datetime, 
                 end_date: datetime, 
                 endpoint: Endpoint,

        endpoint.add_query_parameter('startDate', str(start_date).replace(" ", "T"))
        endpoint.add_query_parameter('endDate', str(end_date).replace(" ", "T"))
        yield from self.endpoint_caller.call_endpoint_and_get_all_pages(

    def get_data_for_timerange(self,
                               start_date: datetime, 
                               end_date: datetime, 
                               endpoint: Endpoint,
        Given an arbitrarily large timerange denoted by `end_date` and `start_date`, this
        function will call the endpoint continuously by breaking up the requested time range into chunks.
        The chunks do not exceed the single request maximum range (endDate - startDate) for the specific endpoint.
        start_date: datetime
        end_date: datetime
        endpoint: Endpoint
            An implementation of Endpoint for a specific class
        http_ok_next_page_url_extractor: function
            The function that extracts the next page url from a given API response

        page: AmberdataResponse
            A single page is yielded for each call to this generator function

        duration = end_date - start_date
        intervals = duration.total_seconds()/endpoint.max_interval_in_seconds
        hours = duration.total_seconds()/3600
        print(f"Getting {hours} hours of data (# of intervals: {intervals})")

        start_date_copy = start_date
        timerange_stack = []
        while start_date_copy + timedelta(seconds=endpoint.max_interval_in_seconds) <= end_date:
            intermediate_end_date = start_date_copy + timedelta(seconds=endpoint.max_interval_in_seconds)
            timerange_stack.append((start_date_copy, intermediate_end_date))
            start_date_copy = intermediate_end_date
        timerange_stack.append((start_date_copy, end_date))

        while len(timerange_stack) > 0:
            timerange = timerange_stack.pop()
            print(f"Retrieving data from {str(timerange[0])} to {str(timerange[1])}")
            yield from self.get_data(timerange[0], timerange[1], endpoint, http_ok_next_page_url_extractor)
import time
import requests
import json


class AmberdataResponse:
    def __init__(self, data, status, duration_seconds, request_url): = data
        self.status = status
        self.duration_seconds = duration_seconds
        self.request_url = request_url
        self.attempts = 0

    def increment_attempt(self, by=None):
        if by is not None:
            self.attempts += by
            self.attempts += 1

Type Hints
AmberdataResponseStack = list[AmberdataResponse]

class EndpointCaller:
    def __init__(self, amberdata_api_key):
        self.x_api_key = amberdata_api_key

    def call_endpoint_and_get_data_as_json(self, path: str, query: dict, headers: dict):
        headers_with_api_key = {
            'x-api-key': self.x_api_key

        if path.startswith(PRODUCTION_BASE_URL):
            When the next page URL is pre-formed and can be used as-is.
            full_api_url = path
            full_api_url = PRODUCTION_BASE_URL_FORMATTABLE.format(path)
        start_time = time.time()
        request = None
            request = requests.get(full_api_url, params=query, headers=headers_with_api_key)
            print("Making HTTP call for: {}".format(request.url))
            end_time = time.time()
            duration = end_time - start_time
            if request.status_code == 200:
                json_data = request.json()
                return AmberdataResponse(json_data, 200, duration, request.url)
                return AmberdataResponse(request.text, request.status_code, duration, request.url)
        except Exception as exc:
            as_5xx_error_json = {
                'status': 500, #default to 500 error
                'message': 'Failed to complete HTTP request.'

            end_time = time.time()
            duration = end_time - start_time
            return AmberdataResponse(json.dumps(as_5xx_error_json), 500, duration, None)

    def call_endpoint_and_get_all_pages(self, path: str, query: dict, headers: dict, http_ok_next_page_url_extractor):
        Iterative, non-recursive way to get all the pages given an initial URL.

        Avoids Python's max recursion depth (~1000).

        This is a generator function and should be used accordingly.

        path: str
            The endpoint path with the path parameters inserted i.e if the path is `/market/defi/trades/{pool}/historical/` then provide `/market/defi/trades/0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640/historical`
        query: dict
            The query parameters in a dict i.e `{'startDate': '2023-05-20', 'endDate': '2023-05-21'}`
        headers: dict
            The request headers in a dict, you do not have to pass in the API key because the `EndpointCaller` class is initialized with it
        http_ok_next_page_url_extractor: function
            The function that extracts the next page url from a given API response

        api_response: AmberdataResponse
            Individual page from the calling the endpoint

        stack: AmberdataResponseStack = []
        current_page_response = self.call_endpoint_and_get_data_as_json(path, query, headers)
        while len(stack) > 0:
            response = stack.pop()
            if response.status == 200:
                next_page_url = http_ok_next_page_url_extractor(
                if len(next_page_url) > 0:
                    next_page_response = self.call_endpoint_and_get_data_as_json(next_page_url, None, headers)
                yield response
                if response.attempts < 3 and response.request_url is not None:
                    retried_page_response = self.call_endpoint_and_get_data_as_json(response.request_url, None, headers)
                    retried_page_response.increment_attempt(by=response.attempts + 1)
                    yield response

        assert len(stack) == 0, "Stack is not empty, more pages need to be retrieved. # of remaining pages is {}".format(len(stack))
        return AmberdataResponse("DONE", -1, 0, "")