Signal¶

def moving_absolute_deviation(x: pl.Expr, com: int = 32) -> pl.Expr:
    """Compute the rolling median absolute deviation (MAD) of log returns.

    A robust alternative to moving standard deviation, less sensitive to outliers.
    Both the center and dispersion use rolling medians, making the estimate doubly
    robust. The result is scaled by 1/0.6745 to be a consistent estimator of std
    under normality.

    Args:
        x: Polars expression representing the price series.
        com: Center of mass used to derive the rolling window as ``window = 2 * com - 1``.

    Returns:
        Polars expression of scaled rolling MAD values consistent with std under normality.
    """
    window = 2 * com - 1
    r = x.log(base=math.e).diff()
    rolling_median = r.rolling_median(window_size=window)
    return (r - rolling_median).abs().rolling_median(window_size=window) / 0.6745

def shrink2id(matrix: np.ndarray, lamb: float = 1.0) -> np.ndarray:
    """Shrink a square matrix towards the identity matrix by a weight factor.

    Args:
        matrix: The input square matrix to be shrunk.
        lamb: Mixing ratio for shrinkage. A value of 1.0 retains the original
            matrix; 0.0 replaces it entirely with the identity matrix. Default is 1.0.

    Returns:
        The resulting matrix after applying the shrinkage transformation.
    """
    return matrix * lamb + (1 - lamb) * np.eye(N=matrix.shape[0])