Source code for skfolio.portfolio._multi_period_portfolio

"""Multi Period Portfolio module.
`MultiPeriodPortfolio` is returned by the `predict` method of Optimization estimators.
`MultiPeriodPortfolio` is a list of `Portfolio`.
"""

# Copyright (c) 2023
# Author: Hugo Delatte <delatte.hugo@gmail.com>
# License: BSD 3 clause

import numbers
from collections.abc import Iterator

import numpy as np
import pandas as pd

import skfolio.typing as skt
from skfolio.portfolio._base import BasePortfolio
from skfolio.portfolio._portfolio import Portfolio
from skfolio.utils.tools import deduplicate_names


[docs] class MultiPeriodPortfolio(BasePortfolio): r"""Multi-Period Portfolio class. A Multi-Period Portfolio is composed of a list of :class:`Portfolio`. Parameters ---------- portfolios : list[Portfolio], optional A list of :class:`Portfolio`. The default (`None`) is to initialize with an empty list. name : str, optional Name of the multi-period portfolio. The default (`None`) is to use the object id. tag : str, optional Tag given to the multi-period portfolio. Tags are used to manipulate groups of portfolios from a `Population`. fitness_measures : list[measures], optional List of fitness measures. Fitness measures are used to compute the portfolio fitness which is used to compute domination. The default (`None`) is to use the list [PerfMeasure.MEAN, RiskMeasure.VARIANCE] annualized_factor : float, default=252.0 Factor used to annualize the below measures using the square-root rule: * Annualized Mean = Mean * factor * Annualized Variance = Variance * factor * Annualized Semi-Variance = Semi-Variance * factor * Annualized Standard-Deviation = Standard-Deviation * sqrt(factor) * Annualized Semi-Deviation = Semi-Deviation * sqrt(factor) * Annualized Sharpe Ratio = Sharpe Ratio * sqrt(factor) * Annualized Sortino Ratio = Sortino Ratio * sqrt(factor) risk_free_rate : float, default=0.0 Risk-free rate. The default value is `0.0`. compounded : bool, default=False If this is set to True, cumulative returns are compounded. The default is `False`. min_acceptable_return : float, optional The minimum acceptable return used to distinguish "downside" and "upside" returns for the computation of lower partial moments: * First Lower Partial Moment * Semi-Variance * Semi-Deviation The default (`None`) is to use the mean. value_at_risk_beta : float, default=0.95 The confidence level of the portfolio VaR (Value At Risk) which represents the return on the worst (1-beta)% observations. The default value is `0.95`. entropic_risk_measure_theta : float, default=1.0 The risk aversion level of the portfolio Entropic Risk Measure. The default value is `1.0`. entropic_risk_measure_beta : float, default=0.95 The confidence level of the portfolio Entropic Risk Measure. The default value is `0.95`. cvar_beta : float, default=0.95 The confidence level of the portfolio CVaR (Conditional Value at Risk) which represents the expected VaR on the worst (1-beta)% observations. The default value is `0.95`. evar_beta : float, default=0.95 The confidence level of the portfolio EVaR (Entropic Value at Risk). The default value is `0.95`. drawdown_at_risk_beta : float, default=0.95 The confidence level of the portfolio Drawdown at Risk (DaR) which represents the drawdown on the worst (1-beta)% observations. The default value is `0.95`. cdar_beta : float, default=0.95 The confidence level of the portfolio CDaR (Conditional Drawdown at Risk) which represents the expected drawdown on the worst (1-beta)% observations. The default value is `0.95`. edar_beta : float, default=0.95 The confidence level of the portfolio EDaR (Entropic Drawdown at Risk). The default value is `0.95`. check_observations_order : bool, default=False If this is set to True, and if the list of portfolios is not chronologically sorted, an error is raised. The chronological order is determined by comparing the first and last observations of each portfolio. The default is `False`. Attributes ---------- n_observations : float Number of observations. mean : float Mean of the portfolio returns. annualized_mean : float Mean annualized by :math:`mean \times annualization\_factor` mean_absolute_deviation : float Mean Absolute Deviation. The deviation is the difference between the return and a minimum acceptable return (`min_acceptable_return`). first_lower_partial_moment : float First Lower Partial Moment. The First Lower Partial Moment is the mean of the returns below a minimum acceptable return (`min_acceptable_return`). variance : float Variance (Second Moment) annualized_variance : float Variance annualized by :math:`variance \times annualization\_factor` semi_variance : float Semi-variance (Second Lower Partial Moment). The semi-variance is the variance of the returns below a minimum acceptable return (`min_acceptable_return`). annualized_semi_variance : float Semi-variance annualized by :math:`semi\_variance \times annualization\_factor` standard_deviation : float Standard Deviation (Square Root of the Second Moment). annualized_standard_deviation : float Standard Deviation annualized by :math:`standard\_deviation \times \sqrt{annualization\_factor}` semi_deviation : float Semi-deviation (Square Root of the Second Lower Partial Moment). The Semi Standard Deviation is the Standard Deviation of the returns below a minimum acceptable return (`min_acceptable_return`). annualized_semi_deviation : float Semi-deviation annualized by :math:`semi\_deviation \times \sqrt{annualization\_factor}` skew : float Skew. The Skew is a measure of the lopsidedness of the distribution. A symmetric distribution have a Skew of zero. Higher Skew corresponds to longer right tail. kurtosis : float Kurtosis. It is a measure of the heaviness of the tail of the distribution. Higher Kurtosis corresponds to greater extremity of deviations (fat tails). fourth_central_moment : float Fourth Central Moment. fourth_lower_partial_moment : float Fourth Lower Partial Moment. It is a measure of the heaviness of the downside tail of the returns below a minimum acceptable return (`min_acceptable_return`). Higher Fourth Lower Partial Moment corresponds to greater extremity of downside deviations (downside fat tail). worst_realization : float Worst Realization which is the worst return. value_at_risk : float Historical VaR (Value at Risk). The VaR is the maximum loss at a given confidence level (`value_at_risk_beta`). cvar : float Historical CVaR (Conditional Value at Risk). The CVaR (or Tail VaR) represents the mean shortfall at a specified confidence level (`cvar_beta`). entropic_risk_measure : float Historical Entropic Risk Measure. It is a risk measure which depends on the risk aversion defined by the investor (`entropic_risk_measure_theta`) through the exponential utility function at a given confidence level (`entropic_risk_measure_beta`). evar : float Historical EVaR (Entropic Value at Risk). It is a coherent risk measure which is an upper bound for the VaR and the CVaR, obtained from the Chernoff inequality at a given confidence level (`evar_beta`). The EVaR can be represented by using the concept of relative entropy. drawdown_at_risk : float Historical Drawdown at Risk. It is the maximum drawdown at a given confidence level (`drawdown_at_risk_beta`). cdar : float Historical CDaR (Conditional Drawdown at Risk) at a given confidence level (`cdar_beta`). max_drawdown : float Maximum Drawdown. average_drawdown : float Average Drawdown. edar : float EDaR (Entropic Drawdown at Risk). It is a coherent risk measure which is an upper bound for the Drawdown at Risk and the CDaR, obtained from the Chernoff inequality at a given confidence level (`edar_beta`). The EDaR can be represented by using the concept of relative entropy. ulcer_index : float Ulcer Index gini_mean_difference : float Gini Mean Difference (GMD). It is the expected absolute difference between two realizations. The GMD is a superior measure of variability for non-normal distribution than the variance. It can be used to form necessary conditions for second-degree stochastic dominance, while the variance cannot. mean_absolute_deviation_ratio : float Mean Absolute Deviation ratio. It is the excess mean (mean - risk_free_rate) divided by the MaD. first_lower_partial_moment_ratio : float First Lower Partial Moment ratio. It is the excess mean (mean - risk_free_rate) divided by the First Lower Partial Moment. sharpe_ratio : float Sharpe ratio. It is the excess mean (mean - risk_free_rate) divided by the standard-deviation. annualized_sharpe_ratio : float Sharpe ratio annualized by :math:`sharpe\_ratio \times \sqrt{annualization\_factor}`. sortino_ratio : float Sortino ratio. It is the excess mean (mean - risk_free_rate) divided by the semi standard-deviation. annualized_sortino_ratio : float Sortino ratio annualized by :math:`sortino\_ratio \times \sqrt{annualization\_factor}`. value_at_risk_ratio : float VaR ratio. It is the excess mean (mean - risk_free_rate) divided by the Value at Risk (VaR). cvar_ratio : float CVaR ratio. It is the excess mean (mean - risk_free_rate) divided by the Conditional Value at Risk (CVaR). entropic_risk_measure_ratio : float Entropic risk measure ratio. It is the excess mean (mean - risk_free_rate) divided by the Entropic risk measure. evar_ratio : float EVaR ratio. It is the excess mean (mean - risk_free_rate) divided by the EVaR (Entropic Value at Risk). worst_realization_ratio : float Worst Realization ratio. It is the excess mean (mean - risk_free_rate) divided by the Worst Realization (worst return). drawdown_at_risk_ratio : float Drawdown at Risk ratio. It is the excess mean (mean - risk_free_rate) divided by the drawdown at risk. cdar_ratio : float CDaR ratio. It is the excess mean (mean - risk_free_rate) divided by the CDaR (conditional drawdown at risk). calmar_ratio : float Calmar ratio. It is the excess mean (mean - risk_free_rate) divided by the Maximum Drawdown. average_drawdown_ratio : float Average Drawdown ratio. It is the excess mean (mean - risk_free_rate) divided by the Average Drawdown. edar_ratio : float EDaR ratio. It is the excess mean (mean - risk_free_rate) divided by the EDaR (Entropic Drawdown at Risk). ulcer_index_ratio : float Ulcer Index ratio. It is the excess mean (mean - risk_free_rate) divided by the Ulcer Index. gini_mean_difference_ratio : float Gini Mean Difference ratio. It is the excess mean (mean - risk_free_rate) divided by the Gini Mean Difference. """ __slots__ = { # read-only "_portfolios", "check_observations_order", } def __init__( self, portfolios: list[Portfolio] | None = None, name: str | None = None, tag: str | None = None, risk_free_rate: float = 0, annualized_factor: float = 252.0, fitness_measures: list[skt.Measure] | None = None, compounded: bool = False, min_acceptable_return: float | None = None, value_at_risk_beta: float = 0.95, entropic_risk_measure_theta: float = 1, entropic_risk_measure_beta: float = 0.95, cvar_beta: float = 0.95, evar_beta: float = 0.95, drawdown_at_risk_beta: float = 0.95, cdar_beta: float = 0.95, edar_beta: float = 0.95, check_observations_order: bool = False, ): super().__init__( returns=np.array([]), observations=np.array([]), name=name, tag=tag, risk_free_rate=risk_free_rate, annualized_factor=annualized_factor, fitness_measures=fitness_measures, compounded=compounded, min_acceptable_return=min_acceptable_return, value_at_risk_beta=value_at_risk_beta, cvar_beta=cvar_beta, entropic_risk_measure_theta=entropic_risk_measure_theta, entropic_risk_measure_beta=entropic_risk_measure_beta, evar_beta=evar_beta, drawdown_at_risk_beta=drawdown_at_risk_beta, cdar_beta=cdar_beta, edar_beta=edar_beta, ) self.check_observations_order = check_observations_order self._set_portfolios(portfolios=portfolios) def __len__(self) -> int: return len(self.portfolios) def __getitem__(self, key: int | slice) -> Portfolio | list[Portfolio]: return self._portfolios[key] def __setitem__(self, key: int, value: Portfolio) -> None: if not isinstance(value, Portfolio): raise TypeError(f"Cannot set a value with type {type(value)}") new_portfolios = self._portfolios.copy() new_portfolios[key] = value self._set_portfolios(portfolios=new_portfolios) self.clear() def __delitem__(self, key: int) -> None: new_portfolios = self._portfolios.copy() del new_portfolios[key] self._set_portfolios(portfolios=new_portfolios) self.clear() def __iter__(self) -> Iterator[Portfolio]: return iter(self._portfolios) def __contains__(self, value: Portfolio) -> bool: if not isinstance(value, Portfolio): return False return value in self._portfolios def __neg__(self): return self.__class__( portfolios=[-p for p in self], tag=self.tag, fitness_measures=self.fitness_measures, ) def __abs__(self): return self.__class__( portfolios=[abs(p) for p in self], tag=self.tag, fitness_measures=self.fitness_measures, ) def __round__(self, n: int): return self.__class__( portfolios=[p.__round__(n) for p in self], tag=self.tag, fitness_measures=self.fitness_measures, ) def __floor__(self): return self.__class__( portfolios=[np.floor(p) for p in self], tag=self.tag, fitness_measures=self.fitness_measures, ) def __trunc__(self): return self.__class__( portfolios=[np.trunc(p) for p in self], tag=self.tag, fitness_measures=self.fitness_measures, ) def __add__(self, other): if not isinstance(other, self.__class__): raise TypeError( "Cannot add a MultiPeriodPortfolio with an object of type" f" {type(other)}" ) if len(self) != len(other): raise TypeError("Cannot add two MultiPeriodPortfolio of different sizes") return self.__class__( portfolios=[p1 + p2 for p1, p2 in zip(self, other, strict=True)], tag=self.tag, fitness_measures=self.fitness_measures, ) def __sub__(self, other): if not isinstance(other, self.__class__): raise TypeError( "Cannot subtract a MultiPeriodPortfolio with an object of type" f" {type(other)}" ) if len(self) != len(other): raise TypeError( "Cannot subtract two MultiPeriodPortfolio of different sizes" ) return self.__class__( portfolios=[p1 - p2 for p1, p2 in zip(self, other, strict=True)], tag=self.tag, fitness_measures=self.fitness_measures, ) def __mul__(self, other: numbers.Number | list[numbers.Number] | np.ndarray): if np.isscalar(other): portfolios = [p * other for p in self] else: portfolios = [p * a for p, a in zip(self, other, strict=True)] return self.__class__( portfolios=portfolios, tag=self.tag, fitness_measures=self.fitness_measures ) __rmul__ = __mul__ def __floordiv__(self, other: numbers.Number | list[numbers.Number] | np.ndarray): if np.isscalar(other): portfolios = [p // other for p in self] else: portfolios = [p // a for p, a in zip(self, other, strict=True)] return self.__class__( portfolios=portfolios, tag=self.tag, fitness_measures=self.fitness_measures ) def __truediv__(self, other: numbers.Number | list[numbers.Number] | np.ndarray): if np.isscalar(other): portfolios = [p / other for p in self] else: portfolios = [p / a for p, a in zip(self, other, strict=True)] return self.__class__( portfolios=portfolios, tag=self.tag, fitness_measures=self.fitness_measures ) # Private method def _set_portfolios(self, portfolios: list[Portfolio] | None = None) -> None: """Set the returns, observations and portfolios list. Parameters ---------- portfolios : list[Portfolio], optional The list of Portfolios. The default (`None`) is to use an empty list. """ returns = [] observations = [] if portfolios is None: portfolios = [] if len(portfolios) != 0: for item in portfolios: if not isinstance(item, BasePortfolio | Portfolio): raise TypeError( "`portfolios` items must be of type `Portfolio`, got" f" {type(item).__name__}" ) returns.append(item.returns) observations.append(item.observations) returns = np.concatenate(returns) observations = np.concatenate(observations) if self.check_observations_order: iteration = iter(portfolios) prev_p = next(iteration) while (p := next(iteration, None)) is not None: if p.observations[0] <= prev_p.observations[-1]: raise ValueError( "Portfolios observations should not overlap:" f" {p} overlapping {prev_p}" ) prev_p = p self._loaded = False self._portfolios = portfolios self.returns = np.asarray(returns) self.observations = np.asarray(observations) self._loaded = True # Custom attribute setter and getter @property def portfolios(self) -> list[Portfolio]: """List of portfolios composing the mutli-period portfolio.""" return self._portfolios @portfolios.setter def portfolios(self, value: list[Portfolio] | None = None): """Set the list of Portfolios and clear the attributes cache linked to the list of portfolios.""" self._set_portfolios(portfolios=value) self.clear() # Classic property @property def assets(self) -> list: """List of assets names in each Portfolio.""" return [p.assets for p in self] @property def composition(self) -> pd.DataFrame: """DataFrame of the Portfolio composition.""" df = pd.concat([p.composition for p in self], axis=1) df.fillna(0, inplace=True) df.columns = deduplicate_names(df.columns) return df @property def weights_per_observation(self) -> pd.DataFrame: """DataFrame of the Portfolio weights per observation.""" return ( pd.concat([p.weights_per_observation for p in self], axis=0) .fillna(0) .sort_index() )
[docs] def contribution( self, measure: skt.Measure, spacing: float | None = None, to_df: bool = True ) -> np.ndarray | pd.DataFrame: r"""Compute the contribution of each asset to a given measure for each portfolio. Parameters ---------- measure : Measure The measure used for the contribution computation. spacing : float, optional Spacing "h" of the finite difference: :math:`contribution(wi)= \frac{measure(wi-h) - measure(wi+h)}{2h}` to_df : bool, default=False If this is set to True, a DataFrame with asset names in index and portfolio names in columns is returned, otherwise a list of numpy array is returned. When a DataFrame is returned, the assets with zero weights are removed. Returns ------- values : list of numpy array of shape (n_assets,) for each portfolio or a DataFrame The measure contribution of each asset for each portfolio. """ contributions = [ ptf.contribution(measure=measure, spacing=spacing, to_df=to_df) for ptf in self ] if not to_df: return contributions df = pd.concat(contributions, axis=1) df.fillna(0, inplace=True) df.columns = deduplicate_names(df.columns) return df
[docs] def summary(self, formatted: bool = True) -> pd.Series: """Portfolio summary of all its measures. Parameters ---------- formatted : bool, default=True If this is set to True, the measures are formatted into rounded string with units. Returns ------- summary : series Portfolio summary of all its measures. """ df = super().summary(formatted=formatted) portfolios_number = len(self) avg_assets_per_portfolio = np.mean([p.n_assets for p in self]) if formatted: portfolios_number = str(int(portfolios_number)) avg_assets_per_portfolio = f"{avg_assets_per_portfolio:0.1f}" df["Portfolios Number"] = portfolios_number df["Avg nb of Assets per Portfolio"] = avg_assets_per_portfolio return df
# Public methods
[docs] def append(self, portfolio: Portfolio) -> None: """Append a Portfolio to the Portfolio list. Parameters ---------- portfolio : Portfolio The Portfolio to append. """ if self.check_observations_order and len(self) != 0: start_date = portfolio.observations[0] prev_last_date = self[-1].observations[-1] if start_date < prev_last_date: raise ValueError( f"Portfolios observations should not overlap: {prev_last_date} ->" f" {start_date} " ) self._loaded = False self._portfolios.append(portfolio) if len(self.observations) == 0: # We don"t concatenate an empty array as we cannot know the dtype before. self.observations = portfolio.observations self.returns = portfolio.returns else: self.observations = np.concatenate( [self.observations, portfolio.observations], axis=0 ) self.returns = np.concatenate([self.returns, portfolio.returns], axis=0) self._loaded = True self.clear()