stable/freqtrade/configuration.py

"""
This module contains the configuration class
"""

import json
import logging
from argparse import Namespace
from typing import Dict, Any
from jsonschema import Draft4Validator, validate
from jsonschema.exceptions import ValidationError, best_match
import ccxt

from freqtrade import OperationalException, constants


logger = logging.getLogger(__name__)


class Configuration(object):
    """
    Class to read and init the bot configuration
    Reuse this class for the bot, backtesting, hyperopt and every script that required configuration
    """
    def __init__(self, args: Namespace) -> None:
        self.args = args
        self.config = None

    def load_config(self) -> Dict[str, Any]:
        """
        Extract information for sys.argv and load the bot configuration
        :return: Configuration dictionary
        """
        logger.info('Using config: %s ...', self.args.config)
        config = self._load_config_file(self.args.config)

        # Set strategy if not specified in config and or if it's non default
        if self.args.strategy != constants.DEFAULT_STRATEGY or not config.get('strategy'):
            config.update({'strategy': self.args.strategy})

        if self.args.strategy_path:
            config.update({'strategy_path': self.args.strategy_path})

        # Load Common configuration
        config = self._load_common_config(config)

        # Load Backtesting
        config = self._load_backtesting_config(config)

        # Load Hyperopt
        config = self._load_hyperopt_config(config)

        return config

    def _load_config_file(self, path: str) -> Dict[str, Any]:
        """
        Loads a config file from the given path
        :param path: path as str
        :return: configuration as dictionary
        """
        try:
            with open(path) as file:
                conf = json.load(file)
        except FileNotFoundError:
            logger.critical(
                'Config file "%s" not found. Please create your config file',
                path
            )
            exit(0)

        if 'internals' not in conf:
            conf['internals'] = {}
        logger.info('Validating configuration ...')

        return self._validate_config(conf)

    def _load_common_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Extract information for sys.argv and load common configuration
        :return: configuration as dictionary
        """

        # Log level
        if 'loglevel' in self.args and self.args.loglevel:
            config.update({'loglevel': self.args.loglevel})
            logging.basicConfig(
                level=config['loglevel'],
                format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            )
            logger.info('Log level set to %s', logging.getLevelName(config['loglevel']))

        # Add dynamic_whitelist if found
        if 'dynamic_whitelist' in self.args and self.args.dynamic_whitelist:
            config.update({'dynamic_whitelist': self.args.dynamic_whitelist})
            logger.info(
                'Parameter --dynamic-whitelist detected. '
                'Using dynamically generated whitelist. '
                '(not applicable with Backtesting and Hyperopt)'
            )

        # Add dry_run_db if found and the bot in dry run
        if self.args.dry_run_db and config.get('dry_run', False):
            config.update({'dry_run_db': True})
            logger.info('Parameter --dry-run-db detected ...')

        if config.get('dry_run_db', False):
            if config.get('dry_run', False):
                logger.info('Dry_run will use the DB file: "tradesv3.dry_run.sqlite"')
            else:
                logger.info('Dry run is disabled. (--dry_run_db ignored)')

        # Check if the exchange set by the user is supported
        self.check_exchange(config)

        return config

    def _load_backtesting_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Extract information for sys.argv and load Backtesting configuration
        :return: configuration as dictionary
        """

        # If -i/--ticker-interval is used we override the configuration parameter
        # (that will override the strategy configuration)
        if 'ticker_interval' in self.args and self.args.ticker_interval:
            config.update({'ticker_interval': self.args.ticker_interval})
            logger.info('Parameter -i/--ticker-interval detected ...')
            logger.info('Using ticker_interval: %s ...', config.get('ticker_interval'))

        # If -l/--live is used we add it to the configuration
        if 'live' in self.args and self.args.live:
            config.update({'live': True})
            logger.info('Parameter -l/--live detected ...')

        # If --realistic-simulation is used we add it to the configuration
        if 'realistic_simulation' in self.args and self.args.realistic_simulation:
            config.update({'realistic_simulation': True})
            logger.info('Parameter --realistic-simulation detected ...')
        logger.info('Using max_open_trades: %s ...', config.get('max_open_trades'))

        # If --timerange is used we add it to the configuration
        if 'timerange' in self.args and self.args.timerange:
            config.update({'timerange': self.args.timerange})
            logger.info('Parameter --timerange detected: %s ...', self.args.timerange)

        # If --datadir is used we add it to the configuration
        if 'datadir' in self.args and self.args.datadir:
            config.update({'datadir': self.args.datadir})
            logger.info('Parameter --datadir detected: %s ...', self.args.datadir)

        # If -r/--refresh-pairs-cached is used we add it to the configuration
        if 'refresh_pairs' in self.args and self.args.refresh_pairs:
            config.update({'refresh_pairs': True})
            logger.info('Parameter -r/--refresh-pairs-cached detected ...')

        # If --export is used we add it to the configuration
        if 'export' in self.args and self.args.export:
            config.update({'export': self.args.export})
            logger.info('Parameter --export detected: %s ...', self.args.export)

        return config

    def _load_hyperopt_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Extract information for sys.argv and load Hyperopt configuration
        :return: configuration as dictionary
        """
        # If --realistic-simulation is used we add it to the configuration
        if 'epochs' in self.args and self.args.epochs:
            config.update({'epochs': self.args.epochs})
            logger.info('Parameter --epochs detected ...')
            logger.info('Will run Hyperopt with for %s epochs ...', config.get('epochs'))

        # If --mongodb is used we add it to the configuration
        if 'mongodb' in self.args and self.args.mongodb:
            config.update({'mongodb': self.args.mongodb})
            logger.info('Parameter --use-mongodb detected ...')

        # If --spaces is used we add it to the configuration
        if 'spaces' in self.args and self.args.spaces:
            config.update({'spaces': self.args.spaces})
            logger.info('Parameter -s/--spaces detected: %s', config.get('spaces'))

        return config

    def _validate_config(self, conf: Dict[str, Any]) -> Dict[str, Any]:
        """
        Validate the configuration follow the Config Schema
        :param conf: Config in JSON format
        :return: Returns the config if valid, otherwise throw an exception
        """
        try:
            validate(conf, constants.CONF_SCHEMA)
            return conf
        except ValidationError as exception:
            logger.fatal(
                'Invalid configuration. See config.json.example. Reason: %s',
                exception
            )
            raise ValidationError(
                best_match(Draft4Validator(constants.CONF_SCHEMA).iter_errors(conf)).message
            )

    def get_config(self) -> Dict[str, Any]:
        """
        Return the config. Use this method to get the bot config
        :return: Dict: Bot config
        """
        if self.config is None:
            self.config = self.load_config()

        return self.config

    def check_exchange(self, config: Dict[str, Any]) -> bool:
        """
        Check if the exchange name in the config file is supported by Freqtrade
        :return: True or raised an exception if the exchange if not supported
        """
        exchange = config.get('exchange', {}).get('name').lower()
        if exchange not in ccxt.exchanges:

            exception_msg = 'Exchange "{}" not supported.\n' \
                            'The following exchanges are supported: {}'\
                .format(exchange, ', '.join(ccxt.exchanges))

            logger.critical(exception_msg)
            raise OperationalException(
                exception_msg
            )

        logger.debug('Exchange "%s" supported', exchange)
        return True