feat(eda): capítulo TIMESERIES del AutomaticEDA (evolución + análisis de serie)
Capítulo nuevo build_timeseries(profile, ctx) -> Chapter|None del motor AutomaticEDA. Cuando la tabla tiene columna de fecha/datetime, grafica la evolución de cada columna numérica por periodo (valor agregado + conteo de filas) y los paneles de descomposición STL y autocorrelación (ACF), con el análisis de la serie: estacionariedad (ADF+KPSS), autocorrelación (Ljung-Box), fuerzas de tendencia/estacionalidad (Hyndman) y la transformación sugerida (retornos o diferencias) para evitar correlaciones espurias. Sin columna temporal devuelve None. Consolida series OHLC casi idénticas en un único gráfico conservando el análisis de cada columna. La serie cruda llega por ctx['timeseries_raw'] (mismo patrón que modelos con raw_numeric); las figuras son perezosas (Figure.make) y el paginador del núcleo garantiza no-corte en PDF y PPTX. CHAPTER_VERSION 1.0.0. Cubre los MUST del diseño (report 2043): MUST-9.1 (línea valor-vs-tiempo + conteo por periodo), MUST-9.2 (paneles STL + ACF), MUST-9.3 (perfil datetime + consolidación OHLC). Funciones nuevas del registry (grupo eda), delegadas a fn-constructor, no inline: - detect_time_column (pure): detecta la columna temporal y las numéricas - profile_datetime (pure): rango/frecuencia/regularidad/huecos de la fecha - resample_timeseries (pure): agrega la serie por periodo + conteo - extract_timeseries_raw (impure): lee la serie cruda ordenada de DuckDB/PG Verificación: 69 tests verdes (capítulo 9 + funciones 28 + núcleo/renderers); golden real sobre seattle-weather (estacional) y aapl (OHLC) con PDF+PPTX sin cortar nada (cols_cortadas=[]). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,275 @@
|
||||
"""Agrega una serie temporal por periodo para el capitulo TIMESERIES (grupo eda).
|
||||
|
||||
Funcion pura y determinista: recibe las fechas y los valores YA leidos (nunca
|
||||
toca una base de datos ni hace I/O) y los agrega por bucket temporal para poder
|
||||
graficar la evolucion de la serie y, en paralelo, el CONTEO de observaciones por
|
||||
periodo (densidad temporal).
|
||||
|
||||
Estilo "dict-no-throw" del grupo eda: NUNCA lanza excepcion, siempre devuelve el
|
||||
mismo conjunto de claves. Lectura y parseo de fechas 100% defensivos. Solo usa la
|
||||
libreria estandar (``datetime``, ``statistics``, ``re``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import re
|
||||
import statistics
|
||||
|
||||
# Frecuencias soportadas, de mas fina a mas gruesa.
|
||||
_FREQS = ("daily", "weekly", "monthly", "quarterly", "yearly")
|
||||
|
||||
# Agregaciones soportadas.
|
||||
_AGGS = ("mean", "sum", "median", "last", "min", "max")
|
||||
|
||||
# Acepta el inicio de una fecha ISO con cualquier separador posterior
|
||||
# (incluido un caracter raro entre la fecha y la hora).
|
||||
_DATE_RE = re.compile(r"(\d{4})-(\d{2})-(\d{2})")
|
||||
|
||||
|
||||
def _to_date(x) -> "datetime.date | None":
|
||||
"""Parsea defensivamente un valor a ``datetime.date``; devuelve None si falla."""
|
||||
if x is None:
|
||||
return None
|
||||
# datetime es subclase de date: comprobarlo primero.
|
||||
if isinstance(x, datetime.datetime):
|
||||
return x.date()
|
||||
if isinstance(x, datetime.date):
|
||||
return x
|
||||
s = str(x).strip()
|
||||
if not s:
|
||||
return None
|
||||
# Camino feliz: ISO completo (con o sin hora, con o sin 'Z' final).
|
||||
try:
|
||||
s2 = s[:-1] if s.endswith("Z") else s
|
||||
return datetime.datetime.fromisoformat(s2).date()
|
||||
except ValueError:
|
||||
pass
|
||||
# Fallback robusto: extrae el prefijo YYYY-MM-DD con cualquier separador.
|
||||
m = _DATE_RE.match(s)
|
||||
if m:
|
||||
try:
|
||||
return datetime.date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
||||
except ValueError:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def _to_number(x) -> "float | None":
|
||||
"""Convierte a float si es numerico finito; devuelve None en otro caso."""
|
||||
if x is None:
|
||||
return None
|
||||
if isinstance(x, bool):
|
||||
# bool es subclase de int: lo tratamos como no-numerico para una serie.
|
||||
return None
|
||||
try:
|
||||
f = float(x)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
# Descarta NaN / Inf (no agregables de forma estable).
|
||||
if f != f or f in (float("inf"), float("-inf")):
|
||||
return None
|
||||
return f
|
||||
|
||||
|
||||
def _infer_freq(dates_sorted: list) -> str:
|
||||
"""Infiere la frecuencia desde el delta mediano (en dias) entre fechas."""
|
||||
if len(dates_sorted) < 2:
|
||||
return "daily"
|
||||
diffs = [
|
||||
(dates_sorted[i + 1] - dates_sorted[i]).days
|
||||
for i in range(len(dates_sorted) - 1)
|
||||
]
|
||||
diffs = [d for d in diffs if d > 0] # ignora duplicados del mismo dia
|
||||
if not diffs:
|
||||
return "daily"
|
||||
med = statistics.median(diffs)
|
||||
if med <= 3:
|
||||
return "daily"
|
||||
if med <= 16:
|
||||
return "weekly"
|
||||
if med <= 75:
|
||||
return "monthly"
|
||||
if med <= 200:
|
||||
return "quarterly"
|
||||
return "yearly"
|
||||
|
||||
|
||||
def _bucket_start(d: "datetime.date", freq: str) -> "datetime.date":
|
||||
"""Trunca una fecha al inicio de su bucket segun la frecuencia."""
|
||||
if freq == "weekly":
|
||||
return d - datetime.timedelta(days=d.weekday()) # lunes ISO
|
||||
if freq == "monthly":
|
||||
return datetime.date(d.year, d.month, 1)
|
||||
if freq == "quarterly":
|
||||
first_month = ((d.month - 1) // 3) * 3 + 1
|
||||
return datetime.date(d.year, first_month, 1)
|
||||
if freq == "yearly":
|
||||
return datetime.date(d.year, 1, 1)
|
||||
return d # daily (o cualquier otra cosa): la propia fecha
|
||||
|
||||
|
||||
def _downsample_indices(n: int, max_points: int) -> list:
|
||||
"""Indices equiespaciados conservando primero y ultimo (<= max_points)."""
|
||||
if max_points <= 0 or max_points >= n:
|
||||
return list(range(n))
|
||||
if max_points == 1:
|
||||
return [0]
|
||||
idx = sorted({round(i * (n - 1) / (max_points - 1)) for i in range(max_points)})
|
||||
return idx
|
||||
|
||||
|
||||
def _empty(freq_req: str, agg: str) -> dict:
|
||||
"""Resultado canonico cuando no hay datos suficientes."""
|
||||
eff_freq = freq_req if freq_req in _FREQS else "auto"
|
||||
return {
|
||||
"t": [],
|
||||
"v": [],
|
||||
"count": [],
|
||||
"freq": eff_freq,
|
||||
"agg": agg if agg in _AGGS else "mean",
|
||||
"n_in": 0,
|
||||
"n_buckets": 0,
|
||||
"downsampled": False,
|
||||
"note": "datos insuficientes",
|
||||
}
|
||||
|
||||
|
||||
def resample_timeseries(
|
||||
t: list,
|
||||
v: list,
|
||||
freq: str = "auto",
|
||||
agg: str = "mean",
|
||||
max_points: int = 400,
|
||||
) -> dict:
|
||||
"""Agrega una serie temporal por periodo (buckets) para graficarla.
|
||||
|
||||
Empareja ``t[i]`` con ``v[i]`` por indice, descarta los pares cuya fecha no
|
||||
parsea, trunca cada fecha al inicio de su bucket segun ``freq`` y agrupa. Por
|
||||
cada bucket devuelve el valor agregado (``agg`` sobre los valores numericos
|
||||
validos) y el CONTEO de observaciones con fecha valida (densidad temporal),
|
||||
independientemente de si su valor numerico es ``None``.
|
||||
|
||||
Funcion pura: no hace I/O, no muta los inputs, es determinista, NUNCA lanza.
|
||||
|
||||
Args:
|
||||
t: lista de fechas paralela a ``v``. Acepta strings ISO
|
||||
(``"YYYY-MM-DD"`` o ``"YYYY-MM-DDTHH:MM:SS"``, con ``Z`` opcional),
|
||||
``datetime.date`` o ``datetime.datetime``. Se parsea defensivamente;
|
||||
las fechas que no parsean se descartan junto con su valor.
|
||||
v: lista de valores numericos (float/int). Puede contener ``None`` o
|
||||
valores no numericos: estos se ignoran en la agregacion, pero la fila
|
||||
sigue contando en ``count`` (siempre que su fecha sea valida).
|
||||
freq: ``"auto"`` (infiere del delta mediano entre fechas) o uno de
|
||||
``"daily"``, ``"weekly"``, ``"monthly"``, ``"quarterly"``,
|
||||
``"yearly"``. Una frecuencia desconocida cae a ``"auto"``.
|
||||
agg: agregacion por bucket: ``"mean"``, ``"sum"``, ``"median"``,
|
||||
``"last"`` (valor de la observacion cronologicamente mas reciente),
|
||||
``"min"`` o ``"max"``. Una agregacion desconocida cae a ``"mean"``.
|
||||
max_points: si tras agregar hay mas buckets que este limite, se hace
|
||||
downsampling uniforme (1 de cada k buckets equiespaciados,
|
||||
conservando el primero y el ultimo) para no saturar el grafico.
|
||||
|
||||
Returns:
|
||||
Siempre un dict con las mismas claves::
|
||||
|
||||
{
|
||||
"t": [str, ...], # etiqueta ISO YYYY-MM-DD de cada bucket
|
||||
"v": [float|None, ...], # valor agregado por bucket (None si vacio)
|
||||
"count": [int, ...], # nº de observaciones con fecha valida
|
||||
"freq": str, # frecuencia efectivamente usada
|
||||
"agg": str, # agregacion usada
|
||||
"n_in": int, # nº de pares (t,v) con fecha valida
|
||||
"n_buckets": int, # nº de buckets antes del downsample
|
||||
"downsampled": bool, # True si se aplico downsampling
|
||||
"note": str, # "" o nota (p.ej. "datos insuficientes")
|
||||
}
|
||||
"""
|
||||
agg = agg if agg in _AGGS else "mean"
|
||||
freq_req = freq if isinstance(freq, str) else "auto"
|
||||
|
||||
# Validacion de entrada: deben ser listas de igual longitud y no vacias.
|
||||
if (
|
||||
not isinstance(t, list)
|
||||
or not isinstance(v, list)
|
||||
or len(t) == 0
|
||||
or len(t) != len(v)
|
||||
):
|
||||
return _empty(freq_req, agg)
|
||||
|
||||
# Empareja por indice y descarta fechas no parseables.
|
||||
parsed: list = [] # (date, original_index, number_or_None)
|
||||
for i, (ti, vi) in enumerate(zip(t, v)):
|
||||
d = _to_date(ti)
|
||||
if d is None:
|
||||
continue
|
||||
parsed.append((d, i, _to_number(vi)))
|
||||
|
||||
n_in = len(parsed)
|
||||
if n_in == 0:
|
||||
return _empty(freq_req, agg)
|
||||
|
||||
# Resuelve la frecuencia efectiva.
|
||||
if freq_req in _FREQS:
|
||||
eff_freq = freq_req
|
||||
else:
|
||||
dates_sorted = sorted(d for d, _, _ in parsed)
|
||||
eff_freq = _infer_freq(dates_sorted)
|
||||
|
||||
# Agrupa por bucket.
|
||||
buckets: dict = {}
|
||||
for d, idx, num in parsed:
|
||||
b = _bucket_start(d, eff_freq)
|
||||
slot = buckets.get(b)
|
||||
if slot is None:
|
||||
slot = {"count": 0, "vals": [], "last_key": None, "last_val": None}
|
||||
buckets[b] = slot
|
||||
slot["count"] += 1
|
||||
if num is not None:
|
||||
slot["vals"].append(num)
|
||||
key = (d, idx)
|
||||
if slot["last_key"] is None or key > slot["last_key"]:
|
||||
slot["last_key"] = key
|
||||
slot["last_val"] = num
|
||||
|
||||
ordered = sorted(buckets.items(), key=lambda kv: kv[0])
|
||||
n_buckets = len(ordered)
|
||||
|
||||
def _aggregate(vals: list, last_val) -> "float | None":
|
||||
if not vals:
|
||||
return None
|
||||
if agg == "sum":
|
||||
return float(sum(vals))
|
||||
if agg == "median":
|
||||
return float(statistics.median(vals))
|
||||
if agg == "last":
|
||||
return float(last_val) if last_val is not None else None
|
||||
if agg == "min":
|
||||
return float(min(vals))
|
||||
if agg == "max":
|
||||
return float(max(vals))
|
||||
return float(statistics.fmean(vals)) # mean (default)
|
||||
|
||||
t_out = [b.isoformat() for b, _ in ordered]
|
||||
v_out = [_aggregate(s["vals"], s["last_val"]) for _, s in ordered]
|
||||
c_out = [s["count"] for _, s in ordered]
|
||||
|
||||
downsampled = False
|
||||
if n_buckets > max_points > 0:
|
||||
keep = _downsample_indices(n_buckets, max_points)
|
||||
t_out = [t_out[i] for i in keep]
|
||||
v_out = [v_out[i] for i in keep]
|
||||
c_out = [c_out[i] for i in keep]
|
||||
downsampled = True
|
||||
|
||||
return {
|
||||
"t": t_out,
|
||||
"v": v_out,
|
||||
"count": c_out,
|
||||
"freq": eff_freq,
|
||||
"agg": agg,
|
||||
"n_in": n_in,
|
||||
"n_buckets": n_buckets,
|
||||
"downsampled": downsampled,
|
||||
"note": "",
|
||||
}
|
||||
Reference in New Issue
Block a user