client

drift_data.client

Client

Bases: BaseModel

Client to interact with Drift Data API.

This client allows you to hit the Drift Data API. The playground is a good place to start exploring the API endpoints. There is also some documentation for the API, but it is not very detailed.

The client supports two clusters: mainnet and devnet. Although support for devnet does not seem to be fully implemented yet, so it may not replicate all functionality available on mainnet.

get_contracts

get_contracts() -> list[ContractRaw]

Get all contracts from the Drift Data API.

This method retrieves all contracts from the Drift Data API. It returns a list of ContractRaw objects.

Returns:

Type	Description
list[ContractRaw]	list[ContractRaw]: A list of raw contract data.

Source code in drift_data/client.py

def get_contracts(self) -> list[ContractRaw]:
    """Get all contracts from the Drift Data API.

    This method retrieves all contracts from the Drift Data API. It returns a list
    of `ContractRaw` objects.

    Returns:
        list[ContractRaw]: A list of raw contract data.
    """
    endpoint = "/contracts"
    logger.info(f"Fetching contracts from `{endpoint}`")
    resp = self._get_api(endpoint=endpoint)
    contracts = resp.get("contracts", [])
    ta = TypeAdapter(list[ContractRaw])
    logger.info(f"Validating {len(contracts)} contracts")
    return ta.validate_python(contracts)

get_funding_rates

get_funding_rates(symbol: str, year: int | None = None, month: int | None = None, day: int | None = None) -> list[FundingRateRecord]

Get funding rates for a given symbol.

This method retrieves funding rates for a specific symbol. You can get the last 30 days of funding rates by default, or you can specify a specific date by providing the year, month, and day parameters.

Parameters:

Name	Type	Description	Default
symbol	str	The symbol to get funding rates for.	required
year	int	The year to get funding rates for. Defaults to None.	None
month	int	The month to get funding rates for. Defaults to None.	None
day	int	The day to get funding rates for. Defaults to None.	None

Returns:

Type	Description
list[FundingRateRecord]	list[FundingRateRecord]: A list of funding rate records for the specified symbol and date.

Source code in drift_data/client.py

def get_funding_rates(
    self,
    symbol: str,
    year: int | None = None,
    month: int | None = None,
    day: int | None = None,
) -> list[FundingRateRecord]:
    """Get funding rates for a given symbol.

    This method retrieves funding rates for a specific symbol. You can get the last
    30 days of funding rates by default, or you can specify a specific date by
    providing the `year`, `month`, and `day` parameters.

    Args:
        symbol (str): The symbol to get funding rates for.
        year (int, optional): The year to get funding rates for. Defaults to None.
        month (int, optional): The month to get funding rates for. Defaults to None.
        day (int, optional): The day to get funding rates for. Defaults to None.

    Returns:
        list[FundingRateRecord]: A list of funding rate records for the specified
            symbol and date.
    """
    if isinstance(year, int) and isinstance(month, int) and isinstance(day, int):
        date_ = date(year=year, month=month, day=day)
        if (date_) > datetime.now(tz=UTC).date():
            msg = "Cannot get funding rates for a future date"
            raise ValueError(msg)
        if date_ < date(year=2023, month=1, day=1):
            msg = "Cannot get funding rates before 2023-01-01"
            raise ValueError(msg)
        endpoint = f"/market/{symbol}/fundingRates/{year}/{month}/{day}"
    elif all(i is None for i in [year, month, day]):
        endpoint = f"/market/{symbol}/fundingRates"
    else:
        msg = "year, month, and day must be either all provided or all None"
        raise ValueError(msg)

    logger.info(f"Fetching funding rates from `{endpoint}`")
    resp = self._get_api(endpoint=endpoint)
    next_page = resp.get("meta", {}).get("nextPage", None)
    records = resp.get("records", [])

    while next_page is not None:
        _endpoint = f"{endpoint}?page={resp['meta']['nextPage']}"
        logger.info(f"Fetching funding rates from `{_endpoint}`")
        resp = self._get_api(endpoint=_endpoint)
        next_page = resp.get("meta", {}).get("nextPage", None)
        records.extend(resp.get("records", []))

    ta = TypeAdapter(list[FundingRateRecord])
    logger.info(f"Validating {len(records)} funding rate records")
    return ta.validate_python(records)

model_post_init

model_post_init(_: Any) -> None

Perform tasks after the model is instantiated and fields are validated.

This is a Pydantic model hook that does the following:

Override this method to perform additional initialization after init and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

Source code in drift_data/client.py

def model_post_init(self, _: Any) -> None:  # noqa: ANN401
    """Perform tasks after the model is instantiated and fields are validated.

    This is a [Pydantic model hook](https://docs.pydantic.dev/latest/api/base_model/#pydantic.BaseModel.model_post_init)
    that does the following:
    > Override this method to perform additional initialization after __init__ and
    > model_construct. This is useful if you want to do some validation that
    > requires the entire model to be initialized.
    """
    if self.cluster == "mainnet":
        self._url = "https://data.api.drift.trade"
    elif self.cluster == "devnet":
        self._url = "https://master-data.drift.trade"
    else:
        msg = f"Not implemented for cluster {self.cluster}"
        raise NotImplementedError(msg)
    logger.info(f"Using Drift Data API cluster: {self.cluster} ({self._url})")