forked from alpha-miner/alpha-mind
-
Notifications
You must be signed in to change notification settings - Fork 0
/
meanvariancebuilder.py
124 lines (103 loc) · 4.5 KB
/
meanvariancebuilder.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
# -*- coding: utf-8 -*-
"""
Created on 2017-6-27
@author: cheng.li
"""
import numpy as np
from typing import Union
from typing import Tuple
from typing import Optional
from typing import Dict
import cvxpy
from alphamind.cython.optimizers import QPOptimizer
from alphamind.cython.optimizers import CVOptimizer
from alphamind.exceptions.exceptions import PortfolioBuilderException
def _create_bounds(lbound,
ubound,
bm,
risk_exposure,
risk_target):
lbound = lbound - bm
ubound = ubound - bm
if risk_exposure is not None:
cons_mat = risk_exposure.T
bm_risk = cons_mat @ bm
clbound = risk_target[0] - bm_risk
cubound = risk_target[1] - bm_risk
else:
cons_mat = None
clbound = None
cubound = None
return lbound, ubound, cons_mat, clbound, cubound
def _create_result(optimizer, bm):
if optimizer.status() == 0 or optimizer.status() == 1:
return 'optimal', optimizer.feval(), optimizer.x_value() + bm
else:
raise PortfolioBuilderException(optimizer.status())
def mean_variance_builder(er: np.ndarray,
risk_model: Dict[str, Union[None, np.ndarray]],
bm: np.ndarray,
lbound: Union[np.ndarray, float],
ubound: Union[np.ndarray, float],
risk_exposure: Optional[np.ndarray],
risk_target: Optional[Tuple[np.ndarray, np.ndarray]],
lam: float=1.,
linear_solver: str='ma27') -> Tuple[str, float, np.ndarray]:
lbound, ubound, cons_mat, clbound, cubound = _create_bounds(lbound, ubound, bm, risk_exposure, risk_target)
if np.all(lbound == -np.inf) and np.all(ubound == np.inf) and cons_mat is None:
# using fast path cvxpy
n = len(er)
w = cvxpy.Variable(n)
cov = risk_model['cov']
special_risk = risk_model['idsync']
risk_cov = risk_model['factor_cov']
risk_exposure = risk_model['factor_loading']
if cov is None:
risk = cvxpy.sum_squares(cvxpy.multiply(cvxpy.sqrt(special_risk), w)) \
+ cvxpy.quad_form((w.T * risk_exposure).T, risk_cov)
else:
risk = cvxpy.quad_form(w, cov)
objective = cvxpy.Minimize(-w.T * er + 0.5 * lam * risk)
prob = cvxpy.Problem(objective)
prob.solve(solver='ECOS', feastol=1e-9, abstol=1e-9, reltol=1e-9)
if prob.status == 'optimal' or prob.status == 'optimal_inaccurate':
return 'optimal', prob.value, np.array(w.value).flatten() + bm
else:
raise PortfolioBuilderException(prob.status)
else:
optimizer = QPOptimizer(er,
risk_model['cov'],
lbound,
ubound,
cons_mat,
clbound,
cubound,
lam,
risk_model['factor_cov'],
risk_model['factor_loading'],
risk_model['idsync'],
linear_solver=linear_solver)
return _create_result(optimizer, bm)
def target_vol_builder(er: np.ndarray,
risk_model: Dict[str, Union[None, np.ndarray]],
bm: np.ndarray,
lbound: Union[np.ndarray, float],
ubound: Union[np.ndarray, float],
risk_exposure: Optional[np.ndarray],
risk_target: Optional[Tuple[np.ndarray, np.ndarray]],
vol_target: float = 1.,
linear_solver: str = 'ma27')-> Tuple[str, float, np.ndarray]:
lbound, ubound, cons_mat, clbound, cubound = _create_bounds(lbound, ubound, bm, risk_exposure, risk_target)
optimizer = CVOptimizer(er,
risk_model['cov'],
lbound,
ubound,
cons_mat,
clbound,
cubound,
vol_target,
risk_model['factor_cov'],
risk_model['factor_loading'],
risk_model['idsync'],
linear_solver=linear_solver)
return _create_result(optimizer, bm)