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.37

/freqtrade/plugins/pairlist/VolatilityFilter.py

"""
Volatility pairlist filter
"""
import logging
import sys
from datetime import timedelta
from typing import Any, Dict, List, Optional

import numpy as np
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 VolatilityFilter(IPairList):
    """
    Filters pairs by volatility
    """

    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_volatility = pairlistconfig.get('min_volatility', 0)
        self._max_volatility = pairlistconfig.get('max_volatility', sys.maxsize)
        self._refresh_period = pairlistconfig.get('refresh_period', 1440)
        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("VolatilityFilter requires lookback_days to be >= 1")
        if self._days > candle_limit:
            raise OperationalException("VolatilityFilter requires lookback_days to not "
                                       f"exceed exchange max request size ({candle_limit})")
        if self._sort_direction not in [None, 'asc', 'desc']:
            raise OperationalException("VolatilityFilter 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
        """
        return (f"{self.name} - Filtering pairs with volatility range "
                f"{self._min_volatility}-{self._max_volatility} "
                f" the last {self._days} {plural(self._days, 'day')}.")

    @staticmethod
    def description() -> str:
        return "Filter pairs by their recent volatility."

    @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_volatility": {
                "type": "number",
                "default": 0,
                "description": "Minimum Volatility",
                "help": "Minimum volatility a pair must have to be considered.",
            },
            "max_volatility": {
                "type": "number",
                "default": None,
                "description": "Maximum Volatility",
                "help": "Maximum volatility a pair must have to be considered.",
            },
            "sort_direction": {
                "type": "option",
                "default": None,
                "options": ["", "asc", "desc"],
                "description": "Sort pairlist",
                "help": "Sort Pairlist ascending or descending by volatility.",
            },
            **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))
        candles = self._exchange.refresh_ohlcv_with_cache(needed_pairs, since_ms=since_ms)

        resulting_pairlist: List[str] = []
        volatilitys: Dict[str, float] = {}
        for p in pairlist:
            daily_candles = candles.get((p, '1d', self._def_candletype), None)

            volatility_avg = self._calculate_volatility(p, daily_candles)

            if volatility_avg is not None:
                if self._validate_pair_loc(p, volatility_avg):
                    resulting_pairlist.append(p)
                    volatilitys[p] = (
                        volatility_avg if volatility_avg and not np.isnan(volatility_avg) else 0
                    )
            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: volatilitys[p],
                                        reverse=self._sort_direction == 'desc')
        return resulting_pairlist

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

        if daily_candles is not None and not daily_candles.empty:
            returns = (np.log(daily_candles["close"].shift(1) / daily_candles["close"]))
            returns.fillna(0, inplace=True)

            volatility_series = returns.rolling(window=self._days).std() * np.sqrt(self._days)
            volatility_avg = volatility_series.mean()
            self._pair_cache[pair] = volatility_avg

            return volatility_avg
        else:
            return None

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

        if self._min_volatility <= volatility_avg <= self._max_volatility:
            result = True
        else:
            self.log_once(f"Removed {pair} from whitelist, because volatility "
                          f"over {self._days} {plural(self._days, 'day')} "
                          f"is: {volatility_avg:.3f} "
                          f"which is not in the configured range of "
                          f"{self._min_volatility}-{self._max_volatility}.",
                          logger.info)
            result = False
        return result

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