""" PairList Handler base class """ import logging from abc import ABC, abstractmethod, abstractproperty from typing import Any, Dict, List from cachetools import TTLCache, cached from freqtrade.exchange import market_is_active logger = logging.getLogger(__name__) class IPairList(ABC): def __init__(self, exchange, pairlistmanager, config: Dict[str, Any], pairlistconfig: Dict[str, Any], pairlist_pos: int) -> None: """ :param exchange: Exchange instance :param pairlistmanager: Instantiated Pairlist manager :param config: Global bot configuration :param pairlistconfig: Configuration for this Pairlist Handler - can be empty. :param pairlist_pos: Position of the Pairlist Handler in the chain """ self._exchange = exchange self._pairlistmanager = pairlistmanager self._config = config self._pairlistconfig = pairlistconfig self._pairlist_pos = pairlist_pos self.refresh_period = self._pairlistconfig.get('refresh_period', 1800) self._last_refresh = 0 self._log_cache = TTLCache(maxsize=1024, ttl=self.refresh_period) @property def name(self) -> str: """ Gets name of the class -> no need to overwrite in subclasses """ return self.__class__.__name__ def log_on_refresh(self, logmethod, message: str) -> None: """ Logs message - not more often than "refresh_period" to avoid log spamming Logs the log-message as debug as well to simplify debugging. :param logmethod: Function that'll be called. Most likely `logger.info`. :param message: String containing the message to be sent to the function. :return: None. """ @cached(cache=self._log_cache) def _log_on_refresh(message: str): logmethod(message) # Log as debug first logger.debug(message) # Call hidden function. _log_on_refresh(message) @abstractproperty def needstickers(self) -> bool: """ Boolean property defining if tickers are necessary. If no Pairlist requries tickers, an empty List is passed as tickers argument to filter_pairlist """ @abstractmethod def short_desc(self) -> str: """ Short whitelist method description - used for startup-messages -> Please overwrite in subclasses """ @abstractmethod def filter_pairlist(self, pairlist: List[str], tickers: Dict) -> List[str]: """ Filters and sorts pairlist and returns the whitelist again. Called on each bot iteration - please use internal caching if necessary -> Please overwrite in subclasses :param pairlist: pairlist to filter or sort :param tickers: Tickers (from exchange.get_tickers()). May be cached. :return: new whitelist """ def verify_blacklist(self, pairlist: List[str], logmethod) -> List[str]: """ Proxy method to verify_blacklist for easy access for child classes. :param pairlist: Pairlist to validate :param logmethod: Function that'll be called, `logger.info` or `logger.warning`. :return: pairlist - blacklisted pairs """ return self._pairlistmanager.verify_blacklist(pairlist, logmethod) def _whitelist_for_active_markets(self, pairlist: List[str]) -> List[str]: """ Check available markets and remove pair from whitelist if necessary :param whitelist: the sorted list of pairs the user might want to trade :return: the list of pairs the user wants to trade without those unavailable or black_listed """ markets = self._exchange.markets sanitized_whitelist: List[str] = [] for pair in pairlist: # pair is not in the generated dynamic market or has the wrong stake currency if pair not in markets: logger.warning(f"Pair {pair} is not compatible with exchange " f"{self._exchange.name}. Removing it from whitelist..") continue if self._exchange.get_pair_quote_currency(pair) != self._config['stake_currency']: logger.warning(f"Pair {pair} is not compatible with your stake currency " f"{self._config['stake_currency']}. Removing it from whitelist..") continue # Check if market is active market = markets[pair] if not market_is_active(market): logger.info(f"Ignoring {pair} from whitelist. Market is not active.") continue if pair not in sanitized_whitelist: sanitized_whitelist.append(pair) # We need to remove pairs that are unknown return sanitized_whitelist