Coverage for src/cvxmarkowitz/problem.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"""The built problem container and its (de)serialization round-trip.""" 

15 

16from __future__ import annotations 

17 

18import pickle # nosec B403 

19from collections.abc import Generator 

20from dataclasses import dataclass, field 

21from os import PathLike 

22from typing import Any 

23 

24import cvxpy as cp 

25import numpy as np 

26 

27from cvxmarkowitz.cvxerror import CvxError 

28from cvxmarkowitz.model import Model 

29from cvxmarkowitz.names import DataNames as D 

30from cvxmarkowitz.types import File, Matrix, Parameter, Variables 

31 

32 

33def deserialize( 

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

35 *, 

36 trusted: bool = False, 

37) -> Any: 

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

39 

40 .. warning:: 

41 

42 This uses :func:`pickle.load`, which executes arbitrary code while 

43 unpickling. Only ever call this on files you produced yourself with 

44 :meth:`_Problem.serialize`. Never deserialize a file received from an 

45 untrusted or unauthenticated source — doing so is equivalent to 

46 running that source's code on your machine. 

47 

48 To make that trust boundary explicit, deserialization is opt-in: you must 

49 pass ``trusted=True`` to confirm the file is one you produced yourself. 

50 Calling without it raises :class:`~cvxmarkowitz.cvxerror.CvxError` rather 

51 than silently unpickling. 

52 

53 Args: 

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

55 trusted: Must be set to ``True`` to confirm the file originates from a 

56 trusted source. Defaults to ``False``, which refuses to load. 

57 

58 Returns: 

59 The deserialized `_Problem` instance. 

60 

61 Raises: 

62 CvxError: If ``trusted`` is not explicitly set to ``True``. 

63 """ 

64 if not trusted: 

65 raise CvxError( # noqa: TRY003 

66 "Refusing to deserialize: pickle.load executes arbitrary code. " 

67 "Pass trusted=True only for a file you produced yourself with " 

68 "_Problem.serialize()." 

69 ) 

70 # nosec B301 / noqa: S301: pickle is the intended format for round-tripping a 

71 # built problem. The trust boundary is guarded by the trusted flag above; the 

72 # input is assumed to be a self-produced serialize() file. 

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

74 return pickle.load(infile) # nosec B301 # noqa: S301 

75 

76 

77@dataclass(frozen=True) 

78class _Problem: 

79 """Frozen container holding a built cvxpy problem and its named models.""" 

80 

81 problem: cp.Problem 

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

83 

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

85 """Update the problem.""" 

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

87 for key in model.data: 

88 if key not in kwargs: 

89 raise CvxError(f"Missing data for {key} in model {name}") # noqa: TRY003 

90 

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

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

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

94 # exactly the correct shape. 

95 model.update(**kwargs) 

96 

97 return self 

98 

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

100 """Solve the problem.""" 

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

102 

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

104 raise CvxError(f"Problem status is {self.problem.status}") # noqa: TRY003 

105 

106 return float(value) 

107 

108 @property 

109 def value(self) -> float: 

110 """Return the current objective value of the solved problem.""" 

111 return float(self.problem.value) 

112 

113 def is_dpp(self) -> bool: 

114 """Return True if the problem satisfies disciplined parameterized programming.""" 

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

116 

117 @property 

118 def data(self) -> Generator[tuple[tuple[str, str], cp.Parameter]]: 

119 """Yield ``((model_name, param_key), parameter)`` pairs for all models.""" 

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

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

122 yield (name, key), value 

123 

124 @property 

125 def parameter(self) -> Parameter: 

126 """Return a mapping of parameter names to cvxpy Parameter objects.""" 

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

128 

129 @property 

130 def variables(self) -> Variables: 

131 """Return a mapping of variable names to cvxpy Variable objects.""" 

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

133 

134 @property 

135 def weights(self) -> Matrix: 

136 """Return the optimal asset weights as a numpy array.""" 

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

138 

139 @property 

140 def factor_weights(self) -> Matrix: 

141 """Return the optimal factor weights as a numpy array.""" 

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

143 

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

145 """Pickle this problem to disk for later reuse with `deserialize`.""" 

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

147 pickle.dump(self, outfile)