9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.54

/freqtrade/data/btanalysis.py

"""
Helpers when analyzing backtest data
"""
import logging
from copy import copy
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Literal, Optional, Union

import numpy as np
import pandas as pd

from freqtrade.constants import LAST_BT_RESULT_FN, IntOrInf
from freqtrade.exceptions import ConfigurationError, OperationalException
from freqtrade.misc import file_dump_json, json_load
from freqtrade.optimize.backtest_caching import get_backtest_metadata_filename
from freqtrade.persistence import LocalTrade, Trade, init_db
from freqtrade.types import BacktestHistoryEntryType, BacktestResultType


logger = logging.getLogger(__name__)

# Newest format
BT_DATA_COLUMNS = ['pair', 'stake_amount', 'max_stake_amount', 'amount',
                   'open_date', 'close_date', 'open_rate', 'close_rate',
                   'fee_open', 'fee_close', 'trade_duration',
                   'profit_ratio', 'profit_abs', 'exit_reason',
                   'initial_stop_loss_abs', 'initial_stop_loss_ratio', 'stop_loss_abs',
                   'stop_loss_ratio', 'min_rate', 'max_rate', 'is_open', 'enter_tag',
                   'leverage', 'is_short', 'open_timestamp', 'close_timestamp', 'orders'
                   ]


def get_latest_optimize_filename(directory: Union[Path, str], variant: str) -> str:
    """
    Get latest backtest export based on '.last_result.json'.
    :param directory: Directory to search for last result
    :param variant: 'backtest' or 'hyperopt' - the method to return
    :return: string containing the filename of the latest backtest result
    :raises: ValueError in the following cases:
        * Directory does not exist
        * `directory/.last_result.json` does not exist
        * `directory/.last_result.json` has the wrong content
    """
    if isinstance(directory, str):
        directory = Path(directory)
    if not directory.is_dir():
        raise ValueError(f"Directory '{directory}' does not exist.")
    filename = directory / LAST_BT_RESULT_FN

    if not filename.is_file():
        raise ValueError(
            f"Directory '{directory}' does not seem to contain backtest statistics yet.")

    with filename.open() as file:
        data = json_load(file)

    if f'latest_{variant}' not in data:
        raise ValueError(f"Invalid '{LAST_BT_RESULT_FN}' format.")

    return data[f'latest_{variant}']


def get_latest_backtest_filename(directory: Union[Path, str]) -> str:
    """
    Get latest backtest export based on '.last_result.json'.
    :param directory: Directory to search for last result
    :return: string containing the filename of the latest backtest result
    :raises: ValueError in the following cases:
        * Directory does not exist
        * `directory/.last_result.json` does not exist
        * `directory/.last_result.json` has the wrong content
    """
    return get_latest_optimize_filename(directory, 'backtest')


def get_latest_hyperopt_filename(directory: Union[Path, str]) -> str:
    """
    Get latest hyperopt export based on '.last_result.json'.
    :param directory: Directory to search for last result
    :return: string containing the filename of the latest hyperopt result
    :raises: ValueError in the following cases:
        * Directory does not exist
        * `directory/.last_result.json` does not exist
        * `directory/.last_result.json` has the wrong content
    """
    try:
        return get_latest_optimize_filename(directory, 'hyperopt')
    except ValueError:
        # Return default (legacy) pickle filename
        return 'hyperopt_results.pickle'


def get_latest_hyperopt_file(
        directory: Union[Path, str], predef_filename: Optional[str] = None) -> Path:
    """
    Get latest hyperopt export based on '.last_result.json'.
    :param directory: Directory to search for last result
    :return: string containing the filename of the latest hyperopt result
    :raises: ValueError in the following cases:
        * Directory does not exist
        * `directory/.last_result.json` does not exist
        * `directory/.last_result.json` has the wrong content
    """
    if isinstance(directory, str):
        directory = Path(directory)
    if predef_filename:
        if Path(predef_filename).is_absolute():
            raise ConfigurationError(
                "--hyperopt-filename expects only the filename, not an absolute path.")
        return directory / predef_filename
    return directory / get_latest_hyperopt_filename(directory)


def load_backtest_metadata(filename: Union[Path, str]) -> Dict[str, Any]:
    """
    Read metadata dictionary from backtest results file without reading and deserializing entire
    file.
    :param filename: path to backtest results file.
    :return: metadata dict or None if metadata is not present.
    """
    filename = get_backtest_metadata_filename(filename)
    try:
        with filename.open() as fp:
            return json_load(fp)
    except FileNotFoundError:
        return {}
    except Exception as e:
        raise OperationalException('Unexpected error while loading backtest metadata.') from e


def load_backtest_stats(filename: Union[Path, str]) -> BacktestResultType:
    """
    Load backtest statistics file.
    :param filename: pathlib.Path object, or string pointing to the file.
    :return: a dictionary containing the resulting file.
    """
    if isinstance(filename, str):
        filename = Path(filename)
    if filename.is_dir():
        filename = filename / get_latest_backtest_filename(filename)
    if not filename.is_file():
        raise ValueError(f"File {filename} does not exist.")
    logger.info(f"Loading backtest result from {filename}")
    with filename.open() as file:
        data = json_load(file)

    # Legacy list format does not contain metadata.
    if isinstance(data, dict):
        data['metadata'] = load_backtest_metadata(filename)
    return data


def load_and_merge_backtest_result(strategy_name: str, filename: Path, results: Dict[str, Any]):
    """
    Load one strategy from multi-strategy result and merge it with results
    :param strategy_name: Name of the strategy contained in the result
    :param filename: Backtest-result-filename to load
    :param results: dict to merge the result to.
    """
    bt_data = load_backtest_stats(filename)
    k: Literal['metadata', 'strategy']
    for k in ('metadata', 'strategy'):  # type: ignore
        results[k][strategy_name] = bt_data[k][strategy_name]
    results['metadata'][strategy_name]['filename'] = filename.stem
    comparison = bt_data['strategy_comparison']
    for i in range(len(comparison)):
        if comparison[i]['key'] == strategy_name:
            results['strategy_comparison'].append(comparison[i])
            break


def _get_backtest_files(dirname: Path) -> List[Path]:
    # Weird glob expression here avoids including .meta.json files.
    return list(reversed(sorted(dirname.glob('backtest-result-*-[0-9][0-9].json'))))


def _extract_backtest_result(filename: Path) -> List[BacktestHistoryEntryType]:
    metadata = load_backtest_metadata(filename)
    return [
        {
            'filename': filename.stem,
            'strategy': s,
            'run_id': v['run_id'],
            'notes': v.get('notes', ''),
            # Backtest "run" time
            'backtest_start_time': v['backtest_start_time'],
            # Backtest timerange
            'backtest_start_ts': v.get('backtest_start_ts', None),
            'backtest_end_ts': v.get('backtest_end_ts', None),
            'timeframe': v.get('timeframe', None),
            'timeframe_detail': v.get('timeframe_detail', None),
        } for s, v in metadata.items()
    ]


def get_backtest_result(filename: Path) -> List[BacktestHistoryEntryType]:
    """
    Get backtest result read from metadata file
    """
    return _extract_backtest_result(filename)


def get_backtest_resultlist(dirname: Path) -> List[BacktestHistoryEntryType]:
    """
    Get list of backtest results read from metadata files
    """
    return [
        result
        for filename in _get_backtest_files(dirname)
        for result in _extract_backtest_result(filename)
    ]


def delete_backtest_result(file_abs: Path):
    """
    Delete backtest result file and corresponding metadata file.
    """
    # *.meta.json
    logger.info(f"Deleting backtest result file: {file_abs.name}")
    file_abs_meta = file_abs.with_suffix('.meta.json')
    file_abs.unlink()
    file_abs_meta.unlink()


def update_backtest_metadata(filename: Path, strategy: str, content: Dict[str, Any]):
    """
    Updates backtest metadata file with new content.
    :raises: ValueError if metadata file does not exist, or strategy is not in this file.
    """
    metadata = load_backtest_metadata(filename)
    if not metadata:
        raise ValueError("File does not exist.")
    if strategy not in metadata:
        raise ValueError("Strategy not in metadata.")
    metadata[strategy].update(content)
    # Write data again.
    file_dump_json(get_backtest_metadata_filename(filename), metadata)


def get_backtest_market_change(filename: Path, include_ts: bool = True) -> pd.DataFrame:
    """
    Read backtest market change file.
    """
    df = pd.read_feather(filename)
    if include_ts:
        df.loc[:, '__date_ts'] = df.loc[:, 'date'].astype(np.int64) // 1000 // 1000
    return df


def find_existing_backtest_stats(dirname: Union[Path, str], run_ids: Dict[str, str],
                                 min_backtest_date: Optional[datetime] = None) -> Dict[str, Any]:
    """
    Find existing backtest stats that match specified run IDs and load them.
    :param dirname: pathlib.Path object, or string pointing to the file.
    :param run_ids: {strategy_name: id_string} dictionary.
    :param min_backtest_date: do not load a backtest older than specified date.
    :return: results dict.
    """
    # Copy so we can modify this dict without affecting parent scope.
    run_ids = copy(run_ids)
    dirname = Path(dirname)
    results: Dict[str, Any] = {
        'metadata': {},
        'strategy': {},
        'strategy_comparison': [],
    }

    for filename in _get_backtest_files(dirname):
        metadata = load_backtest_metadata(filename)
        if not metadata:
            # Files are sorted from newest to oldest. When file without metadata is encountered it
            # is safe to assume older files will also not have any metadata.
            break

        for strategy_name, run_id in list(run_ids.items()):
            strategy_metadata = metadata.get(strategy_name, None)
            if not strategy_metadata:
                # This strategy is not present in analyzed backtest.
                continue

            if min_backtest_date is not None:
                backtest_date = strategy_metadata['backtest_start_time']
                backtest_date = datetime.fromtimestamp(backtest_date, tz=timezone.utc)
                if backtest_date < min_backtest_date:
                    # Do not use a cached result for this strategy as first result is too old.
                    del run_ids[strategy_name]
                    continue

            if strategy_metadata['run_id'] == run_id:
                del run_ids[strategy_name]
                load_and_merge_backtest_result(strategy_name, filename, results)

        if len(run_ids) == 0:
            break
    return results


def _load_backtest_data_df_compatibility(df: pd.DataFrame) -> pd.DataFrame:
    """
    Compatibility support for older backtest data.
    """
    df['open_date'] = pd.to_datetime(df['open_date'], utc=True)
    df['close_date'] = pd.to_datetime(df['close_date'], utc=True)
    # Compatibility support for pre short Columns
    if 'is_short' not in df.columns:
        df['is_short'] = False
    if 'leverage' not in df.columns:
        df['leverage'] = 1.0
    if 'enter_tag' not in df.columns:
        df['enter_tag'] = df['buy_tag']
        df = df.drop(['buy_tag'], axis=1)
    if 'max_stake_amount' not in df.columns:
        df['max_stake_amount'] = df['stake_amount']
    if 'orders' not in df.columns:
        df['orders'] = None
    return df


def load_backtest_data(filename: Union[Path, str], strategy: Optional[str] = None) -> pd.DataFrame:
    """
    Load backtest data file.
    :param filename: pathlib.Path object, or string pointing to a file or directory
    :param strategy: Strategy to load - mainly relevant for multi-strategy backtests
                     Can also serve as protection to load the correct result.
    :return: a dataframe with the analysis results
    :raise: ValueError if loading goes wrong.
    """
    data = load_backtest_stats(filename)
    if not isinstance(data, list):
        # new, nested format
        if 'strategy' not in data:
            raise ValueError("Unknown dataformat.")

        if not strategy:
            if len(data['strategy']) == 1:
                strategy = list(data['strategy'].keys())[0]
            else:
                raise ValueError("Detected backtest result with more than one strategy. "
                                 "Please specify a strategy.")

        if strategy not in data['strategy']:
            raise ValueError(
                f"Strategy {strategy} not available in the backtest result. "
                f"Available strategies are '{','.join(data['strategy'].keys())}'"
                )

        data = data['strategy'][strategy]['trades']
        df = pd.DataFrame(data)
        if not df.empty:
            df = _load_backtest_data_df_compatibility(df)

    else:
        # old format - only with lists.
        raise OperationalException(
            "Backtest-results with only trades data are no longer supported.")
    if not df.empty:
        df = df.sort_values("open_date").reset_index(drop=True)
    return df


def analyze_trade_parallelism(results: pd.DataFrame, timeframe: str) -> pd.DataFrame:
    """
    Find overlapping trades by expanding each trade once per period it was open
    and then counting overlaps.
    :param results: Results Dataframe - can be loaded
    :param timeframe: Timeframe used for backtest
    :return: dataframe with open-counts per time-period in timeframe
    """
    from freqtrade.exchange import timeframe_to_resample_freq
    timeframe_freq = timeframe_to_resample_freq(timeframe)
    dates = [pd.Series(pd.date_range(row[1]['open_date'], row[1]['close_date'],
                                     freq=timeframe_freq))
             for row in results[['open_date', 'close_date']].iterrows()]
    deltas = [len(x) for x in dates]
    dates = pd.Series(pd.concat(dates).values, name='date')
    df2 = pd.DataFrame(np.repeat(results.values, deltas, axis=0), columns=results.columns)

    df2 = pd.concat([dates, df2], axis=1)
    df2 = df2.set_index('date')
    df_final = df2.resample(timeframe_freq)[['pair']].count()
    df_final = df_final.rename({'pair': 'open_trades'}, axis=1)
    return df_final


def evaluate_result_multi(results: pd.DataFrame, timeframe: str,
                          max_open_trades: IntOrInf) -> pd.DataFrame:
    """
    Find overlapping trades by expanding each trade once per period it was open
    and then counting overlaps
    :param results: Results Dataframe - can be loaded
    :param timeframe: Frequency used for the backtest
    :param max_open_trades: parameter max_open_trades used during backtest run
    :return: dataframe with open-counts per time-period in freq
    """
    df_final = analyze_trade_parallelism(results, timeframe)
    return df_final[df_final['open_trades'] > max_open_trades]


def trade_list_to_dataframe(trades: Union[List[Trade], List[LocalTrade]]) -> pd.DataFrame:
    """
    Convert list of Trade objects to pandas Dataframe
    :param trades: List of trade objects
    :return: Dataframe with BT_DATA_COLUMNS
    """
    df = pd.DataFrame.from_records([t.to_json(True) for t in trades], columns=BT_DATA_COLUMNS)
    if len(df) > 0:
        df['close_date'] = pd.to_datetime(df['close_date'], utc=True)
        df['open_date'] = pd.to_datetime(df['open_date'], utc=True)
        df['close_rate'] = df['close_rate'].astype('float64')
    return df


def load_trades_from_db(db_url: str, strategy: Optional[str] = None) -> pd.DataFrame:
    """
    Load trades from a DB (using dburl)
    :param db_url: Sqlite url (default format sqlite:///tradesv3.dry-run.sqlite)
    :param strategy: Strategy to load - mainly relevant for multi-strategy backtests
                     Can also serve as protection to load the correct result.
    :return: Dataframe containing Trades
    """
    init_db(db_url)

    filters = []
    if strategy:
        filters.append(Trade.strategy == strategy)
    trades = trade_list_to_dataframe(list(Trade.get_trades(filters).all()))

    return trades


def load_trades(source: str, db_url: str, exportfilename: Path,
                no_trades: bool = False, strategy: Optional[str] = None) -> pd.DataFrame:
    """
    Based on configuration option 'trade_source':
    * loads data from DB (using `db_url`)
    * loads data from backtestfile (using `exportfilename`)
    :param source: "DB" or "file" - specify source to load from
    :param db_url: sqlalchemy formatted url to a database
    :param exportfilename: Json file generated by backtesting
    :param no_trades: Skip using trades, only return backtesting data columns
    :return: DataFrame containing trades
    """
    if no_trades:
        df = pd.DataFrame(columns=BT_DATA_COLUMNS)
        return df

    if source == "DB":
        return load_trades_from_db(db_url)
    elif source == "file":
        return load_backtest_data(exportfilename, strategy)


def extract_trades_of_period(dataframe: pd.DataFrame, trades: pd.DataFrame,
                             date_index=False) -> pd.DataFrame:
    """
    Compare trades and backtested pair DataFrames to get trades performed on backtested period
    :return: the DataFrame of a trades of period
    """
    if date_index:
        trades_start = dataframe.index[0]
        trades_stop = dataframe.index[-1]
    else:
        trades_start = dataframe.iloc[0]['date']
        trades_stop = dataframe.iloc[-1]['date']
    trades = trades.loc[(trades['open_date'] >= trades_start) &
                        (trades['close_date'] <= trades_stop)]
    return trades

1	"""
2	Helpers when analyzing backtest data
3	"""
4	import logging	1✔
5	from copy import copy	1✔
6	from datetime import datetime, timezone	1✔
7	from pathlib import Path	1✔
8	from typing import Any, Dict, List, Literal, Optional, Union	1✔
9
10	import numpy as np	1✔
11	import pandas as pd	1✔
12
13	from freqtrade.constants import LAST_BT_RESULT_FN, IntOrInf	1✔
14	from freqtrade.exceptions import ConfigurationError, OperationalException	1✔
15	from freqtrade.misc import file_dump_json, json_load	1✔
16	from freqtrade.optimize.backtest_caching import get_backtest_metadata_filename	1✔
17	from freqtrade.persistence import LocalTrade, Trade, init_db	1✔
18	from freqtrade.types import BacktestHistoryEntryType, BacktestResultType	1✔
19
20
21	logger = logging.getLogger(__name__)	1✔
22
23	# Newest format
24	BT_DATA_COLUMNS = ['pair', 'stake_amount', 'max_stake_amount', 'amount',	1✔
25	'open_date', 'close_date', 'open_rate', 'close_rate',
26	'fee_open', 'fee_close', 'trade_duration',
27	'profit_ratio', 'profit_abs', 'exit_reason',
28	'initial_stop_loss_abs', 'initial_stop_loss_ratio', 'stop_loss_abs',
29	'stop_loss_ratio', 'min_rate', 'max_rate', 'is_open', 'enter_tag',
30	'leverage', 'is_short', 'open_timestamp', 'close_timestamp', 'orders'
31	]
32
33
34	def get_latest_optimize_filename(directory: Union[Path, str], variant: str) -> str:	1✔
35	"""
36	Get latest backtest export based on '.last_result.json'.
37	:param directory: Directory to search for last result
38	:param variant: 'backtest' or 'hyperopt' - the method to return
39	:return: string containing the filename of the latest backtest result
40	:raises: ValueError in the following cases:
41	* Directory does not exist
42	* `directory/.last_result.json` does not exist
43	* `directory/.last_result.json` has the wrong content
44	"""
45	if isinstance(directory, str):	1✔
46	directory = Path(directory)	1✔
47	if not directory.is_dir():	1✔
48	raise ValueError(f"Directory '{directory}' does not exist.")	1✔
49	filename = directory / LAST_BT_RESULT_FN	1✔
50
51	if not filename.is_file():	1✔
52	raise ValueError(	1✔
53	f"Directory '{directory}' does not seem to contain backtest statistics yet.")
54
55	with filename.open() as file:	1✔
56	data = json_load(file)	1✔
57
58	if f'latest_{variant}' not in data:	1✔
59	raise ValueError(f"Invalid '{LAST_BT_RESULT_FN}' format.")	1✔
60
61	return data[f'latest_{variant}']	1✔
62
63
64	def get_latest_backtest_filename(directory: Union[Path, str]) -> str:	1✔
65	"""
66	Get latest backtest export based on '.last_result.json'.
67	:param directory: Directory to search for last result
68	:return: string containing the filename of the latest backtest result
69	:raises: ValueError in the following cases:
70	* Directory does not exist
71	* `directory/.last_result.json` does not exist
72	* `directory/.last_result.json` has the wrong content
73	"""
74	return get_latest_optimize_filename(directory, 'backtest')	1✔
75
76
77	def get_latest_hyperopt_filename(directory: Union[Path, str]) -> str:	1✔
78	"""
79	Get latest hyperopt export based on '.last_result.json'.
80	:param directory: Directory to search for last result
81	:return: string containing the filename of the latest hyperopt result
82	:raises: ValueError in the following cases:
83	* Directory does not exist
84	* `directory/.last_result.json` does not exist
85	* `directory/.last_result.json` has the wrong content
86	"""
87	try:	1✔
88	return get_latest_optimize_filename(directory, 'hyperopt')	1✔
89	except ValueError:	1✔
90	# Return default (legacy) pickle filename
91	return 'hyperopt_results.pickle'	1✔
92
93
94	def get_latest_hyperopt_file(	1✔
95	directory: Union[Path, str], predef_filename: Optional[str] = None) -> Path:
96	"""
97	Get latest hyperopt export based on '.last_result.json'.
98	:param directory: Directory to search for last result
99	:return: string containing the filename of the latest hyperopt result
100	:raises: ValueError in the following cases:
101	* Directory does not exist
102	* `directory/.last_result.json` does not exist
103	* `directory/.last_result.json` has the wrong content
104	"""
105	if isinstance(directory, str):	1✔
106	directory = Path(directory)	1✔
107	if predef_filename:	1✔
108	if Path(predef_filename).is_absolute():	1✔
109	raise ConfigurationError(	1✔
110	"--hyperopt-filename expects only the filename, not an absolute path.")
111	return directory / predef_filename	1✔
112	return directory / get_latest_hyperopt_filename(directory)	1✔
113
114
115	def load_backtest_metadata(filename: Union[Path, str]) -> Dict[str, Any]:	1✔
116	"""
117	Read metadata dictionary from backtest results file without reading and deserializing entire
118	file.
119	:param filename: path to backtest results file.
120	:return: metadata dict or None if metadata is not present.
121	"""
122	filename = get_backtest_metadata_filename(filename)	1✔
123	try:	1✔
124	with filename.open() as fp:	1✔
125	return json_load(fp)	1✔
126	except FileNotFoundError:	1✔
127	return {}	1✔
128	except Exception as e:	1✔
129	raise OperationalException('Unexpected error while loading backtest metadata.') from e	1✔
130
131
132	def load_backtest_stats(filename: Union[Path, str]) -> BacktestResultType:	1✔
133	"""
134	Load backtest statistics file.
135	:param filename: pathlib.Path object, or string pointing to the file.
136	:return: a dictionary containing the resulting file.
137	"""
138	if isinstance(filename, str):	1✔
139	filename = Path(filename)	1✔
140	if filename.is_dir():	1✔
141	filename = filename / get_latest_backtest_filename(filename)	1✔
142	if not filename.is_file():	1✔
143	raise ValueError(f"File {filename} does not exist.")	1✔
144	logger.info(f"Loading backtest result from {filename}")	1✔
145	with filename.open() as file:	1✔
146	data = json_load(file)	1✔
147
148	# Legacy list format does not contain metadata.
149	if isinstance(data, dict):	1✔
150	data['metadata'] = load_backtest_metadata(filename)	1✔
151	return data	1✔
152
153
154	def load_and_merge_backtest_result(strategy_name: str, filename: Path, results: Dict[str, Any]):	1✔
155	"""
156	Load one strategy from multi-strategy result and merge it with results
157	:param strategy_name: Name of the strategy contained in the result
158	:param filename: Backtest-result-filename to load
159	:param results: dict to merge the result to.
160	"""
161	bt_data = load_backtest_stats(filename)	1✔
162	k: Literal['metadata', 'strategy']
163	for k in ('metadata', 'strategy'): # type: ignore	1✔
164	results[k][strategy_name] = bt_data[k][strategy_name]	1✔
165	results['metadata'][strategy_name]['filename'] = filename.stem	1✔
166	comparison = bt_data['strategy_comparison']	1✔
167	for i in range(len(comparison)):	1✔
168	if comparison[i]['key'] == strategy_name:	1✔
169	results['strategy_comparison'].append(comparison[i])	1✔
170	break	1✔
171
172
173	def _get_backtest_files(dirname: Path) -> List[Path]:	1✔
174	# Weird glob expression here avoids including .meta.json files.
175	return list(reversed(sorted(dirname.glob('backtest-result-*-[0-9][0-9].json'))))	1✔
176
177
178	def _extract_backtest_result(filename: Path) -> List[BacktestHistoryEntryType]:	1✔
179	metadata = load_backtest_metadata(filename)	1✔
180	return [	1✔
181	{
182	'filename': filename.stem,
183	'strategy': s,
184	'run_id': v['run_id'],
185	'notes': v.get('notes', ''),
186	# Backtest "run" time
187	'backtest_start_time': v['backtest_start_time'],
188	# Backtest timerange
189	'backtest_start_ts': v.get('backtest_start_ts', None),
190	'backtest_end_ts': v.get('backtest_end_ts', None),
191	'timeframe': v.get('timeframe', None),
192	'timeframe_detail': v.get('timeframe_detail', None),
193	} for s, v in metadata.items()
194	]
195
196
197	def get_backtest_result(filename: Path) -> List[BacktestHistoryEntryType]:	1✔
198	"""
199	Get backtest result read from metadata file
200	"""
201	return _extract_backtest_result(filename)	1✔
202
203
204	def get_backtest_resultlist(dirname: Path) -> List[BacktestHistoryEntryType]:	1✔
205	"""
206	Get list of backtest results read from metadata files
207	"""
208	return [	1✔
209	result
210	for filename in _get_backtest_files(dirname)
211	for result in _extract_backtest_result(filename)
212	]
213
214
215	def delete_backtest_result(file_abs: Path):	1✔
216	"""
217	Delete backtest result file and corresponding metadata file.
218	"""
219	# *.meta.json
220	logger.info(f"Deleting backtest result file: {file_abs.name}")	1✔
221	file_abs_meta = file_abs.with_suffix('.meta.json')	1✔
222	file_abs.unlink()	1✔
223	file_abs_meta.unlink()	1✔
224
225
226	def update_backtest_metadata(filename: Path, strategy: str, content: Dict[str, Any]):	1✔
227	"""
228	Updates backtest metadata file with new content.
229	:raises: ValueError if metadata file does not exist, or strategy is not in this file.
230	"""
231	metadata = load_backtest_metadata(filename)	1✔
232	if not metadata:	1✔
233	raise ValueError("File does not exist.")	×
234	if strategy not in metadata:	1✔
235	raise ValueError("Strategy not in metadata.")	1✔
236	metadata[strategy].update(content)	1✔
237	# Write data again.
238	file_dump_json(get_backtest_metadata_filename(filename), metadata)	1✔
239
240
241	def get_backtest_market_change(filename: Path, include_ts: bool = True) -> pd.DataFrame:	1✔
242	"""
243	Read backtest market change file.
244	"""
245	df = pd.read_feather(filename)	1✔
246	if include_ts:	1✔
247	df.loc[:, '__date_ts'] = df.loc[:, 'date'].astype(np.int64) // 1000 // 1000	1✔
248	return df	1✔
249
250
251	def find_existing_backtest_stats(dirname: Union[Path, str], run_ids: Dict[str, str],	1✔
252	min_backtest_date: Optional[datetime] = None) -> Dict[str, Any]:
253	"""
254	Find existing backtest stats that match specified run IDs and load them.
255	:param dirname: pathlib.Path object, or string pointing to the file.
256	:param run_ids: {strategy_name: id_string} dictionary.
257	:param min_backtest_date: do not load a backtest older than specified date.
258	:return: results dict.
259	"""
260	# Copy so we can modify this dict without affecting parent scope.
261	run_ids = copy(run_ids)	1✔
262	dirname = Path(dirname)	1✔
263	results: Dict[str, Any] = {	1✔
264	'metadata': {},
265	'strategy': {},
266	'strategy_comparison': [],
267	}
268
269	for filename in _get_backtest_files(dirname):	1✔
270	metadata = load_backtest_metadata(filename)	1✔
271	if not metadata:	1✔
272	# Files are sorted from newest to oldest. When file without metadata is encountered it
273	# is safe to assume older files will also not have any metadata.
274	break	×
275
276	for strategy_name, run_id in list(run_ids.items()):	1✔
277	strategy_metadata = metadata.get(strategy_name, None)	1✔
278	if not strategy_metadata:	1✔
279	# This strategy is not present in analyzed backtest.
280	continue	×
281
282	if min_backtest_date is not None:	1✔
283	backtest_date = strategy_metadata['backtest_start_time']	1✔
284	backtest_date = datetime.fromtimestamp(backtest_date, tz=timezone.utc)	1✔
285	if backtest_date < min_backtest_date:	1✔
286	# Do not use a cached result for this strategy as first result is too old.
287	del run_ids[strategy_name]	1✔
288	continue	1✔
289
290	if strategy_metadata['run_id'] == run_id:	1✔
291	del run_ids[strategy_name]	1✔
292	load_and_merge_backtest_result(strategy_name, filename, results)	1✔
293
294	if len(run_ids) == 0:	1✔
295	break	1✔
296	return results	1✔
297
298
299	def _load_backtest_data_df_compatibility(df: pd.DataFrame) -> pd.DataFrame:	1✔
300	"""
301	Compatibility support for older backtest data.
302	"""
303	df['open_date'] = pd.to_datetime(df['open_date'], utc=True)	1✔
304	df['close_date'] = pd.to_datetime(df['close_date'], utc=True)	1✔
305	# Compatibility support for pre short Columns
306	if 'is_short' not in df.columns:	1✔
307	df['is_short'] = False	1✔
308	if 'leverage' not in df.columns:	1✔
309	df['leverage'] = 1.0	1✔
310	if 'enter_tag' not in df.columns:	1✔
311	df['enter_tag'] = df['buy_tag']	1✔
312	df = df.drop(['buy_tag'], axis=1)	1✔
313	if 'max_stake_amount' not in df.columns:	1✔
314	df['max_stake_amount'] = df['stake_amount']	1✔
315	if 'orders' not in df.columns:	1✔
316	df['orders'] = None	1✔
317	return df	1✔
318
319
320	def load_backtest_data(filename: Union[Path, str], strategy: Optional[str] = None) -> pd.DataFrame:	1✔
321	"""
322	Load backtest data file.
323	:param filename: pathlib.Path object, or string pointing to a file or directory
324	:param strategy: Strategy to load - mainly relevant for multi-strategy backtests
325	Can also serve as protection to load the correct result.
326	:return: a dataframe with the analysis results
327	:raise: ValueError if loading goes wrong.
328	"""
329	data = load_backtest_stats(filename)	1✔
330	if not isinstance(data, list):	1✔
331	# new, nested format
332	if 'strategy' not in data:	1✔
333	raise ValueError("Unknown dataformat.")	1✔
334
335	if not strategy:	1✔
336	if len(data['strategy']) == 1:	1✔
337	strategy = list(data['strategy'].keys())[0]	1✔
338	else:
339	raise ValueError("Detected backtest result with more than one strategy. "	1✔
340	"Please specify a strategy.")
341
342	if strategy not in data['strategy']:	1✔
343	raise ValueError(	1✔
344	f"Strategy {strategy} not available in the backtest result. "
345	f"Available strategies are '{','.join(data['strategy'].keys())}'"
346	)
347
348	data = data['strategy'][strategy]['trades']	1✔
349	df = pd.DataFrame(data)	1✔
350	if not df.empty:	1✔
351	df = _load_backtest_data_df_compatibility(df)	1✔
352
353	else:
354	# old format - only with lists.
355	raise OperationalException(	1✔
356	"Backtest-results with only trades data are no longer supported.")
357	if not df.empty:	1✔
358	df = df.sort_values("open_date").reset_index(drop=True)	1✔
359	return df	1✔
360
361
362	def analyze_trade_parallelism(results: pd.DataFrame, timeframe: str) -> pd.DataFrame:	1✔
363	"""
364	Find overlapping trades by expanding each trade once per period it was open
365	and then counting overlaps.
366	:param results: Results Dataframe - can be loaded
367	:param timeframe: Timeframe used for backtest
368	:return: dataframe with open-counts per time-period in timeframe
369	"""
370	from freqtrade.exchange import timeframe_to_resample_freq	1✔
371	timeframe_freq = timeframe_to_resample_freq(timeframe)	1✔
372	dates = [pd.Series(pd.date_range(row[1]['open_date'], row[1]['close_date'],	1✔
373	freq=timeframe_freq))
374	for row in results[['open_date', 'close_date']].iterrows()]
375	deltas = [len(x) for x in dates]	1✔
376	dates = pd.Series(pd.concat(dates).values, name='date')	1✔
377	df2 = pd.DataFrame(np.repeat(results.values, deltas, axis=0), columns=results.columns)	1✔
378
379	df2 = pd.concat([dates, df2], axis=1)	1✔
380	df2 = df2.set_index('date')	1✔
381	df_final = df2.resample(timeframe_freq)[['pair']].count()	1✔
382	df_final = df_final.rename({'pair': 'open_trades'}, axis=1)	1✔
383	return df_final	1✔
384
385
386	def evaluate_result_multi(results: pd.DataFrame, timeframe: str,	1✔
387	max_open_trades: IntOrInf) -> pd.DataFrame:
388	"""
389	Find overlapping trades by expanding each trade once per period it was open
390	and then counting overlaps
391	:param results: Results Dataframe - can be loaded
392	:param timeframe: Frequency used for the backtest
393	:param max_open_trades: parameter max_open_trades used during backtest run
394	:return: dataframe with open-counts per time-period in freq
395	"""
396	df_final = analyze_trade_parallelism(results, timeframe)	1✔
397	return df_final[df_final['open_trades'] > max_open_trades]	1✔
398
399
400	def trade_list_to_dataframe(trades: Union[List[Trade], List[LocalTrade]]) -> pd.DataFrame:	1✔
401	"""
402	Convert list of Trade objects to pandas Dataframe
403	:param trades: List of trade objects
404	:return: Dataframe with BT_DATA_COLUMNS
405	"""
406	df = pd.DataFrame.from_records([t.to_json(True) for t in trades], columns=BT_DATA_COLUMNS)	1✔
407	if len(df) > 0:	1✔
408	df['close_date'] = pd.to_datetime(df['close_date'], utc=True)	1✔
409	df['open_date'] = pd.to_datetime(df['open_date'], utc=True)	1✔
410	df['close_rate'] = df['close_rate'].astype('float64')	1✔
411	return df	1✔
412
413
414	def load_trades_from_db(db_url: str, strategy: Optional[str] = None) -> pd.DataFrame:	1✔
415	"""
416	Load trades from a DB (using dburl)
417	:param db_url: Sqlite url (default format sqlite:///tradesv3.dry-run.sqlite)
418	:param strategy: Strategy to load - mainly relevant for multi-strategy backtests
419	Can also serve as protection to load the correct result.
420	:return: Dataframe containing Trades
421	"""
422	init_db(db_url)	1✔
423
424	filters = []	1✔
425	if strategy:	1✔
426	filters.append(Trade.strategy == strategy)	1✔
427	trades = trade_list_to_dataframe(list(Trade.get_trades(filters).all()))	1✔
428
429	return trades	1✔
430
431
432	def load_trades(source: str, db_url: str, exportfilename: Path,	1✔
433	no_trades: bool = False, strategy: Optional[str] = None) -> pd.DataFrame:
434	"""
435	Based on configuration option 'trade_source':
436	* loads data from DB (using `db_url`)
437	* loads data from backtestfile (using `exportfilename`)
438	:param source: "DB" or "file" - specify source to load from
439	:param db_url: sqlalchemy formatted url to a database
440	:param exportfilename: Json file generated by backtesting
441	:param no_trades: Skip using trades, only return backtesting data columns
442	:return: DataFrame containing trades
443	"""
444	if no_trades:	1✔
445	df = pd.DataFrame(columns=BT_DATA_COLUMNS)	1✔
446	return df	1✔
447
448	if source == "DB":	1✔
449	return load_trades_from_db(db_url)	1✔
450	elif source == "file":	1✔
451	return load_backtest_data(exportfilename, strategy)	1✔
452
453
454	def extract_trades_of_period(dataframe: pd.DataFrame, trades: pd.DataFrame,	1✔
455	date_index=False) -> pd.DataFrame:
456	"""
457	Compare trades and backtested pair DataFrames to get trades performed on backtested period
458	:return: the DataFrame of a trades of period
459	"""
460	if date_index:	1✔
461	trades_start = dataframe.index[0]	1✔
462	trades_stop = dataframe.index[-1]	1✔
463	else:
464	trades_start = dataframe.iloc[0]['date']	1✔
465	trades_stop = dataframe.iloc[-1]['date']	1✔
466	trades = trades.loc[(trades['open_date'] >= trades_start) &	1✔
467	(trades['close_date'] <= trades_stop)]
468	return trades	1✔

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous