"""This module contains the address subhandler."""
import warnings
from typing import Dict
from web3data.chains import Chains
from web3data.handlers.base import BaseHandler
[docs]class MarketHandler(BaseHandler):
"""The subhandler for market-related queries."""
def __init__(self, initial_headers: Dict[str, str], chain: Chains):
"""Return a new :code:`MarketHandler` instance.
:param initial_headers: Base headers to attach to every request
:param chain: The blockchain to fetch the information for
"""
super().__init__(chain)
self.initial_headers = initial_headers
self.base_url = "https://web3api.io/api/v2/"
[docs] def exchanges(self, **kwargs) -> Dict:
"""Retrieves information about supported exchange-pairs.
These types of data are supported:
- ticker
- ohlc (open-high-low-close)
- trade
- order_book
- order_book_update
:key exchange: only return data for the given exchanges (comma separated) (str)
:key pair: only return data for the given pairs (comma separated) (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route="market/exchanges",
headers=self.initial_headers,
params=kwargs,
)
[docs] def ohlcv(self, **kwargs) -> Dict:
"""Retrieves information about supported exchange-pairs for ohlcv.
:key exchange: Filter by data for the given exchanges (comma separated). (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route="market/ohlcv/information",
headers=self.initial_headers,
params=kwargs,
)
[docs] def pairs(self, **kwargs) -> Dict:
"""Retrieves information about supported exchange-pairs. These types of
data are supported:
- ticker
- ohlc (open-high-low-close)
- trade
- order_book
- order_book_update
:key exchange: only return data for the given exchanges (comma separated) (str)
:key pair: only return data for the given pairs (comma separated) (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route="market/pairs",
headers=self.initial_headers,
params=kwargs,
)
[docs] def rankings(self, **kwargs) -> Dict:
"""Retrieves the top ranked assets by a specific metric.
:key direction: The sort order in which assets are ranked ascending
or descending. (str)
:key sortType: The metric used to rank the assets.
Options: changeInPrice, currentPrice, liquidMarketCap, marketCap,
tokenVelocity, tradeVolume, transactionVolume, uniqueAddresses (str)
:key timeInterval: The time interval in which to return
the historical data days or hours. (str)
:key type: The type(s) of assets to include in the rankings:
erc20|, erc721, erc777, erc884, erc998. Note: leaving this
parameter empty means all tokens will be included. (str)
:key page: The page number to return. (int)
:key size: The number of records per page. (int)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route="market/rankings",
headers=self.initial_headers,
params=kwargs,
)
[docs] def price_pairs(self) -> Dict:
"""Retrieves the assets for which latest prices are available.
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route="market/prices/pairs",
headers=self.initial_headers,
params={},
)
[docs] def ticker_pairs(self, **kwargs) -> Dict:
"""Retrieves the list of all available market tickers.
:key exchange: Only return data for the given exchanges (comma separated). (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route="market/tickers/information",
headers=self.initial_headers,
params=kwargs,
)
[docs] def trades(self, **kwargs) -> Dict:
"""Retrieves the list of all available market trade data sets.
:key exchange: Only return data for the given exchanges (comma separated). (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route="market/trades/information",
headers=self.initial_headers,
params=kwargs,
)
[docs] def order_best_bid_historical(self, pair: str, **kwargs) -> Dict:
"""Retrieves historical best bid and offer information for the
specified pair.
:param pair: The asset pair to look up
:key exchange: The exchange(s) for which to retrieve order book data.
Example: exchange=bitfinex,bitstamp (str)
:key timeFormat: The timestamp format to use for the timestamps:
milliseconds/ms or iso/iso8611. (str)
:key startDate: Filter by pairs after this date. (int)
:key endDate: Filter by pairs before this date. (int)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/orders/{pair}/bbo/historical",
headers=self.initial_headers,
params=kwargs,
)
[docs] def order_best_bid_latest(self, pair: str, **kwargs) -> Dict:
"""Retrieves the latest best bid and offer information for the
specified pair and exchange.
:param pair: The asset pair to look up
:key exchange: The exchange(s) for which to retrieve order book data.
Example: exchange=bitfinex,bitstamp (str)
:key timeFormat: The timestamp format to use for the timestamps:
milliseconds/ms or iso/iso8611. (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/orders/{pair}/bbo",
headers=self.initial_headers,
params=kwargs,
)
[docs] def order_book_updates(self, pair: str, **kwargs) -> Dict:
"""
:param pair: The asset pair to look up
:key exchange: The exchange(s) for which to retrieve order book data.
Example: exchange=bitfinex,bitstamp (str)
:key timeFormat: The timestamp format to use for the timestamps:
milliseconds/ms or iso/iso8611. (str)
:key startDate: Filter by pairs after this date. (int)
:key endDate: Filter by pairs before this date. (int)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/orders/{pair}/update",
headers=self.initial_headers,
params=kwargs,
)
[docs] def order_book(self, pair: str, **kwargs) -> Dict:
"""Retrieves the order book data for the specified pair.
:param pair: The asset pair to look up
:key exchange: The exchange(s) for which to retrieve order book data.
Example: exchange=bitfinex,bitstamp (str)
:key timestamp: The timestamp at which to return the order book information
(closest match, lower or equal to the timestamp specified). (str)
:key timeFormat: The timestamp format to use for the timestamps:
milliseconds/ms or iso/iso8611. (str)
:key startDate: Filter by pairs after this date. Formats:
milliseconds, iso, or iso8611 (str)
:key endDate: Filter by pairs before this date. Formats:
milliseconds, iso, or iso8611.
Note: Must be greater than startDate and cannot exceed 10 minutes. (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/orders/{pair}",
headers=self.initial_headers,
params=kwargs,
)
[docs] def uniswap_liquidity(self, pair: str, **kwargs) -> Dict:
"""Retrieves the Uniswap-specific ether and token balance pairs over
time.
Note: This endpoint returns a max of 6 months of historical data. In order
to get more than 6 months you must use the startDate & endDate parameters
to move the time frame window to get the next n days/months of data.
:param pair: The asset pair to look up
:key timeFormat: The timestamp format to use for the timestamps:
milliseconds/ms or iso/iso8611. (str)
:key startDate: Filter to liquidity changes after this date (int)
:key endDate: Filter to liquidity changes before this date (int)
:return: The API response parsed into a dict
"""
warnings.warn(
"The Uniswap liquidity endpoint is scheduled for deprecation",
PendingDeprecationWarning,
)
return self.raw_query(
base_url=self.base_url,
route=f"market/orders/uniswap/{pair}/liquidity",
headers=self.initial_headers,
params=kwargs,
)
[docs] def trade_pairs_historical(self, pair: str, **kwargs) -> Dict:
"""Retrieves the historical (time series) trade data for the specified
pair.
Note: This endpoint returns a max of 6 months of historical data. In order to
get more than 6 months you must use the startDate & endDate parameters to move
the time frame window to get the next n days/months of data.
:param pair: The asset pair to look up
:key exchange: The exchange(s) for which to retrieve market trades. (str)
:key startDate: Filter by trades after this date. (int)
:key endDate: Filter by trades before this date. (int)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/trades/{pair}/historical",
headers=self.initial_headers,
params=kwargs,
)
[docs] def ohlcv_pair_latest(self, pair: str, **kwargs) -> Dict:
"""Retrieves the latest open-high-low-close for the specified pair.
:param pair: The asset pair to look up
:key exchange: The exchange(s) for which to retrieve OHLCV.
Example: exchange=bitfinex,bitstamp (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/ohlcv/{pair}/latest",
headers=self.initial_headers,
params=kwargs,
)
[docs] def ohlcv_pair_historical(self, pair: str, **kwargs) -> Dict:
"""Retrieves the historical (time series) open-high-low-close for the
specified pair.
Note: This endpoint returns a max of 6 months of historical data. In order to get more
than 6 months you must use the startDate & endDate parameters to move the time frame
window to get the next n days/months of data.
:param pair: The asset pair to look up
:key exchange: The exchange(s) for which to retrieve OHLCV.
Example: exchange=bitfinex,bitstamp (str)
:key startDate: Filter by pairs after this date. (int)
:key endDate: Filter by pairs before this date. (int)
:key timeInterval: Time interval to return the historical data
in ("days" | "hours" | "minutes") (str)
:key timeFormat: Time format to use for the timestamps
( "milliseconds" | "ms" | "iso" | "iso8611" ) (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/ohlcv/{pair}/historical",
headers=self.initial_headers,
params=kwargs,
)
[docs] def price_pair_historical(self, pair: str, **kwargs) -> Dict:
"""Retrieves the historical prices for the specified asset.
:param pair: The asset pair to look up
:key timeFormat: Time format to use for the timestamps
(Options: milliseconds, ms, iso, iso8611) (str)
:key startDate: Filter by prices after this date. (str)
:key endDate: Filter by prices before this date. (str)
:key timeInterval: Time interval to return the historical data
in ("days" | "hours" | "minutes") (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/spot/prices/pairs/{pair}/historical",
headers=self.initial_headers,
params=kwargs,
)
[docs] def price_pair_latest(self, pair: str, **kwargs) -> Dict:
"""Retrieves the latest price for the specified asset.
:param pair: The asset pair to look up
:key timeFormat: Time format to use for the timestamps
(Options: milliseconds, ms, iso, iso8611) (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/spot/prices/pairs/{pair}/latest",
headers=self.initial_headers,
params=kwargs,
)
[docs] def base_wap_latest(self, base: str, **kwargs) -> Dict:
"""Retrieves the latest VWAP & TWAP price for the specified base.
:param base: The pair's base
:key quote: The currency of the pair. Example: if pair is "eth_usd",
then quote is "usd" (str)
:key timeFormat: Time format to use for the timestamps
(Options: milliseconds, ms, iso, iso8611) (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/prices/{base}/wap/latest",
headers=self.initial_headers,
params=kwargs,
)
[docs] def ticker_bid_ask_latest(self, pair: str, **kwargs) -> Dict:
"""Retrieves the latest market ticker Bid/Ask/Mid/Last for the
specified pair.
:param pair: The asset pair to look up
:key exchange: The exchange(s) for which to retrieve market tickers.
Example: exchange=bitfinex,bitstamp (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/tickers/{pair}/latest",
headers=self.initial_headers,
params=kwargs,
)
[docs] def ticker_bid_ask_historical(self, pair: str, **kwargs) -> Dict:
"""Retrieves the historical ticker, bid/ask/mid/last, for the specified
pair.
Note: This endpoint returns a max of 6 months of historical data. In order to get more
than 6 months you must use the startDate & endDate parameters to move the time frame
window to get the next n days/months of data.
:param pair: The asset pair to look up
:key exchange: The exchange(s) for which to retrieve market tickers.
Example: exchange=bitfinex,bitstamp (str)
:key startDate: Filter by ticker pairs after this date. (int)
:key endDate: Filter by ticker pairs before this date. (int)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/tickers/{pair}/historical",
headers=self.initial_headers,
params=kwargs,
)
[docs] def token_price_historical(self, address: str, **kwargs) -> Dict:
"""
:param address: The token's smart contract address
:key currency: The additional currency (other than ETH and USD) for which
to return price info. (str)
:key timeFormat: The time format to use for the timestamps: milliseconds/ms
or iso/iso861. (str)
:key timeInterval: The time interval to return the historical data in:
by day (d),
by hour (h), or
by minute (m). (str)
:key startDate: Filter by prices after this date. Note that the interval can not exceed
6 months (d), 30 days (h) or 24 hours (m). (int)
:key endDate: Filter by prices before this date. Note that the interval can not exceed
6 months (d), 30 days (h) or 24 hours (m). (int)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/tokens/prices/{address}/historical",
headers=self.initial_headers,
params=kwargs,
)
[docs] def token_price_latest(self, address: str, **kwargs) -> Dict:
"""Retrieves the latest price (and other market information) for the
specified token.
:param address: The token's smart contract address
:key currency: The additional currency (other than ETH and USD)
for which to return price info. (str)
:return: The API response parsed into a dict
"""
return self.raw_query(
base_url=self.base_url,
route=f"market/tokens/prices/{address}/latest",
headers=self.initial_headers,
params=kwargs,
)
[docs] def token_rankings_historical(self, **kwargs) -> Dict:
"""Retrieves the top ranked tokens by a specific metric, with a
lookback window.
Useful for viewing token trends.
:key direction: The sort order in which tokens are ranked
(ascending or descending). (str)
:key sortType: The metric used to rank the tokens (changeInPrice,
currentPrice, marketCap, tokenVelocity, transactionVolume &
uniqueAddresses). (str)
:key topN: Number denominating top ranking tokens to return.
Example: If given "5", results with return top 5 token
rankings for past timeframe, where there are 5 results
per day. (str)
:return: The API response parsed into a dict
"""
self._check_chain_supported()
return self.raw_query(
base_url=self.base_url,
route="tokens/rankings/historical",
headers=self.initial_headers,
params=kwargs,
)
[docs] def token_rankings_latest(self, **kwargs) -> Dict:
"""Retrieves the top ranked tokens by a specific metric.
:key direction: The sort order in which tokens are ranked
(ascending or descending). (str)
:key sortType: The metric used to rank the tokens (changeInPrice,
currentPrice, marketCap, tokenVelocity, transactionVolume &
uniqueAddresses). (str)
:key timeInterval: The time interval to return the historical
data in: by day (days) or by hour (hours). (str)
:key type: The type(s) of tokens to include in the rankings
(erc20, erc721, erc777, erc884, erc998) (str)
:key page: The page number to return. (int)
:key size: Number of records per page (int)
:return: The API response parsed into a dict
"""
self._check_chain_supported()
return self.raw_query(
base_url=self.base_url,
route="tokens/rankings",
headers=self.initial_headers,
params=kwargs,
)