update quantile analysis

alphamind/analysis/quantileanalysis.py

@@ -12,7 +12,7 @@ from alphamind.utilities import agg_mean
 from alphamind.data.quantile import quantile
 from alphamind.data.standardize import standardize
 from alphamind.data.winsorize import winsorize_normal
-from alphamind.analysis.utilities import FDataPack
+from alphamind.data.processing import factor_processing
 
 
 def quantile_analysis(factors: pd.DataFrame,
@@ -21,7 +21,6 @@ def quantile_analysis(factors: pd.DataFrame,
                       n_bins: int=5,
                       benchmark: Optional[np.ndarray]=None,
                       risk_exp: Optional[np.ndarray]=None,
-                      do_neutralize=True,
                       **kwargs):
 
     if 'pre_process' in kwargs:
@@ -36,10 +35,7 @@ def quantile_analysis(factors: pd.DataFrame,
     else:
         post_process = [standardize]
 
-    data_pack = FDataPack(raw_factors=factors.values,
-                          risk_exp=risk_exp)
-
-    er = data_pack.factor_processing(pre_process, post_process, do_neutralize) @ factor_weights
+    er = factor_processing(factors.values, pre_process, risk_exp, post_process) @ factor_weights
     raw_return = q_anl_impl(er, n_bins, dx_return)
 
     if benchmark is not None:

alphamind/analysis/utilities.py

-# -*- coding: utf-8 -*-
-"""
-Created on 2017-8-16
-
-@author: cheng.li
-"""
-
-from typing import List
-from typing import Optional
-import numpy as np
-import pandas as pd
-from alphamind.data.neutralize import neutralize
-
-
-def factor_processing(raw_factors: np.ndarray,
-                      pre_process: Optional[List]=None,
-                      risk_factors: Optional[np.ndarray]=None,
-                      post_process: Optional[List]=None,
-                      do_neutralize: Optional[bool]=True) -> np.ndarray:
-
-    new_factors = raw_factors
-
-    if pre_process:
-        for p in pre_process:
-            new_factors = p(new_factors)
-
-    if risk_factors is not None and do_neutralize:
-        new_factors = neutralize(risk_factors, new_factors)
-
-    if post_process:
-        for p in pre_process:
-            new_factors = p(new_factors)
-
-    return new_factors
-
-
-class FDataPack(object):
-
-    def __init__(self,
-                 raw_factors: np.ndarray,
-                 factor_names: List[str]=None,
-                 codes: List=None,
-                 groups: Optional[np.ndarray]=None,
-                 benchmark: Optional[np.ndarray]=None,
-                 constraints: Optional[np.ndarray]=None,
-                 risk_exp: Optional[np.ndarray]=None,
-                 risk_names: List[str]=None):
-
-        self.raw_factors = raw_factors
-
-        if factor_names:
-            self.factor_names = factor_names
-        else:
-            self.factor_names = ['factor' + str(i) for i in range(raw_factors.shape[1])]
-        self.codes = codes
-
-        if groups is not None:
-            self.groups = groups.flatten()
-        else:
-            self.groups = None
-
-        if benchmark is not None:
-            self.benchmark = benchmark.flatten()
-        else:
-            self.benchmark = None
-
-        if risk_exp is not None:
-            risk_exp = risk_exp[:, risk_exp.sum(axis=0) != 0]
-
-        self.risk_exp = risk_exp
-        self.constraints = constraints
-        self.risk_names = risk_names
-
-    def benchmark_constraints(self) -> np.ndarray:
-        return self.benchmark @ self.constraints
-
-    def settle(self, weights: np.ndarray, dx_return: np.ndarray) -> pd.DataFrame:
-
-        weights = weights.flatten()
-        dx_return = dx_return.flatten()
-
-        if self.benchmark is not None:
-            net_pos = weights - self.benchmark
-        else:
-            net_pos = weights
-
-        ret_arr = net_pos * dx_return
-
-        if self.groups is not None:
-            ret_agg = pd.Series(ret_arr).groupby(self.groups).sum()
-            ret_agg.loc['total'] = ret_agg.sum()
-        else:
-            ret_agg = pd.Series(ret_arr.sum(), index=['total'])
-
-        ret_agg.index.name = 'industry'
-        ret_agg.name = 'er'
-
-        pos_table = pd.DataFrame(net_pos, columns=['weight'])
-        pos_table['ret'] = dx_return
-
-        if self.groups is not None:
-            ic_table = pos_table.groupby(self.groups).corr()['ret'].loc[(slice(None), 'weight')]
-            ic_table.loc['total'] = pos_table.corr().iloc[0, 1]
-        else:
-            ic_table = pd.Series(pos_table.corr().iloc[0, 1], index=['total'])
-
-        return pd.DataFrame({'er': ret_agg.values,
-                             'ic': ic_table.values},
-                            index=ret_agg.index)
-
-    def factor_processing(self, pre_process, pos_process, do_neutralize) -> np.ndarray:
-
-        if self.risk_exp is None:
-            return factor_processing(self.raw_factors,
-                                     pre_process,
-                                     pos_process,
-                                     do_neutralize=do_neutralize)
-        else:
-            return factor_processing(self.raw_factors,
-                                     pre_process,
-                                     self.risk_exp,
-                                     pos_process,
-                                     do_neutralize)
\ No newline at end of file

alphamind/data/processing.py

@@ -13,7 +13,7 @@ from alphamind.data.neutralize import neutralize
 
 def factor_processing(raw_factors: np.ndarray,
                       pre_process: Optional[List]=None,
-                      risk_exp: Optional[np.ndarray]=None,
+                      risk_factors: Optional[np.ndarray]=None,
                       post_process: Optional[List]=None) -> np.ndarray:
 
     new_factors = raw_factors
@@ -22,9 +22,9 @@ def factor_processing(raw_factors: np.ndarray,
         for p in pre_process:
             new_factors = p(new_factors)
 
-    if risk_exp is not None:
-        risk_exp = risk_exp[:, risk_exp.sum(axis=0) != 0]
-        new_factors = neutralize(risk_exp, new_factors)
+    if risk_factors is not None:
+        risk_factors = risk_factors[:, risk_factors.sum(axis=0) != 0]
+        new_factors = neutralize(risk_factors, new_factors)
 
     if post_process:
         for p in post_process:

alphamind/tests/analysis/test_factoranalysis.py

@@ -12,7 +12,7 @@ from alphamind.data.winsorize import winsorize_normal
 from alphamind.data.standardize import standardize
 from alphamind.data.neutralize import neutralize
 from alphamind.portfolio.constraints import Constraints
-from alphamind.analysis.utilities import factor_processing
+from alphamind.data.processing import factor_processing
 from alphamind.analysis.factoranalysis import factor_analysis
 
 

alphamind/tests/analysis/test_quantilieanalysis.py

@@ -10,7 +10,7 @@ import numpy as np
 import pandas as pd
 from alphamind.analysis.quantileanalysis import q_anl_impl
 from alphamind.analysis.quantileanalysis import quantile_analysis
-from alphamind.analysis.utilities import factor_processing
+from alphamind.data.processing import factor_processing
 from alphamind.data.standardize import standardize
 from alphamind.data.winsorize import winsorize_normal
 from alphamind.data.quantile import quantile
@@ -69,8 +69,7 @@ class TestQuantileAnalysis(unittest.TestCase):
         er = self.x_w @ factor_processing(self.x,
                                           [winsorize_normal, standardize],
                                           self.risk_exp,
-                                          [standardize],
-                                          True).T
+                                          [standardize]).T
         expected = q_anl_impl(er, self.n_bins, self.r)
         np.testing.assert_array_almost_equal(calculated, expected)
 
@@ -89,8 +88,7 @@ class TestQuantileAnalysis(unittest.TestCase):
         er = self.x_w @ factor_processing(self.x,
                                           [winsorize_normal, standardize],
                                           self.risk_exp,
-                                          [standardize],
-                                          True).T
+                                          [standardize]).T
         raw_er = q_anl_impl(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)