added constraints class

alphamind/analysis/factoranalysis.py

@@ -13,6 +13,7 @@ import pandas as pd
 from alphamind.data.standardize import standardize
 from alphamind.data.winsorize import winsorize_normal
 from alphamind.data.neutralize import neutralize
+from alphamind.portfolio.constraints import Constraints
 from alphamind.portfolio.longshortbulder import long_short_build
 from alphamind.portfolio.rankbuilder import rank_build
 from alphamind.portfolio.percentbuilder import percent_build
@@ -153,7 +154,7 @@ def factor_analysis(factors: pd.DataFrame,
                     benchmark: Optional[np.ndarray]=None,
                     risk_exp: Optional[np.ndarray]=None,
                     is_tradable: Optional[np.ndarray]=None,
-                    constraints: Optional[np.ndarray]=None,
+                    constraints: Optional[Constraints]=None,
                     method='risk_neutral',
                     **kwargs) -> Tuple[pd.DataFrame, Optional[pd.DataFrame]]:
 
@@ -177,16 +178,17 @@ def factor_analysis(factors: pd.DataFrame,
         if is_tradable is not None:
             ubound[~is_tradable] = 0.
 
-        if 'risk_bound' in kwargs:
-            risk_lbound = kwargs['risk_bound'][0]
-            risk_ubound = kwargs['risk_bound'][1]
+        if constraints:
+            risk_lbound, risk_ubound = constraints.risk_targets()
+            cons_exp = constraints.risk_exp
         else:
+            cons_exp = risk_exp
             risk_lbound = data_pack.benchmark_constraints()
             risk_ubound = data_pack.benchmark_constraints()
 
         weights = build_portfolio(er,
                                   builder='linear',
-                                  risk_constraints=constraints,
+                                  risk_constraints=cons_exp,
                                   lbound=lbound,
                                   ubound=ubound,
                                   risk_target=(risk_lbound, risk_ubound),

alphamind/portfolio/constraints.py

@@ -8,24 +8,54 @@ Created on 2017-7-21
 from math import inf
 import numpy as np
 from typing import Tuple
+from typing import Optional
 
 
 class Constraints(object):
 
     def __init__(self,
-                 risk_exp: np.ndarray,
-                 risk_names: np.ndarray):
+                 risk_exp: Optional[np.ndarray]=None,
+                 risk_names: Optional[np.ndarray]=None):
         self.risk_exp = risk_exp
-        self.risk_names = risk_names
-        self.risk_maps = dict(zip(risk_names, range(len(risk_names))))
-        self.lower_bounds = -inf * np.ones(len(risk_names))
-        self.upper_bounds = inf * np.ones(len(risk_names))
+
+        if risk_names is not None:
+            self.risk_names = np.array(risk_names)
+        else:
+            self.risk_names = np.array([])
+
+        n = len(self.risk_names)
+
+        self.risk_maps = dict(zip(self.risk_names, range(n)))
+        self.lower_bounds = -inf * np.ones(n)
+        self.upper_bounds = inf * np.ones(n)
 
     def set_constraints(self, tag: str, lower_bound: float, upper_bound: float):
         index = self.risk_maps[tag]
         self.lower_bounds[index] = lower_bound
         self.upper_bounds[index] = upper_bound
 
+    def add_exposure(self, tags: np.ndarray, new_exp: np.ndarray):
+        if len(tags) != new_exp.shape[1]:
+            raise ValueError('new dags length is not compatible with exposure shape {1}'.format(len(tags),
+                                                                                                new_exp.shape))
+
+        for tag in tags:
+            if tag in self.risk_maps:
+                raise ValueError('tag {0} is already in risk table'.format(tag))
+
+        self.risk_names = np.concatenate((self.risk_names, tags))
+
+        if self.risk_exp is not None:
+            self.risk_exp = np.concatenate((self.risk_exp, new_exp), axis=1)
+        else:
+            self.risk_exp = new_exp
+
+        n = len(self.risk_names)
+        self.risk_maps = dict(zip(self.risk_names, range(n)))
+
+        self.lower_bounds = np.concatenate((self.lower_bounds, -inf * np.ones(len(tags))))
+        self.upper_bounds = np.concatenate((self.upper_bounds, inf * np.ones(len(tags))))
+
     def risk_targets(self) -> Tuple[np.ndarray, np.ndarray]:
         return self.lower_bounds, self.upper_bounds
 

alphamind/tests/analysis/test_factoranalysis.py

@@ -11,6 +11,7 @@ import pandas as pd
 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.factoranalysis import factor_processing
 from alphamind.analysis.factoranalysis import factor_analysis
 
@@ -45,13 +46,20 @@ class TestFactorAnalysis(unittest.TestCase):
         factor_df = pd.DataFrame(self.raw_factor.flatten(), index=range(len(self.raw_factor)))
         factor_weights = np.array([1.])
 
+        constraints = Constraints()
+        names = np.array(['a', 'b', 'c'])
+        constraints.add_exposure(names, self.risk_factor)
+        targets = self.risk_factor.T @ benchmark
+        for i, name in enumerate(names):
+            constraints.set_constraints(name, targets[i], targets[i])
+
         weight_table, analysis_table = factor_analysis(factor_df,
                                                        factor_weights,
                                                        d1returns=self.d1returns,
                                                        industry=industry,
                                                        benchmark=benchmark,
                                                        risk_exp=self.risk_factor,
-                                                       constraints=self.risk_factor)
+                                                       constraints=constraints)
 
         weight = weight_table.weight
 
@@ -66,13 +74,21 @@ class TestFactorAnalysis(unittest.TestCase):
 
         factor_df = pd.DataFrame(np.random.randn(1000, 2), index=range(len(self.raw_factor)))
         factor_weights = np.array([0.2, 0.8])
+
+        constraints = Constraints()
+        names = np.array(['a', 'b', 'c'])
+        constraints.add_exposure(names, self.risk_factor)
+        targets = self.risk_factor.T @ benchmark
+        for i, name in enumerate(names):
+            constraints.set_constraints(name, targets[i], targets[i])
+
         weight_table, analysis_table = factor_analysis(factor_df,
                                                        factor_weights,
                                                        d1returns=self.d1returns,
                                                        industry=industry,
                                                        benchmark=benchmark,
                                                        risk_exp=self.risk_factor,
-                                                       constraints=self.risk_factor)
+                                                       constraints=constraints)
 
         weight = weight_table.weight
         self.assertEqual(analysis_table['er'].sum() / analysis_table['er'][-1], 2.0)

alphamind/tests/portfolio/test_constraints.py

+# -*- coding: utf-8 -*-
+"""
+Created on 2017-7-20
+
+@author: cheng.li
+"""
+
+import unittest
+import numpy as np
+from alphamind.portfolio.constraints import Constraints
+
+
+class TestConstraints(unittest.TestCase):
+
+    def test_constraints(self):
+        cons = Constraints()
+
+        test_exp = np.array([[1., 2.],
+                             [3., 4.]])
+        test_names = np.array(['a', 'b'])
+        cons.add_exposure(test_names, test_exp)
+
+        test_exp = np.array([[5.],
+                             [6.]])
+        test_names = ['c']
+        cons.add_exposure(test_names, test_exp)
+
+        np.testing.assert_array_almost_equal(np.array([[1., 2., 5.], [3., 4., 6.]]),
+                                             cons.risk_exp)
+
+        risk_targets = cons.risk_targets()
+        np.testing.assert_array_almost_equal(risk_targets[0], np.array([-np.inf, -np.inf, -np.inf]))
+        np.testing.assert_array_almost_equal(risk_targets[1], np.array([np.inf, np.inf, np.inf]))
+
+        cons.set_constraints('a', -0.1, 0.1)
+        risk_targets = cons.risk_targets()
+        np.testing.assert_array_almost_equal(risk_targets[0], np.array([-0.1, -np.inf, -np.inf]))
+        np.testing.assert_array_almost_equal(risk_targets[1], np.array([0.1, np.inf, np.inf]))
+
+        cons.set_constraints('c', -0.1, 0.1)
+        risk_targets = cons.risk_targets()
+        np.testing.assert_array_almost_equal(risk_targets[0], np.array([-0.1, -np.inf, -0.1]))
+        np.testing.assert_array_almost_equal(risk_targets[1], np.array([0.1, np.inf, 0.1]))
+
+
+if __name__ == '__main__':
+    unittest.main()

alphamind/tests/test_suite.py

@@ -14,6 +14,7 @@ from alphamind.utilities import alpha_logger
 from alphamind.tests.data.test_neutralize import TestNeutralize
 from alphamind.tests.data.test_standardize import TestStandardize
 from alphamind.tests.data.test_winsorize import TestWinsorize
+from alphamind.tests.portfolio.test_constraints import TestConstraints
 from alphamind.tests.portfolio.test_longshortbuild import TestLongShortBuild
 from alphamind.tests.portfolio.test_rankbuild import TestRankBuild
 from alphamind.tests.portfolio.test_percentbuild import TestPercentBuild
@@ -29,6 +30,7 @@ if __name__ == '__main__':
     runner = TestRunner([TestNeutralize,
                          TestStandardize,
                          TestWinsorize,
+                         TestConstraints,
                          TestLongShortBuild,
                          TestRankBuild,
                          TestPercentBuild,