Coverage for src/cvxcla/types.py: 100%
113 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-27 06:45 +0000
1"""Type definitions and classes for the Critical Line Algorithm.
3This module defines the core data structures used in the Critical Line Algorithm:
4- FrontierPoint: Represents a point on the efficient frontier.
5- TurningPoint: Represents a turning point on the efficient frontier.
6- Frontier: Represents the entire efficient frontier.
8It also defines type aliases for commonly used types.
9"""
11from __future__ import annotations
13from collections.abc import Iterator
14from dataclasses import dataclass, field
15from typing import TYPE_CHECKING
17if TYPE_CHECKING:
18 import plotly.graph_objects as go
20import numpy as np
21from numpy.typing import NDArray
23from .operators import CovarianceOperator
26def _covariance_matvec(
27 covariance: NDArray[np.float64] | CovarianceOperator, x: NDArray[np.float64]
28) -> NDArray[np.float64]:
29 """Compute ``Sigma @ x`` for a dense matrix or a ``CovarianceOperator`` backend."""
30 if isinstance(covariance, CovarianceOperator):
31 return covariance.matvec(x)
32 return covariance @ x
35@dataclass(frozen=True)
36class FrontierPoint:
37 """A point on the efficient frontier.
39 This class represents a portfolio on the efficient frontier, defined by its weights.
40 It provides methods to compute the expected return and variance of the portfolio.
42 Attributes:
43 weights: Vector of portfolio weights for each asset.
45 """
47 weights: NDArray[np.float64]
49 def mean(self, mean: NDArray[np.float64]) -> float:
50 """Compute the expected return of the portfolio.
52 Args:
53 mean: Vector of expected returns for each asset.
55 Returns:
56 The expected return of the portfolio.
58 Examples:
59 >>> import numpy as np
60 >>> fp = FrontierPoint(weights=np.array([0.5, 0.5]))
61 >>> fp.mean(np.array([0.1, 0.2]))
62 0.15000000000000002
64 """
65 return float(mean.T @ self.weights)
67 def variance(self, covariance: NDArray[np.float64] | CovarianceOperator) -> float:
68 """Compute the expected variance of the portfolio.
70 Args:
71 covariance: Covariance matrix of asset returns, either as a dense
72 matrix or as a ``CovarianceOperator`` backend.
74 Returns:
75 The expected variance of the portfolio.
77 """
78 return float(self.weights.T @ _covariance_matvec(covariance, self.weights))
81@dataclass(frozen=True)
82class TurningPoint(FrontierPoint):
83 """Turning point.
85 A turning point is a vector of weights, a lambda value, and a boolean vector
86 indicating which assets are free. All assets that are not free are blocked.
88 For problems with general inequality constraints ``G w <= h`` the turning
89 point also records which inequality *rows* are active (held at equality
90 ``g_i w = h_i``) via ``active_ineq``. These are the row analogue of the box
91 active set: a free weight on a bound is a per-variable active constraint, an
92 active inequality row is a per-row one. The default is an empty mask, so
93 box-and-equality problems (and the LASSO path) are unaffected.
94 """
96 free: NDArray[np.bool_]
97 lamb: float = np.inf
98 active_ineq: NDArray[np.bool_] = field(default_factory=lambda: np.zeros(0, dtype=bool))
100 @property
101 def free_indices(self) -> np.ndarray:
102 """Returns the indices of the free assets."""
103 return np.where(self.free)[0]
105 @property
106 def blocked_indices(self) -> np.ndarray:
107 """Returns the indices of the blocked assets."""
108 return np.where(~self.free)[0]
111@dataclass(frozen=True)
112class Frontier:
113 """A frontier is a list of frontier points. Some of them might be turning points."""
115 mean: NDArray[np.float64]
116 covariance: NDArray[np.float64] | CovarianceOperator
117 frontier: list[FrontierPoint] = field(default_factory=list)
119 def interpolate(self, num: int = 100) -> Frontier:
120 """Interpolate the frontier with additional points between existing points.
122 This method creates a new Frontier object with additional points interpolated
123 between the existing points. This is useful for creating a smoother representation
124 of the efficient frontier for visualization or analysis.
126 Args:
127 num: The number of points to use in the interpolation. The method will create
128 num-1 new points between each pair of adjacent existing points.
130 Returns:
131 A new Frontier object with the interpolated points.
133 """
135 def _interpolate() -> Iterator[FrontierPoint]:
136 """Yield interpolated frontier points between each adjacent pair."""
137 for w_right, w_left in zip(self.weights[0:-1], self.weights[1:], strict=False): # pragma: no mutate
138 for lamb in np.linspace(0, 1, num):
139 if lamb > 0:
140 yield FrontierPoint(weights=lamb * w_left + (1 - lamb) * w_right)
142 points = list(_interpolate())
143 return Frontier(frontier=points, mean=self.mean, covariance=self.covariance)
145 def __iter__(self) -> Iterator[FrontierPoint]:
146 """Iterate over all frontier points."""
147 yield from self.frontier
149 def __len__(self) -> int:
150 """Give number of frontier points."""
151 return len(self.frontier)
153 @property
154 def weights(self) -> np.ndarray:
155 """Matrix of weights. One row per point."""
156 return np.array([point.weights for point in self])
158 @property
159 def returns(self) -> np.ndarray:
160 """Vector of expected returns."""
161 return np.array([point.mean(self.mean) for point in self])
163 @property
164 def variance(self) -> np.ndarray:
165 """Vector of expected variances."""
166 return np.array([point.variance(self.covariance) for point in self])
168 @property
169 def sharpe_ratio(self) -> np.ndarray:
170 """Vector of expected Sharpe ratios."""
171 ratios: np.ndarray = self.returns / self.volatility
172 return ratios
174 @property
175 def volatility(self) -> np.ndarray:
176 """Vector of expected volatilities."""
177 vol: np.ndarray = np.sqrt(self.variance)
178 return vol
180 @property
181 def max_sharpe(self) -> tuple[float, np.ndarray]:
182 """Maximal Sharpe ratio on the frontier.
184 The maximiser lies on one of the two affine segments adjacent to the
185 turning point of largest discrete Sharpe ratio. On each segment the Sharpe
186 ratio has a closed-form maximiser (see :meth:`_segment_max_sharpe`), so the
187 result is exact rather than the product of a numerical line search.
189 Returns:
190 Tuple of maximal Sharpe ratio and the weights to achieve it
192 """
193 weights = self.weights
194 sharpe_ratios = self.sharpe_ratio
196 # The discrete maximum brackets the continuous one: the optimum sits on a
197 # segment touching the turning point of largest Sharpe ratio.
198 sr_position_max = int(np.argmax(sharpe_ratios))
199 right = min(sr_position_max + 1, len(self) - 1)
200 left = max(0, sr_position_max - 1)
202 # Look to the left and to the right of the discrete maximum.
203 if right > sr_position_max:
204 sharpe_ratio_right, w_right = self._segment_max_sharpe(weights[sr_position_max], weights[right])
205 else:
206 w_right = weights[sr_position_max]
207 sharpe_ratio_right = sharpe_ratios[sr_position_max]
209 if left < sr_position_max:
210 sharpe_ratio_left, w_left = self._segment_max_sharpe(weights[left], weights[sr_position_max])
211 else:
212 w_left = weights[sr_position_max]
213 sharpe_ratio_left = sharpe_ratios[sr_position_max]
215 if sharpe_ratio_left > sharpe_ratio_right:
216 return sharpe_ratio_left, w_left
218 return sharpe_ratio_right, w_right
220 def _segment_max_sharpe(self, w0: np.ndarray, w1: np.ndarray) -> tuple[float, np.ndarray]:
221 """Closed-form maximum Sharpe ratio on the affine segment between two points.
223 Parametrise the segment as ``w(t) = (1 - t) w0 + t w1`` for ``t`` in
224 ``[0, 1]``. The expected return is affine and the variance quadratic in
225 ``t``, so the Sharpe ratio is::
227 S(t) = (a0 + a1 t) / sqrt(c0 + c1 t + c2 t**2)
229 Its derivative has a *linear* numerator (the ``t**2`` terms cancel), so
230 there is a single stationary point
231 ``t* = (a0 c1 - 2 a1 c0) / (a1 c1 - 2 a0 c2)``. The maximiser over the
232 segment is therefore whichever of ``{0, 1, clamp(t*)}`` yields the largest
233 Sharpe ratio, evaluated in closed form rather than by a bounded line search.
235 Args:
236 w0: Weights at the ``t = 0`` end of the segment.
237 w1: Weights at the ``t = 1`` end of the segment.
239 Returns:
240 Tuple of the maximal Sharpe ratio on the segment and its weights.
242 """
243 delta = w1 - w0
244 sigma_w0 = _covariance_matvec(self.covariance, w0)
245 sigma_delta = _covariance_matvec(self.covariance, delta)
246 a0 = float(self.mean @ w0)
247 a1 = float(self.mean @ delta)
248 c0 = float(w0 @ sigma_w0)
249 c1 = 2.0 * float(w0 @ sigma_delta)
250 c2 = float(delta @ sigma_delta)
252 def sharpe_at(t: float) -> tuple[float, np.ndarray]:
253 """Sharpe ratio and weights at position ``t`` along the segment."""
254 weight = w0 + t * delta
255 sharpe = (a0 + a1 * t) / np.sqrt(c0 + c1 * t + c2 * t * t)
256 return float(sharpe), weight
258 # Candidate positions: the two endpoints and the interior stationary point
259 # (only when it falls strictly inside the segment).
260 candidates = [0.0, 1.0]
261 denominator = a1 * c1 - 2.0 * a0 * c2
262 if denominator != 0.0:
263 t_star = (a0 * c1 - 2.0 * a1 * c0) / denominator
264 if 0.0 < t_star < 1.0:
265 candidates.append(t_star)
267 return max((sharpe_at(t) for t in candidates), key=lambda item: item[0])
269 def plot(self, volatility: bool = False, markers: bool = True) -> go.Figure:
270 """Plot the efficient frontier.
272 This function creates a line plot of the efficient frontier, with expected return
273 on the y-axis and either variance or volatility on the x-axis.
275 Args:
276 volatility: If True, plot volatility (standard deviation) on the x-axis.
277 If False, plot variance on the x-axis.
278 markers: If True, show markers at each point on the frontier.
280 Returns:
281 A plotly Figure object that can be displayed or saved.
283 """
284 try:
285 import plotly.graph_objects as go
286 except ImportError as e:
287 msg = "Plotting requires plotly. Install it with: pip install cvxcla[plot]"
288 raise ImportError(msg) from e
290 fig = go.Figure()
292 x = self.volatility if volatility else self.variance
293 axis_title = "Expected volatility" if volatility else "Expected variance"
295 fig.add_trace(
296 go.Scatter(x=x, y=self.returns, mode="lines+markers" if markers else "lines", name="Efficient Frontier")
297 )
299 fig.update_layout(
300 xaxis_title=axis_title,
301 yaxis_title="Expected Return",
302 )
304 return fig