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

/freqtrade/exchange/exchange_utils.py

"""
Exchange support utils
"""
from datetime import datetime, timedelta, timezone
from math import ceil, floor
from typing import Any, Dict, List, Optional, Tuple

import ccxt
from ccxt import (DECIMAL_PLACES, ROUND, ROUND_DOWN, ROUND_UP, SIGNIFICANT_DIGITS, TICK_SIZE,
                  TRUNCATE, decimal_to_precision)

from freqtrade.exchange.common import (BAD_EXCHANGES, EXCHANGE_HAS_OPTIONAL, EXCHANGE_HAS_REQUIRED,
                                       SUPPORTED_EXCHANGES)
from freqtrade.exchange.exchange_utils_timeframe import timeframe_to_minutes, timeframe_to_prev_date
from freqtrade.types import ValidExchangesType
from freqtrade.util import FtPrecise


CcxtModuleType = Any


def is_exchange_known_ccxt(
        exchange_name: str, ccxt_module: Optional[CcxtModuleType] = None) -> bool:
    return exchange_name in ccxt_exchanges(ccxt_module)


def ccxt_exchanges(ccxt_module: Optional[CcxtModuleType] = None) -> List[str]:
    """
    Return the list of all exchanges known to ccxt
    """
    return ccxt_module.exchanges if ccxt_module is not None else ccxt.exchanges


def available_exchanges(ccxt_module: Optional[CcxtModuleType] = None) -> List[str]:
    """
    Return exchanges available to the bot, i.e. non-bad exchanges in the ccxt list
    """
    exchanges = ccxt_exchanges(ccxt_module)
    return [x for x in exchanges if validate_exchange(x)[0]]


def validate_exchange(exchange: str) -> Tuple[bool, str]:
    """
    returns: can_use, reason
        with Reason including both missing and missing_opt
    """
    ex_mod = getattr(ccxt, exchange.lower())()
    result = True
    reason = ''
    if not ex_mod or not ex_mod.has:
        return False, ''
    missing = [
        k for k, v in EXCHANGE_HAS_REQUIRED.items()
        if ex_mod.has.get(k) is not True
        and not (all(ex_mod.has.get(x) for x in v))
    ]
    if missing:
        result = False
        reason += f"missing: {', '.join(missing)}"

    missing_opt = [k for k in EXCHANGE_HAS_OPTIONAL if not ex_mod.has.get(k)]

    if exchange.lower() in BAD_EXCHANGES:
        result = False
        reason = BAD_EXCHANGES.get(exchange.lower(), '')

    if missing_opt:
        reason += f"{'. ' if reason else ''}missing opt: {', '.join(missing_opt)}. "

    return result, reason


def _build_exchange_list_entry(
        exchange_name: str, exchangeClasses: Dict[str, Any]) -> ValidExchangesType:
    valid, comment = validate_exchange(exchange_name)
    result: ValidExchangesType = {
        'name': exchange_name,
        'valid': valid,
        'supported': exchange_name.lower() in SUPPORTED_EXCHANGES,
        'comment': comment,
        'trade_modes': [{'trading_mode': 'spot', 'margin_mode': ''}],
    }
    if resolved := exchangeClasses.get(exchange_name.lower()):
        supported_modes = [{'trading_mode': 'spot', 'margin_mode': ''}] + [
            {'trading_mode': tm.value, 'margin_mode': mm.value}
            for tm, mm in resolved['class']._supported_trading_mode_margin_pairs
        ]
        result.update({
            'trade_modes': supported_modes,
        })

    return result


def list_available_exchanges(all_exchanges: bool) -> List[ValidExchangesType]:
    """
    :return: List of tuples with exchangename, valid, reason.
    """
    exchanges = ccxt_exchanges() if all_exchanges else available_exchanges()
    from freqtrade.resolvers.exchange_resolver import ExchangeResolver

    subclassed = {e['name'].lower(): e for e in ExchangeResolver.search_all_objects({}, False)}

    exchanges_valid: List[ValidExchangesType] = [
        _build_exchange_list_entry(e, subclassed) for e in exchanges
    ]

    return exchanges_valid


def date_minus_candles(
        timeframe: str, candle_count: int, date: Optional[datetime] = None) -> datetime:
    """
    subtract X candles from a date.
    :param timeframe: timeframe in string format (e.g. "5m")
    :param candle_count: Amount of candles to subtract.
    :param date: date to use. Defaults to now(utc)

    """
    if not date:
        date = datetime.now(timezone.utc)

    tf_min = timeframe_to_minutes(timeframe)
    new_date = timeframe_to_prev_date(timeframe, date) - timedelta(minutes=tf_min * candle_count)
    return new_date


def market_is_active(market: Dict) -> bool:
    """
    Return True if the market is active.
    """
    # "It's active, if the active flag isn't explicitly set to false. If it's missing or
    # true then it's true. If it's undefined, then it's most likely true, but not 100% )"
    # See https://github.com/ccxt/ccxt/issues/4874,
    # https://github.com/ccxt/ccxt/issues/4075#issuecomment-434760520
    return market.get('active', True) is not False


def amount_to_contracts(amount: float, contract_size: Optional[float]) -> float:
    """
    Convert amount to contracts.
    :param amount: amount to convert
    :param contract_size: contract size - taken from exchange.get_contract_size(pair)
    :return: num-contracts
    """
    if contract_size and contract_size != 1:
        return float(FtPrecise(amount) / FtPrecise(contract_size))
    else:
        return amount


def contracts_to_amount(num_contracts: float, contract_size: Optional[float]) -> float:
    """
    Takes num-contracts and converts it to contract size
    :param num_contracts: number of contracts
    :param contract_size: contract size - taken from exchange.get_contract_size(pair)
    :return: Amount
    """

    if contract_size and contract_size != 1:
        return float(FtPrecise(num_contracts) * FtPrecise(contract_size))
    else:
        return num_contracts


def amount_to_precision(amount: float, amount_precision: Optional[float],
                        precisionMode: Optional[int]) -> float:
    """
    Returns the amount to buy or sell to a precision the Exchange accepts
    Re-implementation of ccxt internal methods - ensuring we can test the result is correct
    based on our definitions.
    :param amount: amount to truncate
    :param amount_precision: amount precision to use.
                             should be retrieved from markets[pair]['precision']['amount']
    :param precisionMode: precision mode to use. Should be used from precisionMode
                          one of ccxt's DECIMAL_PLACES, SIGNIFICANT_DIGITS, or TICK_SIZE
    :return: truncated amount
    """
    if amount_precision is not None and precisionMode is not None:
        precision = int(amount_precision) if precisionMode != TICK_SIZE else amount_precision
        # precision must be an int for non-ticksize inputs.
        amount = float(decimal_to_precision(amount, rounding_mode=TRUNCATE,
                                            precision=precision,
                                            counting_mode=precisionMode,
                                            ))

    return amount


def amount_to_contract_precision(
        amount, amount_precision: Optional[float], precisionMode: Optional[int],
        contract_size: Optional[float]) -> float:
    """
    Returns the amount to buy or sell to a precision the Exchange accepts
    including calculation to and from contracts.
    Re-implementation of ccxt internal methods - ensuring we can test the result is correct
    based on our definitions.
    :param amount: amount to truncate
    :param amount_precision: amount precision to use.
                             should be retrieved from markets[pair]['precision']['amount']
    :param precisionMode: precision mode to use. Should be used from precisionMode
                          one of ccxt's DECIMAL_PLACES, SIGNIFICANT_DIGITS, or TICK_SIZE
    :param contract_size: contract size - taken from exchange.get_contract_size(pair)
    :return: truncated amount
    """
    if amount_precision is not None and precisionMode is not None:
        contracts = amount_to_contracts(amount, contract_size)
        amount_p = amount_to_precision(contracts, amount_precision, precisionMode)
        return contracts_to_amount(amount_p, contract_size)
    return amount


def __price_to_precision_significant_digits(
    price: float,
    price_precision: float,
    *,
    rounding_mode: int = ROUND,
) -> float:
    """
    Implementation of ROUND_UP/Round_down for significant digits mode.
    """
    from decimal import ROUND_DOWN as dec_ROUND_DOWN
    from decimal import ROUND_UP as dec_ROUND_UP
    from decimal import Decimal
    dec = Decimal(str(price))
    string = f'{dec:f}'
    precision = round(price_precision)

    q = precision - dec.adjusted() - 1
    sigfig = Decimal('10') ** -q
    if q < 0:
        string_to_precision = string[:precision]
        # string_to_precision is '' when we have zero precision
        below = sigfig * Decimal(string_to_precision if string_to_precision else '0')
        above = below + sigfig
        res = above if rounding_mode == ROUND_UP else below
        precise = f'{res:f}'
    else:
        precise = '{:f}'.format(dec.quantize(
            sigfig,
            rounding=dec_ROUND_DOWN if rounding_mode == ROUND_DOWN else dec_ROUND_UP)
        )
    return float(precise)


def price_to_precision(
    price: float,
    price_precision: Optional[float],
    precisionMode: Optional[int],
    *,
    rounding_mode: int = ROUND,
) -> float:
    """
    Returns the price rounded to the precision the Exchange accepts.
    Partial Re-implementation of ccxt internal method decimal_to_precision(),
    which does not support rounding up.
    For stoploss calculations, must use ROUND_UP for longs, and ROUND_DOWN for shorts.

    TODO: If ccxt supports ROUND_UP for decimal_to_precision(), we could remove this and
    align with amount_to_precision().
    :param price: price to convert
    :param price_precision: price precision to use. Used from markets[pair]['precision']['price']
    :param precisionMode: precision mode to use. Should be used from precisionMode
                          one of ccxt's DECIMAL_PLACES, SIGNIFICANT_DIGITS, or TICK_SIZE
    :param rounding_mode: rounding mode to use. Defaults to ROUND
    :return: price rounded up to the precision the Exchange accepts
    """
    if price_precision is not None and precisionMode is not None:
        if rounding_mode not in (ROUND_UP, ROUND_DOWN):
            # Use CCXT code where possible.
            return float(decimal_to_precision(price, rounding_mode=rounding_mode,
                                              precision=price_precision,
                                              counting_mode=precisionMode
                                              ))

        if precisionMode == TICK_SIZE:
            precision = FtPrecise(price_precision)
            price_str = FtPrecise(price)
            missing = price_str % precision
            if not missing == FtPrecise("0"):
                if rounding_mode == ROUND_UP:
                    res = price_str - missing + precision
                elif rounding_mode == ROUND_DOWN:
                    res = price_str - missing
                return round(float(str(res)), 14)
            return price
        elif precisionMode == DECIMAL_PLACES:

            ndigits = round(price_precision)
            ticks = price * (10**ndigits)
            if rounding_mode == ROUND_UP:
                return ceil(ticks) / (10**ndigits)
            if rounding_mode == ROUND_DOWN:
                return floor(ticks) / (10**ndigits)

            raise ValueError(f"Unknown rounding_mode {rounding_mode}")
        elif precisionMode == SIGNIFICANT_DIGITS:
            if rounding_mode in (ROUND_UP, ROUND_DOWN):
                return __price_to_precision_significant_digits(
                    price, price_precision, rounding_mode=rounding_mode
                )

        raise ValueError(f"Unknown precisionMode {precisionMode}")
    return price

1	"""
2	Exchange support utils
3	"""
4	from datetime import datetime, timedelta, timezone	1✔
5	from math import ceil, floor	1✔
6	from typing import Any, Dict, List, Optional, Tuple	1✔
7
8	import ccxt	1✔
9	from ccxt import (DECIMAL_PLACES, ROUND, ROUND_DOWN, ROUND_UP, SIGNIFICANT_DIGITS, TICK_SIZE,	1✔
10	TRUNCATE, decimal_to_precision)
11
12	from freqtrade.exchange.common import (BAD_EXCHANGES, EXCHANGE_HAS_OPTIONAL, EXCHANGE_HAS_REQUIRED,	1✔
13	SUPPORTED_EXCHANGES)
14	from freqtrade.exchange.exchange_utils_timeframe import timeframe_to_minutes, timeframe_to_prev_date	1✔
15	from freqtrade.types import ValidExchangesType	1✔
16	from freqtrade.util import FtPrecise	1✔
17
18
19	CcxtModuleType = Any	1✔
20
21
22	def is_exchange_known_ccxt(	1✔
23	exchange_name: str, ccxt_module: Optional[CcxtModuleType] = None) -> bool:
24	return exchange_name in ccxt_exchanges(ccxt_module)	1✔
25
26
27	def ccxt_exchanges(ccxt_module: Optional[CcxtModuleType] = None) -> List[str]:	1✔
28	"""
29	Return the list of all exchanges known to ccxt
30	"""
31	return ccxt_module.exchanges if ccxt_module is not None else ccxt.exchanges	1✔
32
33
34	def available_exchanges(ccxt_module: Optional[CcxtModuleType] = None) -> List[str]:	1✔
35	"""
36	Return exchanges available to the bot, i.e. non-bad exchanges in the ccxt list
37	"""
38	exchanges = ccxt_exchanges(ccxt_module)	1✔
39	return [x for x in exchanges if validate_exchange(x)[0]]	1✔
40
41
42	def validate_exchange(exchange: str) -> Tuple[bool, str]:	1✔
43	"""
44	returns: can_use, reason
45	with Reason including both missing and missing_opt
46	"""
47	ex_mod = getattr(ccxt, exchange.lower())()	1✔
48	result = True	1✔
49	reason = ''	1✔
50	if not ex_mod or not ex_mod.has:	1✔
51	return False, ''	×
52	missing = [	1✔
53	k for k, v in EXCHANGE_HAS_REQUIRED.items()
54	if ex_mod.has.get(k) is not True
55	and not (all(ex_mod.has.get(x) for x in v))
56	]
57	if missing:	1✔
58	result = False	1✔
59	reason += f"missing: {', '.join(missing)}"	1✔
60
61	missing_opt = [k for k in EXCHANGE_HAS_OPTIONAL if not ex_mod.has.get(k)]	1✔
62
63	if exchange.lower() in BAD_EXCHANGES:	1✔
64	result = False	1✔
65	reason = BAD_EXCHANGES.get(exchange.lower(), '')	1✔
66
67	if missing_opt:	1✔
68	reason += f"{'. ' if reason else ''}missing opt: {', '.join(missing_opt)}. "	1✔
69
70	return result, reason	1✔
71
72
73	def _build_exchange_list_entry(	1✔
74	exchange_name: str, exchangeClasses: Dict[str, Any]) -> ValidExchangesType:
75	valid, comment = validate_exchange(exchange_name)	1✔
76	result: ValidExchangesType = {	1✔
77	'name': exchange_name,
78	'valid': valid,
79	'supported': exchange_name.lower() in SUPPORTED_EXCHANGES,
80	'comment': comment,
81	'trade_modes': [{'trading_mode': 'spot', 'margin_mode': ''}],
82	}
83	if resolved := exchangeClasses.get(exchange_name.lower()):	1✔
84	supported_modes = [{'trading_mode': 'spot', 'margin_mode': ''}] + [	1✔
85	{'trading_mode': tm.value, 'margin_mode': mm.value}
86	for tm, mm in resolved['class']._supported_trading_mode_margin_pairs
87	]
88	result.update({	1✔
89	'trade_modes': supported_modes,
90	})
91
92	return result	1✔
93
94
95	def list_available_exchanges(all_exchanges: bool) -> List[ValidExchangesType]:	1✔
96	"""
97	:return: List of tuples with exchangename, valid, reason.
98	"""
99	exchanges = ccxt_exchanges() if all_exchanges else available_exchanges()	1✔
100	from freqtrade.resolvers.exchange_resolver import ExchangeResolver	1✔
101
102	subclassed = {e['name'].lower(): e for e in ExchangeResolver.search_all_objects({}, False)}	1✔
103
104	exchanges_valid: List[ValidExchangesType] = [	1✔
105	_build_exchange_list_entry(e, subclassed) for e in exchanges
106	]
107
108	return exchanges_valid	1✔
109
110
111	def date_minus_candles(	1✔
112	timeframe: str, candle_count: int, date: Optional[datetime] = None) -> datetime:
113	"""
114	subtract X candles from a date.
115	:param timeframe: timeframe in string format (e.g. "5m")
116	:param candle_count: Amount of candles to subtract.
117	:param date: date to use. Defaults to now(utc)
118
119	"""
120	if not date:	1✔
121	date = datetime.now(timezone.utc)	1✔
122
123	tf_min = timeframe_to_minutes(timeframe)	1✔
124	new_date = timeframe_to_prev_date(timeframe, date) - timedelta(minutes=tf_min * candle_count)	1✔
125	return new_date	1✔
126
127
128	def market_is_active(market: Dict) -> bool:	1✔
129	"""
130	Return True if the market is active.
131	"""
132	# "It's active, if the active flag isn't explicitly set to false. If it's missing or
133	# true then it's true. If it's undefined, then it's most likely true, but not 100% )"
134	# See https://github.com/ccxt/ccxt/issues/4874,
135	# https://github.com/ccxt/ccxt/issues/4075#issuecomment-434760520
136	return market.get('active', True) is not False	1✔
137
138
139	def amount_to_contracts(amount: float, contract_size: Optional[float]) -> float:	1✔
140	"""
141	Convert amount to contracts.
142	:param amount: amount to convert
143	:param contract_size: contract size - taken from exchange.get_contract_size(pair)
144	:return: num-contracts
145	"""
146	if contract_size and contract_size != 1:	1✔
147	return float(FtPrecise(amount) / FtPrecise(contract_size))	1✔
148	else:
149	return amount	1✔
150
151
152	def contracts_to_amount(num_contracts: float, contract_size: Optional[float]) -> float:	1✔
153	"""
154	Takes num-contracts and converts it to contract size
155	:param num_contracts: number of contracts
156	:param contract_size: contract size - taken from exchange.get_contract_size(pair)
157	:return: Amount
158	"""
159
160	if contract_size and contract_size != 1:	1✔
161	return float(FtPrecise(num_contracts) * FtPrecise(contract_size))	1✔
162	else:
163	return num_contracts	1✔
164
165
166	def amount_to_precision(amount: float, amount_precision: Optional[float],	1✔
167	precisionMode: Optional[int]) -> float:
168	"""
169	Returns the amount to buy or sell to a precision the Exchange accepts
170	Re-implementation of ccxt internal methods - ensuring we can test the result is correct
171	based on our definitions.
172	:param amount: amount to truncate
173	:param amount_precision: amount precision to use.
174	should be retrieved from markets[pair]['precision']['amount']
175	:param precisionMode: precision mode to use. Should be used from precisionMode
176	one of ccxt's DECIMAL_PLACES, SIGNIFICANT_DIGITS, or TICK_SIZE
177	:return: truncated amount
178	"""
179	if amount_precision is not None and precisionMode is not None:	1✔
180	precision = int(amount_precision) if precisionMode != TICK_SIZE else amount_precision	1✔
181	# precision must be an int for non-ticksize inputs.
182	amount = float(decimal_to_precision(amount, rounding_mode=TRUNCATE,	1✔
183	precision=precision,
184	counting_mode=precisionMode,
185	))
186
187	return amount	1✔
188
189
190	def amount_to_contract_precision(	1✔
191	amount, amount_precision: Optional[float], precisionMode: Optional[int],
192	contract_size: Optional[float]) -> float:
193	"""
194	Returns the amount to buy or sell to a precision the Exchange accepts
195	including calculation to and from contracts.
196	Re-implementation of ccxt internal methods - ensuring we can test the result is correct
197	based on our definitions.
198	:param amount: amount to truncate
199	:param amount_precision: amount precision to use.
200	should be retrieved from markets[pair]['precision']['amount']
201	:param precisionMode: precision mode to use. Should be used from precisionMode
202	one of ccxt's DECIMAL_PLACES, SIGNIFICANT_DIGITS, or TICK_SIZE
203	:param contract_size: contract size - taken from exchange.get_contract_size(pair)
204	:return: truncated amount
205	"""
206	if amount_precision is not None and precisionMode is not None:	1✔
207	contracts = amount_to_contracts(amount, contract_size)	1✔
208	amount_p = amount_to_precision(contracts, amount_precision, precisionMode)	1✔
209	return contracts_to_amount(amount_p, contract_size)	1✔
210	return amount	1✔
211
212
213	def __price_to_precision_significant_digits(	1✔
214	price: float,
215	price_precision: float,
216	*,
217	rounding_mode: int = ROUND,
218	) -> float:
219	"""
220	Implementation of ROUND_UP/Round_down for significant digits mode.
221	"""
222	from decimal import ROUND_DOWN as dec_ROUND_DOWN	1✔
223	from decimal import ROUND_UP as dec_ROUND_UP	1✔
224	from decimal import Decimal	1✔
225	dec = Decimal(str(price))	1✔
226	string = f'{dec:f}'	1✔
227	precision = round(price_precision)	1✔
228
229	q = precision - dec.adjusted() - 1	1✔
230	sigfig = Decimal('10') ** -q	1✔
231	if q < 0:	1✔
232	string_to_precision = string[:precision]	1✔
233	# string_to_precision is '' when we have zero precision
234	below = sigfig * Decimal(string_to_precision if string_to_precision else '0')	1✔
235	above = below + sigfig	1✔
236	res = above if rounding_mode == ROUND_UP else below	1✔
237	precise = f'{res:f}'	1✔
238	else:
239	precise = '{:f}'.format(dec.quantize(	1✔
240	sigfig,
241	rounding=dec_ROUND_DOWN if rounding_mode == ROUND_DOWN else dec_ROUND_UP)
242	)
243	return float(precise)	1✔
244
245
246	def price_to_precision(	1✔
247	price: float,
248	price_precision: Optional[float],
249	precisionMode: Optional[int],
250	*,
251	rounding_mode: int = ROUND,
252	) -> float:
253	"""
254	Returns the price rounded to the precision the Exchange accepts.
255	Partial Re-implementation of ccxt internal method decimal_to_precision(),
256	which does not support rounding up.
257	For stoploss calculations, must use ROUND_UP for longs, and ROUND_DOWN for shorts.
258
259	TODO: If ccxt supports ROUND_UP for decimal_to_precision(), we could remove this and
260	align with amount_to_precision().
261	:param price: price to convert
262	:param price_precision: price precision to use. Used from markets[pair]['precision']['price']
263	:param precisionMode: precision mode to use. Should be used from precisionMode
264	one of ccxt's DECIMAL_PLACES, SIGNIFICANT_DIGITS, or TICK_SIZE
265	:param rounding_mode: rounding mode to use. Defaults to ROUND
266	:return: price rounded up to the precision the Exchange accepts
267	"""
268	if price_precision is not None and precisionMode is not None:	1✔
269	if rounding_mode not in (ROUND_UP, ROUND_DOWN):	1✔
270	# Use CCXT code where possible.
271	return float(decimal_to_precision(price, rounding_mode=rounding_mode,	1✔
272	precision=price_precision,
273	counting_mode=precisionMode
274	))
275
276	if precisionMode == TICK_SIZE:	1✔
277	precision = FtPrecise(price_precision)	1✔
278	price_str = FtPrecise(price)	1✔
279	missing = price_str % precision	1✔
280	if not missing == FtPrecise("0"):	1✔
281	if rounding_mode == ROUND_UP:	1✔
282	res = price_str - missing + precision	1✔
283	elif rounding_mode == ROUND_DOWN:	1✔
284	res = price_str - missing	1✔
285	return round(float(str(res)), 14)	1✔
286	return price	1✔
287	elif precisionMode == DECIMAL_PLACES:	1✔
288
289	ndigits = round(price_precision)	1✔
290	ticks = price * (10**ndigits)	1✔
291	if rounding_mode == ROUND_UP:	1✔
292	return ceil(ticks) / (10**ndigits)	1✔
293	if rounding_mode == ROUND_DOWN:	1✔
294	return floor(ticks) / (10**ndigits)	1✔
295
296	raise ValueError(f"Unknown rounding_mode {rounding_mode}")	×
297	elif precisionMode == SIGNIFICANT_DIGITS:	1✔
298	if rounding_mode in (ROUND_UP, ROUND_DOWN):	1✔
299	return __price_to_precision_significant_digits(	1✔
300	price, price_precision, rounding_mode=rounding_mode
301	)
302
303	raise ValueError(f"Unknown precisionMode {precisionMode}")	×
304	return price	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