fn_registry/python/functions/datascience/profile_datetime.py

"""Perfil minimo de una columna fecha/datetime para la cabecera TIMESERIES (grupo eda).

Funcion pura y determinista que resume una columna temporal: rango (min/max),
numero de fechas distintas, frecuencia inferida (daily/weekly/monthly/quarterly/
yearly/irregular), regularidad de los pasos, huecos respecto a la rejilla inferida
y paso mediano entre fechas consecutivas. Cierra el `datetime{}=None` que hoy deja
pendiente el TableProfile de AutomaticEDA.

Acepta valores heterogeneos (``datetime.date``, ``datetime.datetime`` y strings
ISO como ``"2021-06-28"``, ``"2021-06-28T00:00:00"`` o ``"2021-06-28 12:00:00"``),
parsea de forma defensiva, ignora lo que no se puede parsear y NUNCA lanza.

Solo usa stdlib (``datetime`` + ``statistics``).
"""

from __future__ import annotations

import statistics
from datetime import date, datetime


def _parse_one(v) -> datetime | None:
    """Parsea un valor a ``datetime`` naive, o devuelve None si no es una fecha.

    Acepta ``datetime.datetime``, ``datetime.date`` y strings ISO. Cualquier
    datetime con zona horaria se normaliza a naive (se descarta el tzinfo) para
    poder mezclarlo con fechas naive sin que las restas lancen ``TypeError``.
    """
    if v is None or isinstance(v, bool):
        return None
    # datetime es subclase de date: comprobar datetime primero.
    if isinstance(v, datetime):
        return v.replace(tzinfo=None)
    if isinstance(v, date):
        return datetime(v.year, v.month, v.day)
    if isinstance(v, str):
        s = v.strip()
        if not s:
            return None
        try:
            dt = datetime.fromisoformat(s)
        except ValueError:
            return None
        return dt.replace(tzinfo=None)
    return None


def _infer_freq(median_step_days: float) -> str:
    """Clasifica la frecuencia a partir del paso mediano (en dias) entre fechas.

    Bandas con tolerancia: ~1 dia -> daily, ~7 -> weekly, 28-31 -> monthly,
    89-92 -> quarterly, 360-366 -> yearly. Cualquier paso fuera de las bandas
    (incluido sub-diario) -> irregular.
    """
    m = median_step_days
    if 0.5 <= m <= 1.5:
        return "daily"
    if 6.0 <= m <= 8.0:
        return "weekly"
    if 28.0 <= m <= 31.0:
        return "monthly"
    if 89.0 <= m <= 92.0:
        return "quarterly"
    if 360.0 <= m <= 366.0:
        return "yearly"
    return "irregular"


def profile_datetime(values: list) -> dict:
    """Perfila una columna de fechas para la cabecera del capitulo TIMESERIES.

    Funcion pura y determinista: no hace I/O, no muta el input y nunca lanza.

    El analisis de frecuencia, regularidad y huecos se hace sobre las **fechas
    distintas ordenadas** (se deduplica antes de calcular los pasos): los valores
    repetidos generarian pasos de 0 dias que distorsionarian el mediano y la
    inferencia. ``n`` cuenta los valores parseables (con duplicados) y
    ``n_distinct`` las fechas unicas.

    Args:
        values: lista de valores fecha. Acepta ``datetime.date``,
            ``datetime.datetime`` y strings ISO (``"2021-06-28"``,
            ``"2021-06-28T00:00:00"``, ``"2021-06-28 12:00:00"``). Los valores
            None, vacios o no parseables se ignoran. Si ``values`` es None o no
            iterable se trata como lista vacia.

    Returns:
        Siempre un dict con esta forma::

            {
              "min": str | None,             # fecha minima ISO date (YYYY-MM-DD)
              "max": str | None,             # fecha maxima ISO date
              "n": int,                      # nº de valores fecha parseables
              "n_distinct": int,             # nº de fechas distintas
              "span_days": float | None,     # (max - min) en dias
              "freq": str,                   # daily|weekly|monthly|quarterly|
                                             #   yearly|irregular|unknown
              "is_regular": bool,            # pasos ~constantes (tolerancia ±25%)
              "n_gaps": int,                 # saltos > ~1.5x el paso mediano
              "median_step_days": float | None,  # paso mediano entre fechas
              "note": str                    # "" o nota corta
            }

        Con menos de 2 valores parseables (o una sola fecha distinta) devuelve
        ``freq="unknown"``, ``is_regular=False``, ``n_gaps=0``,
        ``median_step_days=None`` y la nota correspondiente, manteniendo min/max
        y span_days coherentes cuando hay al menos una fecha.
    """
    base = {
        "min": None,
        "max": None,
        "n": 0,
        "n_distinct": 0,
        "span_days": None,
        "freq": "unknown",
        "is_regular": False,
        "n_gaps": 0,
        "median_step_days": None,
        "note": "",
    }

    if values is None:
        values = []
    try:
        iterator = list(values)
    except TypeError:
        iterator = []

    parsed: list[datetime] = []
    for v in iterator:
        dt = _parse_one(v)
        if dt is not None:
            parsed.append(dt)

    n = len(parsed)
    base["n"] = n

    if n == 0:
        base["note"] = "datos insuficientes"
        return base

    distinct = sorted(set(parsed))
    n_distinct = len(distinct)
    dt_min = min(parsed)
    dt_max = max(parsed)

    base["n_distinct"] = n_distinct
    base["min"] = dt_min.date().isoformat()
    base["max"] = dt_max.date().isoformat()
    base["span_days"] = round((dt_max - dt_min).total_seconds() / 86400.0, 6)

    # Sin al menos dos fechas distintas no hay pasos que medir.
    if n_distinct < 2:
        base["note"] = "datos insuficientes" if n < 2 else "una sola fecha distinta"
        return base

    steps = [
        (distinct[i + 1] - distinct[i]).total_seconds() / 86400.0
        for i in range(n_distinct - 1)
    ]
    median_step = float(statistics.median(steps))
    base["median_step_days"] = round(median_step, 6)

    freq = _infer_freq(median_step)
    base["freq"] = freq

    # Regularidad: >=80% de los pasos dentro de ±25% del paso mediano.
    if median_step > 0:
        tol = 0.25 * median_step
        within = sum(1 for s in steps if abs(s - median_step) <= tol)
        base["is_regular"] = (within / len(steps)) >= 0.8
    else:
        base["is_regular"] = False

    # Huecos: pasos que superan ~1.5x el mediano. Solo tiene sentido cuando la
    # frecuencia es una rejilla regular conocida (no irregular/unknown).
    if freq not in ("unknown", "irregular") and median_step > 0:
        threshold = 1.5 * median_step
        base["n_gaps"] = sum(1 for s in steps if s > threshold)
    else:
        base["n_gaps"] = 0

    return base