# from unittest.mock import MagicMock # from freqtrade.commands.optimize_commands import setup_optimize_configuration, start_edge import copy import platform import shutil from pathlib import Path from unittest.mock import MagicMock from freqtrade.configuration import TimeRange from freqtrade.data.dataprovider import DataProvider # from freqtrade.freqai.data_drawer import FreqaiDataDrawer from freqtrade.freqai.data_kitchen import FreqaiDataKitchen from tests.conftest import get_patched_exchange, log_has from tests.freqai.conftest import freqai_conf, get_patched_freqai_strategy def test_train_model_in_series_LightGBM(mocker, default_conf): freqaiconf = freqai_conf(copy.deepcopy(default_conf)) freqaiconf.update({"timerange": "20180110-20180130"}) strategy = get_patched_freqai_strategy(mocker, freqaiconf) exchange = get_patched_exchange(mocker, freqaiconf) strategy.dp = DataProvider(freqaiconf, exchange) strategy.freqai_info = freqaiconf.get("freqai", {}) freqai = strategy.model.bridge freqai.live = True freqai.dk = FreqaiDataKitchen(freqaiconf, freqai.dd) timerange = TimeRange.parse_timerange("20180110-20180130") freqai.dk.load_all_pair_histories(timerange) freqai.dd.pair_dict = MagicMock() data_load_timerange = TimeRange.parse_timerange("20180110-20180130") new_timerange = TimeRange.parse_timerange("20180120-20180130") freqai.train_model_in_series(new_timerange, "ADA/BTC", strategy, freqai.dk, data_load_timerange) assert ( Path(freqai.dk.data_path / str(freqai.dk.model_filename + "_model.joblib")) .resolve() .exists() ) assert ( Path(freqai.dk.data_path / str(freqai.dk.model_filename + "_metadata.json")) .resolve() .exists() ) assert ( Path(freqai.dk.data_path / str(freqai.dk.model_filename + "_trained_df.pkl")) .resolve() .exists() ) assert ( Path(freqai.dk.data_path / str(freqai.dk.model_filename + "_svm_model.joblib")) .resolve() .exists() ) shutil.rmtree(Path(freqai.dk.full_path)) # Catboost not available for ARM architecture. using platform lib to check processor type if "arm" not in platform.uname()[-1]: def test_train_model_in_series_Catboost(mocker, default_conf): freqaiconf = freqai_conf(copy.deepcopy(default_conf)) freqaiconf.update({"timerange": "20180110-20180130"}) freqaiconf.update({"freqaimodel": "CatboostPredictionModel"}) strategy = get_patched_freqai_strategy(mocker, freqaiconf) exchange = get_patched_exchange(mocker, freqaiconf) strategy.dp = DataProvider(freqaiconf, exchange) strategy.freqai_info = freqaiconf.get("freqai", {}) freqai = strategy.model.bridge freqai.live = True freqai.dk = FreqaiDataKitchen(freqaiconf, freqai.dd) timerange = TimeRange.parse_timerange("20180110-20180130") freqai.dk.load_all_pair_histories(timerange) freqai.dd.pair_dict = MagicMock() data_load_timerange = TimeRange.parse_timerange("20180110-20180130") new_timerange = TimeRange.parse_timerange("20180120-20180130") freqai.train_model_in_series( new_timerange, "ADA/BTC", strategy, freqai.dk, data_load_timerange ) assert ( Path(freqai.dk.data_path / str(freqai.dk.model_filename + "_model.joblib")) .resolve() .exists() ) assert ( Path(freqai.dk.data_path / str(freqai.dk.model_filename + "_metadata.json")) .resolve() .exists() ) assert ( Path(freqai.dk.data_path / str(freqai.dk.model_filename + "_trained_df.pkl")) .resolve() .exists() ) assert ( Path(freqai.dk.data_path / str(freqai.dk.model_filename + "_svm_model.joblib")) .resolve() .exists() ) shutil.rmtree(Path(freqai.dk.full_path)) def test_start_backtesting(mocker, default_conf): freqaiconf = freqai_conf(copy.deepcopy(default_conf)) freqaiconf.update({"timerange": "20180120-20180130"}) strategy = get_patched_freqai_strategy(mocker, freqaiconf) exchange = get_patched_exchange(mocker, freqaiconf) strategy.dp = DataProvider(freqaiconf, exchange) strategy.freqai_info = freqaiconf.get("freqai", {}) freqai = strategy.model.bridge freqai.live = False freqai.dk = FreqaiDataKitchen(freqaiconf, freqai.dd) timerange = TimeRange.parse_timerange("20180110-20180130") freqai.dk.load_all_pair_histories(timerange) sub_timerange = TimeRange.parse_timerange("20180110-20180130") corr_df, base_df = freqai.dk.get_base_and_corr_dataframes(sub_timerange, "LTC/BTC") df = freqai.dk.use_strategy_to_populate_indicators(strategy, corr_df, base_df, "LTC/BTC") metadata = {"pair": "ADA/BTC"} freqai.start_backtesting(df, metadata, freqai.dk) model_folders = [x for x in freqai.dd.full_path.iterdir() if x.is_dir()] assert len(model_folders) == 5 shutil.rmtree(Path(freqai.dk.full_path)) def test_start_backtesting_from_existing_folder(mocker, default_conf, caplog): freqaiconf = freqai_conf(copy.deepcopy(default_conf)) freqaiconf.update({"timerange": "20180120-20180130"}) strategy = get_patched_freqai_strategy(mocker, freqaiconf) exchange = get_patched_exchange(mocker, freqaiconf) strategy.dp = DataProvider(freqaiconf, exchange) strategy.freqai_info = freqaiconf.get("freqai", {}) freqai = strategy.model.bridge freqai.live = False freqai.dk = FreqaiDataKitchen(freqaiconf, freqai.dd) timerange = TimeRange.parse_timerange("20180110-20180130") freqai.dk.load_all_pair_histories(timerange) sub_timerange = TimeRange.parse_timerange("20180110-20180130") corr_df, base_df = freqai.dk.get_base_and_corr_dataframes(sub_timerange, "LTC/BTC") df = freqai.dk.use_strategy_to_populate_indicators(strategy, corr_df, base_df, "LTC/BTC") metadata = {"pair": "ADA/BTC"} freqai.start_backtesting(df, metadata, freqai.dk) model_folders = [x for x in freqai.dd.full_path.iterdir() if x.is_dir()] assert len(model_folders) == 5 # without deleting the exiting folder structure, re-run freqaiconf.update({"timerange": "20180120-20180130"}) strategy = get_patched_freqai_strategy(mocker, freqaiconf) exchange = get_patched_exchange(mocker, freqaiconf) strategy.dp = DataProvider(freqaiconf, exchange) strategy.freqai_info = freqaiconf.get("freqai", {}) freqai = strategy.model.bridge freqai.live = False freqai.dk = FreqaiDataKitchen(freqaiconf, freqai.dd) timerange = TimeRange.parse_timerange("20180110-20180130") freqai.dk.load_all_pair_histories(timerange) sub_timerange = TimeRange.parse_timerange("20180110-20180130") corr_df, base_df = freqai.dk.get_base_and_corr_dataframes(sub_timerange, "LTC/BTC") df = freqai.dk.use_strategy_to_populate_indicators(strategy, corr_df, base_df, "LTC/BTC") freqai.start_backtesting(df, metadata, freqai.dk) assert log_has( "Found model at user_data/models/uniqe-id100/sub-train-ADA1517097600/cb_ada_1517097600", caplog, ) shutil.rmtree(Path(freqai.dk.full_path))