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

98.57

/freqtrade/plugins/pairlist/MarketCapPairList.py

"""
Market Cap PairList provider

Provides dynamic pair list based on Market Cap
"""
import logging
from typing import Any, Dict, List

from cachetools import TTLCache
from pycoingecko import CoinGeckoAPI

from freqtrade.constants import Config
from freqtrade.exceptions import OperationalException
from freqtrade.exchange.types import Tickers
from freqtrade.plugins.pairlist.IPairList import IPairList, PairlistParameter


logger = logging.getLogger(__name__)


class MarketCapPairList(IPairList):

    is_pairlist_generator = True

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

        if 'number_assets' not in self._pairlistconfig:
            raise OperationalException(
                '`number_assets` not specified. Please check your configuration '
                'for "pairlist.config.number_assets"')

        self._stake_currency = config['stake_currency']
        self._number_assets = self._pairlistconfig['number_assets']
        self._max_rank = self._pairlistconfig.get('max_rank', 30)
        self._refresh_period = self._pairlistconfig.get('refresh_period', 86400)
        self._marketcap_cache: TTLCache = TTLCache(maxsize=1, ttl=self._refresh_period)
        self._def_candletype = self._config['candle_type_def']
        self._coingecko: CoinGeckoAPI = CoinGeckoAPI()

        if self._max_rank > 250:
            raise OperationalException(
                "This filter only support marketcap rank up to 250."
            )

    @property
    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

    def short_desc(self) -> str:
        """
        Short whitelist method description - used for startup-messages
        """
        num = self._number_assets
        rank = self._max_rank
        msg = f"{self.name} - {num} pairs placed within top {rank} market cap."
        return msg

    @staticmethod
    def description() -> str:
        return "Provides pair list based on CoinGecko's market cap rank."

    @staticmethod
    def available_parameters() -> Dict[str, PairlistParameter]:
        return {
            "number_assets": {
                "type": "number",
                "default": 30,
                "description": "Number of assets",
                "help": "Number of assets to use from the pairlist",
            },
            "max_rank": {
                "type": "number",
                "default": 30,
                "description": "Max rank of assets",
                "help": "Maximum rank of assets to use from the pairlist",
            },
            "refresh_period": {
                "type": "number",
                "default": 86400,
                "description": "Refresh period",
                "help": "Refresh period in seconds",
            }
        }

    def gen_pairlist(self, tickers: Tickers) -> List[str]:
        """
        Generate the pairlist
        :param tickers: Tickers (from exchange.get_tickers). May be cached.
        :return: List of pairs
        """
        # Generate dynamic whitelist
        # Must always run if this pairlist is the first in the list.
        pairlist = self._marketcap_cache.get('pairlist_mc')
        if pairlist:
            # Item found - no refresh necessary
            return pairlist.copy()
        else:
            # Use fresh pairlist
            # Check if pair quote currency equals to the stake currency.
            _pairlist = [k for k in self._exchange.get_markets(
                quote_currencies=[self._stake_currency],
                tradable_only=True, active_only=True).keys()]
            # No point in testing for blacklisted pairs...
            _pairlist = self.verify_blacklist(_pairlist, logger.info)

            pairlist = self.filter_pairlist(_pairlist, tickers)
            self._marketcap_cache['pairlist_mc'] = pairlist.copy()

        return pairlist

    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
        :param pairlist: pairlist to filter or sort
        :param tickers: Tickers (from exchange.get_tickers). May be cached.
        :return: new whitelist
        """
        marketcap_list = self._marketcap_cache.get('marketcap')

        if marketcap_list is None:
            data = self._coingecko.get_coins_markets(vs_currency='usd', order='market_cap_desc',
                                                     per_page='250', page='1', sparkline='false',
                                                     locale='en')
            if data:
                marketcap_list = [row['symbol'] for row in data]
                self._marketcap_cache['marketcap'] = marketcap_list

        if marketcap_list:
            filtered_pairlist = []

            market = self._config['trading_mode']
            pair_format = f"{self._stake_currency.upper()}"
            if (market == 'futures'):
                pair_format += f":{self._stake_currency.upper()}"

            top_marketcap = marketcap_list[:self._max_rank:]

            for mc_pair in top_marketcap:
                test_pair = f"{mc_pair.upper()}/{pair_format}"
                if test_pair in pairlist:
                    filtered_pairlist.append(test_pair)
                    if len(filtered_pairlist) == self._number_assets:
                        break

            if len(filtered_pairlist) > 0:
                return filtered_pairlist

        return pairlist

1	"""
2	Market Cap PairList provider
3
4	Provides dynamic pair list based on Market Cap
5	"""
6	import logging	1✔
7	from typing import Any, Dict, List	1✔
8
9	from cachetools import TTLCache	1✔
10	from pycoingecko import CoinGeckoAPI	1✔
11
12	from freqtrade.constants import Config	1✔
13	from freqtrade.exceptions import OperationalException	1✔
14	from freqtrade.exchange.types import Tickers	1✔
15	from freqtrade.plugins.pairlist.IPairList import IPairList, PairlistParameter	1✔
16
17
18	logger = logging.getLogger(__name__)	1✔
19
20
21	class MarketCapPairList(IPairList):	1✔
22
23	is_pairlist_generator = True	1✔
24
25	def __init__(self, exchange, pairlistmanager,	1✔
26	config: Config, pairlistconfig: Dict[str, Any],
27	pairlist_pos: int) -> None:
28	super().__init__(exchange, pairlistmanager, config, pairlistconfig, pairlist_pos)	1✔
29
30	if 'number_assets' not in self._pairlistconfig:	1✔
31	raise OperationalException(	1✔
32	'`number_assets` not specified. Please check your configuration '
33	'for "pairlist.config.number_assets"')
34
35	self._stake_currency = config['stake_currency']	1✔
36	self._number_assets = self._pairlistconfig['number_assets']	1✔
37	self._max_rank = self._pairlistconfig.get('max_rank', 30)	1✔
38	self._refresh_period = self._pairlistconfig.get('refresh_period', 86400)	1✔
39	self._marketcap_cache: TTLCache = TTLCache(maxsize=1, ttl=self._refresh_period)	1✔
40	self._def_candletype = self._config['candle_type_def']	1✔
41	self._coingecko: CoinGeckoAPI = CoinGeckoAPI()	1✔
42
43	if self._max_rank > 250:	1✔
44	raise OperationalException(	1✔
45	"This filter only support marketcap rank up to 250."
46	)
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 Dict 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	num = self._number_assets	1✔
62	rank = self._max_rank	1✔
63	msg = f"{self.name} - {num} pairs placed within top {rank} market cap."	1✔
64	return msg	1✔
65
66	@staticmethod	1✔
67	def description() -> str:	1✔
68	return "Provides pair list based on CoinGecko's market cap rank."	1✔
69
70	@staticmethod	1✔
71	def available_parameters() -> Dict[str, PairlistParameter]:	1✔
72	return {	1✔
73	"number_assets": {
74	"type": "number",
75	"default": 30,
76	"description": "Number of assets",
77	"help": "Number of assets to use from the pairlist",
78	},
79	"max_rank": {
80	"type": "number",
81	"default": 30,
82	"description": "Max rank of assets",
83	"help": "Maximum rank of assets to use from the pairlist",
84	},
85	"refresh_period": {
86	"type": "number",
87	"default": 86400,
88	"description": "Refresh period",
89	"help": "Refresh period in seconds",
90	}
91	}
92
93	def gen_pairlist(self, tickers: Tickers) -> List[str]:	1✔
94	"""
95	Generate the pairlist
96	:param tickers: Tickers (from exchange.get_tickers). May be cached.
97	:return: List of pairs
98	"""
99	# Generate dynamic whitelist
100	# Must always run if this pairlist is the first in the list.
101	pairlist = self._marketcap_cache.get('pairlist_mc')	1✔
102	if pairlist:	1✔
103	# Item found - no refresh necessary
104	return pairlist.copy()	1✔
105	else:
106	# Use fresh pairlist
107	# Check if pair quote currency equals to the stake currency.
108	_pairlist = [k for k in self._exchange.get_markets(	1✔
109	quote_currencies=[self._stake_currency],
110	tradable_only=True, active_only=True).keys()]
111	# No point in testing for blacklisted pairs...
112	_pairlist = self.verify_blacklist(_pairlist, logger.info)	1✔
113
114	pairlist = self.filter_pairlist(_pairlist, tickers)	1✔
115	self._marketcap_cache['pairlist_mc'] = pairlist.copy()	1✔
116
117	return pairlist	1✔
118
119	def filter_pairlist(self, pairlist: List[str], tickers: Dict) -> List[str]:	1✔
120	"""
121	Filters and sorts pairlist and returns the whitelist again.
122	Called on each bot iteration - please use internal caching if necessary
123	:param pairlist: pairlist to filter or sort
124	:param tickers: Tickers (from exchange.get_tickers). May be cached.
125	:return: new whitelist
126	"""
127	marketcap_list = self._marketcap_cache.get('marketcap')	1✔
128
129	if marketcap_list is None:	1✔
130	data = self._coingecko.get_coins_markets(vs_currency='usd', order='market_cap_desc',	1✔
131	per_page='250', page='1', sparkline='false',
132	locale='en')
133	if data:	1✔
134	marketcap_list = [row['symbol'] for row in data]	1✔
135	self._marketcap_cache['marketcap'] = marketcap_list	1✔
136
137	if marketcap_list:	1✔
138	filtered_pairlist = []	1✔
139
140	market = self._config['trading_mode']	1✔
141	pair_format = f"{self._stake_currency.upper()}"	1✔
142	if (market == 'futures'):	1✔
143	pair_format += f":{self._stake_currency.upper()}"	1✔
144
145	top_marketcap = marketcap_list[:self._max_rank:]	1✔
146
147	for mc_pair in top_marketcap:	1✔
148	test_pair = f"{mc_pair.upper()}/{pair_format}"	1✔
149	if test_pair in pairlist:	1✔
150	filtered_pairlist.append(test_pair)	1✔
151	if len(filtered_pairlist) == self._number_assets:	1✔
152	break	1✔
153
154	if len(filtered_pairlist) > 0:	1✔
155	return filtered_pairlist	1✔
156
157	return pairlist	×

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