stable/freqtrade/commands/hyperopt_commands.py

import logging
from operator import itemgetter
from typing import Any, Dict, List

from colorama import init as colorama_init

from freqtrade.configuration import setup_utils_configuration
from freqtrade.data.btanalysis import get_latest_hyperopt_file
from freqtrade.enums import RunMode
from freqtrade.exceptions import OperationalException
from freqtrade.optimize.optimize_reports import show_backtest_result


logger = logging.getLogger(__name__)


def start_hyperopt_list(args: Dict[str, Any]) -> None:
    """
    List hyperopt epochs previously evaluated
    """
    from freqtrade.optimize.hyperopt_tools import HyperoptTools

    config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE)

    print_colorized = config.get('print_colorized', False)
    print_json = config.get('print_json', False)
    export_csv = config.get('export_csv', None)
    no_details = config.get('hyperopt_list_no_details', False)
    no_header = False

    filteroptions = {
        'only_best': config.get('hyperopt_list_best', False),
        'only_profitable': config.get('hyperopt_list_profitable', False),
        'filter_min_trades': config.get('hyperopt_list_min_trades', 0),
        'filter_max_trades': config.get('hyperopt_list_max_trades', 0),
        'filter_min_avg_time': config.get('hyperopt_list_min_avg_time', None),
        'filter_max_avg_time': config.get('hyperopt_list_max_avg_time', None),
        'filter_min_avg_profit': config.get('hyperopt_list_min_avg_profit', None),
        'filter_max_avg_profit': config.get('hyperopt_list_max_avg_profit', None),
        'filter_min_total_profit': config.get('hyperopt_list_min_total_profit', None),
        'filter_max_total_profit': config.get('hyperopt_list_max_total_profit', None),
        'filter_min_objective': config.get('hyperopt_list_min_objective', None),
        'filter_max_objective': config.get('hyperopt_list_max_objective', None),
    }

    results_file = get_latest_hyperopt_file(
        config['user_data_dir'] / 'hyperopt_results',
        config.get('hyperoptexportfilename'))

    # Previous evaluations
    epochs = HyperoptTools.load_previous_results(results_file)
    total_epochs = len(epochs)

    epochs = hyperopt_filter_epochs(epochs, filteroptions)

    if print_colorized:
        colorama_init(autoreset=True)

    if not export_csv:
        try:
            print(HyperoptTools.get_result_table(config, epochs, total_epochs,
                                                 not filteroptions['only_best'],
                                                 print_colorized, 0))
        except KeyboardInterrupt:
            print('User interrupted..')

    if epochs and not no_details:
        sorted_epochs = sorted(epochs, key=itemgetter('loss'))
        results = sorted_epochs[0]
        HyperoptTools.show_epoch_details(results, total_epochs, print_json, no_header)

    if epochs and export_csv:
        HyperoptTools.export_csv_file(
            config, epochs, total_epochs, not filteroptions['only_best'], export_csv
        )


def start_hyperopt_show(args: Dict[str, Any]) -> None:
    """
    Show details of a hyperopt epoch previously evaluated
    """
    from freqtrade.optimize.hyperopt_tools import HyperoptTools

    config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE)

    print_json = config.get('print_json', False)
    no_header = config.get('hyperopt_show_no_header', False)
    results_file = get_latest_hyperopt_file(
        config['user_data_dir'] / 'hyperopt_results',
        config.get('hyperoptexportfilename'))

    n = config.get('hyperopt_show_index', -1)

    filteroptions = {
        'only_best': config.get('hyperopt_list_best', False),
        'only_profitable': config.get('hyperopt_list_profitable', False),
        'filter_min_trades': config.get('hyperopt_list_min_trades', 0),
        'filter_max_trades': config.get('hyperopt_list_max_trades', 0),
        'filter_min_avg_time': config.get('hyperopt_list_min_avg_time', None),
        'filter_max_avg_time': config.get('hyperopt_list_max_avg_time', None),
        'filter_min_avg_profit': config.get('hyperopt_list_min_avg_profit', None),
        'filter_max_avg_profit': config.get('hyperopt_list_max_avg_profit', None),
        'filter_min_total_profit': config.get('hyperopt_list_min_total_profit', None),
        'filter_max_total_profit': config.get('hyperopt_list_max_total_profit', None),
        'filter_min_objective': config.get('hyperopt_list_min_objective', None),
        'filter_max_objective': config.get('hyperopt_list_max_objective', None)
    }

    # Previous evaluations
    epochs = HyperoptTools.load_previous_results(results_file)
    total_epochs = len(epochs)

    epochs = hyperopt_filter_epochs(epochs, filteroptions)
    filtered_epochs = len(epochs)

    if n > filtered_epochs:
        raise OperationalException(
            f"The index of the epoch to show should be less than {filtered_epochs + 1}.")
    if n < -filtered_epochs:
        raise OperationalException(
            f"The index of the epoch to show should be greater than {-filtered_epochs - 1}.")

    # Translate epoch index from human-readable format to pythonic
    if n > 0:
        n -= 1

    if epochs:
        val = epochs[n]

        metrics = val['results_metrics']
        if 'strategy_name' in metrics:
            strategy_name = metrics['strategy_name']
            show_backtest_result(strategy_name, metrics,
                                 metrics['stake_currency'])

            # Export parameters ...
            # TODO: make this optional? otherwise it'll overwrite previous parameters ...
            fn = HyperoptTools.get_strategy_filename(config, strategy_name)
            HyperoptTools.export_params(val, strategy_name, fn.with_suffix('.json'))

        HyperoptTools.show_epoch_details(val, total_epochs, print_json, no_header,
                                         header_str="Epoch details")


def hyperopt_filter_epochs(epochs: List, filteroptions: dict) -> List:
    """
    Filter our items from the list of hyperopt results
    TODO: after 2021.5 remove all "legacy" mode queries.
    """
    if filteroptions['only_best']:
        epochs = [x for x in epochs if x['is_best']]
    if filteroptions['only_profitable']:
        epochs = [x for x in epochs if x['results_metrics'].get(
            'profit', x['results_metrics'].get('profit_total', 0)) > 0]

    epochs = _hyperopt_filter_epochs_trade_count(epochs, filteroptions)

    epochs = _hyperopt_filter_epochs_duration(epochs, filteroptions)

    epochs = _hyperopt_filter_epochs_profit(epochs, filteroptions)

    epochs = _hyperopt_filter_epochs_objective(epochs, filteroptions)

    logger.info(f"{len(epochs)} " +
                ("best " if filteroptions['only_best'] else "") +
                ("profitable " if filteroptions['only_profitable'] else "") +
                "epochs found.")
    return epochs


def _hyperopt_filter_epochs_trade(epochs: List, trade_count: int):
    """
    Filter epochs with trade-counts > trades
    """
    return [
        x for x in epochs
        if x['results_metrics'].get(
            'trade_count', x['results_metrics'].get('total_trades', 0)
        ) > trade_count
    ]


def _hyperopt_filter_epochs_trade_count(epochs: List, filteroptions: dict) -> List:

    if filteroptions['filter_min_trades'] > 0:
        epochs = _hyperopt_filter_epochs_trade(epochs, filteroptions['filter_min_trades'])

    if filteroptions['filter_max_trades'] > 0:
        epochs = [
            x for x in epochs
            if x['results_metrics'].get(
                'trade_count', x['results_metrics'].get('total_trades')
                ) < filteroptions['filter_max_trades']
        ]
    return epochs


def _hyperopt_filter_epochs_duration(epochs: List, filteroptions: dict) -> List:

    def get_duration_value(x):
        # Duration in minutes ...
        if 'duration' in x['results_metrics']:
            return x['results_metrics']['duration']
        else:
            # New mode
            if 'holding_avg_s' in x['results_metrics']:
                avg = x['results_metrics']['holding_avg_s']
                return avg // 60
            raise OperationalException(
                "Holding-average not available. Please omit the filter on average time, "
                "or rerun hyperopt with this version")

    if filteroptions['filter_min_avg_time'] is not None:
        epochs = _hyperopt_filter_epochs_trade(epochs, 0)
        epochs = [
            x for x in epochs
            if get_duration_value(x) > filteroptions['filter_min_avg_time']
        ]
    if filteroptions['filter_max_avg_time'] is not None:
        epochs = _hyperopt_filter_epochs_trade(epochs, 0)
        epochs = [
            x for x in epochs
            if get_duration_value(x) < filteroptions['filter_max_avg_time']
        ]

    return epochs


def _hyperopt_filter_epochs_profit(epochs: List, filteroptions: dict) -> List:

    if filteroptions['filter_min_avg_profit'] is not None:
        epochs = _hyperopt_filter_epochs_trade(epochs, 0)
        epochs = [
            x for x in epochs
            if x['results_metrics'].get(
                'avg_profit', x['results_metrics'].get('profit_mean', 0) * 100
            ) > filteroptions['filter_min_avg_profit']
        ]
    if filteroptions['filter_max_avg_profit'] is not None:
        epochs = _hyperopt_filter_epochs_trade(epochs, 0)
        epochs = [
            x for x in epochs
            if x['results_metrics'].get(
                'avg_profit', x['results_metrics'].get('profit_mean', 0) * 100
                ) < filteroptions['filter_max_avg_profit']
        ]
    if filteroptions['filter_min_total_profit'] is not None:
        epochs = _hyperopt_filter_epochs_trade(epochs, 0)
        epochs = [
            x for x in epochs
            if x['results_metrics'].get(
                'profit', x['results_metrics'].get('profit_total_abs', 0)
                ) > filteroptions['filter_min_total_profit']
        ]
    if filteroptions['filter_max_total_profit'] is not None:
        epochs = _hyperopt_filter_epochs_trade(epochs, 0)
        epochs = [
            x for x in epochs
            if x['results_metrics'].get(
                'profit', x['results_metrics'].get('profit_total_abs', 0)
                ) < filteroptions['filter_max_total_profit']
        ]
    return epochs


def _hyperopt_filter_epochs_objective(epochs: List, filteroptions: dict) -> List:

    if filteroptions['filter_min_objective'] is not None:
        epochs = _hyperopt_filter_epochs_trade(epochs, 0)

        epochs = [x for x in epochs if x['loss'] < filteroptions['filter_min_objective']]
    if filteroptions['filter_max_objective'] is not None:
        epochs = _hyperopt_filter_epochs_trade(epochs, 0)

        epochs = [x for x in epochs if x['loss'] > filteroptions['filter_max_objective']]

    return epochs