Coverage for src/cvxmarkowitz/problem.py: 100%
61 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-15 05:21 +0000
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 returned by :meth:`Builder.build`."""
16from __future__ import annotations
18from collections.abc import Generator
19from dataclasses import dataclass, field
20from typing import Any
22import cvxpy as cp
23import numpy as np
25from cvxmarkowitz.cvxerror import CvxDataError, CvxSolverError
26from cvxmarkowitz.model import Model
27from cvxmarkowitz.names import DataNames as D
28from cvxmarkowitz.types import Matrix, Parameter, Variables
31@dataclass(frozen=True)
32class Problem:
33 """Frozen container holding a built cvxpy problem and its named models."""
35 problem: cp.Problem
36 model: dict[str, Model] = field(default_factory=dict)
38 def update(self, **kwargs: Matrix) -> None:
39 """Overwrite the parameter values of every model, **in place**.
41 This mutates the problem rather than returning a new one. `frozen=True`
42 on this dataclass only stops attribute rebinding; the `model` mapping and
43 the cvxpy Parameters it holds stay mutable, and that is deliberate --
44 writing new values into the same compiled problem is exactly what lets
45 cvxpy reuse its cached canonicalization across solves:
47 problem = MinVar(assets=4).build() # compile once
48 for data in datasets:
49 problem.update(**data) # overwrite in place
50 problem.solve()
52 Consequently there is only ever one problem. Two `update` calls against
53 the same object do not yield two independently parametrized problems --
54 the second overwrites the first. Call `build()` again for that.
56 The whole payload is validated against every model before the first
57 value is written, so a rejected payload leaves the problem exactly as it
58 was rather than half-overwritten. See `_validate`.
60 Returns `None` (like `Model.update`) so the in-place semantics are
61 visible at the call site.
63 Raises:
64 CvxDataError: If any model is missing data for one of its parameters,
65 or if the models disagree about how large the universe is.
66 """
67 self._validate(**kwargs)
69 for model in self.model.values():
70 # It's tempting to operate without the models at this stage.
71 # However, we would give up a lot of convenience. For example,
72 # the models can be prepared to deal with data that has not
73 # exactly the correct shape.
74 model.update(**kwargs)
76 def _validate(self, **kwargs: Matrix) -> None:
77 """Check the payload against every model, writing nothing.
79 Two passes, both over all models, both raising `CvxDataError`:
81 1. every keyword each model declares is present, and
82 2. the models agree on the size of each variable they describe.
84 The second is not redundant with the shape checks inside the models.
85 `Model.update` pads a short input up to the compiled size, so a payload
86 that describes two assets to the risk model and four to the bounds
87 solves without complaint -- and solves wrongly, because the padded tail
88 carries no risk while the bounds leave it free, which the solver reads as
89 two riskless assets. Nothing inside a single model can see that; only
90 comparing the models can.
92 Raises:
93 CvxDataError: On a missing keyword, or on models that disagree about
94 the size of a variable.
95 """
96 for name, model in self.model.items():
97 # `Model.keywords`, not `model.data`: a model may consume a keyword
98 # that `data` does not back (see `ExpectedReturns.keywords`), and
99 # checking `data` alone let those through to a bare KeyError.
100 for key in model.keywords:
101 if key not in kwargs:
102 raise CvxDataError(f"Missing data for {key} in model {name}") # noqa: TRY003
104 claimed: dict[str, tuple[str, int]] = {}
106 for name, model in self.model.items():
107 for variable, size in model.dimensions(**kwargs):
108 first_name, first_size = claimed.setdefault(variable, (name, size))
110 if size != first_size:
111 raise CvxDataError( # noqa: TRY003
112 f"Inconsistent size for {variable}: model {first_name} was given "
113 f"{first_size}, model {name} was given {size}"
114 )
116 def solve(self, solver: str = cp.CLARABEL, **kwargs: Any) -> float:
117 """Solve the problem."""
118 value = self.problem.solve(solver=solver, **kwargs)
120 if self.problem.status is not cp.OPTIMAL:
121 raise CvxSolverError(f"Problem status is {self.problem.status}") # noqa: TRY003
123 return float(value)
125 def get_problem_data(
126 self,
127 solver: str = cp.CLARABEL,
128 gp: bool = False,
129 enforce_dpp: bool = False,
130 ignore_dpp: bool = False,
131 verbose: bool = False,
132 canon_backend: str | None = None,
133 solver_opts: dict[str, Any] | None = None,
134 ) -> Any:
135 """Return the low-level data the solver would be handed for this problem.
137 This forwards to :meth:`cvxpy.Problem.get_problem_data`, exposing the
138 compiled form of the problem without solving it. Useful for inspecting
139 the canonicalization or for driving a solver directly.
141 Args:
142 solver: The target solver to compile for.
143 gp: Whether to parse the problem as a disciplined geometric program.
144 enforce_dpp: Raise if the problem is not DPP-compliant.
145 ignore_dpp: Treat the problem as non-DPP even if it is compliant.
146 verbose: Print compilation progress.
147 canon_backend: Canonicalization backend to use, or None for the default.
148 solver_opts: Extra options forwarded to the solver.
150 Returns:
151 The ``(data, chain, inverse_data)`` triple produced by cvxpy.
152 """
153 return self.problem.get_problem_data(
154 solver,
155 gp=gp,
156 enforce_dpp=enforce_dpp,
157 ignore_dpp=ignore_dpp,
158 verbose=verbose,
159 canon_backend=canon_backend,
160 solver_opts=solver_opts,
161 )
163 @property
164 def value(self) -> float:
165 """Return the current objective value of the solved problem."""
166 return float(self.problem.value)
168 def is_dpp(self) -> bool:
169 """Return True if the problem satisfies disciplined parameterized programming."""
170 return bool(self.problem.is_dpp())
172 @property
173 def data(self) -> Generator[tuple[tuple[str, str], cp.Parameter]]:
174 """Yield ``((model_name, param_key), parameter)`` pairs for all models."""
175 for name, model in self.model.items():
176 for key, value in model.data.items():
177 yield (name, key), value
179 @property
180 def parameter(self) -> Parameter:
181 """Return a mapping of parameter names to cvxpy Parameter objects."""
182 return dict(self.problem.param_dict.items())
184 @property
185 def variables(self) -> Variables:
186 """Return a mapping of variable names to cvxpy Variable objects."""
187 return dict(self.problem.var_dict.items())
189 @property
190 def weights(self) -> Matrix:
191 """Return the optimal asset weights as a numpy array."""
192 return np.array(self.variables[D.WEIGHTS].value)
194 @property
195 def factor_weights(self) -> Matrix:
196 """Return the optimal factor weights as a numpy array.
198 Raises:
199 CvxDataError: If the problem was built without a factor risk model,
200 in which case there is no factor-weight variable.
201 """
202 try:
203 return np.array(self.variables[D.FACTOR_WEIGHTS].value)
204 except KeyError as err:
205 raise CvxDataError( # noqa: TRY003
206 "No factor weights: this problem was built without 'factors'."
207 ) from err