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

97.5

/freqtrade/plugins/pairlist/rangestabilityfilter.py

"""
Rate of change pairlist filter
"""
import logging
from datetime import timedelta
from typing import Any, Dict, List, Optional

from cachetools import TTLCache
from pandas import DataFrame

from freqtrade.constants import Config, ListPairsWithTimeframes
from freqtrade.exceptions import OperationalException
from freqtrade.exchange.types import Tickers
from freqtrade.misc import plural
from freqtrade.plugins.pairlist.IPairList import IPairList, PairlistParameter
from freqtrade.util import dt_floor_day, dt_now, dt_ts


logger = logging.getLogger(__name__)


class RangeStabilityFilter(IPairList):

    def __init__(self, exchange, pairlistmanager,
                 config: Config, pairlistconfig: Dict[str, Any],
                 pairlist_pos: int) -> None:
        super().__init__(exchange, pairlistmanager, config, pairlistconfig, pairlist_pos)

        self._days = pairlistconfig.get('lookback_days', 10)
        self._min_rate_of_change = pairlistconfig.get('min_rate_of_change', 0.01)
        self._max_rate_of_change = pairlistconfig.get('max_rate_of_change')
        self._refresh_period = pairlistconfig.get('refresh_period', 86400)
        self._def_candletype = self._config['candle_type_def']
        self._sort_direction: Optional[str] = pairlistconfig.get('sort_direction', None)

        self._pair_cache: TTLCache = TTLCache(maxsize=1000, ttl=self._refresh_period)

        candle_limit = exchange.ohlcv_candle_limit('1d', self._config['candle_type_def'])
        if self._days < 1:
            raise OperationalException("RangeStabilityFilter requires lookback_days to be >= 1")
        if self._days > candle_limit:
            raise OperationalException("RangeStabilityFilter requires lookback_days to not "
                                       f"exceed exchange max request size ({candle_limit})")
        if self._sort_direction not in [None, 'asc', 'desc']:
            raise OperationalException("RangeStabilityFilter requires sort_direction to be "
                                       "either None (undefined), 'asc' or 'desc'")

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

    def short_desc(self) -> str:
        """
        Short whitelist method description - used for startup-messages
        """
        max_rate_desc = ""
        if self._max_rate_of_change:
            max_rate_desc = (f" and above {self._max_rate_of_change}")
        return (f"{self.name} - Filtering pairs with rate of change below "
                f"{self._min_rate_of_change}{max_rate_desc} over the "
                f"last {plural(self._days, 'day')}.")

    @staticmethod
    def description() -> str:
        return "Filters pairs by their rate of change."

    @staticmethod
    def available_parameters() -> Dict[str, PairlistParameter]:
        return {
            "lookback_days": {
                "type": "number",
                "default": 10,
                "description": "Lookback Days",
                "help": "Number of days to look back at.",
            },
            "min_rate_of_change": {
                "type": "number",
                "default": 0.01,
                "description": "Minimum Rate of Change",
                "help": "Minimum rate of change to filter pairs.",
            },
            "max_rate_of_change": {
                "type": "number",
                "default": None,
                "description": "Maximum Rate of Change",
                "help": "Maximum rate of change to filter pairs.",
            },
            "sort_direction": {
                "type": "option",
                "default": None,
                "options": ["", "asc", "desc"],
                "description": "Sort pairlist",
                "help": "Sort Pairlist ascending or descending by rate of change.",
            },
            **IPairList.refresh_period_parameter()
        }

    def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]:
        """
        Validate trading range
        :param pairlist: pairlist to filter or sort
        :param tickers: Tickers (from exchange.get_tickers). May be cached.
        :return: new allowlist
        """
        needed_pairs: ListPairsWithTimeframes = [
            (p, '1d', self._def_candletype) for p in pairlist if p not in self._pair_cache]

        since_ms = dt_ts(dt_floor_day(dt_now()) - timedelta(days=self._days + 1))
        candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms=since_ms)

        resulting_pairlist: List[str] = []
        pct_changes: Dict[str, float] = {}

        for p in pairlist:
            daily_candles = candles.get((p, '1d', self._def_candletype), None)

            pct_change = self._calculate_rate_of_change(p, daily_candles)

            if pct_change is not None:
                if self._validate_pair_loc(p, pct_change):
                    resulting_pairlist.append(p)
                    pct_changes[p] = pct_change
            else:
                self.log_once(f"Removed {p} from whitelist, no candles found.", logger.info)

        if self._sort_direction:
            resulting_pairlist = sorted(resulting_pairlist,
                                        key=lambda p: pct_changes[p],
                                        reverse=self._sort_direction == 'desc')
        return resulting_pairlist

    def _calculate_rate_of_change(self, pair: str, daily_candles: DataFrame) -> Optional[float]:
        # Check symbol in cache
        if (pct_change := self._pair_cache.get(pair, None)) is not None:
            return pct_change
        if daily_candles is not None and not daily_candles.empty:

            highest_high = daily_candles['high'].max()
            lowest_low = daily_candles['low'].min()
            pct_change = ((highest_high - lowest_low) / lowest_low) if lowest_low > 0 else 0
            self._pair_cache[pair] = pct_change
            return pct_change
        else:
            return None

    def _validate_pair_loc(self, pair: str, pct_change: float) -> bool:
        """
        Validate trading range
        :param pair: Pair that's currently validated
        :param pct_change: Rate of change
        :return: True if the pair can stay, false if it should be removed
        """

        result = True
        if pct_change < self._min_rate_of_change:
            self.log_once(f"Removed {pair} from whitelist, because rate of change "
                          f"over {self._days} {plural(self._days, 'day')} is {pct_change:.3f}, "
                          f"which is below the threshold of {self._min_rate_of_change}.",
                          logger.info)
            result = False
        if self._max_rate_of_change:
            if pct_change > self._max_rate_of_change:
                self.log_once(
                    f"Removed {pair} from whitelist, because rate of change "
                    f"over {self._days} {plural(self._days, 'day')} is {pct_change:.3f}, "
                    f"which is above the threshold of {self._max_rate_of_change}.",
                    logger.info)
                result = False
        return result

1	"""
2	Rate of change pairlist filter
3	"""
4	import logging	1✔
5	from datetime import timedelta	1✔
6	from typing import Any, Dict, List, Optional	1✔
7
8	from cachetools import TTLCache	1✔
9	from pandas import DataFrame	1✔
10
11	from freqtrade.constants import Config, ListPairsWithTimeframes	1✔
12	from freqtrade.exceptions import OperationalException	1✔
13	from freqtrade.exchange.types import Tickers	1✔
14	from freqtrade.misc import plural	1✔
15	from freqtrade.plugins.pairlist.IPairList import IPairList, PairlistParameter	1✔
16	from freqtrade.util import dt_floor_day, dt_now, dt_ts	1✔
17
18
19	logger = logging.getLogger(__name__)	1✔
20
21
22	class RangeStabilityFilter(IPairList):	1✔
23
24	def __init__(self, exchange, pairlistmanager,	1✔
25	config: Config, pairlistconfig: Dict[str, Any],
26	pairlist_pos: int) -> None:
27	super().__init__(exchange, pairlistmanager, config, pairlistconfig, pairlist_pos)	1✔
28
29	self._days = pairlistconfig.get('lookback_days', 10)	1✔
30	self._min_rate_of_change = pairlistconfig.get('min_rate_of_change', 0.01)	1✔
31	self._max_rate_of_change = pairlistconfig.get('max_rate_of_change')	1✔
32	self._refresh_period = pairlistconfig.get('refresh_period', 86400)	1✔
33	self._def_candletype = self._config['candle_type_def']	1✔
34	self._sort_direction: Optional[str] = pairlistconfig.get('sort_direction', None)	1✔
35
36	self._pair_cache: TTLCache = TTLCache(maxsize=1000, ttl=self._refresh_period)	1✔
37
38	candle_limit = exchange.ohlcv_candle_limit('1d', self._config['candle_type_def'])	1✔
39	if self._days < 1:	1✔
40	raise OperationalException("RangeStabilityFilter requires lookback_days to be >= 1")	1✔
41	if self._days > candle_limit:	1✔
42	raise OperationalException("RangeStabilityFilter requires lookback_days to not "	1✔
43	f"exceed exchange max request size ({candle_limit})")
44	if self._sort_direction not in [None, 'asc', 'desc']:	1✔
45	raise OperationalException("RangeStabilityFilter requires sort_direction to be "	1✔
46	"either None (undefined), 'asc' or 'desc'")
47
48	@property	1✔
49	def needstickers(self) -> bool:	1✔
50	"""
51	Boolean property defining if tickers are necessary.
52	If no Pairlist requires tickers, an empty List is passed
53	as tickers argument to filter_pairlist
54	"""
55	return False	1✔
56
57	def short_desc(self) -> str:	1✔
58	"""
59	Short whitelist method description - used for startup-messages
60	"""
61	max_rate_desc = ""	1✔
62	if self._max_rate_of_change:	1✔
63	max_rate_desc = (f" and above {self._max_rate_of_change}")	1✔
64	return (f"{self.name} - Filtering pairs with rate of change below "	1✔
65	f"{self._min_rate_of_change}{max_rate_desc} over the "
66	f"last {plural(self._days, 'day')}.")
67
68	@staticmethod	1✔
69	def description() -> str:	1✔
70	return "Filters pairs by their rate of change."	1✔
71
72	@staticmethod	1✔
73	def available_parameters() -> Dict[str, PairlistParameter]:	1✔
74	return {	1✔
75	"lookback_days": {
76	"type": "number",
77	"default": 10,
78	"description": "Lookback Days",
79	"help": "Number of days to look back at.",
80	},
81	"min_rate_of_change": {
82	"type": "number",
83	"default": 0.01,
84	"description": "Minimum Rate of Change",
85	"help": "Minimum rate of change to filter pairs.",
86	},
87	"max_rate_of_change": {
88	"type": "number",
89	"default": None,
90	"description": "Maximum Rate of Change",
91	"help": "Maximum rate of change to filter pairs.",
92	},
93	"sort_direction": {
94	"type": "option",
95	"default": None,
96	"options": ["", "asc", "desc"],
97	"description": "Sort pairlist",
98	"help": "Sort Pairlist ascending or descending by rate of change.",
99	},
100	**IPairList.refresh_period_parameter()
101	}
102
103	def filter_pairlist(self, pairlist: List[str], tickers: Tickers) -> List[str]:	1✔
104	"""
105	Validate trading range
106	:param pairlist: pairlist to filter or sort
107	:param tickers: Tickers (from exchange.get_tickers). May be cached.
108	:return: new allowlist
109	"""
110	needed_pairs: ListPairsWithTimeframes = [	1✔
111	(p, '1d', self._def_candletype) for p in pairlist if p not in self._pair_cache]
112
113	since_ms = dt_ts(dt_floor_day(dt_now()) - timedelta(days=self._days + 1))	1✔
114	candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms=since_ms)	1✔
115
116	resulting_pairlist: List[str] = []	1✔
117	pct_changes: Dict[str, float] = {}	1✔
118
119	for p in pairlist:	1✔
120	daily_candles = candles.get((p, '1d', self._def_candletype), None)	1✔
121
122	pct_change = self._calculate_rate_of_change(p, daily_candles)	1✔
123
124	if pct_change is not None:	1✔
125	if self._validate_pair_loc(p, pct_change):	1✔
126	resulting_pairlist.append(p)	1✔
127	pct_changes[p] = pct_change	1✔
128	else:
129	self.log_once(f"Removed {p} from whitelist, no candles found.", logger.info)	×
130
131	if self._sort_direction:	1✔
132	resulting_pairlist = sorted(resulting_pairlist,	1✔
133	key=lambda p: pct_changes[p],
134	reverse=self._sort_direction == 'desc')
135	return resulting_pairlist	1✔
136
137	def _calculate_rate_of_change(self, pair: str, daily_candles: DataFrame) -> Optional[float]:	1✔
138	# Check symbol in cache
139	if (pct_change := self._pair_cache.get(pair, None)) is not None:	1✔
140	return pct_change	1✔
141	if daily_candles is not None and not daily_candles.empty:	1✔
142
143	highest_high = daily_candles['high'].max()	1✔
144	lowest_low = daily_candles['low'].min()	1✔
145	pct_change = ((highest_high - lowest_low) / lowest_low) if lowest_low > 0 else 0	1✔
146	self._pair_cache[pair] = pct_change	1✔
147	return pct_change	1✔
148	else:
149	return None	×
150
151	def _validate_pair_loc(self, pair: str, pct_change: float) -> bool:	1✔
152	"""
153	Validate trading range
154	:param pair: Pair that's currently validated
155	:param pct_change: Rate of change
156	:return: True if the pair can stay, false if it should be removed
157	"""
158
159	result = True	1✔
160	if pct_change < self._min_rate_of_change:	1✔
161	self.log_once(f"Removed {pair} from whitelist, because rate of change "	1✔
162	f"over {self._days} {plural(self._days, 'day')} is {pct_change:.3f}, "
163	f"which is below the threshold of {self._min_rate_of_change}.",
164	logger.info)
165	result = False	1✔
166	if self._max_rate_of_change:	1✔
167	if pct_change > self._max_rate_of_change:	1✔
168	self.log_once(	1✔
169	f"Removed {pair} from whitelist, because rate of change "
170	f"over {self._days} {plural(self._days, 'day')} is {pct_change:.3f}, "
171	f"which is above the threshold of {self._max_rate_of_change}.",
172	logger.info)
173	result = False	1✔
174	return result	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