MoniGoManiConfig.py

# -*- coding: utf-8 -*-
# -* vim: syntax=python -*-

# --- ↑↓ Do not remove these libs ↑↓ ---------------------------------------------------------------

"""MoniGoManiConfig is the module responsible for all MGM Config related tasks."""

# ___  ___               _  _____        ___  ___               _  _____                 __  _
# |  \/  |              (_)|  __ \       |  \/  |              (_)/  __ \               / _|(_)
# | .  . |  ___   _ __   _ | |  \/  ___  | .  . |  __ _  _ __   _ | /  \/  ___   _ __  | |_  _   __ _
# | |\/| | / _ \ | '_ \ | || | __  / _ \ | |\/| | / _` || '_ \ | || |     / _ \ | '_ \ |  _|| | / _` |
# | |  | || (_) || | | || || |_\ \| (_) || |  | || (_| || | | || || \__/\| (_) || | | || |  | || (_| |
# \_|  |_/ \___/ |_| |_||_| \____/ \___/ \_|  |_/ \__,_||_| |_||_| \____/ \___/ |_| |_||_|  |_| \__, |
#                                                                                                __/ |
#                                                                                               |___/

import json
import os
import shutil
import sys
from collections import OrderedDict
from operator import xor

import yaml

from user_data.mgm_tools.mgm_hurry.CliColor import Color
from user_data.mgm_tools.mgm_hurry.MoniGoManiLogger import MoniGoManiLogger


# --- ↑ Do not remove these libs ↑ ---------------------------------------------------------------


class MoniGoManiConfig(object):
    """
    MoniGoManiConfig is responsible for all MoniGoMani Config related tasks.

    Attributes:
        __config            Dictionary containing the configuration parameters.
        __basedir           The basedir where the monigomani install lives.
        __full_path_config  Absolute path to .hurry config file.
        __mgm_logger        The logger function of the MoniGoManiCli module.
    """
    __config: dict
    __basedir: str
    __full_path_config: str
    __mgm_logger: MoniGoManiLogger

    def __init__(self, basedir: str):
        """
        MoniGoMani has configuration.

        :param basedir: (str) The base directory where Freqtrade & MoniGoMani are installed
        """
        self.__basedir = basedir
        self.__mgm_logger = MoniGoManiLogger(basedir).get_logger()
        self.__full_path_config = f'{self.__basedir}/.hurry'

        # if .hurry file does not exist
        if self.valid_hurry_dotfile_present() is False:
            self.__create_default_config()

        self.__config = self.read_hurry_config()

    @property
    def config(self) -> dict:
        return self.__config

    @config.setter
    def config(self, data: dict):
        self.__config = data

    def get(self, element: str):
        if element not in self.__config:
            return False

        return self.__config[element]

    @property
    def logger(self) -> MoniGoManiLogger:
        return self.__mgm_logger

    @property
    def basedir(self) -> str:
        return self.__basedir

    def reload(self) -> bool:
        """
        Reload config file and store as property in current object.

        :return bool: True if config is read, False if config could not be read.
        """
        if self.valid_hurry_dotfile_present() is not True:
            self.logger.error(Color.red('Failed to reload config. No valid hurry dotfile present.'))
            return False

        self.config = self.read_hurry_config()

        return True

    def valid_hurry_dotfile_present(self) -> bool:
        """
        Check if the .hurry config file exists on disk.

        :return bool: Return true if the config files exist, false if not
        """
        if os.path.isfile(self.__full_path_config) is not True:
            self.logger.warning(Color.yellow(f'Could not find .hurry config file at {self.__full_path_config}'))
            return False

        with open(self.__full_path_config, 'r') as yml_file:
            config = yaml.full_load(yml_file) or {}

        # Check if all required config keys are present in config file
        for key in ['exchange', 'install_type', 'timerange']:
            if not config['config'][key]:
                return False

        return True

    def create_config_files(self, target_dir: str) -> bool:
        """
        Copy example files as def files.

        :param target_dir: (str) The target dir where the "mgm-config.example.json" exists.
        :return bool: True if files are created successfully, false if something failed.
        """
        example_files = [
            {'src': 'mgm-config.example.json', 'dest': 'mgm-config.json'},
            {'src': 'mgm-config-private.example.json', 'dest': 'mgm-config-private.json'}
        ]

        for example_file in example_files:
            src_file = f'{target_dir}/monigomani/user_data/{example_file["src"]}'

            if not os.path.isfile(src_file):
                self.logger.error(Color.red(f'❌ Bummer. Cannot find the example file '
                                            f'"{example_file["src"]}" to copy from.'))
                return False

            dest_file = f'{target_dir}/user_data/{example_file["dest"]}'

            if os.path.isfile(dest_file):
                self.logger.warning(Color.yellow(f'⚠️ The target file "{example_file["dest"]}" '
                                                 f'already exists. Is cool.'))
                continue

            shutil.copyfile(src_file, dest_file)

        self.logger.info(Color.green('👉 MoniGoMani config files prepared √'))
        return True

    def load_config_files(self) -> dict:
        """
        Load & Return all the MoniGoMani Configuration files.

        Including:
            - mgm-config
            - mgm-config-private
            - mgm-config-hyperopt

        :return dict: Dictionary containing all the MoniGoMani Configuration files in format
            {mgm-config: dict, mgm-config-private: dict, mgm-config-hyperopt: dict}
        """

        hurry_config = self.read_hurry_config()

        if hurry_config is None:
            self.logger.error(Color.red('🤷 No Hurry config file found. Please run: mgm-hurry setup'))
            sys.exit(1)

        # Start loading the MoniGoMani config files
        mgm_config_files = {
            'mgm-config': {},
            'mgm-config-private': {},
            'mgm-config-hyperopt': {}
        }

        for mgm_config_filename in mgm_config_files:
            # Check if the MoniGoMani config filename exist in the ".hurry" config file
            if 'mgm_config_names' not in hurry_config or mgm_config_filename not in hurry_config['mgm_config_names']:
                self.logger.critical(Color.red(f'🤷 No "{mgm_config_filename}" filename found in the '
                                               f'".hurry" config file. Please run: mgm-hurry setup'))
                sys.exit(1)

            # Full path to current config file
            mgm_config_filepath = self._get_full_path_for_config_name(hurry_config, mgm_config_filename)

            # Read config file contents
            if mgm_config_filename != 'mgm-config-hyperopt':
                mgm_config_files[mgm_config_filename] = self.load_config_file(filename=mgm_config_filepath)
            else:
                mgm_config_files[mgm_config_filename] = self.load_config_file(filename=mgm_config_filepath, silent=True)

        return mgm_config_files

    def read_hurry_config(self) -> dict:
        """
        Read .hurry configuration dotfile and return its yaml contents as dict.

        :return dict: Dictionary containing the config section of .hurry file. None if failed.
        """
        with open(f'{self.basedir}/.hurry', 'r') as yml_file:
            config = yaml.full_load(yml_file) or {}

        hurry_config = config['config'] if 'config' in config else None

        return hurry_config

    def get_config_filepath(self, cfg_key: str) -> str:
        """
        Transforms given cfg_key into the corresponding absolute config filepath.

        :param cfg_key: (str) The config name (key) to parse.
        :return str: The absolute path to the asked config file.
        """
        hurry_config = self.read_hurry_config()
        return self._get_full_path_for_config_name(hurry_config, cfg_key)

    def load_config_file(self, filename: str, silent: bool = False) -> dict:
        """
        Read json-file contents and return its data.

        :param filename: (str) The absolute path + filename to the json config file.
        :param silent: (bool, Optional) Silently run method (without command line output)
        :return dict: The json content of the file. json.load() return. None if failed.
        """
        if os.path.isfile(filename) is False:
            if silent is False:
                self.logger.warning(Color.yellow(f'🤷 No "{filename}" file found in the "user_data" directory. '
                                                 f'Please run: mgm-hurry setup'))
            return None

        # Load the MoniGoMani config file as an object and parse it as a dictionary
        with open(filename, ) as file_object:
            json_data = json.load(file_object)
            return json_data

    def load_mgm_config_hyperopt(self) -> dict:
        """
        Loads 'mgm-config-hyperopt' as a dictionary object if it exists

        :return: (dict) 'mgm-config-hyperopt' as a dictionary object
        """

        # Check if 'mgm-config-hyperopt' exists
        mgm_config_hyperopt_name = self.config['mgm_config_names']['mgm-config-hyperopt']
        mgm_config_hyperopt_path = f'{self.basedir}/user_data/{mgm_config_hyperopt_name}'

        if os.path.isfile(mgm_config_hyperopt_path) is False:
            self.logger.warning(Color.yellow(f'🤷 No "{mgm_config_hyperopt_name}" file found in the "user_data" '
                                             f'directory. Please run: mgm-hurry hyperopt'))
            return {}

        # Load the needed MoniGoMani Config files and 'mgm-config-hyperopt'
        mgm_config_files = self.load_config_files()
        return mgm_config_files['mgm-config-hyperopt']

    def write_hurry_dotfile(self, config: dict = None):
        """
        Write config-array to ".hurry" config file and load its contents into config-property.
        Writes the passed config dictionary or if nothing passed, it will write default values.

        :param config: (dict, Optional) The config values to store. Defaults to None.
        """
        if config is None:
            config = {
                'config': {
                    'install_type': 'source',
                    'ft_binary': 'freqtrade',
                    'timerange': '20210501-20210616',
                    'exchange': 'binance',
                    'hyperopt': {
                        'strategy': 'MoniGoManiHyperStrategy',
                        'loss': 'MGM_WinRatioAndProfitRatioHyperOptLoss',
                        'spaces': 'buy sell',
                        'stake_currency': 'USDT',
                        'epochs': 1000
                    },
                    'mgm_config_names': {
                        'mgm-config': 'mgm-config.json',
                        'mgm-config-private': 'mgm-config-private.json',
                        'mgm-config-hyperopt': 'mgm-config-hyperopt.json'
                    }
                }
            }

        # Protection to prevent from writing no data at all to mgm-config.
        if len(config) == 0 or 'config' not in config or 'mgm_config_names' not in config['config']:
            self.logger.error(Color.red('🤯 Sorry, but looks like no configuration data would have been written, '
                                        'resulting in an empty config file. I quit.'))
            sys.exit(1)

        with open(self.__full_path_config, 'w+') as cfg_file:
            yaml.dump(config, cfg_file)

        self.reload()

        self.logger.info(Color.green('🍺 Configuration data written to ".hurry" file'))

    def cleanup_hyperopt_files(self, strategy: str = 'MoniGoManiHyperStrategy') -> bool:
        """
        Cleanup leftover strategy HyperOpt files.

        - mgm-config-hyperopt.json (applied results file)
        - {strategy}.json (intermediate results file)

        :param strategy: (str) The strategy used to find the corresponding files. Defaults to 'MoniGoManiHyperStrategy'.
        :return bool: True if one of these files is cleaned up with success. False if no file was cleaned up.
        """
        cleaned_up_cfg = False
        if strategy == 'MoniGoManiHyperStrategy':
            file_abspath = self._get_full_path_for_config_name(self.read_hurry_config(), 'mgm-config-hyperopt')
            cleaned_up_cfg = self._remove_file(file_abspath)

        # Remove the intermediate ho results file if exists
        strategy_ho_intermediate_path = f'{self.basedir}/user_data/strategies/{strategy}.json'
        cleaned_up_intermediate = self._remove_file(strategy_ho_intermediate_path)

        # return true if one of these is true
        return xor(bool(cleaned_up_cfg), bool(cleaned_up_intermediate))

    def get_hyperopted_spaces(self) -> list[str]:
        """
        Fetches a list of the previously hyperopted spaces for MoniGoMani from 'mgm-config-hyperopt' if it exists.

        :return: (list[str]) Returns a list of the previously hyperopted spaces
        """

        mgm_config_hyperopt = self.load_mgm_config_hyperopt()
        if mgm_config_hyperopt == {}:
            return []

        hyperopted_spaces = []
        for hyperopted_space in mgm_config_hyperopt['params']:
            hyperopted_spaces.append(hyperopted_space)

        return hyperopted_spaces

    def _remove_file(self, fil: str) -> bool:
        if os.path.exists(fil) is False:
            return False

        self.logger.info(f'👉 Removing "{os.path.basename(fil)}"')
        os.remove(fil)

        return True

    def _get_full_path_for_config_name(self, hurry_config: dict, cfg_name: str) -> str:
        """
        Parses the full path to given config file based on settings in .hurry.

        :param hurry_config (dict): The dictionary containing the hurry dotfile yaml config.
        :return abs_path: The absolute path to the asked config file.
        """
        # Full path to current config file
        mgm_config_filepath = f'{self.basedir}/user_data/{hurry_config["mgm_config_names"][cfg_name]}'

        return mgm_config_filepath

    def __create_default_config(self):
        """
        Creates default .hurry config file with default values.
        """
        self.write_hurry_dotfile()

    def save_stake_currency(self, stake_currency: str):
        """
        Saves the stake currency to 'mgm-config'

        :param stake_currency: (str) The stake_currency you wish to use
        """
        # Overwrite the new stake currency in mgm-config[stake_currency]
        mgm_config_file = self.get_config_filepath('mgm-config')
        if os.path.isfile(mgm_config_file):
            with open(mgm_config_file, ) as mgm_config:
                mgm_config_object = json.load(mgm_config)
                mgm_config.close()

            with open(mgm_config_file, 'w') as mgm_config:
                mgm_config_object['stake_currency'] = stake_currency
                json.dump(mgm_config_object, mgm_config, indent=4)
                mgm_config.close()

    def save_exchange_credentials(self, cred: dict):
        """
        Save exchange credentials to 'mgm-config-private'

        :param cred: (dict) List containing values for [exchange,api_key,api_secret]
        """
        mgm_config_private_name = self.config['mgm_config_names']['mgm-config-private']
        if len(cred) == 0:
            self.logger.warning(Color.yellow(f'Did not write exchange credentials to "{mgm_config_private_name}" '
                                             f'because no data was passed.'))
            return False

        try:
            with open(f'{self.basedir}/user_data/{mgm_config_private_name}', 'a+') as file:
                data = json.load(file)
        except Exception:
            data = {}

        data['exchange'] = {'name': cred['exchange'], 'key': cred['api_key'], 'secret': cred['api_secret']}

        with open(f'{self.basedir}/user_data/{mgm_config_private_name}', 'w+') as outfile:
            json.dump(data, outfile, indent=4)

        self.logger.info(Color.green(f'🍺 Exchange settings written to "{mgm_config_private_name}"'))

    def save_telegram_credentials(self, opt: dict) -> bool:
        """
        Save Telegram bot settings to 'mgm-config-private'

        :param opt: (dict) Dictionary containing values for [enable_telegram,telegram_token,telegram_chat_id]
        :return bool: True if json data is written. False otherwise.
        """
        mgm_config_private_name = self.config['mgm_config_names']['mgm-config-private']
        if len(opt) == 0:
            self.logger.warning(Color.yellow(f'Did not write telegram credentials to "{mgm_config_private_name}" '
                                             f'because no data was passed.'))
            return False

        with open(f'{self.basedir}/user_data/{mgm_config_private_name}', ) as file:
            data = json.load(file)

        data['telegram'] = {
            'enabled': opt['enable_telegram'], 'token': opt['telegram_token'], 'chat_id': opt['telegram_chat_id']
        }

        with open(f'{self.basedir}/user_data/{mgm_config_private_name}', 'w+') as outfile:
            json.dump(data, outfile, indent=4)

        self.logger.info(Color.green(f'🍺 Telegram bot settings written to "{mgm_config_private_name}"'))

        return True

    def save_weak_strong_signal_overrides(self, previously_hyperopted_spaces: list[str]) -> bool:
        """
        Overrides weak and strong signals to their actual values used by MoniGoMani in 'mgm-config-hyperopt'

        :param previously_hyperopted_spaces: (list[str]) List containing previously HyperOpted spaces
        :return bool: True if json data is overwritten correctly, False otherwise.
        """
        # Check if 'mgm-config-hyperopt' exists
        mgm_config_hyperopt_name = self.config['mgm_config_names']['mgm-config-hyperopt']
        mgm_config_hyperopt_path = f'{self.basedir}/user_data/{mgm_config_hyperopt_name}'

        if os.path.isfile(mgm_config_hyperopt_path) is False:
            self.logger.warning(Color.yellow(f'🤷 No "{mgm_config_hyperopt_name}" file found in the "user_data" '
                                             f'directory. Please run: mgm-hurry hyperopt'))
            return False

        # Load the needed MoniGoMani Config files
        mgm_config_files = self.load_config_files()
        mgm_config_hyperopt = mgm_config_files['mgm-config-hyperopt']
        mgm_config = mgm_config_files['mgm-config']['monigomani_settings']
        signal_triggers_needed_min_value = mgm_config['weighted_signal_spaces']['min_trend_signal_triggers_needed']
        signal_triggers_needed_threshold = mgm_config['weighted_signal_spaces'][
            'search_threshold_trend_signal_triggers_needed']
        total_signal_needed_min_value = mgm_config['weighted_signal_spaces']['min_trend_total_signal_needed_value']
        weighted_signal_max_value = mgm_config['weighted_signal_spaces']['max_weighted_signal_value']
        weighted_signal_min_value = mgm_config['weighted_signal_spaces']['min_weighted_signal_value']
        weighted_signal_threshold = mgm_config['weighted_signal_spaces']['search_threshold_weighted_signal_values']

        # Override the strong and weak signals where above and below the threshold
        for space in ['buy', 'sell']:
            if space in mgm_config_hyperopt['params']:
                mgm_config_hyperopt['params'][space] = \
                    dict(OrderedDict(sorted(mgm_config_hyperopt['params'][space].items())))

                number_of_weighted_signals = 0
                for signal, signal_weight_value in mgm_config_hyperopt['params'][space].items():
                    if (signal.startswith(f'{space}_downwards') is True) and (signal.endswith('_weight') is True):
                        number_of_weighted_signals += 1

                for signal, signal_weight_value in mgm_config_hyperopt['params'][space].items():
                    # Don't override non overrideable parameters
                    if (signal.startswith('sell___unclogger_') is False) and (
                        signal.endswith('_trend_total_signal_needed_candles_lookback_window') is False):

                        if signal.endswith('_weight'):
                            if signal_weight_value <= (weighted_signal_min_value + weighted_signal_threshold):
                                mgm_config_hyperopt['params'][space][signal] = weighted_signal_min_value
                            elif signal_weight_value >= (weighted_signal_max_value - weighted_signal_threshold):
                                mgm_config_hyperopt['params'][space][signal] = weighted_signal_max_value

                        elif signal.endswith('_trend_total_signal_needed'):
                            if signal_weight_value <= (total_signal_needed_min_value + weighted_signal_threshold):
                                mgm_config_hyperopt['params'][space][signal] = total_signal_needed_min_value
                            elif signal_weight_value >= ((weighted_signal_max_value *
                                                          number_of_weighted_signals) - weighted_signal_threshold):
                                mgm_config_hyperopt['params'][space][signal] = (weighted_signal_max_value *
                                                                                number_of_weighted_signals)

                        elif signal.endswith('_trend_signal_triggers_needed'):
                            if signal_weight_value <= (signal_triggers_needed_min_value +
                                                       signal_triggers_needed_threshold):
                                mgm_config_hyperopt['params'][space][signal] = signal_triggers_needed_min_value
                            elif signal_weight_value >= (number_of_weighted_signals - signal_triggers_needed_threshold):
                                mgm_config_hyperopt['params'][space][signal] = number_of_weighted_signals

        # Fetch all used spaces, for the previous + current hyperopt
        all_used_spaces = previously_hyperopted_spaces
        command_object = self.load_config_file(filename=f'{self.basedir}/user_data/.last_command.json')
        currently_used_spaces = list(command_object['properties']['spaces'].split(' '))

        for currently_used_space in currently_used_spaces:
            if currently_used_space not in previously_hyperopted_spaces:
                all_used_spaces.append(currently_used_space)

        # Sort the spaces to the order used during hyperopting
        sorted_spaces = {}
        for space_name in all_used_spaces:
            if space_name in mgm_config_hyperopt['params']:
                sorted_spaces[space_name] = mgm_config_hyperopt['params'][space_name]
        mgm_config_hyperopt['params'] = sorted_spaces

        # Write the updated data to 'mgm-config-hyperopt'
        with open(mgm_config_hyperopt_path, 'w+') as outfile:
            json.dump(mgm_config_hyperopt, outfile, indent=4)

        self.logger.debug(f'Strong and weak signals automatically over-written in "{mgm_config_hyperopt_name}"')
        return True

    def command_configs(self) -> str:
        """
        Returns a string with the 'mgm-config' & 'mgm-config-private' names loaded from '.hurry'
        ready to implement in a freqtrade command.
        :return str: String with 'mgm-config' & 'mgm-config-private' for a freqtrade command
        """
        mgm_json_name = self.config['mgm_config_names']['mgm-config']
        mgm_private_json_name = self.config['mgm_config_names']['mgm-config-private']
        return f'--config ./user_data/{mgm_json_name} --config ./user_data/{mgm_private_json_name} '

    def get_preset_timerange(self, timerange: str) -> str:
        """
        Parses given timerange-string into according timerange dates

        :param timerange: (str) The timerange-string to parse [up, down, side]
        :return str: The parsed timerange string in yyyymmdd-yyyymmdd format
        """

        tr_input = timerange

        if timerange is None:
            timerange = self.config['timerange']
        if timerange == 'down':
            timerange = '20210509-20210524'
        if timerange == 'side':
            timerange = '20210518-20210610'
        if timerange == 'up':
            timerange = '20210127-20210221'

        tr_output = timerange

        self.logger.debug(f'☀️ Timerange string parsed from "{tr_input}" to "{tr_output}"')

        return timerange