markowitz.optimizer.mean_variance

markowitz.optimizer.mean_variance

High-level mean-variance optimizer.

:class:MeanVariance is the public entry point for the numerical layer. It hides the bookkeeping around

• validating / canonicalising inputs (numpy arrays, pandas Series / DataFrames),
• building a fresh :class:cvxpy.Problem per objective call,
• turning the user's declarative constraint list into CVXPY constraints,
• dispatching the Cornuejols--Tutuncu reformulation for max-Sharpe, and
• mapping solver failures to the typed exception hierarchy.

The class is deliberately stateless across solves -- the only mutable state is the constraint list and the most recent solution (kept for convenience in :pyattr:last_weights).

MeanVariance(mu: pd.Series | np.ndarray, sigma: pd.DataFrame | np.ndarray, weight_bounds: tuple[float | None, float | None] = (0.0, 1.0), *, solver: str = 'CLARABEL', solver_options: dict[str, Any] | None = None, regularize: float = _REGULARIZE_DEFAULT)

Convex mean-variance optimizer over a single horizon.

Parameters:

Name	Type	Description	Default
mu	Series | ndarray	Expected returns; either a pandas.Series indexed by ticker or a 1-D numpy array. When an ndarray is passed the assets are named "A0", "A1", ....	required
sigma	DataFrame | ndarray	Covariance matrix; either a pandas.DataFrame whose row / column index matches mu or a 2-D numpy array.	required
weight_bounds	tuple[float | None, float | None]	Default (lower, upper) box constraint applied to every asset. Use (0.0, 1.0) for long-only, (-1.0, 1.0) for unconstrained long-short. Pass (None, None) to disable the box.	(0.0, 1.0)
solver	str	CVXPY solver name forwarded to :func:solve_problem.	'CLARABEL'
solver_options	dict[str, Any] | None	Extra keyword arguments forwarded to the solver backend.	None
regularize	float	Ridge term added to Sigma (Sigma + regularize * I) before building the QP. Helps numerical stability on ill-conditioned covariance matrices; should be left at 0 for parity with analytical baselines.	_REGULARIZE_DEFAULT

Source code in src/markowitz/optimizer/mean_variance.py

def __init__(
    self,
    mu: pd.Series | np.ndarray,
    sigma: pd.DataFrame | np.ndarray,
    weight_bounds: tuple[float | None, float | None] = (0.0, 1.0),
    *,
    solver: str = "CLARABEL",
    solver_options: dict[str, Any] | None = None,
    regularize: float = _REGULARIZE_DEFAULT,
) -> None:
    if not _CVXPY_AVAILABLE:  # pragma: no cover - tested via importorskip
        raise ImportError(
            "MeanVariance requires the optional 'cvxpy' dependency. "
            "Install with: pip install 'markowitz-optimizer[robust]'"
        ) from _CVXPY_IMPORT_ERROR

    tickers, mu_arr = _canonicalise_mu(mu)
    sigma_arr = _canonicalise_sigma(sigma, tickers)

    if regularize < 0:
        raise ValueError("regularize must be non-negative")
    if regularize > 0:
        sigma_arr = sigma_arr + regularize * np.eye(sigma_arr.shape[0])

    self._tickers: tuple[str, ...] = tuple(tickers)
    self._mu: np.ndarray = mu_arr
    self._sigma: np.ndarray = sigma_arr
    self._weight_bounds: tuple[float | None, float | None] = weight_bounds
    self._solver: str = solver
    self._solver_options: dict[str, Any] = dict(solver_options or {})
    self._regularize: float = float(regularize)
    self._constraints: list[Constraint] = []
    self._last_weights: pd.Series | None = None

add_constraint(constraint: Constraint) -> MeanVariance

Append a constraint and return self for chaining.

Source code in src/markowitz/optimizer/mean_variance.py

def add_constraint(self, constraint: Constraint) -> MeanVariance:
    """Append a constraint and return ``self`` for chaining."""
    self._constraints.append(constraint)
    return self

clear_constraints() -> MeanVariance

Drop every previously-added constraint.

Source code in src/markowitz/optimizer/mean_variance.py

def clear_constraints(self) -> MeanVariance:
    """Drop every previously-added constraint."""
    self._constraints.clear()
    return self

efficient_return(target_return: float) -> pd.Series

Minimise variance subject to mu^T w >= target_return.

Source code in src/markowitz/optimizer/mean_variance.py

def efficient_return(self, target_return: float) -> pd.Series:
    """Minimise variance subject to ``mu^T w >= target_return``."""
    w = _cp.Variable(self.n_assets, name="w")
    objective = _cp.Minimize(_cp.quad_form(w, _cp.psd_wrap(self._sigma)))
    constraints = (
        self._base_constraints(w)
        + self._user_constraints(w)
        + [self._mu @ w >= float(target_return)]
    )
    problem = _cp.Problem(objective, constraints)
    try:
        solve_problem(
            problem,
            solver=self._solver,
            solver_options=self._solver_options,
        )
    except OptimizationError as exc:
        if isinstance(exc, InfeasibleError):
            raise InfeasibleError(
                f"No feasible portfolio attains target_return={target_return}.",
                solver_status=exc.solver_status,
            ) from exc
        raise
    return self._finalise(w.value)

efficient_risk(target_volatility: float) -> pd.Series

Maximise mu^T w subject to sqrt(w^T Sigma w) <= target_vol.

Source code in src/markowitz/optimizer/mean_variance.py

def efficient_risk(self, target_volatility: float) -> pd.Series:
    """Maximise ``mu^T w`` subject to ``sqrt(w^T Sigma w) <= target_vol``."""
    if target_volatility <= 0:
        raise ValueError("target_volatility must be strictly positive")
    w = _cp.Variable(self.n_assets, name="w")
    objective = _cp.Maximize(self._mu @ w)
    risk_cap = _cp.quad_form(w, _cp.psd_wrap(self._sigma)) <= (float(target_volatility) ** 2)
    constraints = self._base_constraints(w) + self._user_constraints(w) + [risk_cap]
    problem = _cp.Problem(objective, constraints)
    solve_problem(
        problem,
        solver=self._solver,
        solver_options=self._solver_options,
    )
    return self._finalise(w.value)

max_quadratic_utility(risk_aversion: float = 1.0) -> pd.Series

Maximise mu^T w - (lambda/2) w^T Sigma w.

With risk_aversion = lambda. lambda > 0 is required.

Source code in src/markowitz/optimizer/mean_variance.py

def max_quadratic_utility(self, risk_aversion: float = 1.0) -> pd.Series:
    """Maximise ``mu^T w - (lambda/2) w^T Sigma w``.

    With ``risk_aversion = lambda``.  ``lambda > 0`` is required.
    """
    if risk_aversion <= 0:
        raise ValueError("risk_aversion must be strictly positive")
    w = _cp.Variable(self.n_assets, name="w")
    utility = self._mu @ w - 0.5 * risk_aversion * _cp.quad_form(w, _cp.psd_wrap(self._sigma))
    objective = _cp.Maximize(utility)
    constraints = self._base_constraints(w) + self._user_constraints(w)
    problem = _cp.Problem(objective, constraints)
    solve_problem(
        problem,
        solver=self._solver,
        solver_options=self._solver_options,
    )
    return self._finalise(w.value)

max_sharpe(risk_free_rate: float = 0.0) -> pd.Series

Maximise the Sharpe ratio via the Cornuejols--Tutuncu reformulation.

Source code in src/markowitz/optimizer/mean_variance.py

def max_sharpe(self, risk_free_rate: float = 0.0) -> pd.Series:
    """Maximise the Sharpe ratio via the Cornuejols--Tutuncu reformulation."""
    long_only = bool(self._weight_bounds[0] is not None and self._weight_bounds[0] >= 0)
    detect_degeneracy(self._mu, risk_free_rate, self._weight_bounds, long_only=long_only)

    # Translate the user's linear (in w) constraints into y-space:
    # the substitution w = y / kappa means any "Aw <= b * 1" becomes
    # "Ay <= b * kappa".  Box constraints we already wired in above.
    # For now we only allow user constraints on the *standard* QP path
    # (min-vol / efficient-* / max-quadratic-utility).  If users have
    # added extra constraints we still respect the basic LongOnly /
    # WeightBounds ones via reformulation; richer constraints fall
    # through to a runtime error rather than being silently dropped.
    extra_builders = self._collect_y_space_builders()

    reform = reformulate_max_sharpe(
        self._mu,
        self._sigma,
        risk_free_rate=risk_free_rate,
        weight_bounds=self._weight_bounds,
        long_only=long_only,
        extra_constraint_builders=extra_builders,
    )
    solve_problem(
        reform.problem,
        solver=self._solver,
        solver_options=self._solver_options,
    )
    if reform.y.value is None:  # pragma: no cover - defensive
        raise SolverError(
            "Max-Sharpe reformulation produced no y vector.",
            solver_status="no_solution",
        )
    weights = back_transform(np.asarray(reform.y.value))
    return self._finalise(weights)

min_volatility() -> pd.Series

Minimise w^T Sigma w subject to sum(w) = 1 and constraints.

Source code in src/markowitz/optimizer/mean_variance.py

def min_volatility(self) -> pd.Series:
    """Minimise ``w^T Sigma w`` subject to ``sum(w) = 1`` and constraints."""
    w = _cp.Variable(self.n_assets, name="w")
    objective = _cp.Minimize(_cp.quad_form(w, _cp.psd_wrap(self._sigma)))
    constraints = self._base_constraints(w) + self._user_constraints(w)
    problem = _cp.Problem(objective, constraints)
    solve_problem(
        problem,
        solver=self._solver,
        solver_options=self._solver_options,
    )
    return self._finalise(w.value)

portfolio_performance(weights: pd.Series | np.ndarray | None = None, *, risk_free_rate: float = 0.0) -> tuple[float, float, float]

Return (expected_return, volatility, sharpe_ratio).

If weights is None the most-recently-computed weights are used; an error is raised when no solve has happened yet.

Source code in src/markowitz/optimizer/mean_variance.py

def portfolio_performance(
    self,
    weights: pd.Series | np.ndarray | None = None,
    *,
    risk_free_rate: float = 0.0,
) -> tuple[float, float, float]:
    """Return ``(expected_return, volatility, sharpe_ratio)``.

    If ``weights`` is ``None`` the most-recently-computed weights are
    used; an error is raised when no solve has happened yet.
    """
    if weights is None:
        if self._last_weights is None:
            raise ValueError("No weights available -- call an optimizer first or pass weights.")
        w_arr = self._last_weights.to_numpy(dtype=float)
    else:
        w_arr = _align_weights(weights, self._tickers)

    expected_return = float(self._mu @ w_arr)
    variance = float(w_arr @ self._sigma @ w_arr)
    volatility = float(np.sqrt(max(variance, 0.0)))
    if volatility <= 0.0:
        sharpe = float("nan")
    else:
        sharpe = (expected_return - float(risk_free_rate)) / volatility
    return expected_return, volatility, sharpe