Coverage for src / cvx / markowitz / builder.py: 100%

1# Copyright 2023 Stanford University Convex Optimization Group 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14"""Core builder classes to assemble and solve Markowitz problems.""" 

15 

16from __future__ import annotations 

17 

18import pickle 

19from abc import abstractmethod 

20from collections.abc import Generator 

21from dataclasses import dataclass, field 

22from os import PathLike 

23from typing import Any 

24 

25import cvxpy as cp 

26import numpy as np 

27 

28from cvx.markowitz.cvxerror import CvxError 

29from cvx.markowitz.model import Model 

30from cvx.markowitz.models.bounds import Bounds 

31from cvx.markowitz.names import DataNames as D 

32from cvx.markowitz.names import ModelName as M 

33from cvx.markowitz.risk.factor.factor import FactorModel 

34from cvx.markowitz.risk.sample.sample import SampleCovariance 

35from cvx.markowitz.types import File, Matrix, Parameter, Variables 

36 

37 

38def deserialize( 

39 problem_file: str | bytes | PathLike[str] | PathLike[bytes] | int, 

40) -> Any: 

41 """Load a previously serialized Markowitz problem from disk. 

42 

43 Args: 

44 problem_file: Path to the pickle file created by `_Problem.serialize`. 

45 

46 Returns: 

47 The deserialized `_Problem` instance. 

48 """ 

49 with open(problem_file, "rb") as infile: 

50 return pickle.load(infile) 

51 

52 

53@dataclass(frozen=True) 

54class _Problem: 

55 problem: cp.Problem 

56 model: dict[str, Model] = field(default_factory=dict) 

57 

58 def update(self, **kwargs: Matrix) -> _Problem: 

59 """Update the problem.""" 

60 for name, model in self.model.items(): 

61 for key in model.data.keys(): 

62 if key not in kwargs: 

63 raise CvxError(f"Missing data for {key} in model {name}") 

64 

65 # It's tempting to operate without the models at this stage. 

66 # However, we would give up a lot of convenience. For example, 

67 # the models can be prepared to deal with data that has not 

68 # exactly the correct shape. 

69 model.update(**kwargs) 

70 

71 return self 

72 

73 def solve(self, solver: str = cp.CLARABEL, **kwargs: Any) -> float: 

74 """Solve the problem.""" 

75 value = self.problem.solve(solver=solver, **kwargs) 

76 

77 if self.problem.status is not cp.OPTIMAL: 

78 raise CvxError(f"Problem status is {self.problem.status}") 

79 

80 return float(value) 

81 

82 @property 

83 def value(self) -> float: 

84 return float(self.problem.value) 

85 

86 def is_dpp(self) -> bool: 

87 return bool(self.problem.is_dpp()) 

88 

89 @property 

90 def data(self) -> Generator[tuple[tuple[str, str], Matrix]]: 

91 for name, model in self.model.items(): 

92 for key, value in model.data.items(): 

93 yield (name, key), value 

94 

95 @property 

96 def parameter(self) -> Parameter: 

97 return dict(self.problem.param_dict.items()) 

98 

99 @property 

100 def variables(self) -> Variables: 

101 return dict(self.problem.var_dict.items()) 

102 

103 @property 

104 def weights(self) -> Matrix: 

105 return np.array(self.variables[D.WEIGHTS].value) 

106 

107 @property 

108 def factor_weights(self) -> Matrix: 

109 return np.array(self.variables[D.FACTOR_WEIGHTS].value) 

110 

111 def serialize(self, problem_file: File) -> None: 

112 with open(problem_file, "wb") as outfile: 

113 pickle.dump(self, outfile) 

114 

115 

116@dataclass(frozen=True) 

117class Builder: 

118 """Assemble variables, models, and constraints for Markowitz problems. 

119 

120 Attributes: 

121 assets: Number of asset weights to optimize. 

122 factors: Optional number of factors; if provided, a FactorModel is used, 

123 otherwise a SampleCovariance risk model is configured. 

124 model: Mapping of model components (e.g., bounds, risk) by name. 

125 constraints: Mapping of named cvxpy constraints added during build. 

126 variables: Mapping of problem variables (weights, factor weights, etc.). 

127 parameter: Mapping of cvxpy Parameters used by the builder/models. 

128 """ 

129 

130 assets: int = 0 

131 factors: int | None = None 

132 model: dict[str, Model] = field(default_factory=dict) 

133 constraints: dict[str, cp.Constraint] = field(default_factory=dict) 

134 variables: Variables = field(default_factory=dict) 

135 parameter: Parameter = field(default_factory=dict) 

136 

137 def __post_init__(self) -> None: 

138 """Initialize default risk model, variables, and bounds. 

139 

140 Selects a factor-based or sample-covariance risk model depending on 

141 `factors`, creates the corresponding variables (weights and, if 

142 applicable, factor weights and their absolute values), and registers 

143 per-asset and/or per-factor bound models. 

144 """ 

145 # pick the correct risk model 

146 if self.factors is not None: 

147 self.model[M.RISK] = FactorModel(assets=self.assets, factors=self.factors) 

148 

149 # add variable for factor weights 

150 self.variables[D.FACTOR_WEIGHTS] = cp.Variable(self.factors, name=D.FACTOR_WEIGHTS) 

151 # add bounds for factor weights 

152 self.model[M.BOUND_FACTORS] = Bounds(assets=self.factors, name="factors", acting_on=D.FACTOR_WEIGHTS) 

153 # add variable for absolute factor weights 

154 self.variables[D._ABS] = cp.Variable(self.factors, name=D._ABS, nonneg=True) 

155 

156 else: 

157 self.model[M.RISK] = SampleCovariance(assets=self.assets) 

158 # add variable for absolute weights 

159 self.variables[D._ABS] = cp.Variable(self.assets, name=D._ABS, nonneg=True) 

160 

161 # Note that for the SampleCovariance model the factor_weights are None. 

162 # They are only included for the harmony of the interfaces for both models. 

163 self.variables[D.WEIGHTS] = cp.Variable(self.assets, name=D.WEIGHTS) 

164 

165 # add bounds on assets 

166 self.model[M.BOUND_ASSETS] = Bounds(assets=self.assets, name="assets", acting_on=D.WEIGHTS) 

167 

168 @property 

169 @abstractmethod 

170 def objective(self) -> cp.Expression: 

171 """Return the objective function.""" 

172 

173 def build(self) -> _Problem: 

174 """Build the cvxpy problem.""" 

175 for name_model, model in self.model.items(): 

176 for name_constraint, constraint in model.constraints(self.variables).items(): 

177 self.constraints[f"{name_model}_{name_constraint}"] = constraint 

178 

179 problem = cp.Problem(self.objective, list(self.constraints.values())) 

180 assert problem.is_dpp(), "Problem is not DPP" 

181 

182 return _Problem(problem=problem, model=self.model) 

183 

184 @property 

185 def weights(self) -> cp.Variable: 

186 """Return the asset-weight decision variable (`weights`).""" 

187 return self.variables[D.WEIGHTS] 

188 

189 @property 

190 def risk(self) -> Model: 

191 """Return the configured risk model held under `model[M.RISK]`.""" 

192 return self.model[M.RISK] 

193 

194 @property 

195 def factor_weights(self) -> cp.Variable: 

196 """Return the factor-weight variable. 

197 

198 Note: Only present when a factor risk model is used; accessing this 

199 property without factors configured will raise a KeyError. 

200 """ 

201 return self.variables[D.FACTOR_WEIGHTS]