`markowitz`¶

`markowitz` ¶

markowitz-optimizer -- research-grade mean-variance portfolio toolkit.

Public entry point. Exposes the installed package version via :pydata:markowitz.__version__ and re-exports the headline public API so that from markowitz import MeanVariance works without users having to remember submodule paths.

`AbsoluteView(asset: str, expected_return: float, confidence: float | None = None)` `dataclass` ¶

A point estimate on a single asset's expected excess return.

`AnalyticFrontier(mu: npt.ArrayLike, Sigma: npt.ArrayLike)` ¶

Closed-form mean-variance frontier for (mu, Sigma).

Construction is the only point at which Sigma is factored; all subsequent queries (gmv, tangency, efficient_return, frontier) are O(n) once the cached basis vectors are available.

Source code in src/markowitz/core/frontier.py

def __init__(self, mu: npt.ArrayLike, Sigma: npt.ArrayLike) -> None:
    mu_arr = np.asarray(mu, dtype=np.float64)
    Sigma_arr = np.asarray(Sigma, dtype=np.float64)

    if mu_arr.ndim != 1:
        raise NumericalError(f"mu must be 1-D, got shape {mu_arr.shape!r}")
    if Sigma_arr.ndim != 2 or Sigma_arr.shape[0] != Sigma_arr.shape[1]:
        raise NumericalError(f"Sigma must be square 2-D, got shape {Sigma_arr.shape!r}")
    if mu_arr.shape[0] != Sigma_arr.shape[0]:
        raise NumericalError(
            "Dimension mismatch between mu and Sigma: "
            f"{mu_arr.shape[0]} vs {Sigma_arr.shape[0]}"
        )
    if not np.all(np.isfinite(mu_arr)):
        raise NumericalError("mu contains non-finite values")

    # Validate that Sigma is symmetric positive definite.  This
    # raises SingularCovarianceError on failure, matching the
    # documented contract.
    psd_check(Sigma_arr)

    n = mu_arr.shape[0]
    ones = np.ones(n, dtype=np.float64)
    rhs = np.column_stack((ones, mu_arr))
    sol = cholesky_solve(Sigma_arr, rhs)
    Sinv_one = sol[:, 0]
    Sinv_mu = sol[:, 1]

    A = float(ones @ Sinv_one)
    B = float(ones @ Sinv_mu)
    C = float(mu_arr @ Sinv_mu)
    D = A * C - B * B

    self._mu = mu_arr
    self._Sigma = Sigma_arr
    self._Sinv_one = Sinv_one
    self._Sinv_mu = Sinv_mu
    self._A = A
    self._B = B
    self._C = C
    self._D = D
    self._n = int(mu_arr.shape[0])

`abcd: MertonABCD` `property` ¶

The Merton scalars associated with (mu, Sigma).

`is_degenerate: bool` `property` ¶

True when D is essentially zero.

A degenerate frontier collapses to a single point in mean- variance space because mu is collinear with 1 (every asset has the same expected return). Efficient-return queries are still well-defined at the GMV expected return but are infeasible for any other target.

`n_assets: int` `property` ¶

Number of assets in the investment universe.

`efficient_return(mu_p: float) -> Portfolio` ¶

Return the minimum-variance portfolio with expected return mu_p.

Source code in src/markowitz/core/frontier.py

def efficient_return(self, mu_p: float) -> Portfolio:
    """Return the minimum-variance portfolio with expected return ``mu_p``."""

    from markowitz.core.portfolio import Portfolio  # noqa: PLC0415 - circular-import guard

    mu_target = float(mu_p)
    if self.is_degenerate:
        raise InfeasibleFrontierError(
            "Frontier is degenerate (D ~ 0); efficient_return is "
            "undefined for arbitrary target returns."
        )

    lam1 = (self._A * mu_target - self._B) / self._D
    lam2 = (self._C - self._B * mu_target) / self._D
    w = lam1 * self._Sinv_mu + lam2 * self._Sinv_one
    var = (self._A * mu_target * mu_target - 2.0 * self._B * mu_target + self._C) / self._D
    # Floor to zero to absorb sub-eps rounding at the GMV vertex.
    var = max(var, 0.0)
    return Portfolio(
        weights=w,
        expected_return=mu_target,
        volatility=float(np.sqrt(var)),
        sharpe=None,
    )

`frontier(n_points: int = 100, *, mu_min: float | None = None, mu_max: float | None = None) -> list[Portfolio]` ¶

Return n_points portfolios spanning [mu_min, mu_max].

Defaults straddle the GMV expected return symmetrically: if no bounds are supplied, the frontier extends from mu_GMV to mu_GMV + 2 * (max(mu) - mu_GMV). Callers that need a specific range should pass it explicitly.

Source code in src/markowitz/core/frontier.py

def frontier(
    self,
    n_points: int = 100,
    *,
    mu_min: float | None = None,
    mu_max: float | None = None,
) -> list[Portfolio]:
    """Return ``n_points`` portfolios spanning ``[mu_min, mu_max]``.

    Defaults straddle the GMV expected return symmetrically: if no
    bounds are supplied, the frontier extends from ``mu_GMV`` to
    ``mu_GMV + 2 * (max(mu) - mu_GMV)``.  Callers that need a
    specific range should pass it explicitly.
    """

    if n_points < 2:
        raise ValueError(f"n_points must be >= 2, got {n_points}")

    mu_gmv = self._B / self._A
    if mu_min is None:
        mu_min = float(mu_gmv)
    if mu_max is None:
        top = float(np.max(self._mu))
        mu_max = float(mu_gmv + 2.0 * (top - mu_gmv))
        if mu_max <= mu_min:
            mu_max = mu_min + 1.0

    if mu_max < mu_min:
        raise ValueError(f"mu_max ({mu_max}) must be >= mu_min ({mu_min})")

    grid = np.linspace(float(mu_min), float(mu_max), int(n_points))
    return [self.efficient_return(float(m)) for m in grid]

`gmv() -> Portfolio` ¶

Return the global minimum-variance portfolio.

Source code in src/markowitz/core/frontier.py

def gmv(self) -> Portfolio:
    """Return the global minimum-variance portfolio."""

    from markowitz.core.portfolio import Portfolio  # noqa: PLC0415 - circular-import guard

    w = self._Sinv_one / self._A
    var = 1.0 / self._A
    mu_p = self._B / self._A
    return Portfolio(
        weights=w,
        expected_return=float(mu_p),
        volatility=float(np.sqrt(var)),
        sharpe=None,
    )

`tangency(rf: float) -> Portfolio` ¶

Return the tangency portfolio for risk-free rate rf.

Raises:

Type	Description
`InfeasibleFrontierError`	If `rf` coincides with the GMV expected return (i.e. `B - A * rf` is numerically zero), making the tangency line parallel to the frontier.

Source code in src/markowitz/core/frontier.py

def tangency(self, rf: float) -> Portfolio:
    """Return the tangency portfolio for risk-free rate ``rf``.

    Raises
    ------
    InfeasibleFrontierError
        If ``rf`` coincides with the GMV expected return (i.e.
        ``B - A * rf`` is numerically zero), making the tangency
        line parallel to the frontier.
    """

    from markowitz.core.portfolio import Portfolio  # noqa: PLC0415 - circular-import guard

    rf_f = float(rf)
    denom = self._B - self._A * rf_f
    scale = max(abs(self._B), self._A * abs(rf_f), 1.0)
    if abs(denom) < 1e-12 * scale:
        raise InfeasibleFrontierError(
            "Tangency portfolio is undefined: risk-free rate coincides "
            f"with mu_GMV = {self._B / self._A:.10g} (B - A*rf = {denom:.3e})"
        )

    # w = Sigma^-1 (mu - rf * 1) / (B - A * rf)
    # Reuse cached basis vectors instead of a second Cholesky solve.
    w = (self._Sinv_mu - rf_f * self._Sinv_one) / denom
    excess = self._C - 2.0 * self._B * rf_f + self._A * rf_f * rf_f
    # Algebraically excess >= 0 for a PD Sigma, but floor it to guard
    # against floating-point chatter near zero.
    excess = max(excess, 0.0)
    sharpe = float(np.sqrt(excess))
    er = float(w @ self._mu)
    var = float(w @ self._Sigma @ w)
    vol = float(np.sqrt(max(var, 0.0)))
    return Portfolio(
        weights=w,
        expected_return=er,
        volatility=vol,
        sharpe=sharpe,
    )

`BlackLitterman(cov: pd.DataFrame, market_weights: pd.Series, views: Views | None = None, *, tau: float = 0.05, delta: float = 2.5, omega: np.ndarray | None = None, omega_method: str = 'he_litterman', posterior_covariance_mode: str = 'full', risk_free_rate: float | None = None)` ¶

Construct a Black-Litterman posterior from a covariance prior and views.

Parameters:

Name	Type	Description	Default
`cov`	`DataFrame`	Prior covariance matrix as a labelled `DataFrame`.	required
`market_weights`	`Series`	Market-cap weights series aligned to the covariance index.	required
`views`	`Views \| None`	:class:`Views` container with absolute and/or relative view specs. Pass `None` (or an empty container) for the no-views special case in which the posterior collapses to the prior.	`None`
`tau`	`float`	Prior-scaling parameter. Must be strictly positive. Defaults to `0.05`.	`0.05`
`delta`	`float`	Representative-investor risk-aversion coefficient. Defaults to `2.5`.	`2.5`
`omega`	`ndarray \| None`	Optional pre-computed view-uncertainty matrix; if supplied, `omega_method` is ignored.	`None`
`omega_method`	`str`	Strategy used to derive `Omega` when not supplied: one of `"he_litterman"`, `"idzorek_exact"`, `"idzorek_approx"`.	`'he_litterman'`
`posterior_covariance_mode`	`str`	`"full"` returns `Sigma_BL = Sigma + M` (default), `"prior_only"` returns the prior covariance unchanged.	`'full'`
`risk_free_rate`	`float \| None`	Optional metadata used by :meth:`summary` to label the regime; not used in any numerical step (all quantities are excess returns).	`None`

Source code in src/markowitz/views/black_litterman.py

def __init__(
    self,
    cov: pd.DataFrame,
    market_weights: pd.Series,
    views: Views | None = None,
    *,
    tau: float = 0.05,
    delta: float = 2.5,
    omega: np.ndarray | None = None,
    omega_method: str = "he_litterman",
    posterior_covariance_mode: str = "full",
    risk_free_rate: float | None = None,
) -> None:
    self._validate_inputs(cov, market_weights, tau, delta)
    if omega_method not in _OMEGA_METHODS:
        raise OmegaSpecificationError(
            f"omega_method must be one of {sorted(_OMEGA_METHODS)}, got {omega_method!r}."
        )
    if posterior_covariance_mode not in _COV_MODES:
        raise BlackLittermanError(
            f"posterior_covariance_mode must be one of {sorted(_COV_MODES)}, "
            f"got {posterior_covariance_mode!r}."
        )

    self._cov: pd.DataFrame = cov
    self._assets: list[str] = list(cov.index)
    self._market_weights: pd.Series = market_weights.reindex(cov.index)
    if self._market_weights.isna().any():
        missing = self._market_weights.index[self._market_weights.isna()].tolist()
        raise BlackLittermanError(f"Market weights missing entries for: {missing}.")
    self._tau: float = float(tau)
    self._delta: float = float(delta)
    self._omega_method: str = omega_method
    self._posterior_cov_mode: str = posterior_covariance_mode
    self._risk_free_rate: float | None = risk_free_rate

    self._views: Views | None = views if views is not None and len(views) > 0 else None
    self._omega_user: np.ndarray | None = omega

    # Cached numerical pieces (built lazily).
    self._pi: pd.Series | None = None
    self._mu_bl: pd.Series | None = None
    self._sigma_bl: pd.DataFrame | None = None

`implied_returns() -> pd.Series` ¶

Return the equilibrium-implied prior pi = delta * Sigma * w_mkt.

Source code in src/markowitz/views/black_litterman.py

def implied_returns(self) -> pd.Series:
    """Return the equilibrium-implied prior ``pi = delta * Sigma * w_mkt``."""
    if self._pi is None:
        self._pi = _implied_returns(self._cov, self._market_weights, delta=self._delta)
    return self._pi.copy()

`posterior_covariance() -> pd.DataFrame` ¶

Return the posterior covariance matrix.

Source code in src/markowitz/views/black_litterman.py

def posterior_covariance(self) -> pd.DataFrame:
    """Return the posterior covariance matrix."""
    if self._sigma_bl is None:
        self._compute_posterior()
    assert self._sigma_bl is not None
    out: pd.DataFrame = self._sigma_bl.copy()
    return out

`posterior_returns() -> pd.Series` ¶

Return the Black-Litterman posterior excess-return vector.

Source code in src/markowitz/views/black_litterman.py

def posterior_returns(self) -> pd.Series:
    """Return the Black-Litterman posterior excess-return vector."""
    if self._mu_bl is None:
        self._compute_posterior()
    assert self._mu_bl is not None
    return self._mu_bl.copy()

`posterior_weights(*, constrained: bool = False) -> pd.Series` ¶

Return the unconstrained tangency weights implied by the posterior.

w_BL = (delta * Sigma_BL)^{-1} * mu_BL.

Parameters:

Name	Type	Description	Default
`constrained`	`bool`	If `True`, project to the long-only simplex by clipping at zero and renormalising to sum to 1. This is a convenience for quick inspection -- production code should use the full :mod:`markowitz.optimizer` pipeline.	`False`

Source code in src/markowitz/views/black_litterman.py

def posterior_weights(self, *, constrained: bool = False) -> pd.Series:
    """Return the unconstrained tangency weights implied by the posterior.

    ``w_BL = (delta * Sigma_BL)^{-1} * mu_BL``.

    Parameters
    ----------
    constrained
        If ``True``, project to the long-only simplex by clipping at zero
        and renormalising to sum to 1.  This is a convenience for quick
        inspection -- production code should use the full
        :mod:`markowitz.optimizer` pipeline.
    """
    mu = self.posterior_returns().to_numpy()
    sigma_bl = self.posterior_covariance().to_numpy()
    c_factor = cho_factor(self._delta * sigma_bl)
    w = cho_solve(c_factor, mu)
    if constrained:
        w = np.clip(w, 0.0, None)
        total = w.sum()
        if total > 0.0:
            w = w / total
    return pd.Series(w, index=self._assets, name="w_bl")

`summary() -> pd.DataFrame` ¶

Return a per-asset table with prior, posterior and weight tilts.

Source code in src/markowitz/views/black_litterman.py

def summary(self) -> pd.DataFrame:
    """Return a per-asset table with prior, posterior and weight tilts."""
    pi = self.implied_returns()
    mu_bl = self.posterior_returns()
    w_mkt = self._market_weights.copy()
    w_bl = self.posterior_weights()
    df = pd.DataFrame(
        {
            "pi": pi,
            "mu_bl": mu_bl,
            "delta_mu": mu_bl - pi,
            "w_mkt": w_mkt,
            "w_bl": w_bl,
            "delta_w": w_bl - w_mkt,
        },
        index=self._assets,
    )
    return df

`CAPMReturns(*, risk_free_rate: float = 0.0, market_premium: float | None = None, annualize: bool = True, periods_per_year: int | None = None)` ¶

Bases: _BaseMean

CAPM-implied expected returns from per-asset OLS betas.

Source code in src/markowitz/estimators/means.py

def __init__(
    self,
    *,
    risk_free_rate: float = 0.0,
    market_premium: float | None = None,
    annualize: bool = True,
    periods_per_year: int | None = None,
) -> None:
    self.risk_free_rate = float(risk_free_rate)
    self.market_premium = market_premium
    self.annualize = bool(annualize)
    self.periods_per_year = periods_per_year

`CustomConstraint(builder: Callable[[cp.Variable, ConstraintContext], list[cp.Constraint]], description: str = 'custom')` `dataclass` ¶

Escape hatch for user-defined CVXPY constraints.

builder is a callable (w, ctx) -> list[cvxpy.Constraint]; it is invoked freshly each solve so it should not close over solver state.

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

def __init__(
    self,
    builder: Callable[[cp.Variable, ConstraintContext], list[cp.Constraint]],
    description: str = "custom",
) -> None:
    object.__setattr__(self, "builder", builder)
    object.__setattr__(self, "_description", description)

`EWMACovariance(*, halflife_years: float | None = None, lam: float | None = None, burn_in: int | Literal['auto'] = 'auto', annualize: bool = True, periods_per_year: int | None = None)` ¶

Bases: _BaseCov

Exponentially weighted covariance (RiskMetrics recursion).

Source code in src/markowitz/estimators/covariance.py

def __init__(
    self,
    *,
    halflife_years: float | None = None,
    lam: float | None = None,
    burn_in: int | Literal["auto"] = "auto",
    annualize: bool = True,
    periods_per_year: int | None = None,
) -> None:
    if halflife_years is not None and lam is not None:
        raise EstimatorConfigError("Provide at most one of halflife_years or lam, not both.")
    self.halflife_years = halflife_years
    self.lam = lam
    self.burn_in = burn_in
    self.annualize = bool(annualize)
    self.periods_per_year = periods_per_year

`EWMAMean(*, halflife_years: float | None = None, lam: float | None = None, annualize: bool = True, periods_per_year: int | None = None)` ¶

Bases: _BaseMean

Exponentially weighted mean using RiskMetrics-style decay.

Source code in src/markowitz/estimators/means.py

def __init__(
    self,
    *,
    halflife_years: float | None = None,
    lam: float | None = None,
    annualize: bool = True,
    periods_per_year: int | None = None,
) -> None:
    if (halflife_years is None) == (lam is None):
        raise EstimatorConfigError("Exactly one of halflife_years or lam must be provided.")
    self.halflife_years = halflife_years
    self.lam = lam
    self.annualize = bool(annualize)
    self.periods_per_year = periods_per_year

`ImpliedReturns(*, delta: float = 2.5)` ¶

Bases: _BaseMean

Reverse-optimization implied returns μ = δ · Σ · w.

Source code in src/markowitz/estimators/means.py

def __init__(self, *, delta: float = 2.5) -> None:
    if delta <= 0.0:
        raise EstimatorConfigError(f"delta must be positive; got {delta}")
    self.delta = float(delta)

`InfeasibleError(message: str, *, solver_status: str | None = None, solver_message: str | None = None, constraints_violated: list[str] | None = None)` ¶

Bases: OptimizationError

Raised when the feasible region is empty.

Either the solver returned infeasible / infeasible_inaccurate, or an upfront sanity check (e.g. max(mu) <= rf for max-Sharpe) detected that no admissible portfolio exists.

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

def __init__(
    self,
    message: str,
    *,
    solver_status: str | None = None,
    solver_message: str | None = None,
    constraints_violated: list[str] | None = None,
) -> None:
    super().__init__(message)
    self.solver_status = solver_status
    self.solver_message = solver_message
    self.constraints_violated = constraints_violated

`JorionBayesStein(*, base_cov: np.ndarray | None = None, annualize: bool = True, periods_per_year: int | None = None)` ¶

Bases: _BaseMean

Bayes-Stein shrinkage of the sample mean toward the minimum-variance mean.

Source code in src/markowitz/estimators/means.py

def __init__(
    self,
    *,
    base_cov: np.ndarray | None = None,
    annualize: bool = True,
    periods_per_year: int | None = None,
) -> None:
    self.base_cov = base_cov
    self.annualize = bool(annualize)
    self.periods_per_year = periods_per_year

`LedoitWolfShrinkage(*, target: Literal['identity', 'constant_corr'] = 'identity', annualize: bool = True, periods_per_year: int | None = None)` ¶

Bases: _BaseCov

Ledoit-Wolf shrinkage covariance, sklearn-compatible for the identity target.

Source code in src/markowitz/estimators/covariance.py

def __init__(
    self,
    *,
    target: Literal["identity", "constant_corr"] = "identity",
    annualize: bool = True,
    periods_per_year: int | None = None,
) -> None:
    if target not in ("identity", "constant_corr"):
        raise EstimatorConfigError(
            f"target must be 'identity' or 'constant_corr'; got {target!r}"
        )
    self.target = target
    self.annualize = bool(annualize)
    self.periods_per_year = periods_per_year

`LeverageCap(max_leverage: float)` `dataclass` ¶

Gross-exposure cap sum(|w_i|) <= max_leverage.

`LongOnly()` `dataclass` ¶

No short sales: w_i >= 0 for every asset.