New methods to compare ETFs with index (benchmarks) in AssetList:

- tracking difference
- tracking difference annualized
- tracking error
- index correlation
- index beta

Author: chilango74

okama/assets.py

@@ -4,7 +4,7 @@
 import numpy as np
 
 from .macro import Inflation
-from .helpers import Float, Frame, Rebalance, Date
+from .helpers import Float, Frame, Rebalance, Date, Index
 from .settings import default_ticker, assets_namespaces
 from .data import QueryData
 
@@ -215,6 +215,10 @@ def currency(self):
 
     @property
     def wealth_indexes(self) -> pd.DataFrame:
+        """
+        Wealth index time series for the assets and accumulated inflation.
+        Wealth index is obtained from the accumulated return multiplicated by the initial investments (1000).
+        """
         if hasattr(self, 'inflation'):
             df = pd.concat([self.ror, self.inflation_ts], axis=1, join='inner', copy='false')
         else:
@@ -239,25 +243,42 @@ def risk_annual(self) -> pd.Series:
 
     @property
     def semideviation_monthly(self) -> pd.Series:
+        """
+        Returns semideviation monthly values for each asset (full period).
+        """
         return Frame.get_semideviation(self.ror)
 
     @property
     def semideviation_annual(self) -> float:
+        """
+        Returns semideviation annual values for each asset (full period).
+        """
         return Frame.get_semideviation(self.returns_ts) * 12 ** 0.5
 
     def get_var_historic(self, level: int = 5) -> pd.Series:
+        """
+        Calculates historic VAR for the assets (full period).
+        VAR levels could be set by level attribute (integer).
+        """
         return Frame.get_var_historic(self.ror, level)
 
     def get_cvar_historic(self, level: int = 5) -> pd.Series:
+        """
+        Calculates historic CVAR for the assets (full period).
+        CVAR levels could be set by level attribute (integer).
+        """
         return Frame.get_cvar_historic(self.ror, level)
 
     @property
     def drawdowns(self) -> pd.DataFrame:
+        """
+        Calculates drawdowns time series for the assets.
+        """
         return Frame.get_drawdowns(self.ror)
 
     def get_cagr(self, period: Union[str, int, None] = None) -> pd.Series:
         """
-        Calculates Compound Annual Growth Rate for a given period:
+        Calculates Compound Annual Growth Rate (CAGR) for a given period:
         None: full time
         'YTD': Year To Date compound rate of return (formally not a CAGR)
         Integer: several years
@@ -286,6 +307,9 @@ def get_cagr(self, period: Union[str, int, None] = None) -> pd.Series:
 
     @property
     def annual_return_ts(self) -> pd.DataFrame:
+        """
+        Calculates annual rate of return time series for the assets.
+        """
         return Frame.get_annual_return_ts_from_monthly(self.ror)
 
     def describe(self, years: tuple = (1, 5, 10), tickers: bool = True) -> pd.DataFrame:
@@ -394,6 +418,9 @@ def describe(self, years: tuple = (1, 5, 10), tickers: bool = True) -> pd.DataFr
 
     @property
     def mean_return(self) -> pd.Series:
+        """
+        Calculates mean return (arithmetic mean) for the assets.
+        """
         if hasattr(self, 'inflation'):
             df = pd.concat([self.ror, self.inflation_ts], axis=1, join='inner', copy='false')
         else:
@@ -404,7 +431,7 @@ def mean_return(self) -> pd.Series:
     @property
     def real_mean_return(self) -> pd.Series:
         """
-        Calculates real mean return (arithmetic mean).
+        Calculates real mean return (arithmetic mean) for the assets.
         """
         if hasattr(self, 'inflation'):
             df = pd.concat([self.ror, self.inflation_ts], axis=1, join='inner', copy='false')
@@ -533,6 +560,53 @@ def get_dividend_mean_growth_rate(self, period=5) -> pd.Series:
             raise TypeError(f'{period} is not a valid value for period')
         return mean_growth_rate
 
+    # index methods
+    @property
+    def tracking_difference(self):
+        """
+        Returns tracking difference for the rate of return of assets.
+        Assets are compared with the index or another benchmark.
+        Index should be in the first position (first column).
+        """
+        accumulated_return = Frame.get_wealth_indexes(self.ror)  # we don't need inflation here
+        return Index.tracking_difference(accumulated_return)
+
+    @property
+    def tracking_difference_annualized(self):
+        """
+        Annualizes the values of tracking difference time series.
+        Annual values are available for periods of more than 12 months.
+        Returns for less than 12 months can't be annualized.
+        """
+        return Index.tracking_difference_annualized(self.tracking_difference)
+
+    @property
+    def tracking_error(self):
+        """
+        Returns tracking error for the rate of return time series of assets.
+        Assets are compared with the index or another benchmark.
+        Index should be in the first position (first column).
+        """
+        return Index.tracking_error(self.ror)
+
+    @property
+    def index_corr(self):
+        """
+        Returns the accumulated correlation with the index (or benchmark) time series for the assets.
+        Index should be in the first position (first column).
+        The period should be at least 12 months.
+        """
+        return Index.cov_cor(self.ror, fn='corr')
+
+    @property
+    def index_beta(self):
+        """
+        Returns beta coefficient time series for the assets.
+        Index (or benchmark) should be in the first position (first column).
+        The period should be at least 12 months.
+        """
+        return Index.beta(self.ror)
+
 
 class Portfolio:
     """

okama/frontier_reb.py

@@ -163,7 +163,7 @@ def gmv_annual_values(self) -> Tuple[float]:
     @property
     def max_return(self) -> dict:
         """
-        Returns the weights and risk / CAGR of the maximum return portfolio.
+        Returns the weights and risk / CAGR of the maximum return portfolio point.
         """
         ror = self.ror
         period = self.reb_period

okama/helpers.py

@@ -226,6 +226,8 @@ def get_cvar_historic(ror: Union[pd.DataFrame, pd.Series], level: int = 5) -> Un
         """
         Computes the Conditional VaR (CVaR) of Series or DataFrame at a specified level.
         """
+        if not isinstance(level, int):
+            raise TypeError("Level should be an integer.")
         if isinstance(ror, pd.Series) or isinstance(ror, pd.DataFrame):
             is_beyond = ror <= ror.quantile(level / 100)  # mask: return is less than quantile
             return -ror[is_beyond].mean()
@@ -303,6 +305,25 @@ def rebalanced_portfolio_return_ts(weights: list, ror: pd.DataFrame, *, period:
         # ror.sort_index(ascending=True, inplace=True)
         return ror
 
+    @staticmethod
+    def create_fn_list_ror_ts(ror: pd.DataFrame, *, period: str = 'Y') -> list:
+        """
+        Returns a list of functions of weights.
+        """
+        # Frame.weights_sum_is_one(weights)
+        initial_inv = 1000
+        fn_list = []
+        for x in ror.resample(period):
+            def ror_list_fn(weights, y=x):
+                df = y[1]  # select ror part of the grouped data
+                inv_period_spread = np.asarray(weights) * initial_inv  # rebalancing
+                assets_wealth_indexes = inv_period_spread * (1 + df).cumprod()
+                wealth_index_local = assets_wealth_indexes.sum(axis=1)
+                ror_local = wealth_index_local.pct_change()
+                return ror_local
+            fn_list.append(ror_list_fn)
+        return fn_list
+
 
 class Date:
     @staticmethod
@@ -320,3 +341,72 @@ def subtract_years(dt: pd.Timestamp, years: int) -> pd.Timestamp:
             raise TypeError('The period should be integer')
         return dt
 
+
+class Index:
+    @staticmethod
+    def tracking_difference(accumulated_return: pd.DataFrame) -> pd.DataFrame:
+        """
+        Returns tracking difference for a rate of return time series.
+        Assets are compared with the index or another benchmark.
+        Index should be in the first position (first column).
+        """
+        if accumulated_return.shape[1] < 2:
+            raise ValueError('At least 2 symbols should be provided to calculate Tracking Difference.')
+        initial_value = accumulated_return.iloc[0]
+        difference = accumulated_return.subtract(accumulated_return.iloc[:, 0], axis=0) / initial_value
+        difference.drop(difference.columns[0], axis=1, inplace=True)  # drop the first column (stock index data)
+        return difference
+
+    @staticmethod
+    def tracking_difference_annualized(tracking_diff: pd.DataFrame) -> pd.DataFrame:
+        """
+        Annualizes the values of tracking difference time series.
+        Annual values are available for periods of more than 12 months.
+        Returns for less than 12 months can't be annualized.
+        """
+        pwr = 12 / (1. + np.arange(tracking_diff.shape[0]))
+        y = abs(tracking_diff)
+        diff = np.sign(tracking_diff) * (y + 1.).pow(pwr, axis=0) - 1.
+        return diff.iloc[_MONTHS_PER_YEAR - 1:]  # returns for the first 11 months can't be annualized
+
+    @staticmethod
+    def tracking_error(ror: pd.DataFrame) -> pd.DataFrame:
+        """
+        Returns tracking error for a rate of return time series.
+        Assets are compared with the index or another benchmark.
+        Index should be in the first position (first column).
+        """
+        if ror.shape[1] < 2:
+            raise ValueError('At least 2 symbols should be provided to calculate Tracking Error.')
+        cumsum = ror.subtract(ror.iloc[:, 0], axis=0).pow(2, axis=0).cumsum()
+        cumsum.drop(cumsum.columns[0], axis=1, inplace=True)  # drop the first column (stock index data)
+        tracking_error = cumsum.divide((1. + np.arange(ror.shape[0])), axis=0).pow(0.5, axis=0)
+        return tracking_error * np.sqrt(12)
+
+    @staticmethod
+    def cov_cor(ror: pd.DataFrame, fn: str) -> pd.DataFrame:
+        """
+        Returns the accumulated correlation or covariance time series.
+        The period should be at least 12 months.
+        """
+        if ror.shape[1] < 2:
+            raise ValueError('At least 2 symbols should be provided.')
+        if fn not in ['cov', 'corr']:
+            raise ValueError('fn should be cor or cov')
+        cov_matrix_ts = getattr(ror.expanding(), fn)()
+        cov_matrix_ts = cov_matrix_ts.drop(index=ror.columns[1:], level=1).droplevel(1)
+        cov_matrix_ts.drop(columns=ror.columns[0], inplace=True)
+        return cov_matrix_ts.iloc[_MONTHS_PER_YEAR:]
+
+    @staticmethod
+    def beta(ror: pd.DataFrame) -> pd.DataFrame:
+        """
+        Returns beta coefficient the rate of return time series.
+        Index (or benchmark) should be in the first position (first column).
+        The period should be at least 12 months.
+        """
+        cov = Index.cov_cor(ror, fn='cov')
+        std = ror.expanding().std().drop(columns=ror.columns[0])
+        std = std[_MONTHS_PER_YEAR:]
+        return cov / std
+

okama/plots.py

@@ -201,16 +201,4 @@ def plot_forecast(self,
             else:
                 self.ax.plot(x, y, linestyle='dashed', linewidth=1, label=f'Percentile {percentile}')
         self.ax.legend(loc='upper left')
-        # place a text box in upper left in axes coords
-        # max_value = y_end_values[percentiles[-1]]
-        # min_value = y_end_values[percentiles[0]]
-        # median_value = y_end_values[percentiles[1]]
-        # textstr = '\n'.join((
-        #     f'Значение портфеля сегодня: {today_value}',
-        #     f'Максимальное значение через {years} лет: {max_value}',
-        #     f'Минимальное значение через {years} лет: {min_value}',
-        #     f'Наиболее вероятное значение через {years} лет: {median_value}'
-        # ))
-        # props = dict(boxstyle='round', facecolor='wheat', alpha=0.5)
-        # self.ax.text(0.02, 0.85, textstr, transform=self.ax.transAxes, fontsize=14, verticalalignment='top', bbox=props)
-        return self.ax
+        return self.ax

tests/test_assets.py

@@ -115,16 +115,33 @@ def test_growing_dividend_years(self):
     def test_paying_dividend_years(self):
         assert self.spy.dividend_paying_years.iloc[-2, 0] == 2
 
+    def test_tracking_difference_failing(self):
+        with pytest.raises(Exception, match='At least 2 symbols should be provided to calculate Tracking Difference.'):
+            self.spy.tracking_difference
+
+    def test_tracking_difference(self):
+        assert self.asset_list.tracking_difference.iloc[-1, 0] == approx(0.4832375074686104, rel=1e-2)
+
+    def test_tracking_difference_annualized(self):
+        assert self.asset_list.tracking_difference_annualized.iloc[-1, 0] == approx(0.438933, rel=1e-2)
+
+    def test_tracking_error(self):
+        assert self.asset_list.tracking_error.iloc[-1, 0] == approx(0.203372, rel=1e-2)
+
+    def test_index_corr(self):
+        assert self.asset_list.index_corr.iloc[-1, 0] == approx(-0.62177, rel=1e-2)
+
+    @mark.test
+    def test_index_beta(self):
+        assert self.asset_list.index_beta.iloc[-1, 0] == approx(-0.018598, rel=1e-2)
+
+
 @mark.portfolio
 def test_init_portfolio_failing():
     with pytest.raises(Exception, match=r'Number of tickers \(2\) should be equal to the weights number \(3\)'):
         Portfolio(symbols=['RUB.FX', 'MCFTR.INDX'], weights=[0.1, 0.2, 0.7]).symbols
     with pytest.raises(Exception, match='Weights sum is not equal to one.'):
         Portfolio(symbols=['RUB.FX', 'MCFTR.INDX'], weights=[0.1, 0.2]).symbols
-    # with pytest.raises(Exception, match=r'RUB is not in allowed assets namespaces*'):
-    #     Portfolio(symbols=['RUB.RUB'], weights=[1])
-    # # with pytest.raises(Exception):
-    # #     Portfolio(symbols=['XXXX.RE'], weights=[1])
 
 
 @mark.portfolio