reimplementation of risk model and mean variance builder

alphamind/portfolio/meanvariancebuilder.py

@@ -9,11 +9,9 @@ import numpy as np
 from typing import Union
 from typing import Tuple
 from typing import Optional
+from typing import Dict
 from alphamind.cython.optimizers import QPOptimizer
 from alphamind.cython.optimizers import CVOptimizer
-from alphamind.portfolio.riskmodel import RiskModel
-from alphamind.portfolio.riskmodel import FullRiskModel
-from alphamind.portfolio.riskmodel import FactorRiskModel
 
 
 def _create_bounds(lbound,
@@ -48,7 +46,7 @@ def _create_result(optimizer, bm):
 
 
 def mean_variance_builder(er: np.ndarray,
-                          risk_model: RiskModel,
+                          risk_model: Dict[str, Union[None, np.ndarray]],
                           bm: np.ndarray,
                           lbound: Union[np.ndarray, float],
                           ubound: Union[np.ndarray, float],
@@ -57,40 +55,23 @@ def mean_variance_builder(er: np.ndarray,
                           lam: float=1.) -> Tuple[str, float, np.ndarray]:
     lbound, ubound, cons_mat, clbound, cubound = _create_bounds(lbound, ubound, bm, risk_exposure, risk_target)
 
-    if isinstance(risk_model, FullRiskModel):
-        cov = risk_model.get_cov()
-        optimizer = QPOptimizer(er,
-                                cov,
-                                lbound,
-                                ubound,
-                                cons_mat,
-                                clbound,
-                                cubound,
-                                lam)
-    elif isinstance(risk_model, FactorRiskModel):
-        cov = None
-        factor_cov = risk_model.get_factor_cov()
-        factor_loading = risk_model.get_risk_exp()
-        idsync = risk_model.get_idsync()
-        optimizer = QPOptimizer(er,
-                                cov,
-                                lbound,
-                                ubound,
-                                cons_mat,
-                                clbound,
-                                cubound,
-                                lam,
-                                factor_cov,
-                                factor_loading,
-                                idsync)
-    else:
-        raise ValueError("{0} is not recognized as valid risk model".format(risk_model))
+    optimizer = QPOptimizer(er,
+                            risk_model['cov'],
+                            lbound,
+                            ubound,
+                            cons_mat,
+                            clbound,
+                            cubound,
+                            lam,
+                            risk_model['factor_cov'],
+                            risk_model['factor_loading'],
+                            risk_model['idsync'])
 
     return _create_result(optimizer, bm)
 
 
 def target_vol_builder(er: np.ndarray,
-                       risk_model: RiskModel,
+                       risk_model: Dict[str, Union[None, np.ndarray]],
                        bm: np.ndarray,
                        lbound: Union[np.ndarray, float],
                        ubound: Union[np.ndarray, float],
@@ -99,36 +80,18 @@ def target_vol_builder(er: np.ndarray,
                        vol_target: float = 1.)-> Tuple[str, float, np.ndarray]:
     lbound, ubound, cons_mat, clbound, cubound = _create_bounds(lbound, ubound, bm, risk_exposure, risk_target)
 
-    if isinstance(risk_model, FullRiskModel):
-        cov = risk_model.get_cov()
-        optimizer = CVOptimizer(er,
-                                cov,
-                                lbound,
-                                ubound,
-                                cons_mat,
-                                clbound,
-                                cubound,
-                                0.,
-                                vol_target)
-    elif isinstance(risk_model, FactorRiskModel):
-        cov = None
-        factor_cov = risk_model.get_factor_cov()
-        factor_loading = risk_model.get_risk_exp()
-        idsync = risk_model.get_idsync()
-        optimizer = CVOptimizer(er,
-                                cov,
-                                lbound,
-                                ubound,
-                                cons_mat,
-                                clbound,
-                                cubound,
-                                0.,
-                                vol_target,
-                                factor_cov,
-                                factor_loading,
-                                idsync)
-    else:
-        raise ValueError("{0} is not recognized as valid risk model".format(risk_model))
+    optimizer = CVOptimizer(er,
+                            risk_model['cov'],
+                            lbound,
+                            ubound,
+                            cons_mat,
+                            clbound,
+                            cubound,
+                            0.,
+                            vol_target,
+                            risk_model['factor_cov'],
+                            risk_model['factor_loading'],
+                            risk_model['idsync'])
 
     return _create_result(optimizer, bm)
 

alphamind/portfolio/riskmodel.py

@@ -11,7 +11,9 @@ import pandas as pd
 
 
 class RiskModel(metaclass=abc.ABCMeta):
-    pass
+
+    def get_risk_profile(self):
+        pass
 
 
 class FullRiskModel(RiskModel):
@@ -26,6 +28,14 @@ class FullRiskModel(RiskModel):
         else:
             return self.sec_cov.values
 
+    def get_risk_profile(self, codes: List[int]=None):
+        return dict(
+            cov=self.get_cov(codes),
+            factor_cov=None,
+            factor_loading=None,
+            idsync=None
+        )
+
 
 class FactorRiskModel(RiskModel):
 
@@ -54,4 +64,12 @@ class FactorRiskModel(RiskModel):
         if codes:
             return self.idsync[codes].values
         else:
-            return self.idsync.values
+            return self.idsync.values
\ No newline at end of file
+
+    def get_risk_profile(self, codes: List[int]=None):
+        return dict(
+            cov=None,
+            factor_cov=self.get_factor_cov(),
+            factor_loading=self.get_risk_exp(codes),
+            idsync=self.get_idsync(codes)
+        )
\ No newline at end of file

alphamind/tests/portfolio/test_meanvariancebuild.py

@@ -8,7 +8,6 @@ Created on 2017-6-27
 import unittest
 import numpy as np
 import pandas as pd
-from alphamind.portfolio.riskmodel import FullRiskModel
 from alphamind.portfolio.meanvariancebuilder import mean_variance_builder
 from alphamind.portfolio.meanvariancebuilder import target_vol_builder
 
@@ -31,7 +30,7 @@ class TestMeanVarianceBuild(unittest.TestCase):
                                   [1., 0., 1.]]).T
         risk_target = (np.array([bm.sum(), 0.3]), np.array([bm.sum(), 0.7]))
 
-        model = FullRiskModel(pd.DataFrame(cov))
+        model = dict(cov=cov, factor_cov=None, factor_loading=None, idsync=None)
         status, _, x = mean_variance_builder(er, model, bm, lbound, ubound, risk_exposure, risk_target)
 
         self.assertTrue(status == 'optimal')
@@ -58,7 +57,7 @@ class TestMeanVarianceBuild(unittest.TestCase):
                                   [1., 0., 1.]]).T
         risk_target = (np.array([bm.sum(), 0.3]), np.array([bm.sum(), 0.7]))
 
-        model = FullRiskModel(pd.DataFrame(cov))
+        model = dict(cov=cov, factor_cov=None, factor_loading=None, idsync=None)
         status, _, x = mean_variance_builder(er, model, bm, lbound, ubound, risk_exposure, risk_target, lam=100)
 
         self.assertTrue(status == 'optimal')
@@ -82,7 +81,7 @@ class TestMeanVarianceBuild(unittest.TestCase):
 
         risk_exposure = np.array([[1., 1., 1.]]).T
         risk_target = (np.array([bm.sum()]), np.array([bm.sum()]))
-        model = FullRiskModel(pd.DataFrame(cov))
+        model = dict(cov=cov, factor_cov=None, factor_loading=None, idsync=None)
         status, _, x = target_vol_builder(er, model, bm, lbound, ubound, risk_exposure, risk_target, 0.1)
         self.assertTrue(status == 'optimal')
         self.assertTrue(np.all(x <= ubound + 1.e-6))