diff --git a/README.md b/README.md index a3486d7a..c667e78b 100644 --- a/README.md +++ b/README.md @@ -588,6 +588,7 @@ The Performance module is meant to calculate important performance metrics such - Treynor Ratio - Sharpe Ratio - Sortino Ratio +- Ulcer Performance Index (UPI) - M2 Ratio - Tracking Error - Information Error @@ -598,6 +599,8 @@ The Risk module is meant to calculate important risk metrics such as Value at Ri - Value at Risk (VaR) with distributions Historical, Gaussian, Student-t, Cornish-Fisher. - Conditional Value at Risk (cVaR) with distributions Historical, Gaussian, Laplace, Logistic. +- Entropic Value at Risk (eVaR) with a Gaussian distribution. +- Ulcer Index (UI) - Maximum Drawdown (MDD) - Skewness - Kurtosis diff --git a/examples/Finance Toolkit - 2. Combining the Finance Toolkit with the Finance Database.ipynb b/examples/Finance Toolkit - 2. Combining the Finance Toolkit with the Finance Database.ipynb index 9d1c2e31..9210462e 100644 --- a/examples/Finance Toolkit - 2. Combining the Finance Toolkit with the Finance Database.ipynb +++ b/examples/Finance Toolkit - 2. Combining the Finance Toolkit with the Finance Database.ipynb @@ -44,7 +44,7 @@ "from financetoolkit import Toolkit\n", "import financedatabase as fd\n", "\n", - "API_KEY = \"71b0c24bef80521e936f3eb851630a6a\"" + "API_KEY = \"FINANCIAL_MODELING_PREP_KEY\"" ] }, { diff --git a/financetoolkit/base/performance/performance_controller.py b/financetoolkit/base/performance/performance_controller.py index 9a43e69d..9367a1c0 100644 --- a/financetoolkit/base/performance/performance_controller.py +++ b/financetoolkit/base/performance/performance_controller.py @@ -10,8 +10,9 @@ handle_risk_free_data_periods, ) from financetoolkit.performance import performance +from financetoolkit.risk.risk import get_ui -# pylint: disable=too-many-instance-attributes,too-few-public-methods +# pylint: disable=too-many-instance-attributes,too-few-public-methods,too-many-lines class Performance: @@ -717,6 +718,78 @@ def get_sortino_ratio( return sortino_ratio + @handle_errors + def get_ulcer_performance_index( + self, + period: str | None = None, + rolling: int = 14, + rounding: int | None = None, + growth: bool = False, + lag: int | list[int] = 1, + ): + """ + Calculate the Ulcer Performance Index (UPI), alternatively called Martin ratio, a measure of risk-adjusted + return that evaluates the excess return of an investment portfolio or asset per unit of risk taken. + + It can be used to compare volatilities in different stocks or show stocks go into Ulcer territory. + Similair to the Sharpe Ratio, a higher UPI is better than a lower one (since investors prefer more return + for less risk). + + Args: + period (str, optional): The period to use for the calculation. Defaults to None which + results in basing it off the quarterly parameter as defined in the class instance. + rolling (int): The rolling period to use to calculate the Ulcer Index. Defaults to 14. + rounding (int, optional): The number of decimals to round the results to. Defaults to 4. + growth (bool, optional): Whether to calculate the growth of the ratios. Defaults to False. + lag (int | str, optional): The lag to use for the growth calculation. Defaults to 1. + + Returns: + pd.DataFrame: Ulcer Performance Index values. + + Notes: + - The method retrieves historical data and calculates the UPI for each asset in the Toolkit instance. + - The risk-free rate is often represented by the return of a risk-free investment, such as a Treasury bond. + - If `growth` is set to True, the method calculates the growth of the ratio values using the specified `lag`. + + As an example: + + ```python + from financetoolkit import Toolkit + + toolkit = Toolkit(["AAPL", "TSLA"], api_key=FMP_KEY) + + toolkit.performance.get_ulcer_performance_index() + ``` + """ + + period = period if period else "quarterly" if self._quarterly else "yearly" + + historical_data = handle_return_data_periods(self, period, True) + returns = historical_data.loc[:, "Return"][self._tickers] + historical_data_within_period = handle_return_data_periods(self, period, False) + excess_return = historical_data_within_period.loc[:, "Excess Return"][ + self._tickers + ] + + ulcer_index = get_ui(returns, rolling) + + ulcer_performance_index = performance.get_ulcer_performance_index( + excess_return, ulcer_index + ) + ulcer_performance_index = ulcer_performance_index.round( + rounding if rounding else self._rounding + ).loc[self._start_date : self._end_date] + + if growth: + return calculate_growth( + ulcer_performance_index, + lag=lag, + rounding=rounding if rounding else self._rounding, + axis="index", + ) + + return ulcer_performance_index + @handle_errors def get_m2_ratio( self, diff --git a/financetoolkit/base/ratios/ratios_controller.py b/financetoolkit/base/ratios/ratios_controller.py index 03b56b37..8308e24d 100644 --- a/financetoolkit/base/ratios/ratios_controller.py +++ b/financetoolkit/base/ratios/ratios_controller.py @@ -317,9 +317,11 @@ def collect_custom_ratios( break if formula_adjusted: - calculation = eval(formula_adjusted).astype(np.float64) + calculation = eval(formula_adjusted) # noqa - total_financials.loc[:, name, :] = calculation.to_numpy() # ruff: noqa + total_financials.loc[:, name, :] = calculation.astype( + np.float64 + ).to_numpy() self._custom_ratios = total_financials.loc[ :, list(custom_ratios_dict.keys()), : diff --git a/financetoolkit/base/risk/risk_controller.py b/financetoolkit/base/risk/risk_controller.py index a4084877..f32b746e 100644 --- a/financetoolkit/base/risk/risk_controller.py +++ b/financetoolkit/base/risk/risk_controller.py @@ -207,9 +207,9 @@ def get_conditional_value_at_risk( pd.Series: CVaR values with time as the index. Notes: - - The method retrieves historical return data based on the specified `period` and calculates VaR for each + - The method retrieves historical return data based on the specified `period` and calculates CVaR for each asset in the Toolkit instance. - - If `growth` is set to True, the method calculates the growth of VaR values using the specified `lag`. + - If `growth` is set to True, the method calculates the growth of CVaR values using the specified `lag`. Example: ```python @@ -265,6 +265,88 @@ def get_conditional_value_at_risk( return conditional_value_at_risk.round(rounding if rounding else self._rounding) + @handle_errors + def get_entropic_value_at_risk( + self, + period: str | None = None, + alpha: float = 0.05, + within_period: bool = True, + rounding: int | None = 4, + growth: bool = False, + lag: int | list[int] = 1, + ): + """ + Calculate the Entropic Value at Risk (EVaR) of an investment portfolio or asset's returns. + + Entropic Value at Risk (EVaR) is a risk management metric that quantifies upper bound for the value + at risk (VaR) and the conditional value at risk (CVaR) over a specified time horizon and confidence + level. EVaR is obtained from the Chernoff inequality. It provides insights into the downside risk + associated with an investment and helps investors make informed decisions about risk tolerance. + + The EVaR is calculated as the upper bound of VaR and CVaR with a given confidence level (e.g., 5% for + alpha=0.05). + + Args: + period (str, optional): The data frequency for returns (daily, weekly, quarterly, or yearly). + Defaults to "yearly". + alpha (float, optional): The confidence level for EVaR calculation (e.g., 0.05 for 95% confidence). + Defaults to 0.05. + within_period (bool, optional): Whether to calculate EVaR within the specified period or for the entire + period. Thus whether to look at the CVaR within a specific year (if period = 'yearly') or look at the entirety + of all years. Defaults to True. + rounding (int | None, optional): The number of decimals to round the results to. Defaults to 4. + growth (bool, optional): Whether to calculate the growth of the CVaR values over time. Defaults to False. + lag (int | list[int], optional): The lag to use for the growth calculation. Defaults to 1. + + Returns: + pd.Series: EVaR values with time as the index. + + Notes: + - The method retrieves historical return data based on the specified `period` and calculates EVaR for each + asset in the Toolkit instance. + - If `growth` is set to True, the method calculates the growth of EVaR values using the specified `lag`. + + Example: + ```python + from financetoolkit import Toolkit + + toolkit = Toolkit(["AMZN", "TSLA"], api_key=FMP_KEY) + + toolkit.risk.get_entropic_value_at_risk() + ``` + + Which returns: + + | | AMZN | TSLA | ^GSPC | + |:-----|--------:|--------:|--------:| + | 2012 | -0.0392 | -0.0604 | -0.0177 | + | 2013 | -0.0377 | -0.0928 | -0.0152 | + | 2014 | -0.0481 | -0.0689 | -0.0162 | + | 2015 | -0.046 | -0.0564 | -0.0227 | + | 2016 | -0.043 | -0.0571 | -0.0188 | + | 2017 | -0.0289 | -0.0501 | -0.0091 | + | 2018 | -0.0518 | -0.085 | -0.0252 | + | 2019 | -0.0327 | -0.071 | -0.0173 | + | 2020 | -0.054 | -0.1211 | -0.0497 | + | 2021 | -0.0352 | -0.0782 | -0.0183 | + | 2022 | -0.0758 | -0.1012 | -0.0362 | + | 2023 | -0.0471 | -0.0793 | -0.0188 | + """ + period = period if period else "quarterly" if self._quarterly else "yearly" + returns = helpers.handle_return_data_periods(self, period, within_period) + + entropic_value_at_risk = risk.get_evar_gaussian(returns, alpha) + + if growth: + return calculate_growth( + entropic_value_at_risk, + lag=lag, + rounding=rounding if rounding else self._rounding, + axis="index", + ) + + return entropic_value_at_risk.round(rounding if rounding else self._rounding) + @handle_errors def get_maximum_drawdown( self, @@ -295,12 +377,12 @@ def get_maximum_drawdown( lag (int | list[int], optional): The lag to use for the growth calculation. Defaults to 1. Returns: - pd.Series: CVaR values with time as the index. + pd.Series: Maximum Drawdown values with time as the index. Notes: - - The method retrieves historical return data based on the specified `period` and calculates VaR for each + - The method retrieves historical return data based on the specified `period` and calculates MMD for each asset in the Toolkit instance. - - If `growth` is set to True, the method calculates the growth of VaR values using the specified `lag`. + - If `growth` is set to True, the method calculates the growth of MMD values using the specified `lag`. Example: ```python @@ -343,6 +425,85 @@ def get_maximum_drawdown( return maximum_drawdown.round(rounding if rounding else self._rounding) + @handle_errors + def get_ulcer_index( + self, + period: str | None = None, + rolling: int = 14, + rounding: int | None = 4, + growth: bool = False, + lag: int | list[int] = 1, + ): + """ + The Ulcer Index is a financial metric used to assess the risk and volatility of an + investment portfolio or asset. Developed by Peter Martin in the 1980s, the Ulcer Index + is particularly useful for evaluating the downside risk and drawdowns associated with investments. + + The Ulcer Index differs from traditional volatility measures like standard deviation or variance + because it focuses on the depth and duration of drawdowns rather than the dispersion of + returns. + + The formula is a follows: + + Ulcer Index = SQRT(SUM[(Pn / Highest High)^2] / n) + + Args: + period (str, optional): The data frequency for returns (daily, weekly, quarterly, or yearly). + Defaults to "yearly". + rolling (int, optional): The rolling period to use for the calculation. Defaults to 14. + rounding (int | None, optional): The number of decimals to round the results to. Defaults to 4. + growth (bool, optional): Whether to calculate the growth of the UI values over time. Defaults to False. + lag (int | list[int], optional): The lag to use for the growth calculation. Defaults to 1. + + Returns: + pd.Series: UI values with time as the index. + + Notes: + - The method retrieves historical return data based on the specified `period` and calculates UI for each + asset in the Toolkit instance. + - If `growth` is set to True, the method calculates the growth of VaR values using the specified `lag`. + + Example: + ```python + from financetoolkit import Toolkit + + toolkit = Toolkit(["AMZN", "TSLA"], api_key=FMP_KEY) + + toolkit.risk.get_ulcer_index() + ``` + + Which returns: + + | | AMZN | TSLA | Benchmark | + |:-----|-------:|-------:|------------:| + | 2012 | 0.0497 | 0.0454 | 0.0234 | + | 2013 | 0.035 | 0.0829 | 0.0142 | + | 2014 | 0.0659 | 0.0746 | 0.0174 | + | 2015 | 0.0273 | 0.0624 | 0.0238 | + | 2016 | 0.0519 | 0.0799 | 0.0151 | + | 2017 | 0.0241 | 0.0616 | 0.0067 | + | 2018 | 0.0619 | 0.0892 | 0.0356 | + | 2019 | 0.0373 | 0.0839 | 0.016 | + | 2020 | 0.0536 | 0.1205 | 0.0594 | + | 2021 | 0.0427 | 0.085 | 0.0136 | + | 2022 | 0.1081 | 0.1373 | 0.0492 | + | 2023 | 0.0475 | 0.0815 | 0.0186 | + """ + period = period if period else "quarterly" if self._quarterly else "yearly" + returns = helpers.handle_return_data_periods(self, period, True) + + ulcer_index = risk.get_ui(returns, rolling) + + if growth: + return calculate_growth( + ulcer_index, + lag=lag, + rounding=rounding if rounding else self._rounding, + axis="index", + ) + + return ulcer_index.round(rounding if rounding else self._rounding) + @handle_errors def get_skewness( self, diff --git a/financetoolkit/performance/performance.py b/financetoolkit/performance/performance.py index 45bc1ec0..fb92717d 100644 --- a/financetoolkit/performance/performance.py +++ b/financetoolkit/performance/performance.py @@ -336,6 +336,22 @@ def get_sortino_ratio(excess_returns: pd.Series | pd.DataFrame) -> pd.Series: raise TypeError("Expects pd.DataFrame, pd.Series inputs, no other value.") +def get_ulcer_performance_index( + excess_returns: pd.Series | pd.DataFrame, ulcer_index: pd.Series | pd.DataFrame +) -> pd.Series: + """ + Calculate the Ulcer Performance Index (UPI) of returns. + + Args: + excess_returns (pd.Series | pd.DataFrame): A Series of returns with risk-free rate subtracted. + ulcer_index (pd.Series | pd.DataFrame): The corresponding + + Returns: + pd.Series: A Series of Ulcer Performance Index values with time as index and assets as columns. + """ + return (excess_returns / ulcer_index).dropna() + + def get_m2_ratio( asset_returns: pd.Series | pd.DataFrame, risk_free_rate: pd.Series, diff --git a/financetoolkit/risk/risk.py b/financetoolkit/risk/risk.py index 9d009f5f..ce0032f3 100644 --- a/financetoolkit/risk/risk.py +++ b/financetoolkit/risk/risk.py @@ -6,6 +6,11 @@ ALPHA_CONSTRAINT = 0.5 +# This is meant for calculations in which a Multi Index exists. This is the case +# when calculating a "within period" in which the first index represents the period +# (e.g. 2020Q1) and the second index the days within that period (January to March) +MULTI_PERIOD_INDEX_LEVELS = 2 + def get_var_historic( returns: pd.Series | pd.DataFrame, alpha: float @@ -23,23 +28,24 @@ def get_var_historic( """ returns = returns.dropna() if isinstance(returns, pd.DataFrame): - if returns.index.nlevels == 1: - return returns.aggregate(get_var_historic, alpha=alpha) + if returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS: + periods = returns.index.get_level_values(0).unique() + period_data_list = [] - periods = returns.index.get_level_values(0).unique() - period_data_list = [] + for sub_period in periods: + period_data = returns.loc[sub_period].aggregate( + get_var_historic, alpha=alpha + ) + period_data.name = sub_period - for sub_period in periods: - period_data = returns.loc[sub_period].aggregate( - get_var_historic, alpha=alpha - ) - period_data.name = sub_period + if not period_data.empty: + period_data_list.append(period_data) - if not period_data.empty: - period_data_list.append(period_data) + value_at_risk = pd.concat(period_data_list, axis=1) - value_at_risk = pd.concat(period_data_list, axis=1) - return value_at_risk.T + return value_at_risk.T + + return returns.aggregate(get_var_historic, alpha=alpha) if isinstance(returns, pd.Series): return np.percentile( returns, alpha * 100 @@ -73,16 +79,23 @@ def get_var_gaussian( otherwise as pd.Series or pd.DataFrame with time as index. """ returns = returns.dropna() - if isinstance(returns, pd.DataFrame) and returns.index.nlevels > 1: + if ( + isinstance(returns, pd.DataFrame) + and returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS + ): periods = returns.index.get_level_values(0).unique() - value_at_risk = pd.DataFrame() + period_data_list = [] + for sub_period in periods: period_data = get_var_gaussian( returns.loc[sub_period], alpha, cornish_fisher ) period_data.name = sub_period - value_at_risk = pd.concat([value_at_risk, period_data], axis=1) + if not period_data.empty: + period_data_list.append(period_data) + + value_at_risk = pd.concat(period_data_list, axis=1) return value_at_risk.T @@ -114,14 +127,21 @@ def get_var_studentt(returns, alpha: float) -> pd.Series | pd.DataFrame: otherwise as pd.Series or pd.DataFrame with time as index. """ returns = returns.dropna() - if isinstance(returns, pd.DataFrame) and returns.index.nlevels > 1: + if ( + isinstance(returns, pd.DataFrame) + and returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS + ): periods = returns.index.get_level_values(0).unique() - value_at_risk = pd.DataFrame() + period_data_list = [] + for sub_period in periods: period_data = get_var_studentt(returns.loc[sub_period], alpha) period_data.name = sub_period - value_at_risk = pd.concat([value_at_risk, period_data], axis=1) + if not period_data.empty: + period_data_list.append(period_data) + + value_at_risk = pd.concat(period_data_list, axis=1) return value_at_risk.T @@ -149,20 +169,23 @@ def get_cvar_historic(returns: pd.Series | pd.DataFrame, alpha: float) -> pd.Ser """ returns = returns.dropna() if isinstance(returns, pd.DataFrame): - if returns.index.nlevels == 1: - return returns.aggregate(get_cvar_historic, alpha=alpha) + if returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS: + periods = returns.index.get_level_values(0).unique() + period_data_list = [] - periods = returns.index.get_level_values(0).unique() - value_at_risk = pd.DataFrame() + for sub_period in periods: + period_data = returns.loc[sub_period].aggregate( + get_cvar_historic, alpha=alpha + ) + period_data.name = sub_period - for sub_period in periods: - period_data = returns.loc[sub_period].aggregate( - get_cvar_historic, alpha=alpha - ) - period_data.name = sub_period + if not period_data.empty: + period_data_list.append(period_data) - value_at_risk = pd.concat([value_at_risk, period_data], axis=1) - return value_at_risk.T + value_at_risk = pd.concat(period_data_list, axis=1) + + return value_at_risk.T + return returns.aggregate(get_cvar_historic, alpha=alpha) if isinstance(returns, pd.Series): return returns[ returns <= get_var_historic(returns, alpha) @@ -186,14 +209,21 @@ def get_cvar_gaussian( otherwise as pd.Series or pd.DataFrame with time as index. """ returns = returns.dropna() - if isinstance(returns, pd.DataFrame) and returns.index.nlevels > 1: + if ( + isinstance(returns, pd.DataFrame) + and returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS + ): periods = returns.index.get_level_values(0).unique() - value_at_risk = pd.DataFrame() + period_data_list = [] + for sub_period in periods: period_data = get_cvar_gaussian(returns.loc[sub_period], alpha) period_data.name = sub_period - value_at_risk = pd.concat([value_at_risk, period_data], axis=1) + if not period_data.empty: + period_data_list.append(period_data) + + value_at_risk = pd.concat(period_data_list, axis=1) return value_at_risk.T @@ -216,14 +246,21 @@ def get_cvar_studentt( otherwise as pd.Series or pd.DataFrame with time as index. """ returns = returns.dropna() - if isinstance(returns, pd.DataFrame) and returns.index.nlevels > 1: + if ( + isinstance(returns, pd.DataFrame) + and returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS + ): periods = returns.index.get_level_values(0).unique() - value_at_risk = pd.DataFrame() + period_data_list = [] + for sub_period in periods: period_data = get_cvar_studentt(returns.loc[sub_period], alpha) period_data.name = sub_period - value_at_risk = pd.concat([value_at_risk, period_data], axis=1) + if not period_data.empty: + period_data_list.append(period_data) + + value_at_risk = pd.concat(period_data_list, axis=1) return value_at_risk.T @@ -255,14 +292,21 @@ def get_cvar_laplace( otherwise as pd.Series or pd.DataFrame with time as index. """ returns = returns.dropna() - if isinstance(returns, pd.DataFrame) and returns.index.nlevels > 1: + if ( + isinstance(returns, pd.DataFrame) + and returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS + ): periods = returns.index.get_level_values(0).unique() - value_at_risk = pd.DataFrame() + period_data_list = [] + for sub_period in periods: period_data = get_cvar_laplace(returns.loc[sub_period], alpha) period_data.name = sub_period - value_at_risk = pd.concat([value_at_risk, period_data], axis=1) + if not period_data.empty: + period_data_list.append(period_data) + + value_at_risk = pd.concat(period_data_list, axis=1) return value_at_risk.T @@ -294,14 +338,21 @@ def get_cvar_logistic( otherwise as pd.Series or pd.DataFrame with time as index. """ returns = returns.dropna() - if isinstance(returns, pd.DataFrame) and returns.index.nlevels > 1: + if ( + isinstance(returns, pd.DataFrame) + and returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS + ): periods = returns.index.get_level_values(0).unique() - value_at_risk = pd.DataFrame() + period_data_list = [] + for sub_period in periods: period_data = get_cvar_logistic(returns.loc[sub_period], alpha) period_data.name = sub_period - value_at_risk = pd.concat([value_at_risk, period_data], axis=1) + if not period_data.empty: + period_data_list.append(period_data) + + value_at_risk = pd.concat(period_data_list, axis=1) return value_at_risk.T @@ -314,6 +365,46 @@ def get_cvar_logistic( return -scale * np.log(((1 - alpha) ** (1 - 1 / alpha)) / alpha) + returns.mean() +def get_evar_gaussian( + returns: pd.Series | pd.DataFrame, alpha: float +) -> pd.Series | pd.DataFrame: + """ + Calculate the Entropic Value at Risk (EVaR) of returns based on the gaussian distribution. + + For more information see: https://en.wikipedia.org/wiki/Entropic_value_at_risk + + Args: + returns (pd.Series | pd.DataFrame): A Series or Dataframe of returns. + alpha (float): The confidence level (e.g., 0.05 for 95% confidence). + + Returns: + pd.Series | pd.DataFrame: EVaR values as float if returns is a pd.Series, + otherwise as pd.Series or pd.DataFrame with time as index. + """ + returns = returns.dropna() + if ( + isinstance(returns, pd.DataFrame) + and returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS + ): + periods = returns.index.get_level_values(0).unique() + period_data_list = [] + + for sub_period in periods: + period_data = get_cvar_laplace(returns.loc[sub_period], alpha) + period_data.name = sub_period + + if not period_data.empty: + period_data_list.append(period_data) + + value_at_risk = pd.concat(period_data_list, axis=1) + + return value_at_risk.T + + return returns.mean() + returns.std(ddof=0) * np.sqrt( + -2 * np.log(returns.std(ddof=0)) + ) + + def get_max_drawdown( returns: pd.Series | pd.DataFrame, ) -> pd.Series | pd.DataFrame: @@ -328,14 +419,21 @@ def get_max_drawdown( otherwise as pd.Series or pd.DataFrame with time as index. """ returns = returns.dropna() - if isinstance(returns, pd.DataFrame) and returns.index.nlevels > 1: + if ( + isinstance(returns, pd.DataFrame) + and returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS + ): periods = returns.index.get_level_values(0).unique() - max_drawdown = pd.DataFrame() + period_data_list = [] + for sub_period in periods: period_data = get_max_drawdown(returns.loc[sub_period]) period_data.name = sub_period - max_drawdown = pd.concat([max_drawdown, period_data], axis=1) + if not period_data.empty: + period_data_list.append(period_data) + + max_drawdown = pd.concat(period_data_list, axis=1) return max_drawdown.T @@ -344,6 +442,59 @@ def get_max_drawdown( return (cum_returns / cum_returns.cummax() - 1).min() +def get_ui( + returns: pd.Series | pd.DataFrame, rolling: int = 14 +) -> pd.Series | pd.DataFrame: + """ + Calculates the Ulcer Index (UI), a measure of downside volatility. + + For more information see: + - http://www.tangotools.com/ui/ui.htm + - https://en.wikipedia.org/wiki/Ulcer_index + + Args: + returns (pd.Series | pd.DataFrame): A Series or Dataframe of returns. + rolling (int, optional): The rolling period to use for the calculation. + If you select period = 'monthly' and set rolling to 12 you obtain the rolling + 12-month Ulcer Index. If no value is given, then it calculates it for the + entire period. Defaults to None. + + Returns: + pd.Series | pd.DataFrame: UI values as float if returns is a pd.Series, + otherwise as pd.Series or pd.DataFrame with time as index, if. + """ + returns = returns.dropna() + + if isinstance(returns, pd.DataFrame): + if returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS: + periods = returns.index.get_level_values(0).unique() + period_data_list = [] + + for sub_period in periods: + period_data = returns.loc[sub_period].aggregate(get_ui) + period_data.name = sub_period + + if not period_data.empty: + period_data_list.append(period_data) + + ulcer_index = pd.concat(period_data_list, axis=1) + + return ulcer_index.T + + return returns.aggregate(get_ui) + + if isinstance(returns, pd.Series): + cumulative_returns = (1 + returns).cumprod() + cumulative_max = cumulative_returns.rolling(window=rolling).max() + drawdowns = (cumulative_returns - cumulative_max) / cumulative_max + + ulcer_index_value = np.sqrt((drawdowns**2).mean()) + + return ulcer_index_value + + raise TypeError("Expects pd.DataFrame or pd.Series, no other value.") + + def get_skewness(returns: pd.Series | pd.DataFrame) -> pd.Series | pd.DataFrame: """ Computes the skewness of dataset. @@ -356,17 +507,21 @@ def get_skewness(returns: pd.Series | pd.DataFrame) -> pd.Series | pd.DataFrame: """ returns = returns.dropna() if isinstance(returns, pd.DataFrame): - if returns.index.nlevels == 1: - return returns.aggregate(get_skewness) - periods = returns.index.get_level_values(0).unique() - skewness = pd.DataFrame() + if returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS: + periods = returns.index.get_level_values(0).unique() + period_data_list = [] - for sub_period in periods: - period_data = returns.loc[sub_period].aggregate(get_skewness) - period_data.name = sub_period + for sub_period in periods: + period_data = returns.loc[sub_period].aggregate(get_skewness) + period_data.name = sub_period + + if not period_data.empty: + period_data_list.append(period_data) - skewness = pd.concat([skewness, period_data], axis=1) - return skewness.T + skewness = pd.concat(period_data_list, axis=1) + + return skewness.T + return returns.aggregate(get_skewness) if isinstance(returns, pd.Series): return returns.skew() @@ -387,18 +542,23 @@ def get_kurtosis( """ returns = returns.dropna() if isinstance(returns, pd.DataFrame): - if returns.index.nlevels == 1: - return returns.aggregate(get_kurtosis, fisher=fisher) + if returns.index.nlevels == MULTI_PERIOD_INDEX_LEVELS: + periods = returns.index.get_level_values(0).unique() + period_data_list = [] - periods = returns.index.get_level_values(0).unique() - kurtosis = pd.DataFrame() + for sub_period in periods: + period_data = returns.loc[sub_period].aggregate( + get_kurtosis, fisher=fisher + ) + period_data.name = sub_period - for sub_period in periods: - period_data = returns.loc[sub_period].aggregate(get_kurtosis, fisher=fisher) - period_data.name = sub_period + if not period_data.empty: + period_data_list.append(period_data) + + kurtosis = pd.concat(period_data_list, axis=1) - kurtosis = pd.concat([kurtosis, period_data], axis=1) - return kurtosis.T + return kurtosis.T + return returns.aggregate(get_kurtosis, fisher=fisher) if isinstance(returns, pd.Series): if fisher: return returns.kurtosis() diff --git a/pyproject.toml b/pyproject.toml index 82788da8..2007d473 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "financetoolkit" -version = "1.3.12" +version = "1.3.13" description = "Transparent and Efficient Financial Analysis" license = "MIT" authors = ["Jeroen Bouma"] @@ -67,8 +67,8 @@ skip_gitignore = true combine_as_imports = true [tool.codespell] -ignore-words-list = 'zar,profund,basf,applicatio,ser,mone,vie,wew,ist,tre,ue,nd,fo,nwe,0t,Ot,ot,juni,acn,hve,te,hsa,fof,ege,bu,wya,esy,whn,ned,4rd,wih,mke,caf,fpr,hav,coo' -skip = '*.json,./.git,pyproject.toml,poetry.lock' +ignore-words-list = 'te' +skip = '*.json,./.git,pyproject.toml,poetry.lock,examples' [tool.mypy] disable_error_code = "misc"