stable/freqtrade/data/metrics.py

313 lines
12 KiB
Python
Raw Normal View History

2022-04-30 12:47:27 +00:00
import logging
2022-12-07 09:09:57 +00:00
import math
2022-11-27 23:56:49 +00:00
from datetime import datetime
2022-12-07 09:09:57 +00:00
from typing import Dict, Tuple
2022-04-30 12:47:27 +00:00
import numpy as np
import pandas as pd
2022-12-07 09:09:57 +00:00
2022-04-30 12:47:27 +00:00
logger = logging.getLogger(__name__)
def calculate_market_change(data: Dict[str, pd.DataFrame], column: str = "close") -> float:
"""
Calculate market change based on "column".
Calculation is done by taking the first non-null and the last non-null element of each column
and calculating the pctchange as "(last - first) / first".
Then the results per pair are combined as mean.
:param data: Dict of Dataframes, dict key should be pair.
:param column: Column in the original dataframes to use
:return:
"""
tmp_means = []
for pair, df in data.items():
start = df[column].dropna().iloc[0]
end = df[column].dropna().iloc[-1]
tmp_means.append((end - start) / start)
return float(np.mean(tmp_means))
def combine_dataframes_with_mean(data: Dict[str, pd.DataFrame],
column: str = "close") -> pd.DataFrame:
"""
Combine multiple dataframes "column"
:param data: Dict of Dataframes, dict key should be pair.
:param column: Column in the original dataframes to use
:return: DataFrame with the column renamed to the dict key, and a column
named mean, containing the mean of all pairs.
:raise: ValueError if no data is provided.
"""
df_comb = pd.concat([data[pair].set_index('date').rename(
{column: pair}, axis=1)[pair] for pair in data], axis=1)
df_comb['mean'] = df_comb.mean(axis=1)
return df_comb
def create_cum_profit(df: pd.DataFrame, trades: pd.DataFrame, col_name: str,
timeframe: str) -> pd.DataFrame:
"""
Adds a column `col_name` with the cumulative profit for the given trades array.
:param df: DataFrame with date index
:param trades: DataFrame containing trades (requires columns close_date and profit_abs)
:param col_name: Column name that will be assigned the results
:param timeframe: Timeframe used during the operations
:return: Returns df with one additional column, col_name, containing the cumulative profit.
:raise: ValueError if trade-dataframe was found empty.
"""
if len(trades) == 0:
raise ValueError("Trade dataframe empty.")
from freqtrade.exchange import timeframe_to_minutes
timeframe_minutes = timeframe_to_minutes(timeframe)
# Resample to timeframe to make sure trades match candles
_trades_sum = trades.resample(f'{timeframe_minutes}min', on='close_date'
)[['profit_abs']].sum()
df.loc[:, col_name] = _trades_sum['profit_abs'].cumsum()
# Set first value to 0
df.loc[df.iloc[0].name, col_name] = 0
# FFill to get continuous
df[col_name] = df[col_name].ffill()
return df
def _calc_drawdown_series(profit_results: pd.DataFrame, *, date_col: str, value_col: str,
starting_balance: float) -> pd.DataFrame:
2022-04-30 12:47:27 +00:00
max_drawdown_df = pd.DataFrame()
max_drawdown_df['cumulative'] = profit_results[value_col].cumsum()
max_drawdown_df['high_value'] = max_drawdown_df['cumulative'].cummax()
max_drawdown_df['drawdown'] = max_drawdown_df['cumulative'] - max_drawdown_df['high_value']
max_drawdown_df['date'] = profit_results.loc[:, date_col]
if starting_balance:
cumulative_balance = starting_balance + max_drawdown_df['cumulative']
max_balance = starting_balance + max_drawdown_df['high_value']
max_drawdown_df['drawdown_relative'] = ((max_balance - cumulative_balance) / max_balance)
else:
# NOTE: This is not completely accurate,
# but might good enough if starting_balance is not available
max_drawdown_df['drawdown_relative'] = (
(max_drawdown_df['high_value'] - max_drawdown_df['cumulative'])
/ max_drawdown_df['high_value'])
2022-04-30 12:47:27 +00:00
return max_drawdown_df
def calculate_underwater(trades: pd.DataFrame, *, date_col: str = 'close_date',
value_col: str = 'profit_ratio', starting_balance: float = 0.0
2022-04-30 12:47:27 +00:00
):
"""
Calculate max drawdown and the corresponding close dates
:param trades: DataFrame containing trades (requires columns close_date and profit_ratio)
:param date_col: Column in DataFrame to use for dates (defaults to 'close_date')
:param value_col: Column in DataFrame to use for values (defaults to 'profit_ratio')
:return: Tuple (float, highdate, lowdate, highvalue, lowvalue) with absolute max drawdown,
high and low time and high and low value.
:raise: ValueError if trade-dataframe was found empty.
"""
if len(trades) == 0:
raise ValueError("Trade dataframe empty.")
profit_results = trades.sort_values(date_col).reset_index(drop=True)
max_drawdown_df = _calc_drawdown_series(
profit_results,
date_col=date_col,
value_col=value_col,
starting_balance=starting_balance)
2022-04-30 12:47:27 +00:00
return max_drawdown_df
def calculate_max_drawdown(trades: pd.DataFrame, *, date_col: str = 'close_date',
value_col: str = 'profit_abs', starting_balance: float = 0,
relative: bool = False
2022-04-30 12:47:27 +00:00
) -> Tuple[float, pd.Timestamp, pd.Timestamp, float, float, float]:
"""
Calculate max drawdown and the corresponding close dates
:param trades: DataFrame containing trades (requires columns close_date and profit_ratio)
:param date_col: Column in DataFrame to use for dates (defaults to 'close_date')
:param value_col: Column in DataFrame to use for values (defaults to 'profit_abs')
:param starting_balance: Portfolio starting balance - properly calculate relative drawdown.
:return: Tuple (float, highdate, lowdate, highvalue, lowvalue, relative_drawdown)
with absolute max drawdown, high and low time and high and low value,
and the relative account drawdown
:raise: ValueError if trade-dataframe was found empty.
"""
if len(trades) == 0:
raise ValueError("Trade dataframe empty.")
profit_results = trades.sort_values(date_col).reset_index(drop=True)
max_drawdown_df = _calc_drawdown_series(
profit_results,
date_col=date_col,
value_col=value_col,
starting_balance=starting_balance
)
2022-04-30 12:47:27 +00:00
idxmin = max_drawdown_df['drawdown_relative'].idxmax() if relative \
else max_drawdown_df['drawdown'].idxmin()
2022-04-30 12:47:27 +00:00
if idxmin == 0:
raise ValueError("No losing trade, therefore no drawdown.")
high_date = profit_results.loc[max_drawdown_df.iloc[:idxmin]['high_value'].idxmax(), date_col]
low_date = profit_results.loc[idxmin, date_col]
high_val = max_drawdown_df.loc[max_drawdown_df.iloc[:idxmin]
['high_value'].idxmax(), 'cumulative']
low_val = max_drawdown_df.loc[idxmin, 'cumulative']
max_drawdown_rel = max_drawdown_df.loc[idxmin, 'drawdown_relative']
2022-04-30 12:47:27 +00:00
return (
abs(max_drawdown_df.loc[idxmin, 'drawdown']),
2022-04-30 12:47:27 +00:00
high_date,
low_date,
high_val,
low_val,
max_drawdown_rel
)
def calculate_csum(trades: pd.DataFrame, starting_balance: float = 0) -> Tuple[float, float]:
"""
Calculate min/max cumsum of trades, to show if the wallet/stake amount ratio is sane
:param trades: DataFrame containing trades (requires columns close_date and profit_percent)
:param starting_balance: Add starting balance to results, to show the wallets high / low points
:return: Tuple (float, float) with cumsum of profit_abs
:raise: ValueError if trade-dataframe was found empty.
"""
if len(trades) == 0:
raise ValueError("Trade dataframe empty.")
csum_df = pd.DataFrame()
csum_df['sum'] = trades['profit_abs'].cumsum()
csum_min = csum_df['sum'].min() + starting_balance
csum_max = csum_df['sum'].max() + starting_balance
return csum_min, csum_max
def calculate_cagr(days_passed: int, starting_balance: float, final_balance: float) -> float:
"""
Calculate CAGR
:param days_passed: Days passed between start and ending balance
:param starting_balance: Starting balance
:param final_balance: Final balance to calculate CAGR against
:return: CAGR
"""
return (final_balance / starting_balance) ** (1 / (days_passed / 365)) - 1
2022-11-27 23:56:49 +00:00
def calculate_expectancy(trades: pd.DataFrame) -> float:
"""
Calculate expectancy
:param trades: DataFrame containing trades (requires columns close_date and profit_ratio)
:return: expectancy
"""
if len(trades) == 0:
return 0
expectancy = 1
profit_sum = trades.loc[trades['profit_abs'] > 0, 'profit_abs'].sum()
loss_sum = abs(trades.loc[trades['profit_abs'] < 0, 'profit_abs'].sum())
nb_win_trades = len(trades.loc[trades['profit_abs'] > 0])
nb_loss_trades = len(trades.loc[trades['profit_abs'] < 0])
if (nb_win_trades > 0) and (nb_loss_trades > 0):
average_win = profit_sum / nb_win_trades
average_loss = loss_sum / nb_loss_trades
risk_reward_ratio = average_win / average_loss
winrate = nb_win_trades / len(trades)
expectancy = ((1 + risk_reward_ratio) * winrate) - 1
elif nb_win_trades == 0:
expectancy = 0
return expectancy
2022-12-07 06:47:58 +00:00
2022-12-25 23:19:51 +00:00
def calculate_sortino(trades: pd.DataFrame, min_date: datetime, max_date: datetime,
starting_balance: float) -> float:
2022-11-27 23:56:49 +00:00
"""
Calculate sortino
:param trades: DataFrame containing trades (requires columns profit_ratio)
:return: sortino
"""
2022-12-07 06:47:58 +00:00
if (len(trades) == 0) or (min_date is None) or (max_date is None) or (min_date == max_date):
2022-11-27 23:56:49 +00:00
return 0
2022-12-25 23:19:51 +00:00
total_profit = trades['profit_abs'] / starting_balance
days_period = max(1, (max_date - min_date).days)
2022-11-27 23:56:49 +00:00
expected_returns_mean = total_profit.sum() / days_period
trades['downside_returns'] = 0
2022-12-25 23:19:51 +00:00
trades.loc[total_profit < 0, 'downside_returns'] = (trades['profit_abs'] / starting_balance)
2022-11-27 23:56:49 +00:00
down_stdev = np.std(trades['downside_returns'])
if down_stdev != 0:
sortino_ratio = expected_returns_mean / down_stdev * np.sqrt(365)
else:
# Define high (negative) sortino ratio to be clear that this is NOT optimal.
sortino_ratio = -100
# print(expected_returns_mean, down_stdev, sortino_ratio)
return sortino_ratio
2022-12-07 06:47:58 +00:00
2022-12-25 23:19:51 +00:00
def calculate_sharpe(trades: pd.DataFrame, min_date: datetime, max_date: datetime,
starting_balance: float) -> float:
2022-11-27 23:56:49 +00:00
"""
Calculate sharpe
:param trades: DataFrame containing trades (requires columns close_date and profit_ratio)
:return: sharpe
"""
2022-12-07 06:47:58 +00:00
if (len(trades) == 0) or (min_date is None) or (max_date is None) or (min_date == max_date):
2022-11-27 23:56:49 +00:00
return 0
2022-12-25 23:19:51 +00:00
total_profit = trades['profit_abs'] / starting_balance
days_period = max(1, (max_date - min_date).days)
2022-11-27 23:56:49 +00:00
expected_returns_mean = total_profit.sum() / days_period
up_stdev = np.std(total_profit)
if up_stdev != 0:
sharp_ratio = expected_returns_mean / up_stdev * np.sqrt(365)
else:
# Define high (negative) sharpe ratio to be clear that this is NOT optimal.
sharp_ratio = -100
# print(expected_returns_mean, up_stdev, sharp_ratio)
return sharp_ratio
2022-12-07 06:47:58 +00:00
2022-12-25 20:29:37 +00:00
def calculate_calmar(trades: pd.DataFrame, min_date: datetime, max_date: datetime,
starting_balance: float) -> float:
2022-11-27 23:56:49 +00:00
"""
Calculate calmar
:param trades: DataFrame containing trades (requires columns close_date and profit_ratio)
:return: calmar
"""
2022-12-07 06:47:58 +00:00
if (len(trades) == 0) or (min_date is None) or (max_date is None) or (min_date == max_date):
2022-11-27 23:56:49 +00:00
return 0
2022-12-25 20:29:37 +00:00
total_profit = trades['profit_abs'].sum() / starting_balance
days_period = max(1, (max_date - min_date).days)
2022-11-27 23:56:49 +00:00
# adding slippage of 0.1% per trade
# total_profit = total_profit - 0.0005
2022-12-25 20:29:37 +00:00
expected_returns_mean = total_profit / days_period * 100
2022-11-27 23:56:49 +00:00
# calculate max drawdown
try:
_, _, _, _, _, max_drawdown = calculate_max_drawdown(
2022-12-25 20:29:37 +00:00
trades, value_col="profit_abs", starting_balance=starting_balance
2022-11-27 23:56:49 +00:00
)
except ValueError:
max_drawdown = 0
if max_drawdown != 0:
calmar_ratio = expected_returns_mean / max_drawdown * math.sqrt(365)
else:
# Define high (negative) calmar ratio to be clear that this is NOT optimal.
calmar_ratio = -100
# print(expected_returns_mean, max_drawdown, calmar_ratio)
return calmar_ratio