lychiyu/freqtrade_origin
1
0
Fork 0
mirror of https://github.com/freqtrade/freqtrade.git synced 2024-11-10 10:21:59 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity
a605ae20a7
freqtrade_origin/tests/optimize/test_optimize_reports.py
Matthias a605ae20a7 feat: initial backtest table to rich
2024-07-09 06:43:12 +02:00

590 lines
22 KiB
Python
Raw Blame History

import re
from datetime import timedelta
from pathlib import Path
from shutil import copyfile
import joblib
import pandas as pd
import pytest
from freqtrade.configuration import TimeRange
from freqtrade.constants import BACKTEST_BREAKDOWNS, DATETIME_PRINT_FORMAT, LAST_BT_RESULT_FN
from freqtrade.data import history
from freqtrade.data.btanalysis import (
get_latest_backtest_filename,
load_backtest_data,
load_backtest_stats,
)
from freqtrade.edge import PairInfo
from freqtrade.enums import ExitType
from freqtrade.optimize.optimize_reports import (
generate_backtest_stats,
generate_daily_stats,
generate_edge_table,
generate_pair_metrics,
generate_periodic_breakdown_stats,
generate_strategy_comparison,
generate_trading_stats,
show_sorted_pairlist,
store_backtest_analysis_results,
store_backtest_stats,
text_table_bt_results,
text_table_strategy,
)
from freqtrade.optimize.optimize_reports.bt_output import text_table_tags
from freqtrade.optimize.optimize_reports.optimize_reports import (
_get_resample_from_period,
calc_streak,
generate_tag_metrics,
)
from freqtrade.resolvers.strategy_resolver import StrategyResolver
from freqtrade.util import dt_ts
from freqtrade.util.datetime_helpers import dt_from_ts, dt_utc
from tests.conftest import CURRENT_TEST_STRATEGY
from tests.data.test_history import _clean_test_file
def _backup_file(file: Path, copy_file: bool = False) -> None:
"""
Backup existing file to avoid deleting the user file
:param file: complete path to the file
:param copy_file: keep file in place too.
:return: None
"""
file_swp = str(file) + ".swp"
if file.is_file():
file.rename(file_swp)
if copy_file:
copyfile(file_swp, file)
def test_text_table_bt_results(capsys):
results = pd.DataFrame(
{
"pair": ["ETH/BTC", "ETH/BTC", "ETH/BTC"],
"profit_ratio": [0.1, 0.2, -0.05],
"profit_abs": [0.2, 0.4, -0.1],
"trade_duration": [10, 30, 20],
}
)
pair_results = generate_pair_metrics(
["ETH/BTC"], stake_currency="BTC", starting_balance=4, results=results
)
text_table_bt_results(pair_results, stake_currency="BTC", title="title")
text = capsys.readouterr().out
re.search(
r".* Pair .* Trades .* Avg Profit % .* Tot Profit BTC .* Tot Profit % .* "
r"Avg Duration .* Win Draw Loss Win% .*",
text,
)
re.search(
r".* ETH/BTC .* 3 .* 8.33 .* 0.50000000 .* 12.50 .* 0:20:00 .* 2 0 1 66.7 .*",
text,
)
re.search(
r".* TOTAL .* 3 .* 8.33 .* 0.50000000 .* 12.50 .* 0:20:00 .* 2 0 1 66.7 .*", text
)
def test_generate_backtest_stats(default_conf, testdatadir, tmp_path):
default_conf.update({"strategy": CURRENT_TEST_STRATEGY})
StrategyResolver.load_strategy(default_conf)
results = {
"DefStrat": {
"results": pd.DataFrame(
{
"pair": ["UNITTEST/BTC", "UNITTEST/BTC", "UNITTEST/BTC", "UNITTEST/BTC"],
"profit_ratio": [0.003312, 0.010801, 0.013803, 0.002780],
"profit_abs": [0.000003, 0.000011, 0.000014, 0.000003],
"open_date": [
dt_utc(2017, 11, 14, 19, 32, 00),
dt_utc(2017, 11, 14, 21, 36, 00),
dt_utc(2017, 11, 14, 22, 12, 00),
dt_utc(2017, 11, 14, 22, 44, 00),
],
"close_date": [
dt_utc(2017, 11, 14, 21, 35, 00),
dt_utc(2017, 11, 14, 22, 10, 00),
dt_utc(2017, 11, 14, 22, 43, 00),
dt_utc(2017, 11, 14, 22, 58, 00),
],
"open_rate": [0.002543, 0.003003, 0.003089, 0.003214],
"close_rate": [0.002546, 0.003014, 0.003103, 0.003217],
"trade_duration": [123, 34, 31, 14],
"is_open": [False, False, False, True],
"is_short": [False, False, False, False],
"stake_amount": [0.01, 0.01, 0.01, 0.01],
"exit_reason": [
ExitType.ROI.value,
ExitType.STOP_LOSS.value,
ExitType.ROI.value,
ExitType.FORCE_EXIT.value,
],
}
),
"config": default_conf,
"locks": [],
"final_balance": 1000.02,
"rejected_signals": 20,
"timedout_entry_orders": 0,
"timedout_exit_orders": 0,
"canceled_trade_entries": 0,
"canceled_entry_orders": 0,
"replaced_entry_orders": 0,
"backtest_start_time": dt_ts() // 1000,
"backtest_end_time": dt_ts() // 1000,
"run_id": "123",
}
}
timerange = TimeRange.parse_timerange("1510688220-1510700340")
min_date = dt_from_ts(1510688220)
max_date = dt_from_ts(1510700340)
btdata = history.load_data(
testdatadir, "1m", ["UNITTEST/BTC"], timerange=timerange, fill_up_missing=True
)
stats = generate_backtest_stats(btdata, results, min_date, max_date)
assert isinstance(stats, dict)
assert "strategy" in stats
assert "DefStrat" in stats["strategy"]
assert "strategy_comparison" in stats
strat_stats = stats["strategy"]["DefStrat"]
assert strat_stats["backtest_start"] == min_date.strftime(DATETIME_PRINT_FORMAT)
assert strat_stats["backtest_end"] == max_date.strftime(DATETIME_PRINT_FORMAT)
assert strat_stats["total_trades"] == len(results["DefStrat"]["results"])
# Above sample had no losing trade
assert strat_stats["max_drawdown_account"] == 0.0
# Retry with losing trade
results = {
"DefStrat": {
"results": pd.DataFrame(
{
"pair": ["UNITTEST/BTC", "UNITTEST/BTC", "UNITTEST/BTC", "UNITTEST/BTC"],
"profit_ratio": [0.003312, 0.010801, -0.013803, 0.002780],
"profit_abs": [0.000003, 0.000011, -0.000014, 0.000003],
"open_date": [
dt_utc(2017, 11, 14, 19, 32, 00),
dt_utc(2017, 11, 14, 21, 36, 00),
dt_utc(2017, 11, 14, 22, 12, 00),
dt_utc(2017, 11, 14, 22, 44, 00),
],
"close_date": [
dt_utc(2017, 11, 14, 21, 35, 00),
dt_utc(2017, 11, 14, 22, 10, 00),
dt_utc(2017, 11, 14, 22, 43, 00),
dt_utc(2017, 11, 14, 22, 58, 00),
],
"open_rate": [0.002543, 0.003003, 0.003089, 0.003214],
"close_rate": [0.002546, 0.003014, 0.0032903, 0.003217],
"trade_duration": [123, 34, 31, 14],
"is_open": [False, False, False, True],
"is_short": [False, False, False, False],
"stake_amount": [0.01, 0.01, 0.01, 0.01],
"exit_reason": [
ExitType.ROI.value,
ExitType.ROI.value,
ExitType.STOP_LOSS.value,
ExitType.FORCE_EXIT.value,
],
}
),
"config": default_conf,
"locks": [],
"final_balance": 1000.02,
"rejected_signals": 20,
"timedout_entry_orders": 0,
"timedout_exit_orders": 0,
"canceled_trade_entries": 0,
"canceled_entry_orders": 0,
"replaced_entry_orders": 0,
"backtest_start_time": dt_ts() // 1000,
"backtest_end_time": dt_ts() // 1000,
"run_id": "124",
}
}
stats = generate_backtest_stats(btdata, results, min_date, max_date)
assert isinstance(stats, dict)
assert "strategy" in stats
assert "DefStrat" in stats["strategy"]
assert "strategy_comparison" in stats
strat_stats = stats["strategy"]["DefStrat"]
assert pytest.approx(strat_stats["max_drawdown_account"]) == 1.399999e-08
assert strat_stats["drawdown_start"] == "2017-11-14 22:10:00"
assert strat_stats["drawdown_end"] == "2017-11-14 22:43:00"
assert strat_stats["drawdown_end_ts"] == 1510699380000
assert strat_stats["drawdown_start_ts"] == 1510697400000
assert strat_stats["pairlist"] == ["UNITTEST/BTC"]
# Test storing stats
filename = tmp_path / "btresult.json"
filename_last = tmp_path / LAST_BT_RESULT_FN
_backup_file(filename_last, copy_file=True)
assert not filename.is_file()
store_backtest_stats(filename, stats, "2022_01_01_15_05_13")
# get real Filename (it's btresult-<date>.json)
last_fn = get_latest_backtest_filename(filename_last.parent)
assert re.match(r"btresult-.*\.json", last_fn)
filename1 = tmp_path / last_fn
assert filename1.is_file()
content = filename1.read_text()
assert "max_drawdown_account" in content
assert "strategy" in content
assert "pairlist" in content
assert filename_last.is_file()
_clean_test_file(filename_last)
filename1.unlink()
def test_store_backtest_stats(testdatadir, mocker):
dump_mock = mocker.patch("freqtrade.optimize.optimize_reports.bt_storage.file_dump_json")
data = {"metadata": {}, "strategy": {}, "strategy_comparison": []}
store_backtest_stats(testdatadir, data, "2022_01_01_15_05_13")
assert dump_mock.call_count == 3
assert isinstance(dump_mock.call_args_list[0][0][0], Path)
assert str(dump_mock.call_args_list[0][0][0]).startswith(str(testdatadir / "backtest-result"))
dump_mock.reset_mock()
filename = testdatadir / "testresult.json"
store_backtest_stats(filename, data, "2022_01_01_15_05_13")
assert dump_mock.call_count == 3
assert isinstance(dump_mock.call_args_list[0][0][0], Path)
# result will be testdatadir / testresult-<timestamp>.json
assert str(dump_mock.call_args_list[0][0][0]).startswith(str(testdatadir / "testresult"))
def test_store_backtest_stats_real(tmp_path):
data = {"metadata": {}, "strategy": {}, "strategy_comparison": []}
store_backtest_stats(tmp_path, data, "2022_01_01_15_05_13")
assert (tmp_path / "backtest-result-2022_01_01_15_05_13.json").is_file()
assert (tmp_path / "backtest-result-2022_01_01_15_05_13.meta.json").is_file()
assert not (tmp_path / "backtest-result-2022_01_01_15_05_13_market_change.feather").is_file()
assert (tmp_path / LAST_BT_RESULT_FN).is_file()
fn = get_latest_backtest_filename(tmp_path)
assert fn == "backtest-result-2022_01_01_15_05_13.json"
store_backtest_stats(tmp_path, data, "2024_01_01_15_05_25", market_change_data=pd.DataFrame())
assert (tmp_path / "backtest-result-2024_01_01_15_05_25.json").is_file()
assert (tmp_path / "backtest-result-2024_01_01_15_05_25.meta.json").is_file()
assert (tmp_path / "backtest-result-2024_01_01_15_05_25_market_change.feather").is_file()
assert (tmp_path / LAST_BT_RESULT_FN).is_file()
# Last file reference should be updated
fn = get_latest_backtest_filename(tmp_path)
assert fn == "backtest-result-2024_01_01_15_05_25.json"
def test_store_backtest_candles(testdatadir, mocker):
dump_mock = mocker.patch("freqtrade.optimize.optimize_reports.bt_storage.file_dump_joblib")
candle_dict = {"DefStrat": {"UNITTEST/BTC": pd.DataFrame()}}
# mock directory exporting
store_backtest_analysis_results(testdatadir, candle_dict, {}, "2022_01_01_15_05_13")
assert dump_mock.call_count == 2
assert isinstance(dump_mock.call_args_list[0][0][0], Path)
assert str(dump_mock.call_args_list[0][0][0]).endswith("_signals.pkl")
dump_mock.reset_mock()
# mock file exporting
filename = Path(testdatadir / "testresult")
store_backtest_analysis_results(filename, candle_dict, {}, "2022_01_01_15_05_13")
assert dump_mock.call_count == 2
assert isinstance(dump_mock.call_args_list[0][0][0], Path)
# result will be testdatadir / testresult-<timestamp>_signals.pkl
assert str(dump_mock.call_args_list[0][0][0]).endswith("_signals.pkl")
dump_mock.reset_mock()
def test_write_read_backtest_candles(tmp_path):
candle_dict = {"DefStrat": {"UNITTEST/BTC": pd.DataFrame()}}
# test directory exporting
sample_date = "2022_01_01_15_05_13"
store_backtest_analysis_results(tmp_path, candle_dict, {}, sample_date)
stored_file = tmp_path / f"backtest-result-{sample_date}_signals.pkl"
with stored_file.open("rb") as scp:
pickled_signal_candles = joblib.load(scp)
assert pickled_signal_candles.keys() == candle_dict.keys()
assert pickled_signal_candles["DefStrat"].keys() == pickled_signal_candles["DefStrat"].keys()
assert pickled_signal_candles["DefStrat"]["UNITTEST/BTC"].equals(
pickled_signal_candles["DefStrat"]["UNITTEST/BTC"]
)
_clean_test_file(stored_file)
# test file exporting
filename = tmp_path / "testresult"
store_backtest_analysis_results(filename, candle_dict, {}, sample_date)
stored_file = tmp_path / f"testresult-{sample_date}_signals.pkl"
with stored_file.open("rb") as scp:
pickled_signal_candles = joblib.load(scp)
assert pickled_signal_candles.keys() == candle_dict.keys()
assert pickled_signal_candles["DefStrat"].keys() == pickled_signal_candles["DefStrat"].keys()
assert pickled_signal_candles["DefStrat"]["UNITTEST/BTC"].equals(
pickled_signal_candles["DefStrat"]["UNITTEST/BTC"]
)
_clean_test_file(stored_file)
def test_generate_pair_metrics():
results = pd.DataFrame(
{
"pair": ["ETH/BTC", "ETH/BTC"],
"profit_ratio": [0.1, 0.2],
"profit_abs": [0.2, 0.4],
"trade_duration": [10, 30],
"wins": [2, 0],
"draws": [0, 0],
"losses": [0, 0],
}
)
pair_results = generate_pair_metrics(
["ETH/BTC"], stake_currency="BTC", starting_balance=2, results=results
)
assert isinstance(pair_results, list)
assert len(pair_results) == 2
assert pair_results[-1]["key"] == "TOTAL"
assert (
pytest.approx(pair_results[-1]["profit_mean_pct"]) == pair_results[-1]["profit_mean"] * 100
)
assert pytest.approx(pair_results[-1]["profit_sum_pct"]) == pair_results[-1]["profit_sum"] * 100
def test_generate_daily_stats(testdatadir):
filename = testdatadir / "backtest_results/backtest-result.json"
bt_data = load_backtest_data(filename)
res = generate_daily_stats(bt_data)
assert isinstance(res, dict)
assert round(res["backtest_best_day"], 4) == 0.1796
assert round(res["backtest_worst_day"], 4) == -0.1468
assert res["winning_days"] == 19
assert res["draw_days"] == 0
assert res["losing_days"] == 2
# Select empty dataframe!
res = generate_daily_stats(bt_data.loc[bt_data["open_date"] == "2000-01-01", :])
assert isinstance(res, dict)
assert round(res["backtest_best_day"], 4) == 0.0
assert res["winning_days"] == 0
assert res["draw_days"] == 0
assert res["losing_days"] == 0
def test_generate_trading_stats(testdatadir):
filename = testdatadir / "backtest_results/backtest-result.json"
bt_data = load_backtest_data(filename)
res = generate_trading_stats(bt_data)
assert isinstance(res, dict)
assert res["winner_holding_avg"] == timedelta(seconds=1440)
assert res["loser_holding_avg"] == timedelta(days=1, seconds=21420)
assert "wins" in res
assert "losses" in res
assert "draws" in res
# Select empty dataframe!
res = generate_trading_stats(bt_data.loc[bt_data["open_date"] == "2000-01-01", :])
assert res["wins"] == 0
assert res["losses"] == 0
def test_calc_streak(testdatadir):
df = pd.DataFrame(
{
"profit_ratio": [0.05, -0.02, -0.03, -0.05, 0.01, 0.02, 0.03, 0.04, -0.02, -0.03],
}
)
# 4 consecutive wins, 3 consecutive losses
res = calc_streak(df)
assert res == (4, 3)
assert isinstance(res[0], int)
assert isinstance(res[1], int)
# invert situation
df1 = df.copy()
df1["profit_ratio"] = df1["profit_ratio"] * -1
assert calc_streak(df1) == (3, 4)
df_empty = pd.DataFrame(
{
"profit_ratio": [],
}
)
assert df_empty.empty
assert calc_streak(df_empty) == (0, 0)
filename = testdatadir / "backtest_results/backtest-result.json"
bt_data = load_backtest_data(filename)
assert calc_streak(bt_data) == (7, 18)
def test_text_table_exit_reason():
results = pd.DataFrame(
{
"pair": ["ETH/BTC", "ETH/BTC", "ETH/BTC"],
"profit_ratio": [0.1, 0.2, -0.1],
"profit_abs": [0.2, 0.4, -0.2],
"trade_duration": [10, 30, 10],
"wins": [2, 0, 0],
"draws": [0, 0, 0],
"losses": [0, 0, 1],
"exit_reason": [ExitType.ROI.value, ExitType.ROI.value, ExitType.STOP_LOSS.value],
}
)
exit_reason_stats = generate_tag_metrics(
"exit_reason", starting_balance=22, results=results, skip_nan=False
)
text = text_table_tags("exit_tag", exit_reason_stats, "BTC")
assert re.search(
r".* Exit Reason .* Exits .* Avg Profit % .* Tot Profit BTC .* Tot Profit % .* "
r"Avg Duration .* Win Draw Loss Win% .*",
text,
)
assert re.search(
r".* roi .* 2 .* 15.00 .* 0.60000000 .* 2.73 .* 0:20:00 .* 2 0 0 100 .*",
text,
)
assert re.search(
r".* stop_loss .* 1 .* -10.00 .* -0.20000000 .* -0.91 .* 0:10:00 .* 0 0 1 0 .*",
text,
)
assert re.search(
r".* TOTAL .* 3 .* 6.67 .* 0.40000000 .* 1.82 .* 0:17:00 .* 2 0 1 66.7 .*", text
)
def test_generate_sell_reason_stats():
results = pd.DataFrame(
{
"pair": ["ETH/BTC", "ETH/BTC", "ETH/BTC"],
"profit_ratio": [0.1, 0.2, -0.1],
"profit_abs": [0.2, 0.4, -0.2],
"trade_duration": [10, 30, 10],
"wins": [2, 0, 0],
"draws": [0, 0, 0],
"losses": [0, 0, 1],
"exit_reason": [ExitType.ROI.value, ExitType.ROI.value, ExitType.STOP_LOSS.value],
}
)
exit_reason_stats = generate_tag_metrics(
"exit_reason", starting_balance=22, results=results, skip_nan=False
)
roi_result = exit_reason_stats[0]
assert roi_result["key"] == "roi"
assert roi_result["trades"] == 2
assert pytest.approx(roi_result["profit_mean"]) == 0.15
assert roi_result["profit_mean_pct"] == round(roi_result["profit_mean"] * 100, 2)
assert pytest.approx(roi_result["profit_mean"]) == 0.15
assert roi_result["profit_mean_pct"] == round(roi_result["profit_mean"] * 100, 2)
stop_result = exit_reason_stats[1]
assert stop_result["key"] == "stop_loss"
assert stop_result["trades"] == 1
assert pytest.approx(stop_result["profit_mean"]) == -0.1
assert stop_result["profit_mean_pct"] == round(stop_result["profit_mean"] * 100, 2)
assert pytest.approx(stop_result["profit_mean"]) == -0.1
assert stop_result["profit_mean_pct"] == round(stop_result["profit_mean"] * 100, 2)
def test_text_table_strategy(testdatadir, capsys):
filename = testdatadir / "backtest_results/backtest-result_multistrat.json"
bt_res_data = load_backtest_stats(filename)
bt_res_data_comparison = bt_res_data.pop("strategy_comparison")
strategy_results = generate_strategy_comparison(bt_stats=bt_res_data["strategy"])
assert strategy_results == bt_res_data_comparison
text_table_strategy(strategy_results, "BTC", "STRATEGY SUMMARY")
captured = capsys.readouterr()
text = captured.out
assert re.search(
r".* Strategy .* Trades .* Avg Profit % .* Tot Profit BTC .* Tot Profit % .* "
r"Avg Duration .* Win Draw Loss Win% .* Drawdown .*",
text,
)
assert re.search(
r".*StrategyTestV2 .* 179 .* 0.08 .* 0.02608550 .* "
r"260.85 .* 3:40:00 .* 170 0 9 95.0 .* 0.00308222 BTC 8.67%.*",
text,
)
assert re.search(
r".*TestStrategy .* 179 .* 0.08 .* 0.02608550 .* "
r"260.85 .* 3:40:00 .* 170 0 9 95.0 .* 0.00308222 BTC 8.67%.*",
text,
)
def test_generate_edge_table():
results = {}
results["ETH/BTC"] = PairInfo(-0.01, 0.60, 2, 1, 3, 10, 60)
text = generate_edge_table(results)
assert text.count("+") == 7
assert text.count("| ETH/BTC |") == 1
assert re.search(r".* Risk Reward Ratio .* Required Risk Reward .* Expectancy .*", text)
def test_generate_periodic_breakdown_stats(testdatadir):
filename = testdatadir / "backtest_results/backtest-result.json"
bt_data = load_backtest_data(filename).to_dict(orient="records")
res = generate_periodic_breakdown_stats(bt_data, "day")
assert isinstance(res, list)
assert len(res) == 21
day = res[0]
assert "date" in day
assert "draws" in day
assert "loses" in day
assert "wins" in day
assert "profit_abs" in day
# Select empty dataframe!
res = generate_periodic_breakdown_stats([], "day")
assert res == []
def test__get_resample_from_period():
assert _get_resample_from_period("day") == "1d"
assert _get_resample_from_period("week") == "1W-MON"
assert _get_resample_from_period("month") == "1ME"
with pytest.raises(ValueError, match=r"Period noooo is not supported."):
_get_resample_from_period("noooo")
for period in BACKTEST_BREAKDOWNS:
assert isinstance(_get_resample_from_period(period), str)
def test_show_sorted_pairlist(testdatadir, default_conf, capsys):
filename = testdatadir / "backtest_results/backtest-result.json"
bt_data = load_backtest_stats(filename)
default_conf["backtest_show_pair_list"] = True
show_sorted_pairlist(default_conf, bt_data)
out, _err = capsys.readouterr()
assert "Pairs for Strategy StrategyTestV3: \n[" in out
assert "TOTAL" not in out
assert '"ETH/BTC", // ' in out
Reference in New Issue View Git Blame Copy Permalink