HURST (DFA)

Intuição

O Hurst exponent é um resumo do grau de persistência de uma série temporal:

• H > 0.5: comportamento mais persistente (tendência / “trending”).
• H < 0.5: comportamento mais anti-persistente (reversão à média).

Para dados financeiros com não-estacionaridade e drift, uma forma robusta de estimar H é via DFA (Detrended Fluctuation Analysis).

Definição

A DFA consiste em:

1. (Opcional) trabalhar com log(price).
2. Construir o perfil (soma cumulativa) da série de-meaned dentro de uma janela.
3. Para vários tamanhos de caixa s, ajustar uma tendência linear e medir a flutuação RMS F(s).
4. Estimar H como o coeficiente angular da regressão log(F(s)) ~ H * log(s).

Uso

from quantmaster.features.statistical import hurst_dfa

df["hurst"] = hurst_dfa(df, window=240)

API

Source code in src/quantmaster/features/statistical.py

def hurst_dfa(
    data: pd.DataFrame | pd.Series,
    *,
    window: int = 240,
    price_col: str = "close",
    min_scale: int = 4,
    max_scale: int | None = None,
    n_scales: int = 10,
    log_price: bool = True,
) -> pd.Series:
    window = validate_positive_int(window, name="window")
    min_scale = validate_positive_int(min_scale, name="min_scale")
    n_scales = validate_positive_int(n_scales, name="n_scales")

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

    if max_scale is None:
        max_scale_eff = max(min_scale, window // 4)
    else:
        max_scale_eff = validate_positive_int(max_scale, name="max_scale")

    max_scale_eff = min(max_scale_eff, window // 2)
    if max_scale_eff < min_scale:
        raise ValueError(f"max_scale must be >= min_scale, got max_scale={max_scale_eff}, min_scale={min_scale}")

    x = get_price_series(data, price_col=price_col).astype(float)
    if log_price:
        x = np.log(x.where(x > 0))

    out = pd.Series(np.nan, index=x.index, dtype=float)
    out.name = f"hurst_dfa_{window}"

    if len(x) < window:
        return out

    scales = np.unique(
        np.floor(
            np.logspace(
                np.log10(float(min_scale)),
                np.log10(float(max_scale_eff)),
                int(n_scales),
            )
        ).astype(int)
    )
    scales = scales[(scales >= min_scale) & (scales <= max_scale_eff)]
    if scales.size < 2:
        return out

    x_arr = x.to_numpy(dtype=float)
    windows = np.lib.stride_tricks.sliding_window_view(x_arr, window_shape=window)

    mean_win = np.nanmean(windows, axis=1, keepdims=True)
    prof = np.nancumsum(windows - mean_win, axis=1)

    log_s = np.log(scales.astype(float))
    f_mat = np.full((prof.shape[0], scales.size), np.nan, dtype=float)

    for j, s in enumerate(scales.tolist()):
        m = window // int(s)
        if m < 2:
            continue

        y = prof[:, : m * s].reshape(prof.shape[0], m, s)

        t = np.arange(s, dtype=float)
        t_mean = t.mean()
        t_centered = t - t_mean
        denom = float(np.sum(t_centered**2))
        if denom == 0.0:
            continue

        y_mean = np.nanmean(y, axis=2, keepdims=True)
        slope = np.nansum(t_centered[None, None, :] * (y - y_mean), axis=2, keepdims=True) / denom
        intercept = y_mean - slope * t_mean

        trend = slope * t[None, None, :] + intercept
        res = y - trend
        f_mat[:, j] = np.sqrt(np.nanmean(res**2, axis=(1, 2)))

    log_f = np.log(f_mat)
    valid_scale = np.isfinite(log_s)
    valid_scale &= np.isfinite(log_f).all(axis=0)

    if valid_scale.sum() < 2:
        return out

    xs = log_s[valid_scale]
    ys = log_f[:, valid_scale]

    x_mean = xs.mean()
    x_centered = xs - x_mean
    x_var = float(np.sum(x_centered**2))
    if x_var == 0.0:
        return out

    y_mean = ys.mean(axis=1, keepdims=True)
    cov = np.sum((ys - y_mean) * x_centered[None, :], axis=1)
    hurst = cov / x_var

    out.iloc[window - 1 :] = hurst
    return out