stable/freqtrade/configuration.py

180 lines
7.0 KiB
Python
Raw Normal View History

"""
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