gavindiaz/drift
1
0
Fork 0
mirror of https://github.com/webclinic017/drift.git synced 2026-08-15 20:08:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(Linter): reformatted code with black (#211)

Browse Source 
* chore(Linter): reformatted code with black

* Create black.yaml
This commit is contained in:
Mark Aron Szulyovszky
2022-02-17 19:22:17 +01:00
committed by GitHub
parent f3fee4a4e1
commit 8dd2d88740
101 changed files with 2595 additions and 2319 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/metrics.py
+129 -61
View File
@@ -3,6 +3,7 @@ import pandas as pd
import scipy.stats as ss
import numpy as np
def timing_of_flattening_and_flips(target_positions: pd.Series) -> pd.DatetimeIndex:
"""
Advances in Financial Machine Learning, Snippet 14.1, page 197
@@ -15,8 +16,12 @@ def timing_of_flattening_and_flips(target_positions: pd.Series) -> pd.DatetimeIn
:return: (pd.DatetimeIndex) Timestamps of trades flattening, flipping and last bet
"""
empty_positions = target_positions[(target_positions == 0)].index # Empty positions index
previous_positions = target_positions.shift(1) # Timestamps pointing at previous positions
empty_positions = target_positions[
(target_positions == 0)
].index # Empty positions index
previous_positions = target_positions.shift(
1
) # Timestamps pointing at previous positions
# Index of positions where previous one wasn't empty
previous_positions = previous_positions[(previous_positions != 0)].index
@@ -30,8 +35,12 @@ def timing_of_flattening_and_flips(target_positions: pd.Series) -> pd.DatetimeIn
# FLIPS - if current position has another direction compared to the next
flips = multiplied_posions[(multiplied_posions < 0)].index
flips_and_flattenings = flattening.union(flips).sort_values()
if target_positions.index[-1] not in flips_and_flattenings: # Appending with last bet
flips_and_flattenings = flips_and_flattenings.append(target_positions.index[-1:])
if (
target_positions.index[-1] not in flips_and_flattenings
): # Appending with last bet
flips_and_flattenings = flips_and_flattenings.append(
target_positions.index[-1:]
)
return flips_and_flattenings
@@ -50,19 +59,23 @@ def average_holding_period(target_positions: pd.Series) -> float:
:return: (float) Estimated average holding period, NaN if zero or unpredicted
"""
holding_period = pd.DataFrame(columns=['holding_time', 'weight'])
holding_period = pd.DataFrame(columns=["holding_time", "weight"])
entry_time = 0
position_difference = target_positions.diff()
# Time elapsed from the starting time for each position
time_difference = (target_positions.index - target_positions.index[0]) / np.timedelta64(1, 'D')
time_difference = (
target_positions.index - target_positions.index[0]
) / np.timedelta64(1, "D")
for i in range(1, target_positions.size):
# Increased or unchanged position
if float(position_difference.iloc[i] * target_positions.iloc[i - 1]) >= 0:
if float(target_positions.iloc[i]) != 0: # And not an empty position
entry_time = (entry_time * target_positions.iloc[i - 1] +
time_difference[i] * position_difference.iloc[i]) / target_positions.iloc[i]
entry_time = (
entry_time * target_positions.iloc[i - 1]
+ time_difference[i] * position_difference.iloc[i]
) / target_positions.iloc[i]
# Decreased
if float(position_difference.iloc[i] * target_positions.iloc[i - 1]) < 0:
@@ -71,19 +84,25 @@ def average_holding_period(target_positions: pd.Series) -> float:
# Flip of a position
if float(target_positions.iloc[i] * target_positions.iloc[i - 1]) < 0:
weight = abs(target_positions.iloc[i - 1])
holding_period.loc[target_positions.index[i], ['holding_time', 'weight']] = (hold_time, weight)
holding_period.loc[
target_positions.index[i], ["holding_time", "weight"]
] = (hold_time, weight)
entry_time = time_difference[i] # Reset entry time
# Only a part of position is closed
else:
weight = abs(position_difference.iloc[i])
holding_period.loc[target_positions.index[i], ['holding_time', 'weight']] = (hold_time, weight)
holding_period.loc[
target_positions.index[i], ["holding_time", "weight"]
] = (hold_time, weight)
if float(holding_period['weight'].sum()) > 0: # If there were closed trades at all
avg_holding_period = float((holding_period['holding_time'] * \
holding_period['weight']).sum() / holding_period['weight'].sum())
if float(holding_period["weight"].sum()) > 0: # If there were closed trades at all
avg_holding_period = float(
(holding_period["holding_time"] * holding_period["weight"]).sum()
/ holding_period["weight"].sum()
)
else:
avg_holding_period = float('nan')
avg_holding_period = float("nan")
return avg_holding_period
@@ -99,15 +118,15 @@ def bets_concentration(returns: pd.Series) -> float:
"""
if returns.size <= 2:
return float('nan') # If less than 3 bets
return float("nan") # If less than 3 bets
weights = returns / returns.sum() # Weights of each bet
hhi = (weights ** 2).sum() # Herfindahl-Hirschman Index for weights
hhi = (weights**2).sum() # Herfindahl-Hirschman Index for weights
hhi = float((hhi - returns.size ** (-1)) / (1 - returns.size ** (-1)))
return hhi
def all_bets_concentration(returns: pd.Series, frequency: str = 'M') -> tuple:
def all_bets_concentration(returns: pd.Series, frequency: str = "M") -> tuple:
"""
Advances in Financial Machine Learning, Snippet 14.3, page 201
Given a pd.Series of returns, derives concentration of positive returns, negative returns
@@ -131,7 +150,9 @@ def all_bets_concentration(returns: pd.Series, frequency: str = 'M') -> tuple:
negative_concentration = bets_concentration(returns[returns < 0])
# Concentration of bets/time period (month by default)
time_concentration = bets_concentration(returns.groupby(pd.Grouper(freq=frequency)).count())
time_concentration = bets_concentration(
returns.groupby(pd.Grouper(freq=frequency)).count()
)
return (positive_concentration, negative_concentration, time_concentration)
@@ -157,35 +178,42 @@ def drawdown_and_time_under_water(returns: pd.Series, dollars: bool = False) ->
:return: (tuple of pd.Series) Series of drawdowns and time under water
"""
frame = returns.to_frame('pnl')
frame['hwm'] = returns.expanding().max() # Adding high watermarks as column
frame = returns.to_frame("pnl")
frame["hwm"] = returns.expanding().max() # Adding high watermarks as column
# Grouped as min returns by high watermarks
high_watermarks = frame.groupby('hwm').min().reset_index()
high_watermarks.columns = ['hwm', 'min']
high_watermarks = frame.groupby("hwm").min().reset_index()
high_watermarks.columns = ["hwm", "min"]
# Time high watermark occurred
high_watermarks.index = frame['hwm'].drop_duplicates(keep='first').index
high_watermarks.index = frame["hwm"].drop_duplicates(keep="first").index
# Picking ones that had a drawdown after high watermark
high_watermarks = high_watermarks[high_watermarks['hwm'] > high_watermarks['min']]
high_watermarks = high_watermarks[high_watermarks["hwm"] > high_watermarks["min"]]
if dollars:
drawdown = high_watermarks['hwm'] - high_watermarks['min']
drawdown = high_watermarks["hwm"] - high_watermarks["min"]
else:
drawdown = 1 - high_watermarks['min'] / high_watermarks['hwm']
drawdown = 1 - high_watermarks["min"] / high_watermarks["hwm"]
time_under_water = ((high_watermarks.index[1:] - high_watermarks.index[:-1]) / np.timedelta64(1, 'Y')).values
time_under_water = (
(high_watermarks.index[1:] - high_watermarks.index[:-1])
/ np.timedelta64(1, "Y")
).values
# Adding also period from last High watermark to last return observed.
time_under_water = np.append(time_under_water,
(returns.index[-1] - high_watermarks.index[-1]) / np.timedelta64(1, 'Y'))
time_under_water = np.append(
time_under_water,
(returns.index[-1] - high_watermarks.index[-1]) / np.timedelta64(1, "Y"),
)
time_under_water = pd.Series(time_under_water, index=high_watermarks.index)
return drawdown, time_under_water
def sharpe_ratio(returns: pd.Series, entries_per_year: int = 252, risk_free_rate: float = 0) -> float:
def sharpe_ratio(
returns: pd.Series, entries_per_year: int = 252, risk_free_rate: float = 0
) -> float:
"""
Calculates annualized Sharpe ratio for pd.Series of normal or log returns.
Risk_free_rate should be given for the same period the returns are given.
@@ -197,12 +225,18 @@ def sharpe_ratio(returns: pd.Series, entries_per_year: int = 252, risk_free_rate
:return: (float) Annualized Sharpe ratio
"""
sharpe_r = (returns.mean() - risk_free_rate) / returns.std() * (entries_per_year) ** (1 / 2)
sharpe_r = (
(returns.mean() - risk_free_rate)
/ returns.std()
* (entries_per_year) ** (1 / 2)
)
return sharpe_r
def information_ratio(returns: pd.Series, benchmark: float = 0, entries_per_year: int = 252) -> float:
def information_ratio(
returns: pd.Series, benchmark: float = 0, entries_per_year: int = 252
) -> float:
"""
Calculates annualized information ratio for pd.Series of normal or log returns.
Benchmark should be provided as a return for the same time period as that between
@@ -223,8 +257,13 @@ def information_ratio(returns: pd.Series, benchmark: float = 0, entries_per_year
return information_r
def probabilistic_sharpe_ratio(observed_sr: float, benchmark_sr: float, number_of_returns: int,
skewness_of_returns: float = 0, kurtosis_of_returns: float = 3) -> float:
def probabilistic_sharpe_ratio(
observed_sr: float,
benchmark_sr: float,
number_of_returns: int,
skewness_of_returns: float = 0,
kurtosis_of_returns: float = 3,
) -> float:
"""
Calculates the probabilistic Sharpe ratio (PSR) that provides an adjusted estimate of SR,
by removing the inflationary effect caused by short series with skewed and/or
@@ -241,30 +280,47 @@ def probabilistic_sharpe_ratio(observed_sr: float, benchmark_sr: float, number_o
:return: (float) Probabilistic Sharpe ratio
"""
test_value = ((observed_sr - benchmark_sr) * np.sqrt(number_of_returns - 1)) / \
((1 - skewness_of_returns * observed_sr + (kurtosis_of_returns - 1) / \
4 * observed_sr ** 2)**(1 / 2))
test_value = ((observed_sr - benchmark_sr) * np.sqrt(number_of_returns - 1)) / (
(
1
- skewness_of_returns * observed_sr
+ (kurtosis_of_returns - 1) / 4 * observed_sr**2
)
** (1 / 2)
)
if np.isnan(test_value):
warnings.warn('Test value is nan. Please check the input values.', UserWarning)
warnings.warn("Test value is nan. Please check the input values.", UserWarning)
return test_value
if isinstance(test_value, complex):
warnings.warn('Output is a complex number. You may want to check the input skewness (too high), '
'kurtosis (too low), or observed_sr values.', UserWarning)
warnings.warn(
"Output is a complex number. You may want to check the input skewness (too high), "
"kurtosis (too low), or observed_sr values.",
UserWarning,
)
if np.isinf(test_value):
warnings.warn('Test value is infinite. You may want to check the input skewness, '
'kurtosis, or observed_sr values.', UserWarning)
warnings.warn(
"Test value is infinite. You may want to check the input skewness, "
"kurtosis, or observed_sr values.",
UserWarning,
)
probab_sr = ss.norm.cdf(test_value)
return probab_sr
def deflated_sharpe_ratio(observed_sr: float, sr_estimates: list, number_of_returns: int,
skewness_of_returns: float = 0, kurtosis_of_returns: float = 3,
estimates_param: bool = False, benchmark_out: bool = False) -> float:
def deflated_sharpe_ratio(
observed_sr: float,
sr_estimates: list,
number_of_returns: int,
skewness_of_returns: float = 0,
kurtosis_of_returns: float = 3,
estimates_param: bool = False,
benchmark_out: bool = False,
) -> float:
"""
Calculates the deflated Sharpe ratio (DSR) - a PSR where the rejection threshold is
adjusted to reflect the multiplicity of trials. DSR is estimated as PSR[SR∗], where
@@ -290,18 +346,25 @@ def deflated_sharpe_ratio(observed_sr: float, sr_estimates: list, number_of_retu
# Calculating benchmark_SR from the parameters of estimates
if estimates_param:
benchmark_sr = sr_estimates[0] * \
((1 - np.euler_gamma) * ss.norm.ppf(1 - 1 / sr_estimates[1]) +
np.euler_gamma * ss.norm.ppf(1 - 1 / sr_estimates[1] * np.e ** (-1)))
benchmark_sr = sr_estimates[0] * (
(1 - np.euler_gamma) * ss.norm.ppf(1 - 1 / sr_estimates[1])
+ np.euler_gamma * ss.norm.ppf(1 - 1 / sr_estimates[1] * np.e ** (-1))
)
# Calculating benchmark_SR from a list of estimates
else:
benchmark_sr = np.array(sr_estimates).std() * \
((1 - np.euler_gamma) * ss.norm.ppf(1 - 1 / len(sr_estimates)) +
np.euler_gamma * ss.norm.ppf(1 - 1 / len(sr_estimates) * np.e ** (-1)))
benchmark_sr = np.array(sr_estimates).std() * (
(1 - np.euler_gamma) * ss.norm.ppf(1 - 1 / len(sr_estimates))
+ np.euler_gamma * ss.norm.ppf(1 - 1 / len(sr_estimates) * np.e ** (-1))
)
deflated_sr = probabilistic_sharpe_ratio(observed_sr, benchmark_sr, number_of_returns,
skewness_of_returns, kurtosis_of_returns)
deflated_sr = probabilistic_sharpe_ratio(
observed_sr,
benchmark_sr,
number_of_returns,
skewness_of_returns,
kurtosis_of_returns,
)
if benchmark_out:
return benchmark_sr
@@ -309,10 +372,13 @@ def deflated_sharpe_ratio(observed_sr: float, sr_estimates: list, number_of_retu
return deflated_sr
def minimum_track_record_length(observed_sr: float, benchmark_sr: float,
skewness_of_returns: float = 0,
kurtosis_of_returns: float = 3,
alpha: float = 0.05) -> float:
def minimum_track_record_length(
observed_sr: float,
benchmark_sr: float,
skewness_of_returns: float = 0,
kurtosis_of_returns: float = 3,
alpha: float = 0.05,
) -> float:
"""
Calculates the minimum track record length (MinTRL) - "How long should a track
record be in order to have statistical confidence that its Sharpe ratio is above
@@ -329,8 +395,10 @@ def minimum_track_record_length(observed_sr: float, benchmark_sr: float,
:return: (float) Minimum number of track records
"""
track_rec_length = 1 + (1 - skewness_of_returns * observed_sr +
(kurtosis_of_returns - 1) / 4 * observed_sr ** 2) * \
(ss.norm.ppf(1 - alpha) / (observed_sr - benchmark_sr)) ** (2)
track_rec_length = 1 + (
1
- skewness_of_returns * observed_sr
+ (kurtosis_of_returns - 1) / 4 * observed_sr**2
) * (ss.norm.ppf(1 - alpha) / (observed_sr - benchmark_sr)) ** (2)
return track_rec_length
return track_rec_length