stable/freqtrade/configuration.py

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

import json

from typing import Dict, List, Any
from jsonschema import Draft4Validator, validate
from jsonschema.exceptions import ValidationError, best_match

from freqtrade.constants import Constants
from freqtrade.logger import Logger


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: List[str]) -> None:
        self.args = args
        self.logging = Logger(name=__name__)
        self.logger = self.logging.get_logger()
        self.config = self._load_config()
        self.show_info()

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

        # Add the strategy file to use
        config.update({'strategy': self.args.strategy})

        # Add dynamic_whitelist if found
        if 'dynamic_whitelist' in self.args and self.args.dynamic_whitelist:
            config.update({'dynamic_whitelist': self.args.dynamic_whitelist})

        # 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})

        # Log level
        if 'loglevel' in self.args and self.args.loglevel:
            config.update({'loglevel': self.args.loglevel})
            self.logging.set_level(self.args.loglevel)

        # 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
        """
        with open(path) as file:
            conf = json.load(file)

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

        return self._validate_config(conf)

    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})
            self.logger.info('Parameter -i/--ticker-interval detected ...')
            self.logger.info('Using ticker_interval: %d ...', 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})
            self.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})
            self.logger.info('Parameter --realistic-simulation detected ...')
            self.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})
            self.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})
            self.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})
            self.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})
            self.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})
            self.logger.info('Parameter --epochs detected ...')
            self.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})
            self.logger.info('Parameter --use-mongodb detected ...')

        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:
            self.logger.fatal(
                'Invalid configuration. See config.json.example. Reason: %s',
                exception
            )
            raise ValidationError(
                best_match(Draft4Validator(Constants.CONF_SCHEMA).iter_errors(conf)).message
            )

    def show_info(self) -> None:
        """
        Display info message to user depending of the configuration of the bot
        :return: None
        """
        if self.config.get('dynamic_whitelist', False):
            self.logger.info(
                'Using dynamically generated whitelist. (--dynamic-whitelist detected)'
            )

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

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