quantlab.portfolio.predefined.ledoit_wolf.LedoitWolfEstimator#

class quantlab.portfolio.predefined.ledoit_wolf.LedoitWolfEstimator(config)[source]#

Bases: CovarianceEstimator

Ledoit-Wolf shrunk covariance of the trailing one-bar returns.

A symbol is covered when every one of the lookback_bars returns in the context’s window is finite, they are not all equal, and its staleness at the bar is at most max_stale_bars (when the context carries one); the others have too little history, no measurable risk or no recent price, and are left out of the estimate. A halt inside the window shows as zero returns and then the gap, so a short one keeps the symbol covered. The estimate is the covered symbols’ sample covariance (of the demeaned window, divided by T) shrunk toward a scaled identity with the Ledoit-Wolf (2004) coefficient (sklearn.covariance.ledoit_wolf), converted to correlations C and scaled back by volatilities D, the given ones where volatility is passed (a symbol without a finite positive one is left out), else the shrunk covariance’s own. It is D C D, of one-bar returns, returned as a LowRankCovarianceEstimate: the shrunk covariance is U U' + a I with U the demeaned window scaled by sqrt((1 - s) / T), so D C D is B B' + diag(D a / sd^2) with B = D U / sd, rank lookback_bars, and an optimiser prices risk from factor_form() without the dense matrix once it has LOW_RANK_MIN_RATIO times as many symbols as bars (covariance builds the matrix on request).

Parameters:

configLedoitWolfEstimatorConfig: lookback_bars, at least 2, and max_stale_bars, at least 0.

Raises:

ValueError: If lookback_bars is below 2 or max_stale_bars is negative.

Examples

>>> import numpy as np, pandas as pd, xarray as xr
>>> from quantlab.portfolio.config import LedoitWolfEstimatorConfig
>>> from quantlab.portfolio.base import PortfolioContext
>>> rng = np.random.default_rng(0)
>>> symbols = ["AAA", "BBB", "CCC"]
>>> window = rng.normal(0.0, 0.01, size=(60, 3))
>>> window[:5, 2] = np.nan  # CCC listed five bars into the window
>>> context = PortfolioContext(
...     timestamp=pd.Timestamp("2024-03-25"),
...     predictions=xr.Dataset(coords={"symbol": symbols}),
...     tradable=xr.DataArray([True] * 3, dims="symbol", coords={"symbol": symbols}),
...     current_weights=xr.DataArray(np.zeros(3), dims="symbol", coords={"symbol": symbols}),
...     returns=xr.DataArray(window, dims=("timestamp", "symbol"), coords={
...         "timestamp": pd.bdate_range("2024-01-01", periods=60), "symbol": symbols}),
... )
>>> risk = LedoitWolfEstimator(LedoitWolfEstimatorConfig(lookback_bars=60))
>>> estimate = risk.estimate(context)
>>> estimate.symbols.tolist(), estimate.covariance.shape
(['AAA', 'BBB'], (2, 2))
>>> given = xr.DataArray([0.02, 0.03, 0.04], dims="symbol", coords={"symbol": symbols})
>>> np.sqrt(risk.estimate(context, volatility=given).variance).round(6)
array([0.02, 0.03])

A backtest also hands each symbol’s staleness; BBB, 7 bars without a price, exceeds the default max_stale_bars of 5:

>>> import dataclasses
>>> stale = dataclasses.replace(context, staleness=xr.DataArray(
...     [0.0, 7.0, 0.0], dims="symbol", coords={"symbol": symbols}))
>>> risk.estimate(stale).symbols.tolist()
['AAA']

config_cls#: alias of LedoitWolfEstimatorConfig

declared_inputs()[source]#

A lookback_bars return window from lookback_bars + 1 + max_stale_bars prices.

A symbol is covered with a staleness of at most max_stale_bars, so its last real price before such a halt still falls in the prices read, and its return window is forward-filled as with an unbounded history.

Examples

>>> risk.declared_inputs().history_bars  # 60 + 1 + 5
66

estimate(context, volatility=None)[source]#

Estimate the covariance of one-bar returns from the context’s return window.

See the class docstring for the rule and an example.

Parameters:

contextPortfolioContext: The bar’s context; the last lookback_bars rows of its returns window are the history.
volatilityxr.DataArray, optional: Per-symbol one-bar volatilities on symbol, replacing the historical ones.

Returns:

LowRankCovarianceEstimate: The covariance over the covered symbols, symmetric and positive definite.

Examples

>>> estimate = risk.estimate(context)
>>> bool(np.allclose(estimate.covariance, estimate.covariance.T))
True
>>> bool((np.linalg.eigvalsh(estimate.covariance) > 0).all())
True

pools_symbols = True#: The shrinkage is estimated over every covered symbol of the context.