feat: add standardized returns in bias check

src/fpm_risk_model/accuracy/bias.py

@@ -0,0 +1,41 @@
+from numpy import ndarray, sqrt, sum
+from pandas import Series
+
+from ..rolling_factor_risk_model import RollingFactorRiskModel
+
+
+def compute_standardized_returns(
+    X: ndarray,
+    weights: ndarray,
+    rolling_risk_model: RollingFactorRiskModel,
+) -> Series:
+    """
+    Compute the standardized returns given the rolling risk model.
+
+    Standardized return is defined as
+
+      .. math::
+        b_t = \frac{r_t}{\\sigma_t}
+
+    Parameters
+    ----------
+    X: ndarray
+        The instrument forecast returns.
+    weights: ndarray
+        Weights of the instruments.
+    rolling_risk_model: RollingFactorRiskModel
+        The rolling risk model.
+
+    Returns
+    -------
+    Series
+        A timeseries of standardized returns.
+    """
+    b_t = Series()
+    for index, risk_model in rolling_risk_model.items():
+        index_weights = weights.loc[index, :]
+        returns = sum(X.loc[index, :] * index_weights)
+        vol = sqrt((risk_model.cov().fillna(0.0) @ index_weights) @ index_weights)
+        b_t[index] = returns / vol
+
+    return b_t

src/fpm_risk_model/rolling_factor_risk_model.py

@@ -1,6 +1,7 @@
-from typing import Iterable, Optional
+from typing import Iterable, Optional, Tuple
 
 import pandas as pd
+from pandas import Timestamp
 
 from .factor_risk_model import FactorRiskModel
 
@@ -40,7 +41,7 @@ def values(self) -> Iterable[object]:
         """
         return self._values.values()
 
-    def items(self) -> Iterable[object]:
+    def items(self) -> Iterable[Tuple[Timestamp, FactorRiskModel]]:
         """
         Return a list of tuples with keys and values.
         """

tests/accuracy/conftest.py

@@ -0,0 +1,78 @@
+import pytest
+from pandas import DataFrame, bdate_range
+
+from fpm_risk_model.rolling_factor_risk_model import RollingFactorRiskModel
+from fpm_risk_model.statistical import PCA
+
+
+@pytest.fixture(scope="module")
+def instruments():
+    return ["A", "AAL", "AAP", "AAPL"]
+
+
+@pytest.fixture(scope="module")
+def dates():
+    return bdate_range("2016-01-04", "2016-01-15")
+
+
+@pytest.fixture(scope="module")
+def factors():
+    return [
+        "factor_1",
+        "factor_2",
+    ]
+
+
+@pytest.fixture(scope="module")
+def daily_returns(instruments, dates):
+    return DataFrame(
+        [
+            [-0.02678756, -0.03400254, 0.0, 0.000855],
+            [-0.00344077, -0.00953307, 0.0, -0.02505943],
+            [0.00443915, 0.01752232, 0.0, -0.01956966],
+            [-0.04247514, -0.01891826, 0.0, -0.04220453],
+            [-0.01051272, -0.00197782, 0.0, 0.00528776],
+            [-0.01684373, 0.01758743, 0.0, 0.01619198],
+            [0.00658919, 0.02239528, 0.0, 0.01451376],
+            [-0.03482585, -0.0452383, 0.0, -0.02571051],
+            [0.02034743, 0.01122229, 0.0, 0.02187115],
+            [-0.01329412, -0.04414332, 0.0, -0.02401548],
+        ],
+        columns=instruments,
+        index=dates,
+    )
+
+
+@pytest.fixture(scope="module")
+def weights(instruments, dates):
+    return DataFrame(
+        [
+            [0.0, 0.0, 0.0, 0.0],
+            [0.0, 0.0, 0.0, 0.0],
+            [0.0, 0.0, 0.0, 0.0],
+            [0.0, 0.0, 0.0, 0.0],
+            [0.0, 0.0, 0.0, 0.0],
+            [0.0, 0.0, 0.0, 0.0],
+            [0.5, 0.5, 0.0, 0.0],
+            [0.6, 0.2, 0.0, 0.2],
+            [0.6, 0.2, 0.0, 0.2],
+            [0.6, 0.2, 0.0, 0.2],
+        ],
+        columns=instruments,
+        index=dates,
+    )
+
+
+@pytest.fixture(scope="module")
+def rolling_factor_risk_model(daily_returns):
+    model = PCA(
+        n_components=2,
+        demean=True,
+        speedup=True,
+    )
+    rolling_model = RollingFactorRiskModel(
+        model=model,
+        rolling_timeframe=5,
+        show_progress=False,
+    )
+    return rolling_model.fit(X=daily_returns)

tests/accuracy/test_bias.py

@@ -0,0 +1,20 @@
+from numpy import array, nan
+from pandas import Series
+from pandas.testing import assert_series_equal
+
+from fpm_risk_model.accuracy.bias import compute_standardized_returns
+
+
+def test_compute_standardized_returns(
+    daily_returns, weights, rolling_factor_risk_model
+):
+    standardized_returns = compute_standardized_returns(
+        X=daily_returns,
+        weights=weights,
+        rolling_risk_model=rolling_factor_risk_model,
+    )
+    expected_standardized_returns = Series(
+        array([nan, 0.87928191, -1.7095046, 0.80100368, -1.0760382]),
+        index=list(rolling_factor_risk_model.keys()),
+    )
+    assert_series_equal(standardized_returns, expected_standardized_returns)