stable/tests/optimize/test_hyperopt_tools.py

import logging
import re
from pathlib import Path
from typing import Dict, List

import numpy as np
import pytest
import rapidjson

from freqtrade.constants import FTHYPT_FILEVERSION
from freqtrade.exceptions import OperationalException
from freqtrade.optimize.hyperopt_tools import HyperoptTools, hyperopt_serializer
from tests.conftest import CURRENT_TEST_STRATEGY, log_has


# Functions for recurrent object patching
def create_results() -> List[Dict]:

    return [{'loss': 1, 'result': 'foo', 'params': {}, 'is_best': True}]


def test_save_results_saves_epochs(hyperopt, tmpdir, caplog) -> None:

    hyperopt.results_file = Path(tmpdir / 'ut_results.fthypt')

    hyperopt_epochs = HyperoptTools.load_filtered_results(hyperopt.results_file, {})
    assert hyperopt_epochs == ([], 0)

    # Test writing to temp dir and reading again
    epochs = create_results()

    caplog.set_level(logging.DEBUG)

    for epoch in epochs:
        hyperopt._save_result(epoch)
    assert log_has(f"1 epoch saved to '{hyperopt.results_file}'.", caplog)

    hyperopt._save_result(epochs[0])
    assert log_has(f"2 epochs saved to '{hyperopt.results_file}'.", caplog)

    hyperopt_epochs = HyperoptTools.load_filtered_results(hyperopt.results_file, {})
    assert len(hyperopt_epochs) == 2
    assert hyperopt_epochs[1] == 2
    assert len(hyperopt_epochs[0]) == 2

    result_gen = HyperoptTools._read_results(hyperopt.results_file, 1)
    epoch = next(result_gen)
    assert len(epoch) == 1
    assert epoch[0] == epochs[0]
    epoch = next(result_gen)
    assert len(epoch) == 1
    epoch = next(result_gen)
    assert len(epoch) == 0
    with pytest.raises(StopIteration):
        next(result_gen)


def test_load_previous_results2(mocker, testdatadir, caplog) -> None:
    results_file = testdatadir / 'hyperopt_results_SampleStrategy.pickle'
    with pytest.raises(OperationalException,
                       match=r"Legacy hyperopt results are no longer supported.*"):
        HyperoptTools.load_filtered_results(results_file, {})


@pytest.mark.parametrize("spaces, expected_results", [
    (['buy'],
     {'buy': True, 'sell': False, 'roi': False, 'stoploss': False, 'trailing': False,
      'protection': False}),
    (['sell'],
     {'buy': False, 'sell': True, 'roi': False, 'stoploss': False, 'trailing': False,
      'protection': False}),
    (['roi'],
     {'buy': False, 'sell': False, 'roi': True, 'stoploss': False, 'trailing': False,
      'protection': False}),
    (['stoploss'],
     {'buy': False, 'sell': False, 'roi': False, 'stoploss': True, 'trailing': False,
      'protection': False}),
    (['trailing'],
     {'buy': False, 'sell': False, 'roi': False, 'stoploss': False, 'trailing': True,
      'protection': False}),
    (['buy', 'sell', 'roi', 'stoploss'],
     {'buy': True, 'sell': True, 'roi': True, 'stoploss': True, 'trailing': False,
      'protection': False}),
    (['buy', 'sell', 'roi', 'stoploss', 'trailing'],
     {'buy': True, 'sell': True, 'roi': True, 'stoploss': True, 'trailing': True,
      'protection': False}),
    (['buy', 'roi'],
     {'buy': True, 'sell': False, 'roi': True, 'stoploss': False, 'trailing': False,
      'protection': False}),
    (['all'],
     {'buy': True, 'sell': True, 'roi': True, 'stoploss': True, 'trailing': True,
      'protection': True}),
    (['default'],
     {'buy': True, 'sell': True, 'roi': True, 'stoploss': True, 'trailing': False,
      'protection': False}),
    (['default', 'trailing'],
     {'buy': True, 'sell': True, 'roi': True, 'stoploss': True, 'trailing': True,
      'protection': False}),
    (['all', 'buy'],
     {'buy': True, 'sell': True, 'roi': True, 'stoploss': True, 'trailing': True,
      'protection': True}),
    (['default', 'buy'],
     {'buy': True, 'sell': True, 'roi': True, 'stoploss': True, 'trailing': False,
      'protection': False}),
    (['all'],
     {'buy': True, 'sell': True, 'roi': True, 'stoploss': True, 'trailing': True,
      'protection': True}),
    (['protection'],
     {'buy': False, 'sell': False, 'roi': False, 'stoploss': False, 'trailing': False,
      'protection': True}),
])
def test_has_space(hyperopt_conf, spaces, expected_results):
    for s in ['buy', 'sell', 'roi', 'stoploss', 'trailing', 'protection']:
        hyperopt_conf.update({'spaces': spaces})
        assert HyperoptTools.has_space(hyperopt_conf, s) == expected_results[s]


def test_show_epoch_details(capsys):
    test_result = {
        'params_details': {
            'trailing': {
                'trailing_stop': True,
                'trailing_stop_positive': 0.02,
                'trailing_stop_positive_offset': 0.04,
                'trailing_only_offset_is_reached': True
            },
            'roi': {
                0: 0.18,
                90: 0.14,
                225: 0.05,
                430: 0},
        },
        'results_explanation': 'foo result',
        'is_initial_point': False,
        'total_profit': 0,
        'current_epoch': 2,  # This starts from 1 (in a human-friendly manner)
        'is_best': True
    }

    HyperoptTools.show_epoch_details(test_result, 5, False, no_header=True)
    captured = capsys.readouterr()
    assert '# Trailing stop:' in captured.out
    # re.match(r"Pairs for .*", captured.out)
    assert re.search(r'^\s+trailing_stop = True$', captured.out, re.MULTILINE)
    assert re.search(r'^\s+trailing_stop_positive = 0.02$', captured.out, re.MULTILINE)
    assert re.search(r'^\s+trailing_stop_positive_offset = 0.04$', captured.out, re.MULTILINE)
    assert re.search(r'^\s+trailing_only_offset_is_reached = True$', captured.out, re.MULTILINE)

    assert '# ROI table:' in captured.out
    assert re.search(r'^\s+minimal_roi = \{$', captured.out, re.MULTILINE)
    assert re.search(r'^\s+\"90\"\:\s0.14,\s*$', captured.out, re.MULTILINE)


def test__pprint_dict():
    params = {'buy_std': 1.2, 'buy_rsi': 31, 'buy_enable': True, 'buy_what': 'asdf'}
    non_params = {'buy_notoptimied': 55}

    x = HyperoptTools._pprint_dict(params, non_params)
    assert x == """{
    "buy_std": 1.2,
    "buy_rsi": 31,
    "buy_enable": True,
    "buy_what": "asdf",
    "buy_notoptimied": 55,  # value loaded from strategy
}"""


def test_get_strategy_filename(default_conf):

    x = HyperoptTools.get_strategy_filename(default_conf, 'StrategyTestV3')
    assert isinstance(x, Path)
    assert x == Path(__file__).parents[1] / 'strategy/strats/strategy_test_v3.py'

    x = HyperoptTools.get_strategy_filename(default_conf, 'NonExistingStrategy')
    assert x is None


def test_export_params(tmpdir):

    filename = Path(tmpdir) / f"{CURRENT_TEST_STRATEGY}.json"
    assert not filename.is_file()
    params = {
        "params_details": {
            "buy": {
                "buy_rsi": 30
            },
            "sell": {
                "sell_rsi": 70
            },
            "roi": {
                "0": 0.528,
                "346": 0.08499,
                "507": 0.049,
                "1595": 0
            }
        },
        "params_not_optimized": {
            "stoploss": -0.05,
            "trailing": {
                "trailing_stop": False,
                "trailing_stop_positive": 0.05,
                "trailing_stop_positive_offset": 0.1,
                "trailing_only_offset_is_reached": True
            },
        }

    }
    HyperoptTools.export_params(params, CURRENT_TEST_STRATEGY, filename)

    assert filename.is_file()

    with filename.open('r') as f:
        content = rapidjson.load(f)
    assert content['strategy_name'] == CURRENT_TEST_STRATEGY
    assert 'params' in content
    assert "buy" in content["params"]
    assert "sell" in content["params"]
    assert "roi" in content["params"]
    assert "stoploss" in content["params"]
    assert "trailing" in content["params"]


def test_try_export_params(default_conf, tmpdir, caplog, mocker):
    default_conf['disableparamexport'] = False
    export_mock = mocker.patch("freqtrade.optimize.hyperopt_tools.HyperoptTools.export_params")

    filename = Path(tmpdir) / f"{CURRENT_TEST_STRATEGY}.json"
    assert not filename.is_file()
    params = {
        "params_details": {
            "buy": {
                "buy_rsi": 30
            },
            "sell": {
                "sell_rsi": 70
            },
            "roi": {
                "0": 0.528,
                "346": 0.08499,
                "507": 0.049,
                "1595": 0
            }
        },
        "params_not_optimized": {
            "stoploss": -0.05,
            "trailing": {
                "trailing_stop": False,
                "trailing_stop_positive": 0.05,
                "trailing_stop_positive_offset": 0.1,
                "trailing_only_offset_is_reached": True
            },
        },
        FTHYPT_FILEVERSION: 2,

    }
    HyperoptTools.try_export_params(default_conf, "StrategyTestVXXX", params)

    assert log_has("Strategy not found, not exporting parameter file.", caplog)
    assert export_mock.call_count == 0
    caplog.clear()

    HyperoptTools.try_export_params(default_conf, CURRENT_TEST_STRATEGY, params)

    assert export_mock.call_count == 1
    assert export_mock.call_args_list[0][0][1] == CURRENT_TEST_STRATEGY
    assert export_mock.call_args_list[0][0][2].name == 'strategy_test_v3.json'


def test_params_print(capsys):

    params = {
        "buy": {
            "buy_rsi": 30
        },
        "sell": {
            "sell_rsi": 70
        },
    }
    non_optimized = {
        "buy": {
            "buy_adx": 44
        },
        "sell": {
            "sell_adx": 65
        },
        "stoploss": {
            "stoploss": -0.05,
        },
        "roi": {
            "0": 0.05,
            "20": 0.01,
        },
        "trailing": {
            "trailing_stop": False,
            "trailing_stop_positive": 0.05,
            "trailing_stop_positive_offset": 0.1,
            "trailing_only_offset_is_reached": True
        },

    }
    HyperoptTools._params_pretty_print(params, 'buy', 'No header', non_optimized)

    captured = capsys.readouterr()
    assert re.search("# No header", captured.out)
    assert re.search('"buy_rsi": 30,\n', captured.out)
    assert re.search('"buy_adx": 44,  # value loaded.*\n', captured.out)
    assert not re.search("sell", captured.out)

    HyperoptTools._params_pretty_print(params, 'sell', 'Sell Header', non_optimized)
    captured = capsys.readouterr()
    assert re.search("# Sell Header", captured.out)
    assert re.search('"sell_rsi": 70,\n', captured.out)
    assert re.search('"sell_adx": 65,  # value loaded.*\n', captured.out)

    HyperoptTools._params_pretty_print(params, 'roi', 'ROI Table:', non_optimized)
    captured = capsys.readouterr()
    assert re.search("# ROI Table:  # value loaded.*\n", captured.out)
    assert re.search('minimal_roi = {\n', captured.out)
    assert re.search('"20": 0.01\n', captured.out)

    HyperoptTools._params_pretty_print(params, 'trailing', 'Trailing stop:', non_optimized)
    captured = capsys.readouterr()
    assert re.search("# Trailing stop:", captured.out)
    assert re.search('trailing_stop = False  # value loaded.*\n', captured.out)
    assert re.search('trailing_stop_positive = 0.05  # value loaded.*\n', captured.out)
    assert re.search('trailing_stop_positive_offset = 0.1  # value loaded.*\n', captured.out)
    assert re.search('trailing_only_offset_is_reached = True  # value loaded.*\n', captured.out)


def test_hyperopt_serializer():

    assert isinstance(hyperopt_serializer(np.int_(5)), int)
    assert isinstance(hyperopt_serializer(np.bool_(True)), bool)
    assert isinstance(hyperopt_serializer(np.bool_(False)), bool)