Source code for tradingstrategy.exchange

"""Exchange information and analysis.

Exchanges are presented by

- :py:class:`Exchange` class

To download the pairs dataset see

- :py:meth:`tradingstrategy.client.Client.fetch_exchange_universe`
"""

import enum
from dataclasses import dataclass
from typing import Optional, List, Iterable, Dict

from dataclasses_json import dataclass_json

from tradingstrategy.chain import ChainId
from tradingstrategy.types import NonChecksummedAddress, UNIXTimestamp, PrimaryKey


[docs]class ExchangeType(str, enum.Enum): """What kind of an decentralised exchange, AMM or other the pair is trading on. Note that each type can have multiple implementations. For example QuickSwap, Sushi and Pancake are all Uniswap v2 types. """ #: Uniswap v2 style exchange uniswap_v2 = "uniswap_v2" #: Uniswap v2 style exchange, but with incompatible implementation (e.g. Nomiswap Stable) uniswap_v2_incompatible = "uniswap_v2_incompatible" #: Uniswap v3 style exchange uniswap_v3 = "uniswap_v3" # Uniswap v2 style exchange (same as above `uniswap_v2`) # NOTE: Do not use this member as it is deprecated and only kept for backward # compatibility, it will be removed in the future _deprecated_uni_v2 = "uni_v2"
[docs]@dataclass_json @dataclass class Exchange: """A decentralised exchange. Each chain can have multiple active or abadon decentralised exchanges of different types, like :term:`AMM` based or order book based. The :term:`dataset server` server automatically discovers exchanges and tries to add meaningful label and risk data for them. Most of the fields are optionally and having values them depends on the oracle data indexinb phase. Regarding 30d and life time stats like `buy_volume_30d`: These stats calculated only if exchanged deemed active and we can convert the volume to a supported quote token. Any unsupported token volume does not show up in these stats. Useful mostly for risk assessment, as this data is **not** accurate, but gives some reference information about the popularity of the token. """ #: The chain id on which chain this pair is trading. 1 for Ethereum. #: For JSON, this is serialised as one of the name of enum memmbers of ChainId #: e.g. `"ethereum"`. chain_id: ChainId #: The URL slug derived from the blockchain name. #: Used as the primary key in URLs and other user facing services.] #: Example: "ethereum", "polygon" chain_slug: str #: The exchange where this token trades exchange_id: PrimaryKey #: The URL slug derived from the exchange name. #: Used as the primary key in URLs and other user facing addressers. exchange_slug: str #: The factory smart contract address of Uniswap based exchanges. address: NonChecksummedAddress #: What kind of exchange is this exchange_type: ExchangeType #: How many pairs we have discovered for this exchange so far #: TODO: Make optional - not needed in the tester deployments. pair_count: int #: How many supported trading pairs we have #: See https://tradingstrategy.ai/docs/programming/tracking.html for more information active_pair_count: Optional[int] = None #: When someone traded on this exchange for the first time first_trade_at: Optional[UNIXTimestamp] = None #: When someone traded on this exchange last time last_trade_at: Optional[UNIXTimestamp] = None #: Exchange name - if known or guessed name: Optional[str] = None #: Exchange homepage if available as https:// link homepage: Optional[str] = None #: Denormalised exchange statistics buy_count_all_time: Optional[int] = None #: Denormalised exchange statistics sell_count_all_time: Optional[int] = None #: Denormalised exchange statistics buy_volume_all_time: Optional[float] = None #: Denormalised exchange statistics sell_volume_all_time: Optional[float] = None #: Denormalised exchange statistics buy_count_30d: Optional[int] = None #: Denormalised exchange statistics sell_count_30d: Optional[int] = None #: Denormalised exchange statistics buy_volume_30d: Optional[float] = None #: Denormalised exchange statistics sell_volume_30d: Optional[float] = None def __repr__(self): chain_name = self.chain_id.get_name() name = self.name or "<unknown>" return f"<Exchange {name} at {self.address} on {chain_name}>" def __json__(self, request): """Pyramid JSON renderer compatibility. https://docs.pylonsproject.org/projects/pyramid/en/latest/narr/renderers.html#using-a-custom-json-method """ return self.__dict__ def __hash__(self) -> int: return int(self.address, 16) def __eq__(self, other) -> bool: # https://stackoverflow.com/a/12511715/315168 return isinstance(other, self.__class__) and self.address == other.address @property def vol_30d(self): return (self.buy_volume_30d or 0) + (self.sell_volume_30d or 0)
[docs]@dataclass_json @dataclass class ExchangeUniverse: """Exchange manager. Contains look up for exchanges by their internal primary key ids. """ #: Exchange id -> Exchange data mapping exchanges: Dict[PrimaryKey, Exchange] def get_by_id(self, id) -> Optional[Exchange]: return self.exchanges.get(id)
[docs] def get_top_exchanges_by_30d_volume(self) -> List[Exchange]: """Get top exchanges sorted by their 30d volume. Note that we consider volume only for supported quote tokens. See :py:class:`tradingstrategy.exchange.Exchange` for more details. """ def vol(x: Exchange): return (x.buy_volume_30d or 0) + (x.sell_volume_30d or 0) exchanges = sorted(list(self.exchanges.values()), key=vol, reverse=True) return exchanges
[docs] def get_by_chain_and_name(self, chain_id: ChainId, name: str) -> Optional[Exchange]: """Get the exchange implementation on a specific chain. :param chain_id: Blockchain this exchange is on :param name: Like `sushi` or `uniswap v2`. Case insensitive. """ name = name.lower() assert isinstance(chain_id, ChainId) for xchg in self.exchanges.values(): if xchg.name.lower() == name.lower(): return xchg return None
[docs] def get_by_chain_and_slug(self, chain_id: ChainId, slug: str) -> Optional[Exchange]: """Get the exchange implementation on a specific chain. :param chain_id: Blockchain this exchange is on :param slug: Machine readable exchange name. Like `uniswap-v2`. Case sensitive. """ assert isinstance(chain_id, ChainId) for xchg in self.exchanges.values(): if xchg.exchange_slug == slug: return xchg return None
[docs] def get_by_chain_and_factory(self, chain_id: ChainId, factory_address: str) -> Optional[Exchange]: """Get the exchange implementation on a specific chain. :param chain_id: Blockchain this exchange is on :param factory_address: The smart contract address of the exchange factory """ assert isinstance(chain_id, ChainId) factory_address = factory_address.lower() for xchg in self.exchanges.values(): if xchg.address.lower() == factory_address: return xchg return None