markowitz.views

markowitz.views

Black-Litterman posterior construction from absolute and relative views.

Public re-exports

• :class:BlackLitterman -- the main posterior-construction class.
• :class:Views, :class:AbsoluteView, :class:RelativeView -- view-specification dataclasses and their container.
• :func:implied_returns, :func:infer_delta -- equilibrium helpers.
• :func:idzorek_omega, :func:idzorek_omega_approx -- confidence-based Omega construction.
• Exception hierarchy rooted at :class:BlackLittermanError.

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

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

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

BlackLittermanError

Bases: Exception

Base class for all errors raised by :mod:markowitz.views.

OmegaSpecificationError

Bases: BlackLittermanError

Raised when the view-uncertainty matrix Omega is mis-specified.

Triggered for shape mismatches, non-positive-definite supplied matrices, or inconsistent combinations of omega, omega_method and per-view confidences.

RelativeView(long_leg: Mapping[str, float], short_leg: Mapping[str, float], spread: float, confidence: float | None = None) dataclass

A spread view between a long and a short basket of assets.

The basket weights specify how each leg is constructed: long_leg maps asset names to non-negative coefficients and likewise short_leg. The resulting pick row is long_leg - short_leg and must sum to zero (this is verified at construction time).

TauScaleError

Bases: BlackLittermanError

Raised when the prior-scaling parameter tau is invalid (non-positive).

ViewValidationError

Bases: BlackLittermanError

Raised when a user-supplied view fails structural validation.

Examples:

• A relative view whose pick-row coefficients do not sum to zero.
• A view that references an asset which is not present in the universe.
• A confidence value outside the closed interval [0, 1].

Views(views: Sequence[ViewSpec], assets: Sequence[str])

Container that validates and serialises a list of :class:ViewSpec.

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

def __init__(
    self,
    views: Sequence[ViewSpec],
    assets: Sequence[str],
) -> None:
    self._views: tuple[ViewSpec, ...] = tuple(views)
    self._assets: tuple[str, ...] = tuple(assets)
    self._asset_index: dict[str, int] = {a: i for i, a in enumerate(self._assets)}
    self._validate()

build_omega_he_litterman(tau: float, cov: np.ndarray) -> np.ndarray

Return the He-Litterman diagonal Omega = diag(P tau Sigma P^T).

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

def build_omega_he_litterman(self, tau: float, cov: np.ndarray) -> np.ndarray:
    """Return the He-Litterman diagonal ``Omega = diag(P tau Sigma P^T)``."""
    if tau <= 0.0:
        raise ViewValidationError(f"tau must be > 0, got {tau}.")
    if cov.shape[0] != cov.shape[1] or cov.shape[0] != len(self._assets):
        raise ViewValidationError(
            f"Covariance shape {cov.shape} incompatible with universe "
            f"of size {len(self._assets)}."
        )
    p_mat, _ = self.build_pq()
    diag = np.einsum("ki,ij,kj->k", p_mat, tau * cov, p_mat)
    return np.diag(diag)

build_pq() -> tuple[np.ndarray, np.ndarray]

Return the (P, Q) matrices for the current view set.

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

def build_pq(self) -> tuple[np.ndarray, np.ndarray]:
    """Return the ``(P, Q)`` matrices for the current view set."""
    k = len(self._views)
    n = len(self._assets)
    p_mat = np.zeros((k, n), dtype=float)
    q_vec = np.zeros(k, dtype=float)

    for i, view in enumerate(self._views):
        if isinstance(view, AbsoluteView):
            p_mat[i, self._asset_index[view.asset]] = 1.0
            q_vec[i] = float(view.expected_return)
        else:
            for asset, w in view.long_leg.items():
                p_mat[i, self._asset_index[asset]] += float(w)
            for asset, w in view.short_leg.items():
                p_mat[i, self._asset_index[asset]] -= float(w)
            q_vec[i] = float(view.spread)
    return p_mat, q_vec

confidences() -> np.ndarray | None

Return per-view confidences as an array, or None if unspecified.

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

def confidences(self) -> np.ndarray | None:
    """Return per-view confidences as an array, or ``None`` if unspecified."""
    if not self._views or self._views[0].confidence is None:
        return None
    return np.asarray([v.confidence for v in self._views], dtype=float)

idzorek_omega(P: np.ndarray, tau_Sigma: np.ndarray, confidences: np.ndarray, pi: np.ndarray, delta: float, w_mkt: np.ndarray, *, brent_xtol: float = 1e-10, brent_maxiter: int = 200) -> np.ndarray

Exact Idzorek omega via per-view root finding (with safe fallback).

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

def idzorek_omega(
    P: np.ndarray,
    tau_Sigma: np.ndarray,
    confidences: np.ndarray,
    pi: np.ndarray,
    delta: float,
    w_mkt: np.ndarray,
    *,
    brent_xtol: float = 1e-10,
    brent_maxiter: int = 200,
) -> np.ndarray:
    """Exact Idzorek omega via per-view root finding (with safe fallback)."""
    c = np.asarray(confidences, dtype=float)
    _check_confidences(c)
    if P.shape[0] != c.shape[0]:
        raise OmegaSpecificationError(
            f"P has {P.shape[0]} rows but {c.shape[0]} confidences were supplied."
        )

    q_full = np.einsum("ki,i->k", P, pi)  # placeholder; per-view qk supplied below
    # We need actual q-values: derive them from P @ pi shifted by the
    # caller-implied view targets.  Because this function may be invoked
    # without a separate q vector, we treat the *signal* as the deviation that
    # the single-view computation cares about; q drops out of the *direction*
    # of the tilt-residual so we just need any consistent vector.
    omega_diag = np.empty(P.shape[0])
    approx = idzorek_omega_approx(P, tau_Sigma, c)
    approx_diag = np.diag(approx)

    for k in range(P.shape[0]):
        ck = float(c[k])
        pk = P[k]
        # Use q derived from P @ pi + 1.0 (arbitrary positive innovation) so
        # the tilt is non-zero; the resulting omega depends only on ck and pk
        # because of scale invariance.
        qk = float(q_full[k] + 1.0)

        if ck >= 1.0 - _TINY:
            omega_diag[k] = 0.0
            continue
        if ck <= _TINY:
            omega_diag[k] = _LARGE * max(approx_diag[k], 1.0)
            continue

        target = _target_tilt(ck, pk, qk, pi, tau_Sigma, delta, w_mkt)
        target_norm = float(np.linalg.norm(target))
        if target_norm <= _TINY:
            omega_diag[k] = approx_diag[k]
            continue

        lo, hi = -40.0, 40.0
        f_lo = _tilt_residual(lo, pk, qk, pi, tau_Sigma, delta, w_mkt, target_norm)
        f_hi = _tilt_residual(hi, pk, qk, pi, tau_Sigma, delta, w_mkt, target_norm)
        if f_lo * f_hi > 0.0:
            omega_diag[k] = approx_diag[k]
            continue
        try:
            log_omega = brentq(
                _tilt_residual,
                lo,
                hi,
                args=(pk, qk, pi, tau_Sigma, delta, w_mkt, target_norm),
                xtol=brent_xtol,
                maxiter=brent_maxiter,
            )
            omega_diag[k] = float(np.exp(log_omega))
        except (ValueError, RuntimeError):
            omega_diag[k] = approx_diag[k]

    return np.diag(omega_diag)

idzorek_omega_approx(P: np.ndarray, tau_Sigma: np.ndarray, confidences: np.ndarray) -> np.ndarray

Closed-form Idzorek omega approximation.

Parameters:

Name	Type	Description	Default
P	ndarray	Pick matrix of shape (K, N).	required
tau_Sigma	ndarray	The scaled prior covariance tau * Sigma of shape (N, N).	required
confidences	ndarray	One-dimensional array of length K with entries in [0, 1].	required

Returns:

Type	Description
ndarray	Diagonal (K, K) matrix of view-uncertainty variances.

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

def idzorek_omega_approx(
    P: np.ndarray,
    tau_Sigma: np.ndarray,
    confidences: np.ndarray,
) -> np.ndarray:
    """Closed-form Idzorek omega approximation.

    Parameters
    ----------
    P
        Pick matrix of shape ``(K, N)``.
    tau_Sigma
        The scaled prior covariance ``tau * Sigma`` of shape ``(N, N)``.
    confidences
        One-dimensional array of length ``K`` with entries in ``[0, 1]``.

    Returns
    -------
    numpy.ndarray
        Diagonal ``(K, K)`` matrix of view-uncertainty variances.
    """
    c = np.asarray(confidences, dtype=float)
    _check_confidences(c)
    if P.shape[0] != c.shape[0]:
        raise OmegaSpecificationError(
            f"P has {P.shape[0]} rows but {c.shape[0]} confidences were supplied."
        )

    diag_pkk = np.einsum("ki,ij,kj->k", P, tau_Sigma, P)
    omega_diag = np.empty_like(c)
    for k, ck in enumerate(c):
        if ck >= 1.0 - _TINY:
            omega_diag[k] = 0.0
        elif ck <= _TINY:
            # Effectively no confidence => infinite variance.  We use a very
            # large number so the matrix remains numerically representable.
            omega_diag[k] = _LARGE * max(diag_pkk[k], 1.0)
        else:
            omega_diag[k] = ((1.0 - ck) / ck) * diag_pkk[k]
    return np.diag(omega_diag)

implied_returns(cov: pd.DataFrame, market_weights: pd.Series, delta: float = 2.5) -> pd.Series

Compute the equilibrium-implied excess-return vector.

Parameters:

Name	Type	Description	Default
cov	DataFrame	Asset return covariance matrix as a square DataFrame with matching row and column labels.	required
market_weights	Series	Market-capitalisation weights indexed by asset. The index must align with cov (the function reindexes to the covariance ordering).	required
delta	float	Representative-investor risk-aversion coefficient. Defaults to the commonly cited value of 2.5 from He & Litterman (1999).	2.5

Returns:

Type	Description
Series	Implied excess returns pi = delta * Sigma * w_mkt aligned to the covariance index.

Raises:

Type	Description
ValueError	If the covariance matrix is not square, weights cannot be aligned, or delta is not strictly positive.

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

def implied_returns(
    cov: pd.DataFrame,
    market_weights: pd.Series,
    delta: float = 2.5,
) -> pd.Series:
    """Compute the equilibrium-implied excess-return vector.

    Parameters
    ----------
    cov
        Asset return covariance matrix as a square ``DataFrame`` with matching
        row and column labels.
    market_weights
        Market-capitalisation weights indexed by asset.  The index must align
        with ``cov`` (the function reindexes to the covariance ordering).
    delta
        Representative-investor risk-aversion coefficient.  Defaults to the
        commonly cited value of ``2.5`` from He & Litterman (1999).

    Returns
    -------
    pandas.Series
        Implied excess returns ``pi = delta * Sigma * w_mkt`` aligned to the
        covariance index.

    Raises
    ------
    ValueError
        If the covariance matrix is not square, weights cannot be aligned, or
        ``delta`` is not strictly positive.
    """
    if cov.shape[0] != cov.shape[1]:
        raise ValueError(f"Covariance matrix must be square, got shape {cov.shape}.")
    if not cov.index.equals(cov.columns):
        raise ValueError("Covariance matrix index and columns must be identical.")
    if delta <= 0.0:
        raise ValueError(f"Risk-aversion delta must be > 0, got {delta}.")

    aligned = market_weights.reindex(cov.index)
    if aligned.isna().any():
        missing = aligned.index[aligned.isna()].tolist()
        raise ValueError(f"Market weights missing entries for: {missing}.")

    pi = delta * cov.to_numpy() @ aligned.to_numpy()
    return pd.Series(pi, index=cov.index, name="pi")

infer_delta(market_returns: pd.Series, risk_free_rate: float, ddof: int = 1) -> float

Estimate the risk-aversion coefficient from market-return history.

Uses the standard identity delta = (E[r_m] - rf) / Var[r_m] where r_m is the market portfolio return series. The series is assumed to already be on the same periodicity as risk_free_rate (e.g. both annual or both daily).

Parameters:

Name	Type	Description	Default
market_returns	Series	Realised market returns (not excess).	required
risk_free_rate	float	Periodic risk-free rate matching the periodicity of market_returns.	required
ddof	int	Degrees-of-freedom passed through to :meth:pandas.Series.var. Defaults to 1 (sample variance).	1

Returns:

Type	Description
float	The implied risk-aversion coefficient.

Raises:

Type	Description
ValueError	If the input series has fewer than two observations or has zero variance.

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

def infer_delta(
    market_returns: pd.Series,
    risk_free_rate: float,
    ddof: int = 1,
) -> float:
    """Estimate the risk-aversion coefficient from market-return history.

    Uses the standard identity ``delta = (E[r_m] - rf) / Var[r_m]`` where
    ``r_m`` is the market portfolio return series.  The series is assumed to
    already be on the same periodicity as ``risk_free_rate`` (e.g. both annual
    or both daily).

    Parameters
    ----------
    market_returns
        Realised market returns (not excess).
    risk_free_rate
        Periodic risk-free rate matching the periodicity of
        ``market_returns``.
    ddof
        Degrees-of-freedom passed through to :meth:`pandas.Series.var`.
        Defaults to ``1`` (sample variance).

    Returns
    -------
    float
        The implied risk-aversion coefficient.

    Raises
    ------
    ValueError
        If the input series has fewer than two observations or has zero
        variance.
    """
    if len(market_returns) < 2:
        raise ValueError("At least two market-return observations are required to estimate delta.")
    variance = float(market_returns.var(ddof=ddof))
    if not np.isfinite(variance) or variance <= 0.0:
        raise ValueError(f"Market-return variance must be strictly positive, got {variance}.")
    expected = float(market_returns.mean())
    return (expected - risk_free_rate) / variance