freqtrade_origin/freqtrade/resolvers/exchange_resolver.py

99 lines
3.6 KiB
Python

"""
This module loads custom exchanges
"""
import logging
from inspect import isclass
from typing import Any, Dict, List, Optional
import freqtrade.exchange as exchanges
from freqtrade.constants import Config, ExchangeConfig
from freqtrade.exchange import MAP_EXCHANGE_CHILDCLASS, Exchange
from freqtrade.resolvers import IResolver
logger = logging.getLogger(__name__)
class ExchangeResolver(IResolver):
"""
This class contains all the logic to load a custom exchange class
"""
object_type = Exchange
@staticmethod
def load_exchange(config: Config, *, exchange_config: Optional[ExchangeConfig] = None,
validate: bool = True, load_leverage_tiers: bool = False) -> Exchange:
"""
Load the custom class from config parameter
:param exchange_name: name of the Exchange to load
:param config: configuration dictionary
"""
exchange_name: str = config['exchange']['name']
# Map exchange name to avoid duplicate classes for identical exchanges
exchange_name = MAP_EXCHANGE_CHILDCLASS.get(exchange_name, exchange_name)
exchange_name = exchange_name.title()
exchange = None
try:
exchange = ExchangeResolver._load_exchange(
exchange_name,
kwargs={
'config': config,
'validate': validate,
'exchange_config': exchange_config,
'load_leverage_tiers': load_leverage_tiers}
)
except ImportError:
logger.info(
f"No {exchange_name} specific subclass found. Using the generic class instead.")
if not exchange:
exchange = Exchange(config, validate=validate, exchange_config=exchange_config,)
return exchange
@staticmethod
def _load_exchange(exchange_name: str, kwargs: dict) -> Exchange:
"""
Loads the specified exchange.
Only checks for exchanges exported in freqtrade.exchanges
:param exchange_name: name of the module to import
:return: Exchange instance or None
"""
try:
ex_class = getattr(exchanges, exchange_name)
exchange = ex_class(**kwargs)
if exchange:
logger.info(f"Using resolved exchange '{exchange_name}'...")
return exchange
except AttributeError:
# Pass and raise ImportError instead
pass
raise ImportError(
f"Impossible to load Exchange '{exchange_name}'. This class does not exist "
"or contains Python code errors."
)
@classmethod
def search_all_objects(cls, config: Config, enum_failed: bool,
recursive: bool = False) -> List[Dict[str, Any]]:
"""
Searches for valid objects
:param config: Config object
:param enum_failed: If True, will return None for modules which fail.
Otherwise, failing modules are skipped.
:param recursive: Recursively walk directory tree searching for strategies
:return: List of dicts containing 'name', 'class' and 'location' entries
"""
result = []
for exchange_name in dir(exchanges):
exchange = getattr(exchanges, exchange_name)
if isclass(exchange) and issubclass(exchange, Exchange):
result.append({
'name': exchange_name,
'class': exchange,
'location': exchange.__module__,
'location_rel: ': exchange.__module__.replace('freqtrade.', ''),
})
return result