import logging from pathlib import Path from typing import List import joblib import pandas as pd from tabulate import tabulate from freqtrade.configuration import TimeRange from freqtrade.constants import Config from freqtrade.data.btanalysis import (get_latest_backtest_filename, load_backtest_data, load_backtest_stats) from freqtrade.exceptions import OperationalException logger = logging.getLogger(__name__) def _load_backtest_analysis_data(backtest_dir: Path, name: str): if backtest_dir.is_dir(): scpf = Path(backtest_dir, Path(get_latest_backtest_filename(backtest_dir)).stem + "_" + name + ".pkl" ) else: scpf = Path(backtest_dir.parent / f"{backtest_dir.stem}_{name}.pkl") try: with scpf.open("rb") as scp: loaded_data = joblib.load(scp) logger.info(f"Loaded {name} candles: {str(scpf)}") except Exception as e: logger.error(f"Cannot load {name} data from pickled results: ", e) return None return loaded_data def _load_rejected_signals(backtest_dir: Path): return _load_backtest_analysis_data(backtest_dir, "rejected") def _load_signal_candles(backtest_dir: Path): return _load_backtest_analysis_data(backtest_dir, "signals") def _process_candles_and_indicators(pairlist, strategy_name, trades, signal_candles): analysed_trades_dict = {} analysed_trades_dict[strategy_name] = {} try: logger.info(f"Processing {strategy_name} : {len(pairlist)} pairs") for pair in pairlist: if pair in signal_candles[strategy_name]: analysed_trades_dict[strategy_name][pair] = _analyze_candles_and_indicators( pair, trades, signal_candles[strategy_name][pair]) except Exception as e: print(f"Cannot process entry/exit reasons for {strategy_name}: ", e) return analysed_trades_dict def _analyze_candles_and_indicators(pair, trades: pd.DataFrame, signal_candles: pd.DataFrame): buyf = signal_candles if len(buyf) > 0: buyf = buyf.set_index('date', drop=False) trades_red = trades.loc[trades['pair'] == pair].copy() trades_inds = pd.DataFrame() if trades_red.shape[0] > 0 and buyf.shape[0] > 0: for t, v in trades_red.open_date.items(): allinds = buyf.loc[(buyf['date'] < v)] if allinds.shape[0] > 0: tmp_inds = allinds.iloc[[-1]] trades_red.loc[t, 'signal_date'] = tmp_inds['date'].values[0] trades_red.loc[t, 'enter_reason'] = trades_red.loc[t, 'enter_tag'] tmp_inds.index.rename('signal_date', inplace=True) trades_inds = pd.concat([trades_inds, tmp_inds]) if 'signal_date' in trades_red: trades_red['signal_date'] = pd.to_datetime(trades_red['signal_date'], utc=True) trades_red.set_index('signal_date', inplace=True) try: trades_red = pd.merge(trades_red, trades_inds, on='signal_date', how='outer') except Exception as e: raise e return trades_red else: return pd.DataFrame() def _do_group_table_output(bigdf, glist, csv_path: Path, to_csv=False, ): for g in glist: # 0: summary wins/losses grouped by enter tag if g == "0": group_mask = ['enter_reason'] wins = bigdf.loc[bigdf['profit_abs'] >= 0] \ .groupby(group_mask) \ .agg({'profit_abs': ['sum']}) wins.columns = ['profit_abs_wins'] loss = bigdf.loc[bigdf['profit_abs'] < 0] \ .groupby(group_mask) \ .agg({'profit_abs': ['sum']}) loss.columns = ['profit_abs_loss'] new = bigdf.groupby(group_mask).agg({'profit_abs': [ 'count', lambda x: sum(x > 0), lambda x: sum(x <= 0)]}) new = pd.concat([new, wins, loss], axis=1).fillna(0) new['profit_tot'] = new['profit_abs_wins'] - abs(new['profit_abs_loss']) new['wl_ratio_pct'] = (new.iloc[:, 1] / new.iloc[:, 0] * 100).fillna(0) new['avg_win'] = (new['profit_abs_wins'] / new.iloc[:, 1]).fillna(0) new['avg_loss'] = (new['profit_abs_loss'] / new.iloc[:, 2]).fillna(0) new['exp_ratio'] = ( ( (1 + (new['avg_win'] / abs(new['avg_loss']))) * (new['wl_ratio_pct'] / 100) ) - 1).fillna(0) new.columns = ['total_num_buys', 'wins', 'losses', 'profit_abs_wins', 'profit_abs_loss', 'profit_tot', 'wl_ratio_pct', 'avg_win', 'avg_loss', 'exp_ratio'] sortcols = ['total_num_buys'] _print_table(new, sortcols, show_index=True, name="Group 0:", to_csv=to_csv, csv_path=csv_path) else: agg_mask = {'profit_abs': ['count', 'sum', 'median', 'mean'], 'profit_ratio': ['median', 'mean', 'sum']} agg_cols = ['num_buys', 'profit_abs_sum', 'profit_abs_median', 'profit_abs_mean', 'median_profit_pct', 'mean_profit_pct', 'total_profit_pct'] sortcols = ['profit_abs_sum', 'enter_reason'] # 1: profit summaries grouped by enter_tag if g == "1": group_mask = ['enter_reason'] # 2: profit summaries grouped by enter_tag and exit_tag if g == "2": group_mask = ['enter_reason', 'exit_reason'] # 3: profit summaries grouped by pair and enter_tag if g == "3": group_mask = ['pair', 'enter_reason'] # 4: profit summaries grouped by pair, enter_ and exit_tag (this can get quite large) if g == "4": group_mask = ['pair', 'enter_reason', 'exit_reason'] # 5: profit summaries grouped by exit_tag if g == "5": group_mask = ['exit_reason'] sortcols = ['exit_reason'] if group_mask: new = bigdf.groupby(group_mask).agg(agg_mask).reset_index() new.columns = group_mask + agg_cols new['median_profit_pct'] = new['median_profit_pct'] * 100 new['mean_profit_pct'] = new['mean_profit_pct'] * 100 new['total_profit_pct'] = new['total_profit_pct'] * 100 _print_table(new, sortcols, name=f"Group {g}:", to_csv=to_csv, csv_path=csv_path) else: logger.warning("Invalid group mask specified.") def _do_rejected_signals_output(rejected_signals_df: pd.DataFrame, to_csv: bool = False, csv_path=None) -> None: cols = ['pair', 'date', 'enter_tag'] sortcols = ['date', 'pair', 'enter_tag'] _print_table(rejected_signals_df[cols], sortcols, show_index=False, name="Rejected Signals:", to_csv=to_csv, csv_path=csv_path) def _select_rows_within_dates(df, timerange=None, df_date_col: str = 'date'): if timerange: if timerange.starttype == 'date': df = df.loc[(df[df_date_col] >= timerange.startdt)] if timerange.stoptype == 'date': df = df.loc[(df[df_date_col] < timerange.stopdt)] return df def _select_rows_by_tags(df, enter_reason_list, exit_reason_list): if enter_reason_list and "all" not in enter_reason_list: df = df.loc[(df['enter_reason'].isin(enter_reason_list))] if exit_reason_list and "all" not in exit_reason_list: df = df.loc[(df['exit_reason'].isin(exit_reason_list))] return df def prepare_results(analysed_trades, stratname, enter_reason_list, exit_reason_list, timerange=None): res_df = pd.DataFrame() for pair, trades in analysed_trades[stratname].items(): if (trades.shape[0] > 0): trades.dropna(subset=['close_date'], inplace=True) res_df = pd.concat([res_df, trades], ignore_index=True) res_df = _select_rows_within_dates(res_df, timerange) if res_df is not None and res_df.shape[0] > 0 and ('enter_reason' in res_df.columns): res_df = _select_rows_by_tags(res_df, enter_reason_list, exit_reason_list) return res_df def print_results(res_df: pd.DataFrame, analysis_groups: List[str], indicator_list: List[str], csv_path: Path, rejected_signals=None, to_csv=False): if res_df.shape[0] > 0: if analysis_groups: _do_group_table_output(res_df, analysis_groups, to_csv=to_csv, csv_path=csv_path) if rejected_signals is not None: if rejected_signals.empty: print("There were no rejected signals.") else: _do_rejected_signals_output(rejected_signals, to_csv=to_csv, csv_path=csv_path) # NB this can be large for big dataframes! if "all" in indicator_list: _print_table(res_df, show_index=False, name="Indicators:", to_csv=to_csv, csv_path=csv_path) elif indicator_list is not None and indicator_list: available_inds = [] for ind in indicator_list: if ind in res_df: available_inds.append(ind) ilist = ["pair", "enter_reason", "exit_reason"] + available_inds _print_table(res_df[ilist], sortcols=['exit_reason'], show_index=False, name="Indicators:", to_csv=to_csv, csv_path=csv_path) else: print("\\No trades to show") def _print_table(df: pd.DataFrame, sortcols=None, *, show_index=False, name=None, to_csv=False, csv_path: Path): if (sortcols is not None): data = df.sort_values(sortcols) else: data = df if to_csv: safe_name = Path(csv_path, name.lower().replace(" ", "_").replace(":", "") + ".csv") data.to_csv(safe_name) print(f"Saved {name} to {safe_name}") else: if name is not None: print(name) print( tabulate( data, headers='keys', tablefmt='psql', showindex=show_index ) ) def process_entry_exit_reasons(config: Config): try: analysis_groups = config.get('analysis_groups', []) enter_reason_list = config.get('enter_reason_list', ["all"]) exit_reason_list = config.get('exit_reason_list', ["all"]) indicator_list = config.get('indicator_list', []) do_rejected = config.get('analysis_rejected', False) to_csv = config.get('analysis_to_csv', False) csv_path = Path(config.get('analysis_csv_path', config['exportfilename'])) if to_csv and not csv_path.is_dir(): raise OperationalException(f"Specified directory {csv_path} does not exist.") timerange = TimeRange.parse_timerange(None if config.get( 'timerange') is None else str(config.get('timerange'))) backtest_stats = load_backtest_stats(config['exportfilename']) for strategy_name, results in backtest_stats['strategy'].items(): trades = load_backtest_data(config['exportfilename'], strategy_name) if trades is not None and not trades.empty: signal_candles = _load_signal_candles(config['exportfilename']) rej_df = None if do_rejected: rejected_signals_dict = _load_rejected_signals(config['exportfilename']) rej_df = prepare_results(rejected_signals_dict, strategy_name, enter_reason_list, exit_reason_list, timerange=timerange) analysed_trades_dict = _process_candles_and_indicators( config['exchange']['pair_whitelist'], strategy_name, trades, signal_candles) res_df = prepare_results(analysed_trades_dict, strategy_name, enter_reason_list, exit_reason_list, timerange=timerange) print_results(res_df, analysis_groups, indicator_list, rejected_signals=rej_df, to_csv=to_csv, csv_path=csv_path) except ValueError as e: raise OperationalException(e) from e