import shutil from pathlib import Path from unittest.mock import patch import pandas as pd import pytest from freqtrade.configuration import TimeRange from freqtrade.data.dataprovider import DataProvider from freqtrade.exceptions import OperationalException from freqtrade.freqai.data_kitchen import FreqaiDataKitchen from tests.conftest import get_patched_exchange from tests.freqai.conftest import get_patched_freqai_strategy def test_update_historic_data(mocker, freqai_conf): freqai_conf['runmode'] = 'backtest' strategy = get_patched_freqai_strategy(mocker, freqai_conf) exchange = get_patched_exchange(mocker, freqai_conf) strategy.dp = DataProvider(freqai_conf, exchange) freqai = strategy.freqai freqai.live = True freqai.dk = FreqaiDataKitchen(freqai_conf) freqai.dk.live = True timerange = TimeRange.parse_timerange("20180110-20180114") freqai.dd.load_all_pair_histories(timerange, freqai.dk) historic_candles = len(freqai.dd.historic_data["ADA/BTC"]["5m"]) dp_candles = len(strategy.dp.get_pair_dataframe("ADA/BTC", "5m")) candle_difference = dp_candles - historic_candles freqai.dk.pair = "ADA/BTC" freqai.dd.update_historic_data(strategy, freqai.dk) updated_historic_candles = len(freqai.dd.historic_data["ADA/BTC"]["5m"]) assert updated_historic_candles - historic_candles == candle_difference shutil.rmtree(Path(freqai.dk.full_path)) def test_load_all_pairs_histories(mocker, freqai_conf): strategy = get_patched_freqai_strategy(mocker, freqai_conf) exchange = get_patched_exchange(mocker, freqai_conf) strategy.dp = DataProvider(freqai_conf, exchange) freqai = strategy.freqai freqai.live = True freqai.dk = FreqaiDataKitchen(freqai_conf) freqai.dk.live = True timerange = TimeRange.parse_timerange("20180110-20180114") freqai.dd.load_all_pair_histories(timerange, freqai.dk) assert len(freqai.dd.historic_data.keys()) == len( freqai_conf.get("exchange", {}).get("pair_whitelist") ) assert len(freqai.dd.historic_data["ADA/BTC"]) == len( freqai_conf.get("freqai", {}).get("feature_parameters", {}).get("include_timeframes") ) shutil.rmtree(Path(freqai.dk.full_path)) def test_get_base_and_corr_dataframes(mocker, freqai_conf): strategy = get_patched_freqai_strategy(mocker, freqai_conf) exchange = get_patched_exchange(mocker, freqai_conf) strategy.dp = DataProvider(freqai_conf, exchange) freqai = strategy.freqai freqai.live = True freqai.dk = FreqaiDataKitchen(freqai_conf) freqai.dk.live = True timerange = TimeRange.parse_timerange("20180110-20180114") freqai.dd.load_all_pair_histories(timerange, freqai.dk) sub_timerange = TimeRange.parse_timerange("20180111-20180114") corr_df, base_df = freqai.dd.get_base_and_corr_dataframes(sub_timerange, "LTC/BTC", freqai.dk) num_tfs = len( freqai_conf.get("freqai", {}).get("feature_parameters", {}).get("include_timeframes") ) assert len(base_df.keys()) == num_tfs assert len(corr_df.keys()) == len( freqai_conf.get("freqai", {}).get("feature_parameters", {}).get("include_corr_pairlist") ) assert len(corr_df["ADA/BTC"].keys()) == num_tfs shutil.rmtree(Path(freqai.dk.full_path)) def test_use_strategy_to_populate_indicators(mocker, freqai_conf): strategy = get_patched_freqai_strategy(mocker, freqai_conf) exchange = get_patched_exchange(mocker, freqai_conf) strategy.dp = DataProvider(freqai_conf, exchange) strategy.freqai_info = freqai_conf.get("freqai", {}) freqai = strategy.freqai freqai.live = True freqai.dk = FreqaiDataKitchen(freqai_conf) freqai.dk.live = True timerange = TimeRange.parse_timerange("20180110-20180114") freqai.dd.load_all_pair_histories(timerange, freqai.dk) sub_timerange = TimeRange.parse_timerange("20180111-20180114") corr_df, base_df = freqai.dd.get_base_and_corr_dataframes(sub_timerange, "LTC/BTC", freqai.dk) df = freqai.dk.use_strategy_to_populate_indicators(strategy, corr_df, base_df, 'LTC/BTC') assert len(df.columns) == 33 shutil.rmtree(Path(freqai.dk.full_path)) def test_get_timerange_from_live_historic_predictions(mocker, freqai_conf): strategy = get_patched_freqai_strategy(mocker, freqai_conf) exchange = get_patched_exchange(mocker, freqai_conf) strategy.dp = DataProvider(freqai_conf, exchange) freqai = strategy.freqai freqai.live = False freqai.dk = FreqaiDataKitchen(freqai_conf) freqai.dk.live = False timerange = TimeRange.parse_timerange("20180126-20180130") freqai.dd.load_all_pair_histories(timerange, freqai.dk) sub_timerange = TimeRange.parse_timerange("20180128-20180130") _, base_df = freqai.dd.get_base_and_corr_dataframes(sub_timerange, "ADA/BTC", freqai.dk) base_df["5m"]["date_pred"] = base_df["5m"]["date"] freqai.dd.historic_predictions = {} freqai.dd.historic_predictions["ADA/USDT"] = base_df["5m"] freqai.dd.save_historic_predictions_to_disk() freqai.dd.save_global_metadata_to_disk({"start_dry_live_date": 1516406400}) timerange = freqai.dd.get_timerange_from_live_historic_predictions() assert timerange.startts == 1516406400 assert timerange.stopts == 1517356500 def test_get_timerange_from_backtesting_live_df_pred_not_found(mocker, freqai_conf): strategy = get_patched_freqai_strategy(mocker, freqai_conf) exchange = get_patched_exchange(mocker, freqai_conf) strategy.dp = DataProvider(freqai_conf, exchange) freqai = strategy.freqai with pytest.raises( OperationalException, match=r'Historic predictions not found.*' ): freqai.dd.get_timerange_from_live_historic_predictions() def test_set_initial_return_values(mocker, freqai_conf): """ Simple test of the set initial return values that ensures we are concatenating and ffilling values properly. """ strategy = get_patched_freqai_strategy(mocker, freqai_conf) exchange = get_patched_exchange(mocker, freqai_conf) strategy.dp = DataProvider(freqai_conf, exchange) freqai = strategy.freqai freqai.live = False freqai.dk = FreqaiDataKitchen(freqai_conf) # Setup pair = "BTC/USD" end_x = "2023-08-31" start_x_plus_1 = "2023-08-30" end_x_plus_5 = "2023-09-03" historic_data = { 'date_pred': pd.date_range(end=end_x, periods=5), 'value': range(1, 6) } new_data = { 'date': pd.date_range(start=start_x_plus_1, end=end_x_plus_5), 'value': range(6, 11) } freqai.dd.historic_predictions[pair] = pd.DataFrame(historic_data) new_pred_df = pd.DataFrame(new_data) dataframe = pd.DataFrame(new_data) # Action with patch('logging.Logger.warning') as mock_logger_warning: freqai.dd.set_initial_return_values(pair, new_pred_df, dataframe) # Assertions hist_pred_df = freqai.dd.historic_predictions[pair] model_return_df = freqai.dd.model_return_values[pair] assert hist_pred_df['date_pred'].iloc[-1] == pd.Timestamp(end_x_plus_5) assert 'date_pred' in hist_pred_df.columns assert hist_pred_df.shape[0] == 8 # compare values in model_return_df with hist_pred_df assert (model_return_df["value"].values == hist_pred_df.tail(len(dataframe))["value"].values).all() assert model_return_df.shape[0] == len(dataframe) # Ensure logger error is not called mock_logger_warning.assert_not_called() def test_set_initial_return_values_warning(mocker, freqai_conf): """ Simple test of set_initial_return_values that hits the warning associated with leaving a FreqAI bot offline so long that the exchange candles have no common date with the historic predictions """ strategy = get_patched_freqai_strategy(mocker, freqai_conf) exchange = get_patched_exchange(mocker, freqai_conf) strategy.dp = DataProvider(freqai_conf, exchange) freqai = strategy.freqai freqai.live = False freqai.dk = FreqaiDataKitchen(freqai_conf) # Setup pair = "BTC/USD" end_x = "2023-08-31" start_x_plus_1 = "2023-09-01" end_x_plus_5 = "2023-09-05" historic_data = { 'date_pred': pd.date_range(end=end_x, periods=5), 'value': range(1, 6) } new_data = { 'date': pd.date_range(start=start_x_plus_1, end=end_x_plus_5), 'value': range(6, 11) } freqai.dd.historic_predictions[pair] = pd.DataFrame(historic_data) new_pred_df = pd.DataFrame(new_data) dataframe = pd.DataFrame(new_data) # Action with patch('logging.Logger.warning') as mock_logger_warning: freqai.dd.set_initial_return_values(pair, new_pred_df, dataframe) # Assertions hist_pred_df = freqai.dd.historic_predictions[pair] model_return_df = freqai.dd.model_return_values[pair] assert hist_pred_df['date_pred'].iloc[-1] == pd.Timestamp(end_x_plus_5) assert 'date_pred' in hist_pred_df.columns assert hist_pred_df.shape[0] == 10 # compare values in model_return_df with hist_pred_df assert (model_return_df["value"].values == hist_pred_df.tail( len(dataframe))["value"].values).all() assert model_return_df.shape[0] == len(dataframe) # Ensure logger error is not called mock_logger_warning.assert_called()