re-implement quantile analysis

alphamind/analysis/quantileanalysis.py

@@ -19,7 +19,6 @@ def quantile_analysis(factors: pd.DataFrame,
                       factor_weights: np.ndarray,
                       dx_return: np.ndarray,
                       n_bins: int=5,
-                      benchmark: Optional[np.ndarray]=None,
                       risk_exp: Optional[np.ndarray]=None,
                       **kwargs):
 
@@ -36,13 +35,12 @@ def quantile_analysis(factors: pd.DataFrame,
         post_process = [standardize]
 
     er = factor_processing(factors.values, pre_process, risk_exp, post_process) @ factor_weights
-    return er_quantile_analysis(er, n_bins, dx_return, benchmark)
+    return er_quantile_analysis(er, n_bins, dx_return)
 
 
 def er_quantile_analysis(er: np.ndarray,
                          n_bins: int,
-                         dx_return: np.ndarray,
-                         benchmark: Optional[np.ndarray]=None,) -> np.ndarray:
+                         dx_return: np.ndarray) -> np.ndarray:
 
     er = er.flatten()
     q_groups = quantile(er, n_bins)
@@ -51,12 +49,15 @@ def er_quantile_analysis(er: np.ndarray,
         dx_return.shape = -1, 1
 
     group_return = agg_mean(q_groups, dx_return).flatten()
-    if benchmark is not None:
-        b_ret = np.dot(benchmark, dx_return)
-        b_total = benchmark.sum()
-        return group_return * b_total - b_ret
-    else:
-        return group_return
+    total_return = group_return.sum()
+    ret = group_return.copy()
+
+    resid = n_bins - 1
+    res_weight = 1. / resid
+    for i, value in enumerate(ret):
+        ret[i] = (1. + res_weight) * value - res_weight * total_return
+
+    return ret
 
 
 if __name__ == '__main__':

alphamind/tests/analysis/test_quantilieanalysis.py

@@ -36,10 +36,18 @@ class TestQuantileAnalysis(unittest.TestCase):
         q_groups = quantile(x, n_bins)
 
         s = pd.Series(self.r, index=q_groups)
-        expected_res = s.groupby(level=0).mean()
+        grouped_return = s.groupby(level=0).mean().values.flatten()
+
+        expected_res = grouped_return.copy()
+        res = n_bins - 1
+        res_weight = 1. / res
+
+        for i, value in enumerate(expected_res):
+            expected_res[i] = (1. + res_weight) * value - res_weight * grouped_return.sum()
+
         calculated_res = er_quantile_analysis(x, n_bins, self.r)
 
-        np.testing.assert_array_almost_equal(expected_res.values, calculated_res)
+        np.testing.assert_array_almost_equal(expected_res, calculated_res)
 
     def test_quantile_analysis_simple(self):
         f_df = pd.DataFrame(self.x)
@@ -73,26 +81,6 @@ class TestQuantileAnalysis(unittest.TestCase):
         expected = er_quantile_analysis(er, self.n_bins, self.r)
         np.testing.assert_array_almost_equal(calculated, expected)
 
-    def test_quantile_analysis_with_benchmark(self):
-        f_df = pd.DataFrame(self.x)
-        calculated = quantile_analysis(f_df,
-                                       self.x_w,
-                                       self.r,
-                                       n_bins=self.n_bins,
-                                       do_neutralize=True,
-                                       benchmark=self.b_w,
-                                       risk_exp=self.risk_exp,
-                                       pre_process=[winsorize_normal, standardize],
-                                       post_process=[standardize])
-
-        er = self.x_w @ factor_processing(self.x,
-                                          [winsorize_normal, standardize],
-                                          self.risk_exp,
-                                          [standardize]).T
-        raw_er = er_quantile_analysis(er, self.n_bins, self.r)
-        expected = raw_er * self.b_w.sum() - np.dot(self.b_w, self.r)
-        np.testing.assert_array_almost_equal(calculated, expected)
-
 
 if __name__ == '__main__':
     unittest.main()