freqtrade_origin/tests/optimize/test_hyperopt_tools.py

import logging
import re
from pathlib import Path

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, log_has_re


# 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, tmp_path, caplog) -> None:
    hyperopt.results_file = tmp_path / "ut_results.fthypt"

    hyperopt_epochs = HyperoptTools.load_filtered_results(hyperopt.results_file, {})
    assert log_has_re("Hyperopt file .* not found.", caplog)
    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,
                "trades": False,
            },
        ),
        (
            ["sell"],
            {
                "buy": False,
                "sell": True,
                "roi": False,
                "stoploss": False,
                "trailing": False,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["roi"],
            {
                "buy": False,
                "sell": False,
                "roi": True,
                "stoploss": False,
                "trailing": False,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["stoploss"],
            {
                "buy": False,
                "sell": False,
                "roi": False,
                "stoploss": True,
                "trailing": False,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["trailing"],
            {
                "buy": False,
                "sell": False,
                "roi": False,
                "stoploss": False,
                "trailing": True,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["buy", "sell", "roi", "stoploss"],
            {
                "buy": True,
                "sell": True,
                "roi": True,
                "stoploss": True,
                "trailing": False,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["buy", "sell", "roi", "stoploss", "trailing"],
            {
                "buy": True,
                "sell": True,
                "roi": True,
                "stoploss": True,
                "trailing": True,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["buy", "roi"],
            {
                "buy": True,
                "sell": False,
                "roi": True,
                "stoploss": False,
                "trailing": False,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["all"],
            {
                "buy": True,
                "sell": True,
                "roi": True,
                "stoploss": True,
                "trailing": True,
                "protection": True,
                "trades": True,
            },
        ),
        (
            ["default"],
            {
                "buy": True,
                "sell": True,
                "roi": True,
                "stoploss": True,
                "trailing": False,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["default", "trailing"],
            {
                "buy": True,
                "sell": True,
                "roi": True,
                "stoploss": True,
                "trailing": True,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["all", "buy"],
            {
                "buy": True,
                "sell": True,
                "roi": True,
                "stoploss": True,
                "trailing": True,
                "protection": True,
                "trades": True,
            },
        ),
        (
            ["default", "buy"],
            {
                "buy": True,
                "sell": True,
                "roi": True,
                "stoploss": True,
                "trailing": False,
                "protection": False,
                "trades": False,
            },
        ),
        (
            ["all"],
            {
                "buy": True,
                "sell": True,
                "roi": True,
                "stoploss": True,
                "trailing": True,
                "protection": True,
                "trades": True,
            },
        ),
        (
            ["protection"],
            {
                "buy": False,
                "sell": False,
                "roi": False,
                "stoploss": False,
                "trailing": False,
                "protection": True,
                "trades": False,
            },
        ),
        (
            ["trades"],
            {
                "buy": False,
                "sell": False,
                "roi": False,
                "stoploss": False,
                "trailing": False,
                "protection": False,
                "trades": True,
            },
        ),
        (
            ["default", "trades"],
            {
                "buy": True,
                "sell": True,
                "roi": True,
                "stoploss": True,
                "trailing": False,
                "protection": False,
                "trades": True,
            },
        ),
    ],
)
def test_has_space(hyperopt_conf, spaces, expected_results):
    for s in ["buy", "sell", "roi", "stoploss", "trailing", "protection", "trades"]:
        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, tmp_path):
    default_conf["user_data_dir"] = tmp_path
    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(tmp_path):
    filename = tmp_path / 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},
            "max_open_trades": {"max_open_trades": 5},
        },
        "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"]
    assert "max_open_trades" in content["params"]


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

    filename = tmp_path / 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,
        },
        "max_open_trades": {"max_open_trades": 5},
    }
    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)

    HyperoptTools._params_pretty_print(params, "max_open_trades", "Max Open Trades:", non_optimized)
    captured = capsys.readouterr()

    assert re.search("# Max Open Trades:", captured.out)
    assert re.search("max_open_trades = 5  # 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)