fn_registry/python/functions/datascience/acf_pacf.py

"""Autocorrelacion (ACF) y autocorrelacion parcial (PACF) de una serie (grupo eda).

Funcion pura y determinista que calcula la funcion de autocorrelacion y la
parcial con sus bandas de confianza, mas el test de Ljung-Box de autocorrelacion
global. Motivada por Hyndman ("Forecasting") para identificar el orden de un
modelo ARIMA, y por Lopez de Prado ("Advances in Financial ML"): una serie
autocorrelacionada viola el supuesto IID, de modo que los p-valores de una
regresion OLS estandar sobre ella estan inflados (falsos positivos).
"""

from __future__ import annotations

import math

import numpy as np
from statsmodels.stats.diagnostic import acorr_ljungbox
from statsmodels.tsa.stattools import acf, pacf


def _clean(values: list) -> list[float]:
    """Conserva solo valores numericos finitos, descartando None/NaN/no-numericos.

    Los booleanos se excluyen explicitamente (en Python ``bool`` es subclase de
    ``int``, pero no es un valor de serie temporal valido).
    """
    out: list[float] = []
    for v in values:
        if v is None or isinstance(v, bool):
            continue
        if not isinstance(v, (int, float)):
            continue
        x = float(v)
        if math.isnan(x) or math.isinf(x):
            continue
        out.append(x)
    return out


def acf_pacf(values: list, nlags: int = 40, alpha: float = 0.05) -> dict:
    """Calcula ACF, PACF y el test Ljung-Box de una serie temporal.

    Computa la funcion de autocorrelacion (ACF) y la autocorrelacion parcial
    (PACF) hasta ``nlags`` retardos, con sus bandas de confianza al nivel
    ``1 - alpha``, e identifica que retardos son significativos (cuyo intervalo
    de confianza no contiene 0). Ademas corre el test de **Ljung-Box** sobre el
    conjunto de retardos: H0 = "los datos son independientes" (sin
    autocorrelacion); si ``p < alpha`` se rechaza -> la serie esta
    autocorrelacionada.

    Funcion pura y determinista: no hace I/O, no muta los inputs.

    Args:
        values: serie temporal de valores numericos en orden cronologico.
            None/NaN/infinitos/no-numericos se descartan antes del calculo.
        nlags: numero maximo de retardos a calcular (default 40). Se recorta
            automaticamente a ``n // 2`` para PACF (statsmodels exige
            ``nlags < n/2``) y a ``n - 1`` para ACF.
        alpha: nivel de significancia para las bandas de confianza y para el
            test de Ljung-Box (default 0.05).

    Returns:
        Con menos de 8 puntos validos devuelve
        ``{"n": n, "note": "datos insuficientes", "is_autocorrelated": None}``.

        En otro caso un dict con::

            {
              "n": int,
              "nlags": int,            # retardos efectivamente calculados
              "acf": [float, ...],     # incluye lag 0 (=1.0) en el indice 0
              "pacf": [float, ...],
              "acf_confint": [[low, high], ...],   # banda por lag
              "pacf_confint": [[low, high], ...],
              "significant_acf_lags": [int, ...],  # lags (>=1) significativos
              "significant_pacf_lags": [int, ...],
              "ljung_box": {"stat": float, "p_value": float, "lags": int},
              "is_autocorrelated": bool,  # Ljung-Box rechaza independencia
            }

        ``is_autocorrelated = True`` significa que la serie NO es ruido blanco:
        cuidado al aplicarle inferencia OLS clasica (p-valores inflados).
    """
    clean = _clean(values)
    n = len(clean)

    if n < 8:
        return {"n": n, "note": "datos insuficientes", "is_autocorrelated": None}

    arr = np.asarray(clean, dtype=float)

    # Recorta nlags a los limites de statsmodels: ACF admite hasta n-1, PACF < n/2.
    eff_lags = min(nlags, n - 1, (n // 2) - 1)
    eff_lags = max(eff_lags, 1)

    acf_vals, acf_confint = acf(arr, nlags=eff_lags, alpha=alpha, fft=False)
    pacf_vals, pacf_confint = pacf(arr, nlags=eff_lags, alpha=alpha)

    # Un lag es significativo si su banda de confianza (centrada en el valor) no
    # contiene 0. statsmodels devuelve confint como intervalos centrados en el
    # estimador, asi que comparamos el intervalo desplazado al origen.
    def _significant(vals, confint) -> list[int]:
        out: list[int] = []
        for lag in range(1, len(vals)):
            low = confint[lag][0] - vals[lag]
            high = confint[lag][1] - vals[lag]
            if vals[lag] < low or vals[lag] > high:
                out.append(lag)
        return out

    significant_acf = _significant(acf_vals, acf_confint)
    significant_pacf = _significant(pacf_vals, pacf_confint)

    # Ljung-Box sobre el maximo retardo calculado.
    lb = acorr_ljungbox(arr, lags=[eff_lags], return_df=True)
    lb_stat = float(lb["lb_stat"].iloc[0])
    lb_p = float(lb["lb_pvalue"].iloc[0])
    is_autocorrelated = bool(lb_p < alpha)

    return {
        "n": n,
        "nlags": int(eff_lags),
        "acf": [float(v) for v in acf_vals],
        "pacf": [float(v) for v in pacf_vals],
        "acf_confint": [[float(lo), float(hi)] for lo, hi in acf_confint],
        "pacf_confint": [[float(lo), float(hi)] for lo, hi in pacf_confint],
        "significant_acf_lags": significant_acf,
        "significant_pacf_lags": significant_pacf,
        "ljung_box": {
            "stat": lb_stat,
            "p_value": lb_p,
            "lags": int(eff_lags),
        },
        "is_autocorrelated": is_autocorrelated,
    }