9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.74

/freqtrade/plugins/pairlist/IPairList.py

"""
PairList Handler base class
"""
import logging
from abc import ABC, abstractmethod, abstractproperty
from copy import deepcopy
from typing import Any, Dict, List, Literal, Optional, TypedDict, Union

from freqtrade.constants import Config
from freqtrade.exceptions import OperationalException
from freqtrade.exchange import Exchange, market_is_active
from freqtrade.exchange.types import Ticker, Tickers
from freqtrade.mixins import LoggingMixin


logger = logging.getLogger(__name__)


class __PairlistParameterBase(TypedDict):
    description: str
    help: str


class __NumberPairlistParameter(__PairlistParameterBase):
    type: Literal["number"]
    default: Union[int, float, None]


class __StringPairlistParameter(__PairlistParameterBase):
    type: Literal["string"]
    default: Union[str, None]


class __OptionPairlistParameter(__PairlistParameterBase):
    type: Literal["option"]
    default: Union[str, None]
    options: List[str]


class __BoolPairlistParameter(__PairlistParameterBase):
    type: Literal["boolean"]
    default: Union[bool, None]


PairlistParameter = Union[
    __NumberPairlistParameter,
    __StringPairlistParameter,
    __OptionPairlistParameter,
    __BoolPairlistParameter
    ]


class IPairList(LoggingMixin, ABC):

    is_pairlist_generator = False

    def __init__(self, exchange: Exchange, pairlistmanager,
                 config: Config, 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._enabled = True

        self._exchange: 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)
        LoggingMixin.__init__(self, logger, self.refresh_period)

    @property
    def name(self) -> str:
        """
        Gets name of the class
        -> no need to overwrite in subclasses
        """
        return self.__class__.__name__

    @abstractproperty
    def needstickers(self) -> bool:
        """
        Boolean property defining if tickers are necessary.
        If no Pairlist requires tickers, an empty Dict is passed
        as tickers argument to filter_pairlist
        """
        return False

    @staticmethod
    @abstractmethod
    def description() -> str:
        """
        Return description of this Pairlist Handler
        -> Please overwrite in subclasses
        """
        return ""

    @staticmethod
    def available_parameters() -> Dict[str, PairlistParameter]:
        """
        Return parameters used by this Pairlist Handler, and their type
        contains a dictionary with the parameter name as key, and a dictionary
        with the type and default value.
        -> Please overwrite in subclasses
        """
        return {}

    @staticmethod
    def refresh_period_parameter() -> Dict[str, PairlistParameter]:
        return {
            "refresh_period": {
                "type": "number",
                "default": 1800,
                "description": "Refresh period",
                "help": "Refresh period in seconds",
            }
        }

    @abstractmethod
    def short_desc(self) -> str:
        """
        Short whitelist method description - used for startup-messages
        -> Please overwrite in subclasses
        """

    def _validate_pair(self, pair: str, ticker: Optional[Ticker]) -> bool:
        """
        Check one pair against Pairlist Handler's specific conditions.

        Either implement it in the Pairlist Handler or override the generic
        filter_pairlist() method.

        :param pair: Pair that's currently validated
        :param ticker: ticker dict as returned from ccxt.fetch_ticker
        :return: True if the pair can stay, false if it should be removed
        """
        raise NotImplementedError()

    def gen_pairlist(self, tickers: Tickers) -> List[str]:
        """
        Generate the pairlist.

        This method is called once by the pairlistmanager in the refresh_pairlist()
        method to supply the starting pairlist for the chain of the Pairlist Handlers.
        Pairlist Filters (those Pairlist Handlers that cannot be used at the first
        position in the chain) shall not override this base implementation --
        it will raise the exception if a Pairlist Handler is used at the first
        position in the chain.

        :param tickers: Tickers (from exchange.get_tickers). May be cached.
        :return: List of pairs
        """
        raise OperationalException("This Pairlist Handler should not be used "
                                   "at the first position in the list of Pairlist Handlers.")

    def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]:
        """
        Filters and sorts pairlist and returns the whitelist again.

        Called on each bot iteration - please use internal caching if necessary
        This generic implementation calls self._validate_pair() for each pair
        in the pairlist.

        Some Pairlist Handlers override this generic implementation and employ
        own filtration.

        :param pairlist: pairlist to filter or sort
        :param tickers: Tickers (from exchange.get_tickers). May be cached.
        :return: new whitelist
        """
        if self._enabled:
            # Copy list since we're modifying this list
            for p in deepcopy(pairlist):
                # Filter out assets
                if not self._validate_pair(p, tickers[p] if p in tickers else None):
                    pairlist.remove(p)

        return pairlist

    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 verify_whitelist(self, pairlist: List[str], logmethod,
                         keep_invalid: bool = False) -> List[str]:
        """
        Proxy method to verify_whitelist for easy access for child classes.
        :param pairlist: Pairlist to validate
        :param logmethod: Function that'll be called, `logger.info` or `logger.warning`
        :param keep_invalid: If sets to True, drops invalid pairs silently while expanding regexes.
        :return: pairlist - whitelisted pairs
        """
        return self._pairlistmanager.verify_whitelist(pairlist, logmethod, keep_invalid)

    def _whitelist_for_active_markets(self, pairlist: List[str]) -> List[str]:
        """
        Check available markets and remove pair from whitelist if necessary
        :param pairlist: 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
        if not markets:
            raise OperationalException(
                'Markets not loaded. Make sure that exchange is initialized correctly.')

        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:
                self.log_once(f"Pair {pair} is not compatible with exchange "
                              f"{self._exchange.name}. Removing it from whitelist..",
                              logger.warning)
                continue

            if not self._exchange.market_is_tradable(markets[pair]):
                self.log_once(f"Pair {pair} is not tradable with Freqtrade."
                              "Removing it from whitelist..", logger.warning)
                continue

            if self._exchange.get_pair_quote_currency(pair) != self._config['stake_currency']:
                self.log_once(f"Pair {pair} is not compatible with your stake currency "
                              f"{self._config['stake_currency']}. Removing it from whitelist..",
                              logger.warning)
                continue

            # Check if market is active
            market = markets[pair]
            if not market_is_active(market):
                self.log_once(f"Ignoring {pair} from whitelist. Market is not active.", logger.info)
                continue
            if pair not in sanitized_whitelist:
                sanitized_whitelist.append(pair)

        # We need to remove pairs that are unknown
        return sanitized_whitelist

1	"""
2	PairList Handler base class
3	"""
4	import logging	1✔
5	from abc import ABC, abstractmethod, abstractproperty	1✔
6	from copy import deepcopy	1✔
7	from typing import Any, Dict, List, Literal, Optional, TypedDict, Union	1✔
8
9	from freqtrade.constants import Config	1✔
10	from freqtrade.exceptions import OperationalException	1✔
11	from freqtrade.exchange import Exchange, market_is_active	1✔
12	from freqtrade.exchange.types import Ticker, Tickers	1✔
13	from freqtrade.mixins import LoggingMixin	1✔
14
15
16	logger = logging.getLogger(__name__)	1✔
17
18
19	class __PairlistParameterBase(TypedDict):	1✔
20	description: str	1✔
21	help: str	1✔
22
23
24	class __NumberPairlistParameter(__PairlistParameterBase):	1✔
25	type: Literal["number"]	1✔
26	default: Union[int, float, None]	1✔
27
28
29	class __StringPairlistParameter(__PairlistParameterBase):	1✔
30	type: Literal["string"]	1✔
31	default: Union[str, None]	1✔
32
33
34	class __OptionPairlistParameter(__PairlistParameterBase):	1✔
35	type: Literal["option"]	1✔
36	default: Union[str, None]	1✔
37	options: List[str]	1✔
38
39
40	class __BoolPairlistParameter(__PairlistParameterBase):	1✔
41	type: Literal["boolean"]	1✔
42	default: Union[bool, None]	1✔
43
44
45	PairlistParameter = Union[	1✔
46	__NumberPairlistParameter,
47	__StringPairlistParameter,
48	__OptionPairlistParameter,
49	__BoolPairlistParameter
50	]
51
52
53	class IPairList(LoggingMixin, ABC):	1✔
54
55	is_pairlist_generator = False	1✔
56
57	def __init__(self, exchange: Exchange, pairlistmanager,	1✔
58	config: Config, pairlistconfig: Dict[str, Any],
59	pairlist_pos: int) -> None:
60	"""
61	:param exchange: Exchange instance
62	:param pairlistmanager: Instantiated Pairlist manager
63	:param config: Global bot configuration
64	:param pairlistconfig: Configuration for this Pairlist Handler - can be empty.
65	:param pairlist_pos: Position of the Pairlist Handler in the chain
66	"""
67	self._enabled = True	1✔
68
69	self._exchange: Exchange = exchange	1✔
70	self._pairlistmanager = pairlistmanager	1✔
71	self._config = config	1✔
72	self._pairlistconfig = pairlistconfig	1✔
73	self._pairlist_pos = pairlist_pos	1✔
74	self.refresh_period = self._pairlistconfig.get('refresh_period', 1800)	1✔
75	LoggingMixin.__init__(self, logger, self.refresh_period)	1✔
76
77	@property	1✔
78	def name(self) -> str:	1✔
79	"""
80	Gets name of the class
81	-> no need to overwrite in subclasses
82	"""
83	return self.__class__.__name__	1✔
84
85	@abstractproperty	1✔
86	def needstickers(self) -> bool:	1✔
87	"""
88	Boolean property defining if tickers are necessary.
89	If no Pairlist requires tickers, an empty Dict is passed
90	as tickers argument to filter_pairlist
91	"""
92	return False	×
93
94	@staticmethod	1✔
95	@abstractmethod	1✔
96	def description() -> str:	1✔
97	"""
98	Return description of this Pairlist Handler
99	-> Please overwrite in subclasses
100	"""
101	return ""	×
102
103	@staticmethod	1✔
104	def available_parameters() -> Dict[str, PairlistParameter]:	1✔
105	"""
106	Return parameters used by this Pairlist Handler, and their type
107	contains a dictionary with the parameter name as key, and a dictionary
108	with the type and default value.
109	-> Please overwrite in subclasses
110	"""
111	return {}	1✔
112
113	@staticmethod	1✔
114	def refresh_period_parameter() -> Dict[str, PairlistParameter]:	1✔
115	return {	1✔
116	"refresh_period": {
117	"type": "number",
118	"default": 1800,
119	"description": "Refresh period",
120	"help": "Refresh period in seconds",
121	}
122	}
123
124	@abstractmethod	1✔
125	def short_desc(self) -> str:	1✔
126	"""
127	Short whitelist method description - used for startup-messages
128	-> Please overwrite in subclasses
129	"""
130
131	def _validate_pair(self, pair: str, ticker: Optional[Ticker]) -> bool:	1✔
132	"""
133	Check one pair against Pairlist Handler's specific conditions.
134
135	Either implement it in the Pairlist Handler or override the generic
136	filter_pairlist() method.
137
138	:param pair: Pair that's currently validated
139	:param ticker: ticker dict as returned from ccxt.fetch_ticker
140	:return: True if the pair can stay, false if it should be removed
141	"""
142	raise NotImplementedError()	×
143
144	def gen_pairlist(self, tickers: Tickers) -> List[str]:	1✔
145	"""
146	Generate the pairlist.
147
148	This method is called once by the pairlistmanager in the refresh_pairlist()
149	method to supply the starting pairlist for the chain of the Pairlist Handlers.
150	Pairlist Filters (those Pairlist Handlers that cannot be used at the first
151	position in the chain) shall not override this base implementation --
152	it will raise the exception if a Pairlist Handler is used at the first
153	position in the chain.
154
155	:param tickers: Tickers (from exchange.get_tickers). May be cached.
156	:return: List of pairs
157	"""
158	raise OperationalException("This Pairlist Handler should not be used "	1✔
159	"at the first position in the list of Pairlist Handlers.")
160
161	def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]:	1✔
162	"""
163	Filters and sorts pairlist and returns the whitelist again.
164
165	Called on each bot iteration - please use internal caching if necessary
166	This generic implementation calls self._validate_pair() for each pair
167	in the pairlist.
168
169	Some Pairlist Handlers override this generic implementation and employ
170	own filtration.
171
172	:param pairlist: pairlist to filter or sort
173	:param tickers: Tickers (from exchange.get_tickers). May be cached.
174	:return: new whitelist
175	"""
176	if self._enabled:	1✔
177	# Copy list since we're modifying this list
178	for p in deepcopy(pairlist):	1✔
179	# Filter out assets
180	if not self._validate_pair(p, tickers[p] if p in tickers else None):	1✔
181	pairlist.remove(p)	1✔
182
183	return pairlist	1✔
184
185	def verify_blacklist(self, pairlist: List[str], logmethod) -> List[str]:	1✔
186	"""
187	Proxy method to verify_blacklist for easy access for child classes.
188	:param pairlist: Pairlist to validate
189	:param logmethod: Function that'll be called, `logger.info` or `logger.warning`.
190	:return: pairlist - blacklisted pairs
191	"""
192	return self._pairlistmanager.verify_blacklist(pairlist, logmethod)	1✔
193
194	def verify_whitelist(self, pairlist: List[str], logmethod,	1✔
195	keep_invalid: bool = False) -> List[str]:
196	"""
197	Proxy method to verify_whitelist for easy access for child classes.
198	:param pairlist: Pairlist to validate
199	:param logmethod: Function that'll be called, `logger.info` or `logger.warning`
200	:param keep_invalid: If sets to True, drops invalid pairs silently while expanding regexes.
201	:return: pairlist - whitelisted pairs
202	"""
203	return self._pairlistmanager.verify_whitelist(pairlist, logmethod, keep_invalid)	1✔
204
205	def _whitelist_for_active_markets(self, pairlist: List[str]) -> List[str]:	1✔
206	"""
207	Check available markets and remove pair from whitelist if necessary
208	:param pairlist: the sorted list of pairs the user might want to trade
209	:return: the list of pairs the user wants to trade without those unavailable or
210	black_listed
211	"""
212	markets = self._exchange.markets	1✔
213	if not markets:	1✔
214	raise OperationalException(	1✔
215	'Markets not loaded. Make sure that exchange is initialized correctly.')
216
217	sanitized_whitelist: List[str] = []	1✔
218	for pair in pairlist:	1✔
219	# pair is not in the generated dynamic market or has the wrong stake currency
220	if pair not in markets:	1✔
221	self.log_once(f"Pair {pair} is not compatible with exchange "	1✔
222	f"{self._exchange.name}. Removing it from whitelist..",
223	logger.warning)
224	continue	1✔
225
226	if not self._exchange.market_is_tradable(markets[pair]):	1✔
227	self.log_once(f"Pair {pair} is not tradable with Freqtrade."	1✔
228	"Removing it from whitelist..", logger.warning)
229	continue	1✔
230
231	if self._exchange.get_pair_quote_currency(pair) != self._config['stake_currency']:	1✔
232	self.log_once(f"Pair {pair} is not compatible with your stake currency "	1✔
233	f"{self._config['stake_currency']}. Removing it from whitelist..",
234	logger.warning)
235	continue	1✔
236
237	# Check if market is active
238	market = markets[pair]	1✔
239	if not market_is_active(market):	1✔
240	self.log_once(f"Ignoring {pair} from whitelist. Market is not active.", logger.info)	1✔
241	continue	1✔
242	if pair not in sanitized_whitelist:	1✔
243	sanitized_whitelist.append(pair)	1✔
244
245	# We need to remove pairs that are unknown
246	return sanitized_whitelist	1✔

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous