feat: support exponential weighted factor covariance

src/fpm_risk_model/engine/numpy.py

@@ -3,7 +3,4 @@ class NumpyEngine:
     Numpy Engine.
     """
 
-    from numpy import diagonal, sqrt
-
-    diagonal = diagonal
-    sqrt = sqrt
+    from numpy import array, cov, diagonal, newaxis, sqrt, sum

src/fpm_risk_model/factor_risk_model.py

@@ -220,49 +220,66 @@ def transform(self, y: ndarray, regressor: Optional[object] = None) -> object:
         self._residual_returns = residual_returns
         return self
 
-    def cov(self) -> ndarray:
+    def cov(self, halflife: Optional[float] = None) -> ndarray:
         """
         Get the covariance matrix.
 
+        Parameters
+        ----------
+        halflife : Optional[float]
+            Half life in applying the exponential weighting on factor
+            returns for computing the factor covariance matrix. If
+            None is passed, no exponential weighting is applied.
+
         Returns
         -------
         numpy.ndarray
             A square pairwise covariance matrix which its
             diagonal entries are the variances.
         """
-        cov = (
-            self._factor_exposures.T @ self._factor_covariances @ self._factor_exposures
-        )
+        B = self._factor_exposures
+        F = self._factor_returns
+        if F is None:
+            raise ValueError("Factor return cannot be None")
+
+        if halflife is not None:
+            T = F.shape[0]
+            W = self._engine.array(
+                [2 ** (-(T - 1 - t) / halflife / 2) for t in range(0, T)]
+            )
+            F = F * W[:, self._engine.newaxis]
+
+        factor_covariances = self._engine.cov(F.T)
         specific_variances = self.specific_variances()
 
-        if isinstance(cov, DataFrame):
-            cov_values = cov.values
-            specific_variances = specific_variances.loc[cov.index]
-        elif isinstance(cov, ndarray):
-            cov_values = cov
-        else:
+        R = specific_variances
+        if isinstance(B, DataFrame):
+            instruments = self._factor_exposures.columns
+            B = B.values
+            R = R.loc[instruments].values
+
+        if not isinstance(B, ndarray):
             raise TypeError(
                 "Only pandas DataFrame / numpy ndarray is supported, but not "
-                f"{cov.__class__.__name__}"
+                f"{B.__class__.__name__}"
             )
 
+        cov = B.T @ factor_covariances @ B
+
         # Add the specific variances into the covariance matrix
-        cov_values[diag_indices_from(cov_values)] += specific_variances
+        cov[diag_indices_from(cov)] += R
 
-        valid_instruments = any(cov_values != 0.0, axis=0)
+        # Set zero covariance instruments to nan
+        valid_instruments = any(cov != 0.0, axis=0)
+        cov[~valid_instruments, :] = nan
+        cov[:, ~valid_instruments] = nan
 
-        if self._show_all_instruments:
-            # Set zero covariance instruments to nan
-            cov_values[~valid_instruments, :] = nan
-            cov_values[:, ~valid_instruments] = nan
-        elif isinstance(cov, DataFrame):
-            cov = cov.loc[valid_instruments, valid_instruments]
-        elif isinstance(cov, ndarray):
+        if not self._show_all_instruments:
             cov = cov[valid_instruments, :][:, valid_instruments]
-        else:
-            raise TypeError(
-                "Only pandas DataFrame / numpy ndarray is supported, but not "
-                f"{cov.__class__.__name__}"
-            )
+            if isinstance(self._factor_exposures, DataFrame):
+                instruments = instruments[valid_instruments]
+
+        if isinstance(self._factor_exposures, DataFrame):
+            cov = DataFrame(cov, index=instruments, columns=instruments)
 
         return cov

src/fpm_risk_model/risk_model.py

@@ -32,7 +32,7 @@ def __init__(self, engine: Any = None, show_all_instruments: bool = False):
         self._show_all_instruments = show_all_instruments
 
     @abstractmethod
-    def cov(self) -> ndarray:
+    def cov(self, **kwargs) -> ndarray:
         """
         Get the covariance matrix.
 
@@ -43,7 +43,7 @@ def cov(self) -> ndarray:
             diagonal entries are the variances.
         """
 
-    def corr(self):
+    def corr(self, **kwargs):
         """
         Get the correlation matrix.
 
@@ -53,6 +53,6 @@ def corr(self):
             A square pairwise correlation matrix which its
             diagonal entries are all ones.
         """
-        cov = self.cov()
+        cov = self.cov(**kwargs)
         vol = self._engine.sqrt(self._engine.diagonal(cov))
         return ((cov / vol).T / vol).T

tests/test_factor_risk_model.py

@@ -150,6 +150,17 @@ def expected_covariances():
     )
 
 
+@pytest.fixture(scope="module")
+def expected_covariances_halflife():
+    return array(
+        [
+            [0.00029782, 0.00027028, 0.00020145],
+            [0.00027028, 0.00045064, 0.00025519],
+            [0.00020145, 0.00025519, 0.00030442],
+        ]
+    )
+
+
 @pytest.fixture(scope="module")
 def expected_correlations():
     return array(
@@ -161,6 +172,17 @@ def expected_correlations():
     )
 
 
+@pytest.fixture(scope="module")
+def expected_correlations_halflife():
+    return array(
+        [
+            [1.0, 0.48164216, 0.46908114],
+            [0.48164216, 1.0, 0.75251128],
+            [0.46908114, 0.75251128, 1.0],
+        ]
+    )
+
+
 def test_factor_risk_model_np_covariances(factor_risk_model_np, expected_covariances):
     cov = factor_risk_model_np.cov()
     np.testing.assert_allclose(cov, expected_covariances, atol=1e-7)
@@ -192,3 +214,30 @@ def test_factor_risk_model_pd_correlations(
         expected_correlations, index=valid_instruments, columns=valid_instruments
     )
     pd.testing.assert_frame_equal(corr, expected_correlations)
+
+
+def test_factor_risk_model_pd_covariances_halflife(
+    factor_risk_model_pd, expected_covariances_halflife, valid_instruments
+):
+    cov = factor_risk_model_pd.cov(halflife=5)
+    expected_cov = pd.DataFrame(
+        expected_covariances_halflife,
+        index=valid_instruments,
+        columns=valid_instruments,
+    )
+    pd.testing.assert_frame_equal(
+        cov,
+        expected_cov,
+    )
+
+
+def test_factor_risk_model_pd_correlations_halflife(
+    factor_risk_model_pd, expected_correlations_halflife, valid_instruments
+):
+    corr = factor_risk_model_pd.corr(halflife=0.5)
+    expected_corr = pd.DataFrame(
+        expected_correlations_halflife,
+        index=valid_instruments,
+        columns=valid_instruments,
+    )
+    pd.testing.assert_frame_equal(corr, expected_corr)