mohammadian7/freqtrade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #8824 from freqtrade/refactor/optimize_reports

Browse Source 
Refactor/optimize reports
This commit is contained in:
Matthias
2023-06-25 19:29:22 +02:00
committed by GitHub
parent 2c36a09b4f 1717f86702
commit 4a886e1b97
6 changed files with 506 additions and 468 deletions
Download Patch File Download Diff File Expand all files Collapse all files
freqtrade/optimize/optimize_reports/__init__.py
+18
View File
@@ -0,0 +1,18 @@
# flake8: noqa: F401
from freqtrade.optimize.optimize_reports.bt_output import (generate_edge_table,
show_backtest_result,
show_backtest_results,
show_sorted_pairlist,
text_table_add_metrics,
text_table_bt_results,
text_table_exit_reason,
text_table_periodic_breakdown,
text_table_strategy, text_table_tags)
from freqtrade.optimize.optimize_reports.bt_storage import (store_backtest_analysis_results,
store_backtest_stats)
from freqtrade.optimize.optimize_reports.optimize_reports import (
generate_all_periodic_breakdown_stats, generate_backtest_stats, generate_daily_stats,
generate_exit_reason_stats, generate_pair_metrics, generate_periodic_breakdown_stats,
generate_rejected_signals, generate_strategy_comparison, generate_strategy_stats,
generate_tag_metrics, generate_trade_signal_candles, generate_trading_stats,
generate_wins_draws_losses)
freqtrade/optimize/optimize_reports/bt_output.py
+405
View File
@@ -0,0 +1,405 @@
import logging
from typing import Any, Dict, List
from tabulate import tabulate
from freqtrade.constants import UNLIMITED_STAKE_AMOUNT, Config
from freqtrade.misc import decimals_per_coin, round_coin_value
from freqtrade.optimize.optimize_reports.optimize_reports import (generate_periodic_breakdown_stats,
generate_wins_draws_losses)
logger = logging.getLogger(__name__)
def _get_line_floatfmt(stake_currency: str) -> List[str]:
"""
Generate floatformat (goes in line with _generate_result_line())
"""
return ['s', 'd', '.2f', '.2f', f'.{decimals_per_coin(stake_currency)}f',
'.2f', 'd', 's', 's']
def _get_line_header(first_column: str, stake_currency: str,
direction: str = 'Entries') -> List[str]:
"""
Generate header lines (goes in line with _generate_result_line())
"""
return [first_column, direction, 'Avg Profit %', 'Cum Profit %',
f'Tot Profit {stake_currency}', 'Tot Profit %', 'Avg Duration',
'Win Draw Loss Win%']
def text_table_bt_results(pair_results: List[Dict[str, Any]], stake_currency: str) -> str:
"""
Generates and returns a text table for the given backtest data and the results dataframe
:param pair_results: List of Dictionaries - one entry per pair + final TOTAL row
:param stake_currency: stake-currency - used to correctly name headers
:return: pretty printed table with tabulate as string
"""
headers = _get_line_header('Pair', stake_currency)
floatfmt = _get_line_floatfmt(stake_currency)
output = [[
t['key'], t['trades'], t['profit_mean_pct'], t['profit_sum_pct'], t['profit_total_abs'],
t['profit_total_pct'], t['duration_avg'],
generate_wins_draws_losses(t['wins'], t['draws'], t['losses'])
] for t in pair_results]
# Ignore type as floatfmt does allow tuples but mypy does not know that
return tabulate(output, headers=headers,
floatfmt=floatfmt, tablefmt="orgtbl", stralign="right")
def text_table_exit_reason(exit_reason_stats: List[Dict[str, Any]], stake_currency: str) -> str:
"""
Generate small table outlining Backtest results
:param sell_reason_stats: Exit reason metrics
:param stake_currency: Stakecurrency used
:return: pretty printed table with tabulate as string
"""
headers = [
'Exit Reason',
'Exits',
'Win Draws Loss Win%',
'Avg Profit %',
'Cum Profit %',
f'Tot Profit {stake_currency}',
'Tot Profit %',
]
output = [[
t.get('exit_reason', t.get('sell_reason')), t['trades'],
generate_wins_draws_losses(t['wins'], t['draws'], t['losses']),
t['profit_mean_pct'], t['profit_sum_pct'],
round_coin_value(t['profit_total_abs'], stake_currency, False),
t['profit_total_pct'],
] for t in exit_reason_stats]
return tabulate(output, headers=headers, tablefmt="orgtbl", stralign="right")
def text_table_tags(tag_type: str, tag_results: List[Dict[str, Any]], stake_currency: str) -> str:
"""
Generates and returns a text table for the given backtest data and the results dataframe
:param pair_results: List of Dictionaries - one entry per pair + final TOTAL row
:param stake_currency: stake-currency - used to correctly name headers
:return: pretty printed table with tabulate as string
"""
if (tag_type == "enter_tag"):
headers = _get_line_header("TAG", stake_currency)
else:
headers = _get_line_header("TAG", stake_currency, 'Exits')
floatfmt = _get_line_floatfmt(stake_currency)
output = [
[
t['key'] if t['key'] is not None and len(
t['key']) > 0 else "OTHER",
t['trades'],
t['profit_mean_pct'],
t['profit_sum_pct'],
t['profit_total_abs'],
t['profit_total_pct'],
t['duration_avg'],
generate_wins_draws_losses(
t['wins'],
t['draws'],
t['losses'])] for t in tag_results]
# Ignore type as floatfmt does allow tuples but mypy does not know that
return tabulate(output, headers=headers,
floatfmt=floatfmt, tablefmt="orgtbl", stralign="right")
def text_table_periodic_breakdown(days_breakdown_stats: List[Dict[str, Any]],
stake_currency: str, period: str) -> str:
"""
Generate small table with Backtest results by days
:param days_breakdown_stats: Days breakdown metrics
:param stake_currency: Stakecurrency used
:return: pretty printed table with tabulate as string
"""
headers = [
period.capitalize(),
f'Tot Profit {stake_currency}',
'Wins',
'Draws',
'Losses',
]
output = [[
d['date'], round_coin_value(d['profit_abs'], stake_currency, False),
d['wins'], d['draws'], d['loses'],
] for d in days_breakdown_stats]
return tabulate(output, headers=headers, tablefmt="orgtbl", stralign="right")
def text_table_strategy(strategy_results, stake_currency: str) -> str:
"""
Generate summary table per strategy
:param strategy_results: Dict of <Strategyname: DataFrame> containing results for all strategies
:param stake_currency: stake-currency - used to correctly name headers
:return: pretty printed table with tabulate as string
"""
floatfmt = _get_line_floatfmt(stake_currency)
headers = _get_line_header('Strategy', stake_currency)
# _get_line_header() is also used for per-pair summary. Per-pair drawdown is mostly useless
# therefore we slip this column in only for strategy summary here.
headers.append('Drawdown')
# Align drawdown string on the center two space separator.
if 'max_drawdown_account' in strategy_results[0]:
drawdown = [f'{t["max_drawdown_account"] * 100:.2f}' for t in strategy_results]
else:
# Support for prior backtest results
drawdown = [f'{t["max_drawdown_per"]:.2f}' for t in strategy_results]
dd_pad_abs = max([len(t['max_drawdown_abs']) for t in strategy_results])
dd_pad_per = max([len(dd) for dd in drawdown])
drawdown = [f'{t["max_drawdown_abs"]:>{dd_pad_abs}} {stake_currency} {dd:>{dd_pad_per}}%'
for t, dd in zip(strategy_results, drawdown)]
output = [[
t['key'], t['trades'], t['profit_mean_pct'], t['profit_sum_pct'], t['profit_total_abs'],
t['profit_total_pct'], t['duration_avg'],
generate_wins_draws_losses(t['wins'], t['draws'], t['losses']), drawdown]
for t, drawdown in zip(strategy_results, drawdown)]
# Ignore type as floatfmt does allow tuples but mypy does not know that
return tabulate(output, headers=headers,
floatfmt=floatfmt, tablefmt="orgtbl", stralign="right")
def text_table_add_metrics(strat_results: Dict) -> str:
if len(strat_results['trades']) > 0:
best_trade = max(strat_results['trades'], key=lambda x: x['profit_ratio'])
worst_trade = min(strat_results['trades'], key=lambda x: x['profit_ratio'])
short_metrics = [
('', ''), # Empty line to improve readability
('Long / Short',
f"{strat_results.get('trade_count_long', 'total_trades')} / "
f"{strat_results.get('trade_count_short', 0)}"),
('Total profit Long %', f"{strat_results['profit_total_long']:.2%}"),
('Total profit Short %', f"{strat_results['profit_total_short']:.2%}"),
('Absolute profit Long', round_coin_value(strat_results['profit_total_long_abs'],
strat_results['stake_currency'])),
('Absolute profit Short', round_coin_value(strat_results['profit_total_short_abs'],
strat_results['stake_currency'])),
] if strat_results.get('trade_count_short', 0) > 0 else []
drawdown_metrics = []
if 'max_relative_drawdown' in strat_results:
# Compatibility to show old hyperopt results
drawdown_metrics.append(
('Max % of account underwater', f"{strat_results['max_relative_drawdown']:.2%}")
)
drawdown_metrics.extend([
('Absolute Drawdown (Account)', f"{strat_results['max_drawdown_account']:.2%}")
if 'max_drawdown_account' in strat_results else (
'Drawdown', f"{strat_results['max_drawdown']:.2%}"),
('Absolute Drawdown', round_coin_value(strat_results['max_drawdown_abs'],
strat_results['stake_currency'])),
('Drawdown high', round_coin_value(strat_results['max_drawdown_high'],
strat_results['stake_currency'])),
('Drawdown low', round_coin_value(strat_results['max_drawdown_low'],
strat_results['stake_currency'])),
('Drawdown Start', strat_results['drawdown_start']),
('Drawdown End', strat_results['drawdown_end']),
])
entry_adjustment_metrics = [
('Canceled Trade Entries', strat_results.get('canceled_trade_entries', 'N/A')),
('Canceled Entry Orders', strat_results.get('canceled_entry_orders', 'N/A')),
('Replaced Entry Orders', strat_results.get('replaced_entry_orders', 'N/A')),
] if strat_results.get('canceled_entry_orders', 0) > 0 else []
# Newly added fields should be ignored if they are missing in strat_results. hyperopt-show
# command stores these results and newer version of freqtrade must be able to handle old
# results with missing new fields.
metrics = [
('Backtesting from', strat_results['backtest_start']),
('Backtesting to', strat_results['backtest_end']),
('Max open trades', strat_results['max_open_trades']),
('', ''), # Empty line to improve readability
('Total/Daily Avg Trades',
f"{strat_results['total_trades']} / {strat_results['trades_per_day']}"),
('Starting balance', round_coin_value(strat_results['starting_balance'],
strat_results['stake_currency'])),
('Final balance', round_coin_value(strat_results['final_balance'],
strat_results['stake_currency'])),
('Absolute profit ', round_coin_value(strat_results['profit_total_abs'],
strat_results['stake_currency'])),
('Total profit %', f"{strat_results['profit_total']:.2%}"),
('CAGR %', f"{strat_results['cagr']:.2%}" if 'cagr' in strat_results else 'N/A'),
('Sortino', f"{strat_results['sortino']:.2f}" if 'sortino' in strat_results else 'N/A'),
('Sharpe', f"{strat_results['sharpe']:.2f}" if 'sharpe' in strat_results else 'N/A'),
('Calmar', f"{strat_results['calmar']:.2f}" if 'calmar' in strat_results else 'N/A'),
('Profit factor', f'{strat_results["profit_factor"]:.2f}' if 'profit_factor'
in strat_results else 'N/A'),
('Expectancy', f"{strat_results['expectancy']:.2f}" if 'expectancy'
in strat_results else 'N/A'),
('Trades per day', strat_results['trades_per_day']),
('Avg. daily profit %',
f"{(strat_results['profit_total'] / strat_results['backtest_days']):.2%}"),
('Avg. stake amount', round_coin_value(strat_results['avg_stake_amount'],
strat_results['stake_currency'])),
('Total trade volume', round_coin_value(strat_results['total_volume'],
strat_results['stake_currency'])),
*short_metrics,
('', ''), # Empty line to improve readability
('Best Pair', f"{strat_results['best_pair']['key']} "
f"{strat_results['best_pair']['profit_sum']:.2%}"),
('Worst Pair', f"{strat_results['worst_pair']['key']} "
f"{strat_results['worst_pair']['profit_sum']:.2%}"),
('Best trade', f"{best_trade['pair']} {best_trade['profit_ratio']:.2%}"),
('Worst trade', f"{worst_trade['pair']} "
f"{worst_trade['profit_ratio']:.2%}"),
('Best day', round_coin_value(strat_results['backtest_best_day_abs'],
strat_results['stake_currency'])),
('Worst day', round_coin_value(strat_results['backtest_worst_day_abs'],
strat_results['stake_currency'])),
('Days win/draw/lose', f"{strat_results['winning_days']} / "
f"{strat_results['draw_days']} / {strat_results['losing_days']}"),
('Avg. Duration Winners', f"{strat_results['winner_holding_avg']}"),
('Avg. Duration Loser', f"{strat_results['loser_holding_avg']}"),
('Rejected Entry signals', strat_results.get('rejected_signals', 'N/A')),
('Entry/Exit Timeouts',
f"{strat_results.get('timedout_entry_orders', 'N/A')} / "
f"{strat_results.get('timedout_exit_orders', 'N/A')}"),
*entry_adjustment_metrics,
('', ''), # Empty line to improve readability
('Min balance', round_coin_value(strat_results['csum_min'],
strat_results['stake_currency'])),
('Max balance', round_coin_value(strat_results['csum_max'],
strat_results['stake_currency'])),
*drawdown_metrics,
('Market change', f"{strat_results['market_change']:.2%}"),
]
return tabulate(metrics, headers=["Metric", "Value"], tablefmt="orgtbl")
else:
start_balance = round_coin_value(strat_results['starting_balance'],
strat_results['stake_currency'])
stake_amount = round_coin_value(
strat_results['stake_amount'], strat_results['stake_currency']
) if strat_results['stake_amount'] != UNLIMITED_STAKE_AMOUNT else 'unlimited'
message = ("No trades made. "
f"Your starting balance was {start_balance}, "
f"and your stake was {stake_amount}."
)
return message
def show_backtest_result(strategy: str, results: Dict[str, Any], stake_currency: str,
backtest_breakdown=[]):
"""
Print results for one strategy
"""
# Print results
print(f"Result for strategy {strategy}")
table = text_table_bt_results(results['results_per_pair'], stake_currency=stake_currency)
if isinstance(table, str):
print(' BACKTESTING REPORT '.center(len(table.splitlines()[0]), '='))
print(table)
table = text_table_bt_results(results['left_open_trades'], stake_currency=stake_currency)
if isinstance(table, str) and len(table) > 0:
print(' LEFT OPEN TRADES REPORT '.center(len(table.splitlines()[0]), '='))
print(table)
if (results.get('results_per_enter_tag') is not None
or results.get('results_per_buy_tag') is not None):
# results_per_buy_tag is deprecated and should be removed 2 versions after short golive.
table = text_table_tags(
"enter_tag",
results.get('results_per_enter_tag', results.get('results_per_buy_tag')),
stake_currency=stake_currency)
if isinstance(table, str) and len(table) > 0:
print(' ENTER TAG STATS '.center(len(table.splitlines()[0]), '='))
print(table)
exit_reasons = results.get('exit_reason_summary', results.get('sell_reason_summary'))
table = text_table_exit_reason(exit_reason_stats=exit_reasons,
stake_currency=stake_currency)
if isinstance(table, str) and len(table) > 0:
print(' EXIT REASON STATS '.center(len(table.splitlines()[0]), '='))
print(table)
for period in backtest_breakdown:
if period in results.get('periodic_breakdown', {}):
days_breakdown_stats = results['periodic_breakdown'][period]
else:
days_breakdown_stats = generate_periodic_breakdown_stats(
trade_list=results['trades'], period=period)
table = text_table_periodic_breakdown(days_breakdown_stats=days_breakdown_stats,
stake_currency=stake_currency, period=period)
if isinstance(table, str) and len(table) > 0:
print(f' {period.upper()} BREAKDOWN '.center(len(table.splitlines()[0]), '='))
print(table)
table = text_table_add_metrics(results)
if isinstance(table, str) and len(table) > 0:
print(' SUMMARY METRICS '.center(len(table.splitlines()[0]), '='))
print(table)
if isinstance(table, str) and len(table) > 0:
print('=' * len(table.splitlines()[0]))
print()
def show_backtest_results(config: Config, backtest_stats: Dict):
stake_currency = config['stake_currency']
for strategy, results in backtest_stats['strategy'].items():
show_backtest_result(
strategy, results, stake_currency,
config.get('backtest_breakdown', []))
if len(backtest_stats['strategy']) > 0:
# Print Strategy summary table
table = text_table_strategy(backtest_stats['strategy_comparison'], stake_currency)
print(f"Backtested {results['backtest_start']} -> {results['backtest_end']} |"
f" Max open trades : {results['max_open_trades']}")
print(' STRATEGY SUMMARY '.center(len(table.splitlines()[0]), '='))
print(table)
print('=' * len(table.splitlines()[0]))
print('\nFor more details, please look at the detail tables above')
def show_sorted_pairlist(config: Config, backtest_stats: Dict):
if config.get('backtest_show_pair_list', False):
for strategy, results in backtest_stats['strategy'].items():
print(f"Pairs for Strategy {strategy}: \n[")
for result in results['results_per_pair']:
if result["key"] != 'TOTAL':
print(f'"{result["key"]}", // {result["profit_mean"]:.2%}')
print("]")
def generate_edge_table(results: dict) -> str:
floatfmt = ('s', '.10g', '.2f', '.2f', '.2f', '.2f', 'd', 'd', 'd')
tabular_data = []
headers = ['Pair', 'Stoploss', 'Win Rate', 'Risk Reward Ratio',
'Required Risk Reward', 'Expectancy', 'Total Number of Trades',
'Average Duration (min)']
for result in results.items():
if result[1].nb_trades > 0:
tabular_data.append([
result[0],
result[1].stoploss,
result[1].winrate,
result[1].risk_reward_ratio,
result[1].required_risk_reward,
result[1].expectancy,
result[1].nb_trades,
round(result[1].avg_trade_duration)
])
# Ignore type as floatfmt does allow tuples but mypy does not know that
return tabulate(tabular_data, headers=headers,
floatfmt=floatfmt, tablefmt="orgtbl", stralign="right")
freqtrade/optimize/optimize_reports/bt_storage.py
+71
View File
@@ -0,0 +1,71 @@
import logging
from pathlib import Path
from typing import Dict
from pandas import DataFrame
from freqtrade.constants import LAST_BT_RESULT_FN
from freqtrade.misc import file_dump_joblib, file_dump_json
from freqtrade.optimize.backtest_caching import get_backtest_metadata_filename
logger = logging.getLogger(__name__)
def store_backtest_stats(
recordfilename: Path, stats: Dict[str, DataFrame], dtappendix: str) -> None:
"""
Stores backtest results
:param recordfilename: Path object, which can either be a filename or a directory.
Filenames will be appended with a timestamp right before the suffix
while for directories, <directory>/backtest-result-<datetime>.json will be used as filename
:param stats: Dataframe containing the backtesting statistics
:param dtappendix: Datetime to use for the filename
"""
if recordfilename.is_dir():
filename = (recordfilename / f'backtest-result-{dtappendix}.json')
else:
filename = Path.joinpath(
recordfilename.parent, f'{recordfilename.stem}-{dtappendix}'
).with_suffix(recordfilename.suffix)
# Store metadata separately.
file_dump_json(get_backtest_metadata_filename(filename), stats['metadata'])
del stats['metadata']
file_dump_json(filename, stats)
latest_filename = Path.joinpath(filename.parent, LAST_BT_RESULT_FN)
file_dump_json(latest_filename, {'latest_backtest': str(filename.name)})
def _store_backtest_analysis_data(
recordfilename: Path, data: Dict[str, Dict],
dtappendix: str, name: str) -> Path:
"""
Stores backtest trade candles for analysis
:param recordfilename: Path object, which can either be a filename or a directory.
Filenames will be appended with a timestamp right before the suffix
while for directories, <directory>/backtest-result-<datetime>_<name>.pkl will be used
as filename
:param candles: Dict containing the backtesting data for analysis
:param dtappendix: Datetime to use for the filename
:param name: Name to use for the file, e.g. signals, rejected
"""
if recordfilename.is_dir():
filename = (recordfilename / f'backtest-result-{dtappendix}_{name}.pkl')
else:
filename = Path.joinpath(
recordfilename.parent, f'{recordfilename.stem}-{dtappendix}_{name}.pkl'
)
file_dump_joblib(filename, data)
return filename
def store_backtest_analysis_results(
recordfilename: Path, candles: Dict[str, Dict], trades: Dict[str, Dict],
dtappendix: str) -> None:
_store_backtest_analysis_data(recordfilename, candles, dtappendix, "signals")
_store_backtest_analysis_data(recordfilename, trades, dtappendix, "rejected")
freqtrade/optimize/optimize_reports.py → freqtrade/optimize/optimize_reports/optimize_reports.py
+2 -462
View File
@@ -1,83 +1,20 @@
import logging import logging
from copy import deepcopy from copy import deepcopy
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Union from typing import Any, Dict, List, Union
from pandas import DataFrame, concat, to_datetime from pandas import DataFrame, concat, to_datetime
from tabulate import tabulate
from freqtrade.constants import (BACKTEST_BREAKDOWNS, DATETIME_PRINT_FORMAT, LAST_BT_RESULT_FN, from freqtrade.constants import BACKTEST_BREAKDOWNS, DATETIME_PRINT_FORMAT, IntOrInf
UNLIMITED_STAKE_AMOUNT, Config, IntOrInf)
from freqtrade.data.metrics import (calculate_cagr, calculate_calmar, calculate_csum, from freqtrade.data.metrics import (calculate_cagr, calculate_calmar, calculate_csum,
calculate_expectancy, calculate_market_change, calculate_expectancy, calculate_market_change,
calculate_max_drawdown, calculate_sharpe, calculate_sortino) calculate_max_drawdown, calculate_sharpe, calculate_sortino)
from freqtrade.misc import decimals_per_coin, file_dump_joblib, file_dump_json, round_coin_value from freqtrade.misc import decimals_per_coin, round_coin_value
from freqtrade.optimize.backtest_caching import get_backtest_metadata_filename
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def store_backtest_stats(
recordfilename: Path, stats: Dict[str, DataFrame], dtappendix: str) -> None:
"""
Stores backtest results
:param recordfilename: Path object, which can either be a filename or a directory.
Filenames will be appended with a timestamp right before the suffix
while for directories, <directory>/backtest-result-<datetime>.json will be used as filename
:param stats: Dataframe containing the backtesting statistics
:param dtappendix: Datetime to use for the filename
"""
if recordfilename.is_dir():
filename = (recordfilename / f'backtest-result-{dtappendix}.json')
else:
filename = Path.joinpath(
recordfilename.parent, f'{recordfilename.stem}-{dtappendix}'
).with_suffix(recordfilename.suffix)
# Store metadata separately.
file_dump_json(get_backtest_metadata_filename(filename), stats['metadata'])
del stats['metadata']
file_dump_json(filename, stats)
latest_filename = Path.joinpath(filename.parent, LAST_BT_RESULT_FN)
file_dump_json(latest_filename, {'latest_backtest': str(filename.name)})
def _store_backtest_analysis_data(
recordfilename: Path, data: Dict[str, Dict],
dtappendix: str, name: str) -> Path:
"""
Stores backtest trade candles for analysis
:param recordfilename: Path object, which can either be a filename or a directory.
Filenames will be appended with a timestamp right before the suffix
while for directories, <directory>/backtest-result-<datetime>_<name>.pkl will be used
as filename
:param candles: Dict containing the backtesting data for analysis
:param dtappendix: Datetime to use for the filename
:param name: Name to use for the file, e.g. signals, rejected
"""
if recordfilename.is_dir():
filename = (recordfilename / f'backtest-result-{dtappendix}_{name}.pkl')
else:
filename = Path.joinpath(
recordfilename.parent, f'{recordfilename.stem}-{dtappendix}_{name}.pkl'
)
file_dump_joblib(filename, data)
return filename
def store_backtest_analysis_results(
recordfilename: Path, candles: Dict[str, Dict], trades: Dict[str, Dict],
dtappendix: str) -> None:
_store_backtest_analysis_data(recordfilename, candles, dtappendix, "signals")
_store_backtest_analysis_data(recordfilename, trades, dtappendix, "rejected")
def generate_trade_signal_candles(preprocessed_df: Dict[str, DataFrame], def generate_trade_signal_candles(preprocessed_df: Dict[str, DataFrame],
bt_results: Dict[str, Any]) -> DataFrame: bt_results: Dict[str, Any]) -> DataFrame:
signal_candles_only = {} signal_candles_only = {}
@@ -120,24 +57,6 @@ def generate_rejected_signals(preprocessed_df: Dict[str, DataFrame],
return rejected_candles_only return rejected_candles_only
def _get_line_floatfmt(stake_currency: str) -> List[str]:
"""
Generate floatformat (goes in line with _generate_result_line())
"""
return ['s', 'd', '.2f', '.2f', f'.{decimals_per_coin(stake_currency)}f',
'.2f', 'd', 's', 's']
def _get_line_header(first_column: str, stake_currency: str,
direction: str = 'Entries') -> List[str]:
"""
Generate header lines (goes in line with _generate_result_line())
"""
return [first_column, direction, 'Avg Profit %', 'Cum Profit %',
f'Tot Profit {stake_currency}', 'Tot Profit %', 'Avg Duration',
'Win Draw Loss Win%']
def generate_wins_draws_losses(wins, draws, losses): def generate_wins_draws_losses(wins, draws, losses):
if wins > 0 and losses == 0: if wins > 0 and losses == 0:
wl_ratio = '100' wl_ratio = '100'
@@ -295,31 +214,6 @@ def generate_strategy_comparison(bt_stats: Dict) -> List[Dict]:
return tabular_data return tabular_data
def generate_edge_table(results: dict) -> str:
floatfmt = ('s', '.10g', '.2f', '.2f', '.2f', '.2f', 'd', 'd', 'd')
tabular_data = []
headers = ['Pair', 'Stoploss', 'Win Rate', 'Risk Reward Ratio',
'Required Risk Reward', 'Expectancy', 'Total Number of Trades',
'Average Duration (min)']
for result in results.items():
if result[1].nb_trades > 0:
tabular_data.append([
result[0],
result[1].stoploss,
result[1].winrate,
result[1].risk_reward_ratio,
result[1].required_risk_reward,
result[1].expectancy,
result[1].nb_trades,
round(result[1].avg_trade_duration)
])
# Ignore type as floatfmt does allow tuples but mypy does not know that
return tabulate(tabular_data, headers=headers,
floatfmt=floatfmt, tablefmt="orgtbl", stralign="right")
def _get_resample_from_period(period: str) -> str: def _get_resample_from_period(period: str) -> str:
if period == 'day': if period == 'day':
return '1d' return '1d'
@@ -652,357 +546,3 @@ def generate_backtest_stats(btdata: Dict[str, DataFrame],
result['strategy_comparison'] = strategy_results result['strategy_comparison'] = strategy_results
return result return result
###
# Start output section
###
def text_table_bt_results(pair_results: List[Dict[str, Any]], stake_currency: str) -> str:
"""
Generates and returns a text table for the given backtest data and the results dataframe
:param pair_results: List of Dictionaries - one entry per pair + final TOTAL row
:param stake_currency: stake-currency - used to correctly name headers
:return: pretty printed table with tabulate as string
"""
headers = _get_line_header('Pair', stake_currency)
floatfmt = _get_line_floatfmt(stake_currency)
output = [[
t['key'], t['trades'], t['profit_mean_pct'], t['profit_sum_pct'], t['profit_total_abs'],
t['profit_total_pct'], t['duration_avg'],
generate_wins_draws_losses(t['wins'], t['draws'], t['losses'])
] for t in pair_results]
# Ignore type as floatfmt does allow tuples but mypy does not know that
return tabulate(output, headers=headers,
floatfmt=floatfmt, tablefmt="orgtbl", stralign="right")
def text_table_exit_reason(exit_reason_stats: List[Dict[str, Any]], stake_currency: str) -> str:
"""
Generate small table outlining Backtest results
:param sell_reason_stats: Exit reason metrics
:param stake_currency: Stakecurrency used
:return: pretty printed table with tabulate as string
"""
headers = [
'Exit Reason',
'Exits',
'Win Draws Loss Win%',
'Avg Profit %',
'Cum Profit %',
f'Tot Profit {stake_currency}',
'Tot Profit %',
]
output = [[
t.get('exit_reason', t.get('sell_reason')), t['trades'],
generate_wins_draws_losses(t['wins'], t['draws'], t['losses']),
t['profit_mean_pct'], t['profit_sum_pct'],
round_coin_value(t['profit_total_abs'], stake_currency, False),
t['profit_total_pct'],
] for t in exit_reason_stats]
return tabulate(output, headers=headers, tablefmt="orgtbl", stralign="right")
def text_table_tags(tag_type: str, tag_results: List[Dict[str, Any]], stake_currency: str) -> str:
"""
Generates and returns a text table for the given backtest data and the results dataframe
:param pair_results: List of Dictionaries - one entry per pair + final TOTAL row
:param stake_currency: stake-currency - used to correctly name headers
:return: pretty printed table with tabulate as string
"""
if (tag_type == "enter_tag"):
headers = _get_line_header("TAG", stake_currency)
else:
headers = _get_line_header("TAG", stake_currency, 'Exits')
floatfmt = _get_line_floatfmt(stake_currency)
output = [
[
t['key'] if t['key'] is not None and len(
t['key']) > 0 else "OTHER",
t['trades'],
t['profit_mean_pct'],
t['profit_sum_pct'],
t['profit_total_abs'],
t['profit_total_pct'],
t['duration_avg'],
generate_wins_draws_losses(
t['wins'],
t['draws'],
t['losses'])] for t in tag_results]
# Ignore type as floatfmt does allow tuples but mypy does not know that
return tabulate(output, headers=headers,
floatfmt=floatfmt, tablefmt="orgtbl", stralign="right")
def text_table_periodic_breakdown(days_breakdown_stats: List[Dict[str, Any]],
stake_currency: str, period: str) -> str:
"""
Generate small table with Backtest results by days
:param days_breakdown_stats: Days breakdown metrics
:param stake_currency: Stakecurrency used
:return: pretty printed table with tabulate as string
"""
headers = [
period.capitalize(),
f'Tot Profit {stake_currency}',
'Wins',
'Draws',
'Losses',
]
output = [[
d['date'], round_coin_value(d['profit_abs'], stake_currency, False),
d['wins'], d['draws'], d['loses'],
] for d in days_breakdown_stats]
return tabulate(output, headers=headers, tablefmt="orgtbl", stralign="right")
def text_table_strategy(strategy_results, stake_currency: str) -> str:
"""
Generate summary table per strategy
:param strategy_results: Dict of <Strategyname: DataFrame> containing results for all strategies
:param stake_currency: stake-currency - used to correctly name headers
:return: pretty printed table with tabulate as string
"""
floatfmt = _get_line_floatfmt(stake_currency)
headers = _get_line_header('Strategy', stake_currency)
# _get_line_header() is also used for per-pair summary. Per-pair drawdown is mostly useless
# therefore we slip this column in only for strategy summary here.
headers.append('Drawdown')
# Align drawdown string on the center two space separator.
if 'max_drawdown_account' in strategy_results[0]:
drawdown = [f'{t["max_drawdown_account"] * 100:.2f}' for t in strategy_results]
else:
# Support for prior backtest results
drawdown = [f'{t["max_drawdown_per"]:.2f}' for t in strategy_results]
dd_pad_abs = max([len(t['max_drawdown_abs']) for t in strategy_results])
dd_pad_per = max([len(dd) for dd in drawdown])
drawdown = [f'{t["max_drawdown_abs"]:>{dd_pad_abs}} {stake_currency} {dd:>{dd_pad_per}}%'
for t, dd in zip(strategy_results, drawdown)]
output = [[
t['key'], t['trades'], t['profit_mean_pct'], t['profit_sum_pct'], t['profit_total_abs'],
t['profit_total_pct'], t['duration_avg'],
generate_wins_draws_losses(t['wins'], t['draws'], t['losses']), drawdown]
for t, drawdown in zip(strategy_results, drawdown)]
# Ignore type as floatfmt does allow tuples but mypy does not know that
return tabulate(output, headers=headers,
floatfmt=floatfmt, tablefmt="orgtbl", stralign="right")
def text_table_add_metrics(strat_results: Dict) -> str:
if len(strat_results['trades']) > 0:
best_trade = max(strat_results['trades'], key=lambda x: x['profit_ratio'])
worst_trade = min(strat_results['trades'], key=lambda x: x['profit_ratio'])
short_metrics = [
('', ''), # Empty line to improve readability
('Long / Short',
f"{strat_results.get('trade_count_long', 'total_trades')} / "
f"{strat_results.get('trade_count_short', 0)}"),
('Total profit Long %', f"{strat_results['profit_total_long']:.2%}"),
('Total profit Short %', f"{strat_results['profit_total_short']:.2%}"),
('Absolute profit Long', round_coin_value(strat_results['profit_total_long_abs'],
strat_results['stake_currency'])),
('Absolute profit Short', round_coin_value(strat_results['profit_total_short_abs'],
strat_results['stake_currency'])),
] if strat_results.get('trade_count_short', 0) > 0 else []
drawdown_metrics = []
if 'max_relative_drawdown' in strat_results:
# Compatibility to show old hyperopt results
drawdown_metrics.append(
('Max % of account underwater', f"{strat_results['max_relative_drawdown']:.2%}")
)
drawdown_metrics.extend([
('Absolute Drawdown (Account)', f"{strat_results['max_drawdown_account']:.2%}")
if 'max_drawdown_account' in strat_results else (
'Drawdown', f"{strat_results['max_drawdown']:.2%}"),
('Absolute Drawdown', round_coin_value(strat_results['max_drawdown_abs'],
strat_results['stake_currency'])),
('Drawdown high', round_coin_value(strat_results['max_drawdown_high'],
strat_results['stake_currency'])),
('Drawdown low', round_coin_value(strat_results['max_drawdown_low'],
strat_results['stake_currency'])),
('Drawdown Start', strat_results['drawdown_start']),
('Drawdown End', strat_results['drawdown_end']),
])
entry_adjustment_metrics = [
('Canceled Trade Entries', strat_results.get('canceled_trade_entries', 'N/A')),
('Canceled Entry Orders', strat_results.get('canceled_entry_orders', 'N/A')),
('Replaced Entry Orders', strat_results.get('replaced_entry_orders', 'N/A')),
] if strat_results.get('canceled_entry_orders', 0) > 0 else []
# Newly added fields should be ignored if they are missing in strat_results. hyperopt-show
# command stores these results and newer version of freqtrade must be able to handle old
# results with missing new fields.
metrics = [
('Backtesting from', strat_results['backtest_start']),
('Backtesting to', strat_results['backtest_end']),
('Max open trades', strat_results['max_open_trades']),
('', ''), # Empty line to improve readability
('Total/Daily Avg Trades',
f"{strat_results['total_trades']} / {strat_results['trades_per_day']}"),
('Starting balance', round_coin_value(strat_results['starting_balance'],
strat_results['stake_currency'])),
('Final balance', round_coin_value(strat_results['final_balance'],
strat_results['stake_currency'])),
('Absolute profit ', round_coin_value(strat_results['profit_total_abs'],
strat_results['stake_currency'])),
('Total profit %', f"{strat_results['profit_total']:.2%}"),
('CAGR %', f"{strat_results['cagr']:.2%}" if 'cagr' in strat_results else 'N/A'),
('Sortino', f"{strat_results['sortino']:.2f}" if 'sortino' in strat_results else 'N/A'),
('Sharpe', f"{strat_results['sharpe']:.2f}" if 'sharpe' in strat_results else 'N/A'),
('Calmar', f"{strat_results['calmar']:.2f}" if 'calmar' in strat_results else 'N/A'),
('Profit factor', f'{strat_results["profit_factor"]:.2f}' if 'profit_factor'
in strat_results else 'N/A'),
('Expectancy', f"{strat_results['expectancy']:.2f}" if 'expectancy'
in strat_results else 'N/A'),
('Trades per day', strat_results['trades_per_day']),
('Avg. daily profit %',
f"{(strat_results['profit_total'] / strat_results['backtest_days']):.2%}"),
('Avg. stake amount', round_coin_value(strat_results['avg_stake_amount'],
strat_results['stake_currency'])),
('Total trade volume', round_coin_value(strat_results['total_volume'],
strat_results['stake_currency'])),
*short_metrics,
('', ''), # Empty line to improve readability
('Best Pair', f"{strat_results['best_pair']['key']} "
f"{strat_results['best_pair']['profit_sum']:.2%}"),
('Worst Pair', f"{strat_results['worst_pair']['key']} "
f"{strat_results['worst_pair']['profit_sum']:.2%}"),
('Best trade', f"{best_trade['pair']} {best_trade['profit_ratio']:.2%}"),
('Worst trade', f"{worst_trade['pair']} "
f"{worst_trade['profit_ratio']:.2%}"),
('Best day', round_coin_value(strat_results['backtest_best_day_abs'],
strat_results['stake_currency'])),
('Worst day', round_coin_value(strat_results['backtest_worst_day_abs'],
strat_results['stake_currency'])),
('Days win/draw/lose', f"{strat_results['winning_days']} / "
f"{strat_results['draw_days']} / {strat_results['losing_days']}"),
('Avg. Duration Winners', f"{strat_results['winner_holding_avg']}"),
('Avg. Duration Loser', f"{strat_results['loser_holding_avg']}"),
('Rejected Entry signals', strat_results.get('rejected_signals', 'N/A')),
('Entry/Exit Timeouts',
f"{strat_results.get('timedout_entry_orders', 'N/A')} / "
f"{strat_results.get('timedout_exit_orders', 'N/A')}"),
*entry_adjustment_metrics,
('', ''), # Empty line to improve readability
('Min balance', round_coin_value(strat_results['csum_min'],
strat_results['stake_currency'])),
('Max balance', round_coin_value(strat_results['csum_max'],
strat_results['stake_currency'])),
*drawdown_metrics,
('Market change', f"{strat_results['market_change']:.2%}"),
]
return tabulate(metrics, headers=["Metric", "Value"], tablefmt="orgtbl")
else:
start_balance = round_coin_value(strat_results['starting_balance'],
strat_results['stake_currency'])
stake_amount = round_coin_value(
strat_results['stake_amount'], strat_results['stake_currency']
) if strat_results['stake_amount'] != UNLIMITED_STAKE_AMOUNT else 'unlimited'
message = ("No trades made. "
f"Your starting balance was {start_balance}, "
f"and your stake was {stake_amount}."
)
return message
def show_backtest_result(strategy: str, results: Dict[str, Any], stake_currency: str,
backtest_breakdown=[]):
"""
Print results for one strategy
"""
# Print results
print(f"Result for strategy {strategy}")
table = text_table_bt_results(results['results_per_pair'], stake_currency=stake_currency)
if isinstance(table, str):
print(' BACKTESTING REPORT '.center(len(table.splitlines()[0]), '='))
print(table)
table = text_table_bt_results(results['left_open_trades'], stake_currency=stake_currency)
if isinstance(table, str) and len(table) > 0:
print(' LEFT OPEN TRADES REPORT '.center(len(table.splitlines()[0]), '='))
print(table)
if (results.get('results_per_enter_tag') is not None
or results.get('results_per_buy_tag') is not None):
# results_per_buy_tag is deprecated and should be removed 2 versions after short golive.
table = text_table_tags(
"enter_tag",
results.get('results_per_enter_tag', results.get('results_per_buy_tag')),
stake_currency=stake_currency)
if isinstance(table, str) and len(table) > 0:
print(' ENTER TAG STATS '.center(len(table.splitlines()[0]), '='))
print(table)
exit_reasons = results.get('exit_reason_summary', results.get('sell_reason_summary'))
table = text_table_exit_reason(exit_reason_stats=exit_reasons,
stake_currency=stake_currency)
if isinstance(table, str) and len(table) > 0:
print(' EXIT REASON STATS '.center(len(table.splitlines()[0]), '='))
print(table)
for period in backtest_breakdown:
if period in results.get('periodic_breakdown', {}):
days_breakdown_stats = results['periodic_breakdown'][period]
else:
days_breakdown_stats = generate_periodic_breakdown_stats(
trade_list=results['trades'], period=period)
table = text_table_periodic_breakdown(days_breakdown_stats=days_breakdown_stats,
stake_currency=stake_currency, period=period)
if isinstance(table, str) and len(table) > 0:
print(f' {period.upper()} BREAKDOWN '.center(len(table.splitlines()[0]), '='))
print(table)
table = text_table_add_metrics(results)
if isinstance(table, str) and len(table) > 0:
print(' SUMMARY METRICS '.center(len(table.splitlines()[0]), '='))
print(table)
if isinstance(table, str) and len(table) > 0:
print('=' * len(table.splitlines()[0]))
print()
def show_backtest_results(config: Config, backtest_stats: Dict):
stake_currency = config['stake_currency']
for strategy, results in backtest_stats['strategy'].items():
show_backtest_result(
strategy, results, stake_currency,
config.get('backtest_breakdown', []))
if len(backtest_stats['strategy']) > 0:
# Print Strategy summary table
table = text_table_strategy(backtest_stats['strategy_comparison'], stake_currency)
print(f"Backtested {results['backtest_start']} -> {results['backtest_end']} |"
f" Max open trades : {results['max_open_trades']}")
print(' STRATEGY SUMMARY '.center(len(table.splitlines()[0]), '='))
print(table)
print('=' * len(table.splitlines()[0]))
print('\nFor more details, please look at the detail tables above')
def show_sorted_pairlist(config: Config, backtest_stats: Dict):
if config.get('backtest_show_pair_list', False):
for strategy, results in backtest_stats['strategy'].items():
print(f"Pairs for Strategy {strategy}: \n[")
for result in results['results_per_pair']:
if result["key"] != 'TOTAL':
print(f'"{result["key"]}", // {result["profit_mean"]:.2%}')
print("]")
tests/optimize/test_backtesting.py
+3 -1
View File
@@ -1437,9 +1437,11 @@ def test_backtest_start_multi_strat(default_conf, mocker, caplog, testdatadir):
strattable_mock = MagicMock() strattable_mock = MagicMock()
strat_summary = MagicMock() strat_summary = MagicMock()
mocker.patch.multiple('freqtrade.optimize.optimize_reports', mocker.patch.multiple('freqtrade.optimize.optimize_reports.bt_output',
text_table_bt_results=text_table_mock, text_table_bt_results=text_table_mock,
text_table_strategy=strattable_mock, text_table_strategy=strattable_mock,
)
mocker.patch.multiple('freqtrade.optimize.optimize_reports.optimize_reports',
generate_pair_metrics=MagicMock(), generate_pair_metrics=MagicMock(),
generate_exit_reason_stats=sell_reason_mock, generate_exit_reason_stats=sell_reason_mock,
generate_strategy_comparison=strat_summary, generate_strategy_comparison=strat_summary,
tests/optimize/test_optimize_reports.py
+7 -5
View File
@@ -14,15 +14,16 @@ from freqtrade.data.btanalysis import (get_latest_backtest_filename, load_backte
load_backtest_stats) load_backtest_stats)
from freqtrade.edge import PairInfo from freqtrade.edge import PairInfo
from freqtrade.enums import ExitType from freqtrade.enums import ExitType
from freqtrade.optimize.optimize_reports import (_get_resample_from_period, generate_backtest_stats, from freqtrade.optimize.optimize_reports import (generate_backtest_stats, generate_daily_stats,
generate_daily_stats, generate_edge_table, generate_edge_table, generate_exit_reason_stats,
generate_exit_reason_stats, generate_pair_metrics, generate_pair_metrics,
generate_periodic_breakdown_stats, generate_periodic_breakdown_stats,
generate_strategy_comparison, generate_strategy_comparison,
generate_trading_stats, show_sorted_pairlist, generate_trading_stats, show_sorted_pairlist,
store_backtest_analysis_results, store_backtest_analysis_results,
store_backtest_stats, text_table_bt_results, store_backtest_stats, text_table_bt_results,
text_table_exit_reason, text_table_strategy) text_table_exit_reason, text_table_strategy)
from freqtrade.optimize.optimize_reports.optimize_reports import _get_resample_from_period
from freqtrade.resolvers.strategy_resolver import StrategyResolver from freqtrade.resolvers.strategy_resolver import StrategyResolver
from freqtrade.util import dt_ts from freqtrade.util import dt_ts
from freqtrade.util.datetime_helpers import dt_from_ts, dt_utc from freqtrade.util.datetime_helpers import dt_from_ts, dt_utc
@@ -209,7 +210,7 @@ def test_generate_backtest_stats(default_conf, testdatadir, tmpdir):
def test_store_backtest_stats(testdatadir, mocker): def test_store_backtest_stats(testdatadir, mocker):
dump_mock = mocker.patch('freqtrade.optimize.optimize_reports.file_dump_json') dump_mock = mocker.patch('freqtrade.optimize.optimize_reports.bt_storage.file_dump_json')
store_backtest_stats(testdatadir, {'metadata': {}}, '2022_01_01_15_05_13') store_backtest_stats(testdatadir, {'metadata': {}}, '2022_01_01_15_05_13')
@@ -228,7 +229,8 @@ def test_store_backtest_stats(testdatadir, mocker):
def test_store_backtest_candles(testdatadir, mocker): def test_store_backtest_candles(testdatadir, mocker):
dump_mock = mocker.patch('freqtrade.optimize.optimize_reports.file_dump_joblib') dump_mock = mocker.patch(
'freqtrade.optimize.optimize_reports.bt_storage.file_dump_joblib')
candle_dict = {'DefStrat': {'UNITTEST/BTC': pd.DataFrame()}} candle_dict = {'DefStrat': {'UNITTEST/BTC': pd.DataFrame()}}