Skip to content

FracDiff (Fractional Differentiation)

Intuição

Séries de preços costumam ser não-estacionárias (tendem a ter raiz unitária), o que dificulta o uso direto em muitos modelos estatísticos e de machine learning.

A diferenciação inteira (d = 1) ajuda a tornar a série mais estacionária, mas pode remover informação de longo prazo. A diferenciação fracionária busca um meio-termo: reduzir a não-estacionariedade preservando mais memory do sinal.

Definição

A diferenciação fracionária de ordem d pode ser escrita como a expansão de:

  • (1 - L)^d x_t = sum_{k=0..∞} w_k x_{t-k}

onde L é o operador de defasagem e os pesos são dados recursivamente por:

  • w_0 = 1
  • w_k = -w_{k-1} * (d - k + 1) / k

Na prática, truncamos a soma quando |w_k| < thresh (ou usando max_lags).

Uso

from quantmaster.features.statistical import fracdiff

df["fracdiff_0.5"] = fracdiff(df, d=0.5, thresh=1e-5)

API

Source code in src/quantmaster/features/statistical.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def fracdiff(
    data: pd.DataFrame | pd.Series,
    *,
    d: float = 0.4,
    thresh: float = 1e-5,
    max_lags: int | None = None,
    price_col: str = "close",
) -> pd.Series:
    try:
        d = float(d)
    except (TypeError, ValueError) as exc:
        raise TypeError(f"d must be float, got {type(d).__name__}") from exc

    if not (0.0 <= d <= 1.0):
        raise ValueError(f"d must be between 0 and 1 (inclusive), got {d}")

    try:
        thresh = float(thresh)
    except (TypeError, ValueError) as exc:
        raise TypeError(f"thresh must be float, got {type(thresh).__name__}") from exc

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

    if max_lags is not None:
        max_lags = validate_positive_int(max_lags, name="max_lags")

    x = get_price_series(data, price_col=price_col).astype(float)

    out_name = f"fracdiff_{d:g}"

    if len(x) < 2:
        out = pd.Series(np.nan, index=x.index, name=out_name, dtype=float)
        if d == 0.0:
            out = x.astype(float).copy()
            out.name = out_name
        return out

    x_arr = x.to_numpy(dtype=float)
    finite = np.isfinite(x_arr)

    y = np.full(x_arr.shape[0], np.nan, dtype=float)

    if finite.any():
        idxs = np.flatnonzero(finite)
        start = int(idxs[0])
        prev = int(idxs[0])
        for i in idxs[1:]:
            i = int(i)
            if i != prev + 1:
                _apply_fracdiff_segment(y, x_arr, start=start, end=prev + 1, d=d, thresh=thresh, max_lags=max_lags)
                start = i
            prev = i
        _apply_fracdiff_segment(y, x_arr, start=start, end=prev + 1, d=d, thresh=thresh, max_lags=max_lags)

    out = pd.Series(y, index=x.index)
    out.name = out_name
    return out