Compare commits

..

1 Commits

Author SHA1 Message Date
egutierrez 7fa19d65db feat(eda): capítulo MISSINGNESS — patrones de datos faltantes (co-ocurrencia + MCAR/MAR)
Añade el capítulo `missingness` al motor AutomaticEDA, complemento natural de
`calidad`: donde calidad reporta cuánto falta por columna, este capítulo analiza
el PATRÓN de los nulos — dónde faltan y si las columnas faltan juntas
(co-ocurrencia de ausencias), la señal que distingue MCAR de MAR antes de imputar.

Capítulo (`chapters/missingness.py`), registrado en `chapters_registry.py` justo
tras `calidad`:
- Resumen global: % de celdas faltantes, columnas con nulos, filas completas vs
  incompletas.
- Ranking por columna (tabla + barras horizontales).
- Co-ocurrencia: correlación de las máscaras is-null entre columnas (heatmap +
  tabla de los pares que co-faltan, con co-faltantes y Jaccard).
- Patrones de fila más frecuentes (estilo matriz de missingno).
- Lectura MCAR/MAR exploratoria (heurística por correlación/solape de ausencias,
  no confirmatoria), que cita la evidencia concreta.
- Términos de glosario clicables: missingness, MCAR, MAR.

La máscara is-null por fila de TODAS las columnas (numéricas y categóricas) se
construye con un push-down DuckDB sobre ctx['db_path']/table (mismo patrón que el
capítulo agregación), con fallback a ctx['raw_numeric'] cuando no hay BD. Activa
solo si la tabla tiene nulos; si no, devuelve None.

Funciones nuevas del grupo `eda` (dominio datascience):
- extract_null_mask (impura): máscara is-null por fila vía query_fn.
- missingness_overview (pura): resumen global + filas completas/incompletas.
- missingness_correlation (pura): correlación de ausencias + pares + Jaccard,
  reutiliza pearson.
- missingness_row_patterns (pura): patrones de fila más comunes.
- missingness_corr_heatmap_figure / missingness_rank_bar_figure (impuras): figuras.

Verificado: EDA de titanic genera el capítulo en PDF + PPTX + MD con Cabin 77.1%,
Age 19.9% y la co-ocurrencia Age↔Cabin (158 filas). Suite completa de AutomaticEDA
+ render_automatic_eda en verde (125 passed); tests por función y por capítulo;
fn index sin error.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 20:38:39 +02:00
46 changed files with 2624 additions and 2880 deletions
@@ -0,0 +1,594 @@
"""Missingness chapter (MISSINGNESS) — patterns of missing data.
Complements the CALIDAD chapter: where CALIDAD reports *how much* is missing per
column (the null percentage that lowers the completeness score), this chapter
reports the **pattern** of the missing data — whether columns tend to be missing
*together* (co-occurrence of absences) or independently. That distinction is what
separates data that is missing completely at random ([[term:mcar]]MCAR[[/term]])
from data missing as a function of another variable ([[term:mar]]MAR[[/term]]),
which is the key question to settle before imputing or modelling.
The chapter activates only when the table actually has missing data (at least one
column with a null in the aggregated profile); otherwise it returns ``None`` and
disappears from the document.
Sections, in order:
1. **Resumen global** — % of missing cells in the dataset, number of columns with
nulls, and complete rows (no missing) vs incomplete rows (≥1 missing).
2. **Ranking por columna** — columns sorted by their null percentage, with a
horizontal bar figure.
3. **Co-ocurrencia de ausencias** — the correlation of the binary is-null masks
between columns (which columns tend to be missing together): a heatmap plus a
table of the top column pairs that co-miss.
4. **Patrones de fila** — the most frequent "which columns are missing together"
row patterns, in the style of missingno's pattern matrix.
5. **Lectura MCAR/MAR** — an interpretive, *exploratory* note (not a confirmatory
test such as Little's) reading the absence correlations as a hint of MCAR
(independent absences) vs MAR (co-occurring absences).
The aggregate per-column null counts come from the ``eda`` group ``TableProfile``
(``columns[i]['null_count'] / 'null_pct'`` and the table-level ``null_cell_pct``).
The per-row is-null mask needed for co-occurrence is built from raw data: a single
DuckDB push-down over ``ctx['db_path'] / ctx['table']`` (same pattern as the
AGREGACION chapter) covering ALL columns, with a fallback to the numeric-only
``ctx['raw_numeric']`` when no database is reachable. All the heavy lifting is
delegated to pure registry functions (``missingness_overview``,
``missingness_correlation``, ``missingness_row_patterns``) and two figure helpers
(``missingness_rank_bar_figure``, ``missingness_corr_heatmap_figure``); every one
is imported lazily and degrades to an honest note so this chapter never raises.
Contract: build_<id>(profile, ctx) -> Chapter | None ; CHAPTER_VERSION = "x.y.z".
"""
from __future__ import annotations
from .. import model
CHAPTER_VERSION = "1.0.0"
CHAPTER_ID = "missingness"
CHAPTER_TITLE = "Datos faltantes"
# Sample cap for the per-row is-null mask push-down. Co-occurrence and row
# patterns are computed on this sample; the global % of missing cells and the
# per-column ranking come from the (exact) aggregated profile instead.
MASK_SAMPLE = 5000
# Thresholds for the MCAR/MAR heuristic note. A pair counts as a *strong*
# co-occurrence when the absence correlation alone is high; as a *partial*
# co-occurrence when the absences overlap materially (high Jaccard) even if the
# Pearson correlation is modest — the usual case when one column is missing far
# more often than the other (e.g. Cabin 77% vs Age 20% in Titanic), which dilutes
# the correlation while the rows still co-miss in absolute terms.
_CORR_STRONG = 0.30
_JACCARD_NOTABLE = 0.20
# Rows shown in the top-pairs and row-patterns tables (bounded, never silently
# truncated: the table note reports the full count).
_TOP_PAIRS = 12
_TOP_PATTERNS = 12
# Truncate long column names in tables (the renderer also wraps).
_LABEL_MAX = 28
# Glossary terms this chapter explains (contract §11.1). Registered in the shared
# collector and marked clickable on their first appearance.
_TERMS = {
"missingness": (
"Patrón de datos faltantes (missingness)",
"El patrón con el que faltan los datos: cuánto falta, en qué columnas y "
"si las ausencias de unas columnas coinciden (co-ocurren) con las de "
"otras. Analizarlo —no solo contar nulos— distingue datos que faltan al "
"azar (MCAR) de los que faltan en función de otra variable (MAR), lo que "
"decide cómo imputar o si descartar filas sin sesgar el análisis.",
),
"mcar": (
"MCAR (Missing Completely At Random)",
"Los valores faltan de forma independiente de cualquier dato, observado o "
"no: las ausencias de unas columnas no se relacionan entre sí ni con los "
"valores. Es el caso más benigno —descartar filas o imputar la media no "
"introduce sesgo—, pero rara vez se cumple del todo en datos reales.",
),
"mar": (
"MAR (Missing At Random)",
"La probabilidad de que un valor falte depende de OTRAS variables "
"observadas (p. ej. una medición que falta más en cierto grupo). Las "
"ausencias co-ocurren entre columnas o se relacionan con los valores de "
"otras; imputar exige condicionar en esas variables para no sesgar. La "
"co-ocurrencia fuerte de ausencias es un indicio (exploratorio) de MAR.",
),
}
# --------------------------------------------------------------------------- #
# Small defensive formatters (own copy: the chapter never imports siblings).
# --------------------------------------------------------------------------- #
def _fmt_int(value) -> str:
if value is None:
return ""
try:
return f"{int(round(float(value))):,}".replace(",", ".")
except (TypeError, ValueError):
return model._safe_str(value)
def _fmt_pct(value, decimals: int = 1) -> str:
"""Format an already-0-100 value as a percentage. None -> placeholder."""
if value is None:
return ""
try:
return f"{float(value):.{decimals}f}%"
except (TypeError, ValueError):
return model._safe_str(value)
def _fmt_num(value, decimals: int = 3) -> str:
if value is None:
return ""
try:
f = float(value)
except (TypeError, ValueError):
return model._safe_str(value)
if f != f: # NaN
return ""
text = f"{f:.{decimals}f}".rstrip("0").rstrip(".")
return text if text else "0"
def _truncate(text, limit: int = _LABEL_MAX) -> str:
s = model._safe_str(text)
if len(s) <= limit:
return s
return s[: max(1, limit - 1)].rstrip() + ""
def _term(key: str, label: str, mark: bool) -> str:
if mark:
return f"[[term:{key}]]**{label}**[[/term]]"
return f"**{label}**"
# --------------------------------------------------------------------------- #
# Profile reads (exact, all rows).
# --------------------------------------------------------------------------- #
def _null_count_of(col: dict):
"""Best-effort null count of a column: ``null_count`` or null_pct*n_rows."""
nc = col.get("null_count")
if isinstance(nc, (int, float)) and not isinstance(nc, bool):
return int(nc)
np_ = col.get("null_pct")
nr = col.get("n_rows")
if isinstance(np_, (int, float)) and isinstance(nr, (int, float)):
return int(round(float(np_) * float(nr)))
return 0
def _columns_with_nulls(profile: dict):
"""Return ``[(name, null_count, null_pct_0_100)]`` for columns with nulls,
sorted by null percentage descending. Reads the aggregated profile (exact)."""
cols = profile.get("columns") or []
out = []
for c in cols:
if not isinstance(c, dict):
continue
nc = _null_count_of(c)
if nc <= 0:
continue
np_ = c.get("null_pct")
nr = c.get("n_rows") or profile.get("n_rows")
if isinstance(np_, (int, float)) and not isinstance(np_, bool):
pct = float(np_) * 100.0 if np_ <= 1.0 else float(np_)
elif nr:
pct = nc / float(nr) * 100.0
else:
pct = None
out.append((c.get("name") or "(col)", nc, pct))
out.sort(key=lambda t: (t[2] if t[2] is not None else -1.0), reverse=True)
return out
def _global_missing_pct(profile: dict):
"""Table-level % of missing cells (0-100), exact, from the profile."""
v = profile.get("null_cell_pct")
if isinstance(v, (int, float)) and not isinstance(v, bool):
return float(v) * 100.0 if v <= 1.0 else float(v)
return None
# --------------------------------------------------------------------------- #
# Per-row is-null mask (sample): DuckDB push-down, fallback to raw_numeric.
# --------------------------------------------------------------------------- #
def _build_query_fn(ctx: dict):
"""Return ``(query_fn, table)`` for a DuckDB-backed ctx, or ``(None, None)``.
Mirrors build_eda_render_ctx: a read-only closure over the registry wrapper.
Only DuckDB is supported here; any other backend degrades to raw_numeric."""
db_path = ctx.get("db_path")
table = ctx.get("table")
if not db_path or not table:
return None, None
try:
from infra import duckdb_query_readonly
except Exception: # noqa: BLE001 — wrapper unavailable -> degrade.
return None, None
def query_fn(sql):
return duckdb_query_readonly(db_path, sql)
return query_fn, table
def _null_mask(profile: dict, ctx: dict):
"""Build the per-row is-null mask ``{col: [0/1, ...]}``.
Tries a single DuckDB push-down over ALL columns first (so categorical
columns like Cabin are covered, not only numeric ones); falls back to the
numeric-only ``ctx['raw_numeric']`` (None -> missing); returns ``(None, 0,
None)`` when neither is reachable. Never raises.
Returns ``(mask, n_sampled, source)`` with source in {"db","raw_numeric"}.
"""
cols = profile.get("columns") or []
names = [c.get("name") for c in cols
if isinstance(c, dict) and c.get("name")]
# 1) DuckDB push-down over every column (covers categoricals too).
query_fn, table = _build_query_fn(ctx)
if query_fn is not None and names:
try:
from datascience.extract_null_mask import extract_null_mask
res = extract_null_mask(query_fn, table, names, max_rows=MASK_SAMPLE)
if isinstance(res, dict) and res.get("status") == "ok":
mask = res.get("mask") or {}
if mask:
return mask, int(res.get("n") or 0), "db"
except Exception: # noqa: BLE001 — degrade to raw_numeric.
pass
# 2) Fallback: numeric-only mask derived from raw_numeric (None -> missing).
rn = ctx.get("raw_numeric")
if isinstance(rn, dict) and rn:
mask = {}
for col, vals in rn.items():
if isinstance(vals, (list, tuple)):
mask[col] = [1 if v is None else 0 for v in vals]
if mask:
n = max((len(v) for v in mask.values()), default=0)
return mask, n, "raw_numeric"
return None, 0, None
# --------------------------------------------------------------------------- #
# Lazy registry delegations (each degrades to None on any failure).
# --------------------------------------------------------------------------- #
def _overview(mask: dict):
try:
from datascience.missingness_overview import missingness_overview
out = missingness_overview(mask)
return out if isinstance(out, dict) else None
except Exception: # noqa: BLE001
return None
def _correlation(mask: dict, top_k: int):
try:
from datascience.missingness_correlation import missingness_correlation
out = missingness_correlation(mask, top_k=top_k)
return out if isinstance(out, dict) else None
except Exception: # noqa: BLE001
return None
def _row_patterns(mask: dict, top_n: int):
try:
from datascience.missingness_row_patterns import missingness_row_patterns
out = missingness_row_patterns(mask, top_n=top_n)
return out if isinstance(out, dict) else None
except Exception: # noqa: BLE001
return None
def _rank_bar_make(names, pcts, title):
def make():
try:
from datascience.missingness_rank_bar_figure import (
missingness_rank_bar_figure,
)
return missingness_rank_bar_figure(names, pcts, title=title)
except Exception: # noqa: BLE001 — minimal fallback figure.
return _fallback_fig("ranking de nulos no disponible")
return make
def _heatmap_make(matrix, labels, title):
def make():
try:
from datascience.missingness_corr_heatmap_figure import (
missingness_corr_heatmap_figure,
)
return missingness_corr_heatmap_figure(matrix, labels, title=title)
except Exception: # noqa: BLE001 — minimal fallback figure.
return _fallback_fig("heatmap de co-ocurrencia no disponible")
return make
def _fallback_fig(message: str):
import matplotlib
matplotlib.use("Agg")
from matplotlib.figure import Figure
fig = Figure(figsize=(5.0, 2.2))
ax = fig.add_subplot(111)
ax.text(0.5, 0.5, message, ha="center", va="center")
ax.axis("off")
return fig
# --------------------------------------------------------------------------- #
# Block builders.
# --------------------------------------------------------------------------- #
def _summary_block(profile: dict, with_nulls: list, overview, sampled, n_total):
rows = []
gpct = _global_missing_pct(profile)
rows.append(("Celdas faltantes (global)", _fmt_pct(gpct)))
rows.append(("Columnas con faltantes", str(len(with_nulls))))
all_null = profile.get("all_null_cols")
if isinstance(all_null, (list, tuple)) and all_null:
rows.append(("Columnas 100% faltantes", str(len(all_null))))
if isinstance(overview, dict):
cr = overview.get("complete_rows")
ir = overview.get("incomplete_rows")
suffix = ""
if (isinstance(sampled, int) and isinstance(n_total, (int, float))
and sampled and n_total and sampled < n_total):
suffix = f" (sobre muestra de {_fmt_int(sampled)} filas)"
if cr is not None:
rows.append(("Filas completas (sin faltantes)",
f"{_fmt_int(cr)} ({_fmt_pct(overview.get('complete_pct'))})"
+ suffix))
if ir is not None:
rows.append(("Filas con ≥1 faltante",
f"{_fmt_int(ir)} "
f"({_fmt_pct(overview.get('incomplete_pct'))})" + suffix))
return model.KVTable(rows=rows, title="Resumen de datos faltantes")
def _ranking_block(with_nulls: list):
header = ["Columna", "Faltantes", "% faltante"]
rows = [[_truncate(n), _fmt_int(c), _fmt_pct(p)] for (n, c, p) in with_nulls]
if not rows:
return None
return model.DataTable(
header=header, rows=rows, title="Faltantes por columna",
note="ordenado de más a menos faltante")
def _ranking_figure(with_nulls: list):
names = [n for (n, _, p) in with_nulls if p is not None]
pcts = [p for (_, _, p) in with_nulls if p is not None]
if not names:
return None
return model.Figure(
make=_rank_bar_make(names, pcts, "% de valores faltantes por columna"),
caption="Porcentaje de valores faltantes por columna (barras).")
def _pairs_block(corr: dict):
"""Top column pairs whose absences co-occur, as a table, or None."""
pairs = (corr or {}).get("pairs") or []
header = ["Columna A", "Columna B", "Corr. ausencia", "Co-faltan", "Jaccard"]
rows = []
for p in pairs[:_TOP_PAIRS]:
if not isinstance(p, dict):
continue
rows.append([
_truncate(p.get("a")),
_truncate(p.get("b")),
_fmt_num(p.get("corr")),
_fmt_int(p.get("co_missing")),
_fmt_num(p.get("jaccard")),
])
if not rows:
return None
shown = len(rows)
total = len(pairs)
note = ("correlación de las máscaras is-null entre columnas; "
"«Co-faltan» = nº de filas en que ambas faltan a la vez")
if total > shown:
note += f" — top {shown} de {total} pares"
return model.DataTable(header=header, rows=rows,
title="Pares de columnas que co-faltan", note=note)
def _heatmap_block(corr: dict):
cols = (corr or {}).get("columns") or []
matrix = (corr or {}).get("matrix") or []
if len(cols) < 2 or not matrix:
return None
labels = [_truncate(c, 16) for c in cols]
return model.Figure(
make=_heatmap_make(matrix, labels, "Co-ocurrencia de ausencias"),
caption=("Correlación de las ausencias entre columnas (azul = faltan "
"juntas; rojo = cuando una falta la otra tiende a estar)."))
def _patterns_block(patterns_res: dict):
patterns = (patterns_res or {}).get("patterns") or []
header = ["Columnas que faltan juntas", "Filas", "%"]
rows = []
for p in patterns[:_TOP_PATTERNS]:
if not isinstance(p, dict):
continue
cols = p.get("missing_cols") or []
if cols:
label = ", ".join(_truncate(c, 18) for c in cols)
else:
label = "(fila completa — sin faltantes)"
rows.append([label, _fmt_int(p.get("n_rows")), _fmt_pct(p.get("pct"))])
if not rows:
return None
total = (patterns_res or {}).get("n_patterns")
shown = len(rows)
note = "cada fila es un patrón de «qué columnas faltan juntas»"
if isinstance(total, int) and total > shown:
note += f" — top {shown} de {total} patrones distintos"
return model.DataTable(header=header, rows=rows,
title="Patrones de fila más comunes", note=note)
def _mcar_mar_note(corr: dict, mark: bool):
"""Interpretive, exploratory MCAR/MAR note from the absence correlations.
Reads the absence correlations at two levels so the verdict never contradicts
the visible evidence: a *strong* correlation flags a clear non-random (MAR)
pattern; a *partial* overlap (many rows co-miss — high Jaccard — even if the
correlation is diluted by one column being missing far more often) flags a
localized possible-MAR and cites the concrete co-missing pair; only when
neither holds does it read the absences as compatible with MCAR."""
def _pairs_with(attr_ok):
out = []
for p in (corr or {}).get("pairs") or []:
if isinstance(p, dict) and attr_ok(p):
out.append(p)
return out
def _cf(v):
try:
return float(v)
except (TypeError, ValueError):
return 0.0
strong = _pairs_with(lambda p: abs(_cf(p.get("corr"))) >= _CORR_STRONG)
partial = _pairs_with(
lambda p: _cf(p.get("corr")) > 0 and _cf(p.get("jaccard")) >= _JACCARD_NOTABLE)
mcar = _term("mcar", "MCAR", mark)
mar = _term("mar", "MAR", mark)
head = (
"**Lectura exploratoria MCAR/MAR.** Esta es una heurística basada en la "
"correlación de las ausencias entre columnas, NO un test confirmatorio "
"(como el de Little); orienta, no demuestra. ")
if strong:
top = strong[0]
ev = (f"«{model._safe_str(top.get('a'))}» y "
f"«{model._safe_str(top.get('b'))}» "
f"(corr {_fmt_num(top.get('corr'))})")
body = (
f"Hay ausencias que co-ocurren con fuerza —{ev}—: las columnas no "
f"faltan de forma independiente, lo que es un indicio de un patrón no "
f"aleatorio ({mar}). Antes de imputar o descartar filas conviene "
f"comprobar si la ausencia depende de otra variable observada; en ese "
f"caso la imputación debería condicionar en ella para no sesgar.")
elif partial:
top = max(partial, key=lambda p: _cf(p.get("jaccard")))
ev = (f"«{model._safe_str(top.get('a'))}» y "
f"«{model._safe_str(top.get('b'))}» faltan a la vez en "
f"{_fmt_int(top.get('co_missing'))} filas "
f"(Jaccard {_fmt_num(top.get('jaccard'))})")
body = (
f"Hay co-ocurrencia parcial de ausencias —{ev}—: algunas columnas "
f"tienden a faltar juntas aunque la correlación global sea modesta "
f"(habitual cuando una columna falta mucho más que la otra). Es un "
f"indicio de un posible patrón localizado no aleatorio ({mar}); "
f"conviene revisar si esa ausencia depende de otra variable observada "
f"antes de imputar, en lugar de asumir que faltan al azar.")
else:
body = (
f"Las ausencias entre columnas no muestran correlación ni solape "
f"relevante: parecen independientes, lo que es compatible con que "
f"falten al azar ({mcar}). Aun así, la ausencia podría depender de "
f"variables no observadas (la heurística no lo descarta).")
return model.Markdown(text=head + body)
def _intro_block(mark: bool, source):
missingness = _term("missingness", "missingness", mark)
text = (
f"Este capítulo analiza el {missingness} de la tabla: no solo cuánto "
"falta (eso lo cubre la calidad), sino DÓNDE falta y si las columnas "
"faltan juntas. La co-ocurrencia de ausencias se calcula sobre la matriz "
"binaria «is-null» por fila.")
if source == "raw_numeric":
text += (" Nota: no se pudo leer la tabla cruda completa, así que la "
"co-ocurrencia se limita a las columnas numéricas disponibles.")
return model.Markdown(text=text)
# --------------------------------------------------------------------------- #
# Entry point.
# --------------------------------------------------------------------------- #
def build_missingness(profile: dict, ctx: dict):
"""Build the missingness Chapter, or None if the table has no missing data."""
if not isinstance(profile, dict):
profile = {}
ctx = ctx or {}
with_nulls = _columns_with_nulls(profile)
if not with_nulls:
return None # no missing data anywhere -> chapter does not apply.
# Register glossary terms (if a collector is present) and mark them clickable.
glossary = ctx.get("glossary")
mark = False
if isinstance(glossary, model.GlossaryCollector):
for key, (label, definition) in _TERMS.items():
glossary.add(key, label, definition)
mark = True
# Per-row is-null mask (sample) for co-occurrence and row patterns.
mask, sampled, source = _null_mask(profile, ctx)
overview = _overview(mask) if mask else None
n_total = profile.get("n_rows")
blocks = [
model.Heading(text="Cuánto y dónde faltan datos", level=2),
_intro_block(mark, source),
_summary_block(profile, with_nulls, overview, sampled, n_total),
model.Heading(text="Faltantes por columna", level=2),
]
ranking = _ranking_block(with_nulls)
if ranking is not None:
blocks.append(ranking)
rank_fig = _ranking_figure(with_nulls)
if rank_fig is not None:
blocks.append(rank_fig)
# Co-occurrence + row patterns need the per-row mask. Without it, say so.
if not mask:
blocks.append(model.Note(
"No se pudo construir la matriz «is-null» por fila (sin acceso a los "
"datos crudos), así que no se analiza la co-ocurrencia de ausencias "
"ni los patrones de fila en este informe."))
return model.Chapter(id=CHAPTER_ID, title=CHAPTER_TITLE,
version=CHAPTER_VERSION, blocks=blocks)
corr = _correlation(mask, _TOP_PAIRS) or {}
co_blocks = [model.Heading(text="Co-ocurrencia de ausencias", level=2)]
heatmap = _heatmap_block(corr)
if heatmap is not None:
co_blocks.append(heatmap)
pairs = _pairs_block(corr)
if pairs is not None:
co_blocks.append(pairs)
if heatmap is None and pairs is None:
co_blocks.append(model.Note(
"Ninguna pareja de columnas comparte ausencias con variación "
"suficiente para correlacionarlas (p. ej. una sola columna con "
"faltantes), así que no hay co-ocurrencia que mostrar."))
# Keep the co-occurrence heading next to its heatmap and table.
blocks.append(model.Group(blocks=co_blocks))
patterns_res = _row_patterns(mask, _TOP_PATTERNS) or {}
patterns = _patterns_block(patterns_res)
if patterns is not None:
blocks.append(model.Heading(text="Patrones de fila", level=2))
blocks.append(patterns)
blocks.append(model.Heading(text="Lectura MCAR / MAR", level=2))
blocks.append(_mcar_mar_note(corr, mark))
return model.Chapter(id=CHAPTER_ID, title=CHAPTER_TITLE,
version=CHAPTER_VERSION, blocks=blocks)
@@ -0,0 +1,162 @@
"""Tests for the MISSINGNESS chapter.
Covers the Definition of Done for this chapter:
* Activates (non-None Chapter with the expected sections) when the profile has
missing data, building the co-occurrence from the per-row is-null mask.
* Returns None when the table has no missing data at all (edge case).
* Registers the MCAR/MAR/missingness glossary terms.
* The DuckDB push-down path covers categorical columns (not only numeric),
so a categorical column that co-misses with a numeric one is detected.
"""
import os
import sys
_HERE = os.path.dirname(os.path.abspath(__file__))
_FUNCTIONS = os.path.abspath(os.path.join(_HERE, "..", "..", "..")) # python/functions
if _FUNCTIONS not in sys.path:
sys.path.insert(0, _FUNCTIONS)
from datascience.automatic_eda import model # noqa: E402
from datascience.automatic_eda.chapters.missingness import ( # noqa: E402
build_missingness,
)
def _titles(chapter):
"""Collect heading texts and table/figure titles for assertions."""
out = []
for b in chapter.blocks:
kind = getattr(b, "kind", None)
if kind == "heading":
out.append(("heading", getattr(b, "text", "")))
elif kind in ("data_table", "kv_table"):
out.append((kind, getattr(b, "title", "")))
elif kind == "group":
for inner in getattr(b, "blocks", []):
ik = getattr(inner, "kind", None)
if ik == "heading":
out.append(("heading", getattr(inner, "text", "")))
elif ik in ("data_table", "kv_table"):
out.append((ik, getattr(inner, "title", "")))
elif ik == "figure":
out.append(("figure", getattr(inner, "caption", "")))
elif kind == "figure":
out.append(("figure", getattr(b, "caption", "")))
return out
def _all_text(chapter):
parts = []
def walk(blocks):
for b in blocks:
for attr in ("text", "title", "note", "caption"):
v = getattr(b, attr, None)
if v:
parts.append(str(v))
if getattr(b, "kind", None) == "group":
walk(getattr(b, "blocks", []))
walk(chapter.blocks)
return "\n".join(parts)
def test_returns_none_when_no_missing_data():
profile = {
"n_rows": 4,
"null_cell_pct": 0.0,
"columns": [
{"name": "a", "null_count": 0, "null_pct": 0.0, "n_rows": 4},
{"name": "b", "null_count": 0, "null_pct": 0.0, "n_rows": 4},
],
}
assert build_missingness(profile, {}) is None
def test_activates_with_cooccurrence_via_raw_numeric():
# a and b are missing in EXACTLY the same rows (0,1,2) -> perfect absence
# correlation. c has no nulls. No db_path -> the chapter falls back to the
# numeric raw_numeric mask.
profile = {
"n_rows": 6,
"null_cell_pct": (0.5 + 0.5 + 0.0) / 3.0,
"columns": [
{"name": "a", "null_count": 3, "null_pct": 0.5, "n_rows": 6},
{"name": "b", "null_count": 3, "null_pct": 0.5, "n_rows": 6},
{"name": "c", "null_count": 0, "null_pct": 0.0, "n_rows": 6},
],
}
glossary = model.GlossaryCollector()
ctx = {
"raw_numeric": {
"a": [None, None, None, 1.0, 2.0, 3.0],
"b": [None, None, None, 4.0, 5.0, 6.0],
},
"glossary": glossary,
}
ch = build_missingness(profile, ctx)
assert ch is not None
assert ch.id == "missingness"
assert ch.blocks
titles = _titles(ch)
headings = {t for (k, t) in titles if k == "heading"}
# Core sections present.
assert any("Cuánto y dónde" in h for h in headings)
assert any("Faltantes por columna" in h for h in headings)
assert any("Co-ocurrencia" in h for h in headings)
assert any("MCAR" in h for h in headings)
# A summary KVTable, a ranking DataTable, a co-occurrence figure and the
# pairs table all exist.
kinds = {k for (k, _) in titles}
assert "kv_table" in kinds
assert "data_table" in kinds
assert "figure" in kinds
# Glossary terms registered.
keys = {t["key"] for t in glossary.terms()}
assert {"missingness", "mcar", "mar"} <= keys
# The MCAR/MAR note reads the co-occurrence; with a perfect overlap it must
# flag the non-random (MAR) reading.
text = _all_text(ch)
assert "MAR" in text
def test_db_pushdown_covers_categorical_column(tmp_path):
"""The is-null mask push-down must cover a categorical column, so a
categorical that co-misses with a numeric one shows up in the pairs."""
import duckdb
db = str(tmp_path / "miss.duckdb")
con = duckdb.connect(db)
con.execute("CREATE TABLE t (num1 DOUBLE, num2 DOUBLE, cat VARCHAR)")
# num1 and cat are NULL together in the first 4 of 10 rows; num2 never null.
rows = []
for i in range(10):
if i < 4:
rows.append((None, float(i), None))
else:
rows.append((float(i), float(i), f"c{i}"))
con.executemany("INSERT INTO t VALUES (?,?,?)", rows)
con.close()
profile = {
"n_rows": 10,
"null_cell_pct": (0.4 + 0.0 + 0.4) / 3.0,
"columns": [
{"name": "num1", "null_count": 4, "null_pct": 0.4, "n_rows": 10},
{"name": "num2", "null_count": 0, "null_pct": 0.0, "n_rows": 10},
{"name": "cat", "null_count": 4, "null_pct": 0.4, "n_rows": 10},
],
}
ctx = {"db_path": db, "table": "t", "glossary": model.GlossaryCollector()}
ch = build_missingness(profile, ctx)
assert ch is not None
# The pairs table must mention both num1 and cat (they co-miss perfectly),
# which is only possible if the mask covered the categorical column.
text = _all_text(ch)
assert "num1" in text and "cat" in text
# Co-occurrence section + a pairs data table exist.
titles = _titles(ch)
assert any("co-faltan" in (t or "").lower() for (k, t) in titles)
@@ -1,559 +0,0 @@
"""Free-text / NLP distributions chapter (TEXT DISTR) for AutomaticEDA.
First chapter for **non-tabular** content: it profiles the linguistic content of
any column holding long free text (reviews, descriptions, comments, tickets) that
the categorical chapter cannot meaningfully summarize (high cardinality, many
words per value). It is the cheap, model-free counterpart to ``cat_distr`` for
columns that are prose rather than discrete labels.
Activation (returns ``None`` when it does not apply):
1. Cheap gate from the aggregated profile: at least one non-numeric column whose
``categorical.len_mean`` (mean character length) is ``>= _MIN_LEN_CHARS``.
A dataset whose only string columns are short labels (e.g. titanic's
``Name``, ~27 chars) never passes this gate, so the chapter disappears with
zero extra work and the existing report is untouched.
2. Confirmation from a raw sample: each candidate column is sampled (push-down
``extract_text_sample`` over ``ctx['db_path']``/``ctx['table']``, or an
in-memory ``ctx['text_raw']`` for tests) and kept only if the **median word
count is ``>= _MIN_WORDS``** — i.e. it is genuinely long text, not a long
single token. If no column survives, the chapter returns ``None``.
Per surviving column the chapter emits, kept together on its own page/slide
(``Group(page_break_before=...)``):
- a key/value summary (documents, length percentiles, vocabulary richness with
**[[term:ttr]]TTR[[/term]]** and **[[term:hapax]]hapax legomena[[/term]]**,
dominant language, exact-duplicate %, readability when available);
- a word-count histogram figure;
- a top-terms table + a horizontal bar figure;
- bigram and trigram frequency tables;
- a detected-language bar figure (when ``langdetect`` is available);
- an optional word-cloud figure (only when ``wordcloud`` is installed);
- a closing note on duplicates / readability degradation.
Every metric is delegated to pure ``eda`` registry functions
(``compute_text_length_stats``, ``compute_vocabulary_stats``,
``compute_top_ngrams``, ``detect_corpus_language``, ``compute_text_duplicates``,
``compute_text_readability``) and the raw sample to ``extract_text_sample``; all
are imported defensively so a missing function or optional library degrades that
single piece to a note instead of aborting the chapter. Optional libraries
(``langdetect``, ``textstat``, ``wordcloud``, ``datasketch``) are never required:
the piece is silently omitted when they are absent.
Contract: build_<id>(profile, ctx) -> Chapter | None ; CHAPTER_VERSION = "x.y.z".
"""
from __future__ import annotations
from .. import model
CHAPTER_VERSION = "1.0.0"
CHAPTER_ID = "text_distr"
CHAPTER_TITLE = "Texto libre (NLP)"
# Cheap activation gate (characters): a non-numeric column whose mean string
# length reaches this is a candidate for "long text". Short labels (titanic's
# Name ≈ 27 chars) stay below it, so the chapter does not fire on them.
_MIN_LEN_CHARS = 50
# Confirmation gate (words): a candidate is kept only if its median document has
# at least this many words — genuine prose, not a long id/URL token.
_MIN_WORDS = 20
# Bound the document so very wide datasets stay readable.
_MAX_TEXT_COLS = 5
# Raw text rows to sample per column when the chapter must extract them itself.
_SAMPLE_ROWS = 2000
# Rows shown in the frequency tables.
_TOP_TERMS = 15
_TOP_NGRAMS = 10
# Glossary terms this chapter explains (registered in the shared collector and
# marked clickable on first appearance — same mechanism as cat_distr's entropía).
_TERMS = {
"ttr": (
"TTR (type-token ratio)",
"Riqueza léxica de un texto: número de palabras distintas (tipos) "
"dividido por el número total de palabras (tokens). Vale 1 cuando no se "
"repite ninguna palabra (máxima variedad) y baja hacia 0 cuando el "
"vocabulario se repite mucho. Depende de la longitud del corpus, así que "
"compara mejor textos de tamaño parecido."),
"hapax": (
"Hapax legomena",
"Palabras que aparecen una sola vez en todo el corpus. Un porcentaje "
"alto de hapax indica vocabulario muy variado o, a veces, ruido "
"(erratas, identificadores, tokens raros). Se expresa como porcentaje "
"sobre el número de palabras distintas."),
}
def _fmt_int(value) -> str:
if value is None:
return ""
try:
return f"{int(value):,}".replace(",", ".")
except (TypeError, ValueError):
return str(value)
def _fmt_num(value, decimals: int = 2) -> str:
if value is None:
return ""
if isinstance(value, bool):
return str(value)
if isinstance(value, int):
return f"{value:,}".replace(",", ".")
if isinstance(value, float):
if value != value: # NaN
return "NaN"
if value in (float("inf"), float("-inf")):
return str(value)
text = f"{value:.{decimals}f}".rstrip("0").rstrip(".")
return text if text else "0"
return str(value)
def _fmt_pct(value, decimals: int = 1) -> str:
if value is None:
return ""
try:
return f"{float(value):.{decimals}f}%"
except (TypeError, ValueError):
return str(value)
def _truncate(text, limit: int = 40) -> str:
s = model._safe_str(text)
return s if len(s) <= limit else s[: max(1, limit - 1)].rstrip() + ""
# --------------------------------------------------------------------------- #
# Defensive wrappers around the registry functions: each returns the function's
# output dict or a safe empty default, never raising and never importing at
# module load (so the chapter stays importable even if a function is missing).
# --------------------------------------------------------------------------- #
def _length_stats(texts) -> dict:
try:
from datascience.compute_text_length_stats import compute_text_length_stats
out = compute_text_length_stats(texts)
if isinstance(out, dict):
return out
except Exception: # noqa: BLE001
pass
return {}
def _vocab_stats(texts) -> dict:
try:
from datascience.compute_vocabulary_stats import compute_vocabulary_stats
out = compute_vocabulary_stats(texts, top_k=_TOP_TERMS)
if isinstance(out, dict):
return out
except Exception: # noqa: BLE001
pass
return {}
def _ngrams(texts, n) -> list:
try:
from datascience.compute_top_ngrams import compute_top_ngrams
out = compute_top_ngrams(texts, n=n, top_k=_TOP_NGRAMS)
if isinstance(out, dict):
return out.get("top") or []
except Exception: # noqa: BLE001
pass
return []
def _language(texts) -> dict:
try:
from datascience.detect_corpus_language import detect_corpus_language
out = detect_corpus_language(texts)
if isinstance(out, dict):
return out
except Exception: # noqa: BLE001
pass
return {"available": False, "distribution": [], "dominant": None}
def _duplicates(texts) -> dict:
try:
from datascience.compute_text_duplicates import compute_text_duplicates
out = compute_text_duplicates(texts)
if isinstance(out, dict):
return out
except Exception: # noqa: BLE001
pass
return {}
def _readability(texts) -> dict:
try:
from datascience.compute_text_readability import compute_text_readability
out = compute_text_readability(texts)
if isinstance(out, dict):
return out
except Exception: # noqa: BLE001
pass
return {"available": False, "flesch": {}}
# --------------------------------------------------------------------------- #
# Candidate detection + raw sample acquisition.
# --------------------------------------------------------------------------- #
def _candidate_columns(profile: dict) -> list:
"""Cheap gate: non-numeric columns whose mean char length reaches the
threshold. Returns the list of column names (possibly empty)."""
out = []
for col in profile.get("columns") or []:
if not isinstance(col, dict):
continue
if col.get("inferred_type") == "numeric":
continue
cat = col.get("categorical")
if not isinstance(cat, dict):
continue
len_mean = cat.get("len_mean")
if isinstance(len_mean, (int, float)) and not isinstance(len_mean, bool) \
and len_mean >= _MIN_LEN_CHARS:
name = col.get("name")
if name:
out.append(str(name))
return out
def _get_samples(profile: dict, ctx: dict, columns: list) -> dict:
"""Return {col: [str, ...]} raw text samples for the candidate columns.
Prefers an in-memory ``ctx['text_raw']`` (used by tests); otherwise pushes a
sample down to the database via ``extract_text_sample`` using ctx db_path /
table. Never raises: returns {} when no sample can be obtained."""
text_raw = ctx.get("text_raw")
if isinstance(text_raw, dict) and text_raw:
return {c: [str(v) for v in (text_raw.get(c) or []) if v is not None]
for c in columns if text_raw.get(c)}
db_path = ctx.get("db_path")
table = ctx.get("table")
if not db_path or not table:
return {}
backend = ctx.get("backend") or "duckdb"
sample = ctx.get("sample") or _SAMPLE_ROWS
try:
from datascience.extract_text_sample import extract_text_sample
out = extract_text_sample(db_path, table, columns, backend=backend,
sample=sample)
if isinstance(out, dict) and out.get("status") == "ok":
cols = out.get("columns")
if isinstance(cols, dict):
return {c: list(v) for c, v in cols.items() if v}
except Exception: # noqa: BLE001 — dict-no-throw: no sample → chapter omits.
pass
return {}
def _confirm_long_text(samples: dict) -> dict:
"""Keep only columns whose median word count reaches _MIN_WORDS. Returns
{col: length_stats_dict} for the survivors, in input order."""
survivors = {}
for col, texts in samples.items():
stats = _length_stats(texts)
words = stats.get("words") if isinstance(stats, dict) else None
median = words.get("p50") if isinstance(words, dict) else None
if isinstance(median, (int, float)) and not isinstance(median, bool) \
and median >= _MIN_WORDS:
survivors[col] = stats
return survivors
# --------------------------------------------------------------------------- #
# Figures (lazy matplotlib, scaled by the renderers — same style as num_distr).
# --------------------------------------------------------------------------- #
def _hist_figure(name: str, length_stats: dict):
def make():
import matplotlib
matplotlib.use("Agg")
from matplotlib.figure import Figure
fig = Figure(figsize=(6.2, 3.0))
ax = fig.add_subplot(111)
bins = (length_stats or {}).get("word_hist") or []
drew = False
for b in bins:
if not isinstance(b, dict):
continue
lo, hi, count = b.get("lo"), b.get("hi"), b.get("count") or 0
if lo is None or hi is None:
continue
width = (hi - lo) if hi > lo else max(abs(lo) * 1e-3, 1e-6)
ax.bar(lo, count, width=width, align="edge", color="#9ec6df",
edgecolor="#5b8aa6", linewidth=0.4)
drew = True
if not drew:
ax.text(0.5, 0.5, "(sin datos de longitud)", ha="center",
va="center", color="#8a8a8a", transform=ax.transAxes)
ax.set_xlabel("palabras por documento", fontsize=8)
ax.set_ylabel("nº de documentos", fontsize=8)
ax.tick_params(labelsize=7)
for spine in ("top", "right"):
ax.spines[spine].set_visible(False)
ax.set_title(f"Longitud de «{_truncate(name, 30)}»", fontsize=10,
loc="left")
fig.tight_layout()
return fig
return make
def _barh_figure(title: str, items: list, label_key: str, value_key: str,
xlabel: str):
"""Horizontal bar chart from [{label_key:..., value_key:...}, ...]."""
def make():
import matplotlib
matplotlib.use("Agg")
from matplotlib.figure import Figure
rows = [it for it in (items or []) if isinstance(it, dict)
and isinstance(it.get(value_key), (int, float))]
rows = rows[:12]
fig = Figure(figsize=(6.2, max(2.2, 0.32 * len(rows) + 0.8)))
ax = fig.add_subplot(111)
if not rows:
ax.text(0.5, 0.5, "(sin datos)", ha="center", va="center",
color="#8a8a8a", transform=ax.transAxes)
ax.axis("off")
return fig
labels = [_truncate(r.get(label_key), 28) for r in rows][::-1]
values = [float(r.get(value_key) or 0) for r in rows][::-1]
ypos = range(len(rows))
ax.barh(list(ypos), values, color="#9ec6df", edgecolor="#5b8aa6",
linewidth=0.4)
ax.set_yticks(list(ypos))
ax.set_yticklabels(labels, fontsize=7)
ax.set_xlabel(xlabel, fontsize=8)
ax.tick_params(labelsize=7)
for spine in ("top", "right"):
ax.spines[spine].set_visible(False)
ax.set_title(_truncate(title, 44), fontsize=10, loc="left")
fig.tight_layout()
return fig
return make
def _wordcloud_figure(texts):
"""Word-cloud figure callable, or None if wordcloud is not installed."""
try:
import wordcloud # noqa: F401
except Exception: # noqa: BLE001 — optional dependency: omit the figure.
return None
def make():
import matplotlib
matplotlib.use("Agg")
from matplotlib.figure import Figure
from wordcloud import WordCloud
fig = Figure(figsize=(6.2, 3.2))
ax = fig.add_subplot(111)
joined = " ".join(t for t in texts if isinstance(t, str))
try:
wc = WordCloud(width=800, height=400, background_color="white",
colormap="viridis").generate(joined)
ax.imshow(wc, interpolation="bilinear")
except Exception: # noqa: BLE001
ax.text(0.5, 0.5, "(nube de palabras no disponible)", ha="center",
va="center", color="#8a8a8a", transform=ax.transAxes)
ax.axis("off")
fig.tight_layout()
return fig
return make
# --------------------------------------------------------------------------- #
# Per-column block assembly.
# --------------------------------------------------------------------------- #
def _summary_kv(n_docs, length_stats, vocab, lang, dup, read):
chars = (length_stats or {}).get("chars") or {}
words = (length_stats or {}).get("words") or {}
sents = (length_stats or {}).get("sentences") or {}
rows = [
("Documentos", _fmt_int(n_docs)),
("Caracteres (media · p50 · p90 · p99)",
f"{_fmt_num(chars.get('mean'))} · {_fmt_int(chars.get('p50'))} · "
f"{_fmt_int(chars.get('p90'))} · {_fmt_int(chars.get('p99'))}"),
("Palabras (media · p50 · p90 · p99)",
f"{_fmt_num(words.get('mean'))} · {_fmt_int(words.get('p50'))} · "
f"{_fmt_int(words.get('p90'))} · {_fmt_int(words.get('p99'))}"),
("Frases (media · máx)",
f"{_fmt_num(sents.get('mean'))} · {_fmt_int(sents.get('max'))}"),
("Vocabulario (tokens · tipos · TTR)",
f"{_fmt_int(vocab.get('n_tokens'))} · {_fmt_int(vocab.get('n_types'))} "
f"· {_fmt_num(vocab.get('ttr'), 3)}"),
("Hapax legomena",
f"{_fmt_int(vocab.get('n_hapax'))} ({_fmt_pct(vocab.get('hapax_pct'))})"),
]
if isinstance(lang, dict) and lang.get("available"):
dom = lang.get("dominant")
n_langs = len(lang.get("distribution") or [])
rows.append(("Idioma dominante · nº idiomas",
f"{model._safe_str(dom) or ''} · {_fmt_int(n_langs)}"))
if isinstance(dup, dict) and dup.get("n_docs"):
rows.append(("Duplicados exactos",
f"{_fmt_int(dup.get('n_exact_dup'))} "
f"({_fmt_pct(dup.get('exact_dup_pct'))})"))
if isinstance(read, dict) and read.get("available"):
flesch = read.get("flesch") or {}
rows.append(("Legibilidad Flesch (media)",
_fmt_num(flesch.get("mean"), 1)))
return model.KVTable(rows=rows, title="Resumen del texto")
def _terms_table(vocab) -> "model.DataTable | None":
top = (vocab or {}).get("top_terms") or []
rows = [[_truncate(t.get("term"), 32), _fmt_int(t.get("count")),
_fmt_pct(t.get("pct"))]
for t in top[:_TOP_TERMS] if isinstance(t, dict)]
if not rows:
return None
return model.DataTable(header=["Término", "Conteo", "% tokens"], rows=rows,
title="Términos más frecuentes",
note="stopwords ES+EN eliminadas")
def _ngram_table(items, n_label) -> "model.DataTable | None":
rows = [[_truncate(it.get("ngram"), 40), _fmt_int(it.get("count"))]
for it in (items or [])[:_TOP_NGRAMS] if isinstance(it, dict)]
if not rows:
return None
return model.DataTable(header=[n_label, "Conteo"], rows=rows,
title=f"{n_label} más frecuentes")
def _dup_note(dup, lang, read) -> "model.Note | None":
bits = []
if isinstance(dup, dict):
nd = dup.get("near_dup") or {}
if nd.get("available"):
bits.append(
f"casi-duplicados detectados (MinHash, umbral "
f"{_fmt_num(nd.get('threshold'))}): "
f"{_fmt_int(nd.get('n_near_dup_docs'))} documentos")
else:
bits.append("near-duplicados no calculados (datasketch no instalado; "
"se reportan solo los duplicados exactos por hash)")
if isinstance(lang, dict) and not lang.get("available"):
bits.append("detección de idioma omitida (langdetect no instalado)")
if isinstance(read, dict) and not read.get("available"):
bits.append("legibilidad omitida (textstat no instalado)")
if not bits:
return None
return model.Note(" · ".join(bits))
def _column_group(name, texts, length_stats, idx, mark_terms):
vocab = _vocab_stats(texts)
lang = _language(texts)
dup = _duplicates(texts)
read = _readability(texts)
n_docs = (length_stats or {}).get("n_docs")
blocks = [
model.Heading(text=str(name), level=2),
_summary_kv(n_docs, length_stats, vocab, lang, dup, read),
model.Figure(make=_hist_figure(name, length_stats),
caption=f"Distribución de la longitud (palabras) de "
f"«{_truncate(name, 30)}»."),
]
terms_tbl = _terms_table(vocab)
if terms_tbl is not None:
blocks.append(terms_tbl)
blocks.append(model.Figure(
make=_barh_figure(f"Top términos de «{_truncate(name, 24)}»",
vocab.get("top_terms"), "term", "count",
"conteo"),
caption="Términos más frecuentes (barras)."))
bi_tbl = _ngram_table(_ngrams(texts, 2), "Bigrama")
if bi_tbl is not None:
blocks.append(bi_tbl)
tri_tbl = _ngram_table(_ngrams(texts, 3), "Trigrama")
if tri_tbl is not None:
blocks.append(tri_tbl)
if isinstance(lang, dict) and lang.get("available") \
and lang.get("distribution"):
blocks.append(model.Figure(
make=_barh_figure(f"Idiomas detectados en «{_truncate(name, 24)}»",
lang.get("distribution"), "lang", "count",
"documentos"),
caption="Distribución de idiomas detectados (langdetect)."))
wc = _wordcloud_figure(texts)
if wc is not None:
blocks.append(model.Figure(
make=wc, caption=f"Nube de palabras de «{_truncate(name, 30)}»."))
note = _dup_note(dup, lang, read)
if note is not None:
blocks.append(note)
return model.Group(blocks=blocks, page_break_before=(idx > 0))
def _intro_blocks(n_cols, mark_terms):
ttr = ("[[term:ttr]]TTR[[/term]]" if mark_terms else "TTR")
hapax = ("[[term:hapax]]hapax legomena[[/term]]" if mark_terms
else "hapax legomena")
text = (
f"Este capítulo perfila las columnas de **texto libre largo** del "
f"dataset (reseñas, descripciones, comentarios): contenido lingüístico "
f"que la distribución categórica no resume bien. Para cada columna se "
f"muestran la longitud de los documentos, la riqueza de vocabulario "
f"(incluido el {ttr} y el porcentaje de {hapax}), los términos y "
f"n-gramas más frecuentes, los idiomas detectados y el nivel de "
f"duplicación. Las métricas son baratas y sin modelos pesados; las "
f"piezas que dependen de una librería opcional se omiten si no está "
f"instalada.")
return [
model.Heading(text=CHAPTER_TITLE, level=1),
model.Markdown(text=text),
]
def build_text_distr(profile: dict, ctx: dict):
"""Build the free-text Chapter, or None if no long-text column applies."""
profile = profile or {}
ctx = ctx or {}
# 1) Cheap gate from the profile (no DB access yet).
candidates = _candidate_columns(profile)
if not candidates:
return None
# 2) Raw sample + 3) confirm genuine long text (median words >= threshold).
samples = _get_samples(profile, ctx, candidates)
if not samples:
return None
survivors = _confirm_long_text(samples)
if not survivors:
return None
# Register glossary terms (clickable) once we know the chapter applies.
glossary = ctx.get("glossary")
mark_terms = False
if isinstance(glossary, model.GlossaryCollector):
for key, (label, definition) in _TERMS.items():
glossary.add(key, label, definition)
mark_terms = True
blocks = list(_intro_blocks(len(survivors), mark_terms))
rendered = list(survivors.items())[:_MAX_TEXT_COLS]
for idx, (name, length_stats) in enumerate(rendered):
texts = samples.get(name) or []
blocks.append(_column_group(name, texts, length_stats, idx, mark_terms))
if len(survivors) > len(rendered):
omitted = len(survivors) - len(rendered)
blocks.append(model.Note(
f"Se muestran las primeras {len(rendered)} columnas de texto; "
f"quedan {omitted} sin mostrar para mantener acotado el informe."))
return model.Chapter(id=CHAPTER_ID, title=CHAPTER_TITLE,
version=CHAPTER_VERSION, blocks=blocks)
@@ -1,256 +0,0 @@
"""Tests for the TEXT DISTR chapter — DoD: golden + edges + degradation.
Self-contained: builds synthetic TableProfiles and feeds the raw text sample
in-memory through ``ctx['text_raw']`` (no DuckDB needed), so the suite is fast
and deterministic. Verifies that ``build_text_distr``:
- GOLDEN: with a long-text column, emits the chapter with its key blocks
(length summary, word histogram, top-terms table, n-gram tables, language
bars) and registers the clickable glossary terms; and that it renders inside
the full document to both PDF and PPTX showing that content.
- EDGE (None): a dataset whose only string column is short labels (titanic-like
``Name``) yields ``None`` without raising — the existing report is untouched.
- EDGE (None): a column that passes the cheap char gate but whose documents are
short (median words below the threshold) is rejected at the confirmation step.
- DEGRADATION: with ``langdetect`` / ``textstat`` / ``wordcloud`` unavailable,
the chapter still builds (those pieces are omitted) and never raises.
"""
import builtins
import os
import tempfile
from pypdf import PdfReader
from pptx import Presentation
from datascience.automatic_eda.model import (
DataTable, Figure, GlossaryCollector, Group, Heading, KVTable, Markdown,
Note,
)
from datascience.automatic_eda.chapters.text_distr import (
CHAPTER_ID, CHAPTER_VERSION, build_text_distr,
)
from datascience.automatic_eda.chapters_registry import build_document
from datascience.render_automatic_eda_pdf import render_automatic_eda_pdf
from datascience.render_automatic_eda_pptx import render_automatic_eda_pptx
# --------------------------------------------------------------------------- #
# Synthetic corpus + profiles.
# --------------------------------------------------------------------------- #
_ES = [
"El producto llegó en perfecto estado y mucho antes de lo previsto por la tienda",
"La calidad de los materiales es realmente excelente y se nota la diferencia al usarlo",
"No me convenció del todo porque esperaba bastante más por el precio que pagué finalmente",
"El servicio de atención al cliente fue rápido amable y resolvió mi problema sin demora",
"Lo recomiendo totalmente ya que ha superado con creces todas mis expectativas iniciales",
]
_EN = [
"The product arrived in perfect condition and much earlier than the store had promised me",
"The build quality is genuinely outstanding and you can really feel the difference using it",
"I was not fully convinced because I expected quite a lot more for the price i finally paid",
"Customer support was fast friendly and solved my whole problem without any delay at all",
"I highly recommend it since it has exceeded by far every one of my initial expectations",
]
def _long_reviews(n=40) -> list:
"""A corpus of long multi-sentence reviews (>= 20 words each), mixing two
languages and including a few exact duplicates."""
out = []
for i in range(n):
base = _ES if i % 3 != 0 else _EN # mostly ES, some EN
a = base[i % len(base)]
b = base[(i + 2) % len(base)]
out.append(f"{a}. {b}.")
# Inject a couple of exact duplicates.
out.append(out[0])
out.append(out[1])
return out
def _text_profile() -> dict:
"""Profile with a long free-text column (review) + a numeric + a short cat."""
return {
"table": "reviews",
"source": "/data/reviews.duckdb",
"profiled_at": "2026-06-30T10:00:00+00:00",
"n_rows": 42,
"n_cols": 3,
"quality_score": 88.0,
"columns": [
{
"name": "review",
"inferred_type": "categorical",
"categorical": {
"top": [{"value": "x", "count": 2, "pct": 0.05}],
"n_distinct": 40,
"len_mean": 180.0,
"len_min": 80,
"len_max": 220,
},
},
{
"name": "rating",
"inferred_type": "numeric",
"numeric": {"mean": 3.1, "median": 3.0, "std": 1.2,
"min": 1, "max": 5},
},
{
"name": "product",
"inferred_type": "categorical",
"categorical": {
"top": [{"value": "teclado", "count": 10, "pct": 0.25}],
"n_distinct": 6,
"len_mean": 7.0,
"len_min": 5, "len_max": 11,
},
},
],
}
def _no_text_profile() -> dict:
"""titanic-like: the only string column is short labels (Name ≈ 27 chars)."""
return {
"table": "titanic",
"n_rows": 891,
"n_cols": 3,
"columns": [
{"name": "Age", "inferred_type": "numeric",
"numeric": {"mean": 29.7, "median": 28.0, "std": 14.5}},
{"name": "Name", "inferred_type": "categorical",
"categorical": {"top": [{"value": "Braund, Mr. Owen Harris",
"count": 1, "pct": 0.001}],
"n_distinct": 891, "len_mean": 27.0,
"len_min": 12, "len_max": 82}},
{"name": "Sex", "inferred_type": "categorical",
"categorical": {"top": [{"value": "male", "count": 577,
"pct": 0.65}],
"n_distinct": 2, "len_mean": 4.6,
"len_min": 4, "len_max": 6}},
],
}
def _flatten(blocks) -> list:
"""Recursively flatten Group blocks so tests can inspect leaf blocks."""
out = []
for b in blocks:
if isinstance(b, Group):
out.extend(_flatten(b.blocks))
else:
out.append(b)
return out
# --------------------------------------------------------------------------- #
# Golden.
# --------------------------------------------------------------------------- #
def test_golden_activa_con_texto():
glossary = GlossaryCollector()
ctx = {"text_raw": {"review": _long_reviews()}, "glossary": glossary}
ch = build_text_distr(_text_profile(), ctx)
assert ch is not None, "el capítulo debe activarse con una columna de texto largo"
assert ch.id == CHAPTER_ID
assert ch.version == CHAPTER_VERSION
leaves = _flatten(ch.blocks)
kinds = [b.kind for b in leaves]
assert "heading" in kinds
assert "kv_table" in kinds # summary
assert "figure" in kinds # histogram / bars
assert "data_table" in kinds # top terms + n-grams
# KV summary mentions vocabulary metrics.
kv = next(b for b in leaves if isinstance(b, KVTable))
labels = " ".join(str(r[0]) for r in kv.rows)
assert "TTR" in labels
assert "Hapax" in labels or "hapax" in labels
# There is a terms table and at least one n-gram table.
titles = [getattr(b, "title", "") or "" for b in leaves
if isinstance(b, DataTable)]
assert any("Términos" in t for t in titles)
assert any("Bigrama" in t for t in titles)
# Glossary terms were registered (clickable destinations).
assert glossary.has("ttr")
assert glossary.has("hapax")
def test_golden_render_pdf_pptx():
profile = _text_profile()
ctx = {"text_raw": {"review": _long_reviews()},
"dataset_name": "reviews"}
chapters = build_document(profile, ctx)
ids = [c.id for c in chapters]
assert "text_distr" in ids, f"text_distr ausente en {ids}"
with tempfile.TemporaryDirectory() as d:
pdf = os.path.join(d, "t.pdf")
pptx = os.path.join(d, "t.pptx")
rp = render_automatic_eda_pdf(profile, pdf, {"title": "EDA", "ctx": ctx})
rx = render_automatic_eda_pptx(profile, pptx, {"title": "EDA", "ctx": ctx})
assert rp.get("path") and os.path.exists(pdf)
assert rx.get("path") and os.path.exists(pptx)
text = "\n".join(p.extract_text() or "" for p in PdfReader(pdf).pages)
assert "Texto libre" in text or "TTR" in text
prs = Presentation(pptx)
ptext = []
for slide in prs.slides:
for shp in slide.shapes:
if shp.has_text_frame:
ptext.append(shp.text_frame.text)
joined = "\n".join(ptext)
assert "Texto libre" in joined or "TTR" in joined
# --------------------------------------------------------------------------- #
# Edges — None.
# --------------------------------------------------------------------------- #
def test_edge_none_sin_texto_largo():
# titanic-like: short labels only → chapter must not apply.
assert build_text_distr(_no_text_profile(), {}) is None
def test_edge_none_palabras_cortas():
# Char gate passes (len_mean high) but documents are short → confirmation
# rejects them (median words below threshold).
profile = _text_profile()
short = ["palabra " * 3] * 30 # 3 words each, < _MIN_WORDS
ctx = {"text_raw": {"review": short}}
assert build_text_distr(profile, ctx) is None
def test_edge_none_empty_profile():
assert build_text_distr({}, {}) is None
assert build_text_distr(None, None) is None
# --------------------------------------------------------------------------- #
# Degradation — optional libs absent.
# --------------------------------------------------------------------------- #
def test_degradacion_sin_libs(monkeypatch):
real_import = builtins.__import__
blocked = ("langdetect", "textstat", "wordcloud", "datasketch")
def fake_import(name, *a, **k):
if name in blocked or any(name.startswith(b + ".") for b in blocked):
raise ImportError(f"simulado: {name}")
return real_import(name, *a, **k)
monkeypatch.setattr(builtins, "__import__", fake_import)
ctx = {"text_raw": {"review": _long_reviews()}}
ch = build_text_distr(_text_profile(), ctx)
# Still builds (the cheap, stdlib-only pieces remain) and never raises.
assert ch is not None
leaves = _flatten(ch.blocks)
assert any(isinstance(b, KVTable) for b in leaves)
assert any(isinstance(b, DataTable) for b in leaves)
# A degradation note is present mentioning the missing optional libs.
notes = " ".join(b.text for b in leaves if isinstance(b, Note))
assert "langdetect" in notes or "textstat" in notes or "datasketch" in notes
@@ -31,8 +31,8 @@ CHAPTER_ORDER = [
"analisis_llm", # LLM interpretation — sits next to overview (user request)
"num_distr", # numeric distributions
"cat_distr", # categorical distributions
"text_distr", # free-text / NLP distributions (non-tabular content)
"calidad", # data quality
"missingness", # missing-data patterns (co-occurrence of absences; MCAR/MAR)
"correlacion", # correlations / associations
"relaciones", # key relations: declared/candidate PK + FK (inter/intra-table)
"modelos", # cheap models (PCA/KMeans/outliers)
@@ -1,102 +0,0 @@
---
id: compute_text_duplicates_py_datascience
name: compute_text_duplicates
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: pure
signature: "def compute_text_duplicates(texts, near_threshold=0.85, sample_max=2000) -> dict"
description: "Detecta documentos duplicados en un corpus de texto. Los duplicados EXACTOS se calculan siempre con la stdlib: cada documento se normaliza (colapsa espacios, strip, lower) y se hashea con SHA-1; n_exact_dup es cuántos docs repiten uno ya visto y exact_dup_pct su porcentaje. Los CASI-duplicados (near-dup) usan la dependencia OPCIONAL datasketch (MinHash + LSH sobre 3-shingles de palabras); si no está instalada, esa parte degrada a available:False sin afectar al resto. Estilo dict-no-throw del grupo eda — nunca lanza."
tags: [eda, datascience, text, nlp, duplicates, minhash, pure, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [hashlib, re]
example: |
from datascience.compute_text_duplicates import compute_text_duplicates
texts = ["El gato come pescado", "El gato come pescado", "Un perro ladra"]
result = compute_text_duplicates(texts)
# {"n_docs": 3, "n_exact_dup": 1, "exact_dup_pct": 33.33, "n_unique": 2,
# "near_dup": {"available": False, "n_near_dup_docs": 0}}
tested: true
tests:
- "test_duplicados_exactos"
- "test_sin_duplicados"
- "test_vacio"
- "test_near_dup_degrada"
test_file_path: "python/functions/datascience/compute_text_duplicates_test.py"
file_path: "python/functions/datascience/compute_text_duplicates.py"
params:
- name: texts
desc: "Lista de documentos de texto. Los elementos None o que no sean str se descartan silenciosamente; n_docs cuenta solo los documentos válidos. None como argumento se trata como lista vacía."
- name: near_threshold
desc: "Umbral de similitud Jaccard (01) para considerar dos documentos casi-duplicados en el cálculo near-dup vía MinHashLSH. Solo aplica si datasketch está instalada. Default 0.85."
- name: sample_max
desc: "Número máximo de documentos muestreados (los primeros) para el cálculo near-dup, que es O(n) en memoria de MinHashes. No afecta al conteo de duplicados exactos, que siempre recorre todo el corpus. Default 2000."
output: "Dict con exactamente 5 claves, siempre presentes: n_docs (int, docs válidos), n_exact_dup (int, docs que repiten un texto normalizado ya visto = n_docs - n_unique), exact_dup_pct (float a 2 decimales = n_exact_dup/n_docs*100, o None si el corpus está vacío), n_unique (int, nº de textos normalizados distintos), y near_dup (sub-dict con available:bool y n_near_dup_docs:int; cuando available es True incluye además threshold con el near_threshold usado). La función nunca lanza: captura toda excepción y degrada."
---
## Ejemplo
```python
from datascience.compute_text_duplicates import compute_text_duplicates
# Tres copias del mismo texto (con espacios/casing distintos) + dos únicos.
texts = [
"El gato come pescado",
"El gato come pescado",
"el GATO come pescado", # mismo tras normalizar
"Un perro ladra",
"La luna brilla",
]
compute_text_duplicates(texts)
# {
# "n_docs": 5,
# "n_exact_dup": 2, # 3 copias del primer texto => 2 repeticiones
# "exact_dup_pct": 40.0, # 2 / 5 * 100
# "n_unique": 3, # 3 textos normalizados distintos
# "near_dup": {"available": False, "n_near_dup_docs": 0}, # datasketch ausente
# }
# Corpus vacío: contrato estable, exact_dup_pct None, sin excepción.
compute_text_duplicates([])
# {"n_docs": 0, "n_exact_dup": 0, "exact_dup_pct": None, "n_unique": 0,
# "near_dup": {"available": False, "n_near_dup_docs": 0}}
```
## Cuando usarla
Úsala en la fase de calidad de un EDA de texto, cuando quieras saber cuánto de
tu corpus es ruido duplicado antes de entrenar, vectorizar o muestrear: te da
el porcentaje de duplicados exactos (`exact_dup_pct`), el número de documentos
únicos (`n_unique`) y, si tienes `datasketch` instalada, una estimación de
casi-duplicados (paráfrasis, copias con pequeñas ediciones) vía MinHash + LSH.
Pásale directamente la columna/lista de textos crudos; la función filtra None y
no-str por ti y nunca lanza, así que es segura para encadenar en pipelines de
perfilado.
## Gotchas
- **Near-dup requiere `datasketch` (opcional).** Si la librería no está
instalada, `near_dup` degrada a `{"available": False, "n_near_dup_docs": 0}`
(sin clave `threshold`) y el resto del resultado se calcula igual. Los
duplicados **exactos** funcionan siempre porque solo usan la stdlib (hash).
- **Normalización de exactos.** Dos textos cuentan como el mismo duplicado
exacto si coinciden tras `" ".join(doc.split()).strip().lower()`: se colapsan
espacios/tabuladores/saltos, se recortan extremos y se ignora el caso. Cambios
de puntuación o acentos SÍ los distinguen (no se eliminan).
- **`n_exact_dup` cuenta repeticiones, no grupos.** Con 3 copias de un mismo
texto, `n_exact_dup` es 2 (las dos copias extra), no 1. Equivale a
`n_docs - n_unique`.
- **`exact_dup_pct` es `None` con corpus vacío** (no `ZeroDivisionError`); en
cualquier otro caso es un float redondeado a 2 decimales.
- **`sample_max` solo limita el near-dup.** El conteo de duplicados exactos
recorre todo el corpus; el near-dup muestrea los primeros `sample_max`
documentos para acotar memoria. Si el corpus está ordenado, considera barajar
antes para que la muestra sea representativa.
- **Elementos no-str se descartan.** `True`/`False` no cuentan como str y se
ignoran igual que `None`; `n_docs` refleja solo los documentos válidos.
@@ -1,128 +0,0 @@
"""Detección de documentos duplicados en un corpus de texto.
Función pura, estilo dict-no-throw del grupo `eda`: nunca lanza, siempre
devuelve el mismo contrato de claves. Los duplicados EXACTOS se calculan
siempre con la stdlib (normalización + hash SHA-1). Los CASI-duplicados
(near-dup) requieren la dependencia opcional `datasketch`; si no está
instalada, esa parte degrada limpiamente a ``available: False`` sin afectar
al resto del cálculo.
"""
import hashlib
import re
def _compute_near_dup(valid, near_threshold, sample_max):
"""Cuenta documentos con al menos otro casi-duplicado vía MinHash + LSH.
Import perezoso de ``datasketch``. Si la librería no está disponible (o
cualquier paso falla), degrada a ``{"available": False, "n_near_dup_docs": 0}``
sin propagar la excepción.
Args:
valid: lista de str ya filtrada (sin None ni no-str).
near_threshold: umbral de similitud Jaccard para LSH.
sample_max: número máximo de documentos a muestrear.
Returns:
dict con ``available`` (bool) y ``n_near_dup_docs`` (int). Cuando
``available`` es True, incluye además ``threshold``.
"""
try:
from datasketch import MinHash, MinHashLSH
except Exception:
return {"available": False, "n_near_dup_docs": 0}
try:
docs = valid[:sample_max]
num_perm = 128
lsh = MinHashLSH(threshold=near_threshold, num_perm=num_perm)
minhashes = {}
for i, doc in enumerate(docs):
tokens = re.findall(r"\w+", doc.lower())
shingles = set()
for j in range(len(tokens) - 2):
shingles.add(" ".join(tokens[j:j + 3]))
# Documentos con menos de 3 tokens no generan 3-shingles: caemos a
# los tokens sueltos para no perderlos del todo.
if not shingles:
shingles = set(tokens)
if not shingles:
# Documento sin tokens (cadena vacía / solo símbolos): se omite.
continue
m = MinHash(num_perm=num_perm)
for sh in shingles:
m.update(sh.encode("utf-8"))
key = "d{}".format(i)
minhashes[key] = m
lsh.insert(key, m)
n_near = 0
for key, m in minhashes.items():
matches = lsh.query(m)
if len(matches) > 1:
n_near += 1
return {
"available": True,
"n_near_dup_docs": int(n_near),
"threshold": near_threshold,
}
except Exception:
return {"available": False, "n_near_dup_docs": 0}
def compute_text_duplicates(texts, near_threshold=0.85, sample_max=2000) -> dict:
"""Detecta duplicados exactos y casi-duplicados en un corpus de texto.
Args:
texts: lista de documentos. Los elementos None o que no sean str se
descartan; ``n_docs`` cuenta solo los válidos.
near_threshold: umbral de similitud Jaccard para considerar dos
documentos casi-duplicados (solo near-dup, requiere datasketch).
sample_max: tope de documentos muestreados para el cálculo near-dup.
Returns:
dict con las claves ``n_docs``, ``n_exact_dup``, ``exact_dup_pct``
(float redondeado a 2 decimales, o None si el corpus está vacío),
``n_unique`` y ``near_dup`` (sub-dict con ``available`` y
``n_near_dup_docs``, más ``threshold`` cuando está disponible).
Nunca lanza: captura toda excepción y degrada.
"""
# Filtrado defensivo de documentos válidos.
try:
valid = [t for t in texts if isinstance(t, str)] if texts is not None else []
except Exception:
valid = []
n_docs = len(valid)
# Duplicados exactos: normalizar + hash SHA-1 (stdlib, siempre disponible).
try:
seen = set()
n_exact_dup = 0
for doc in valid:
norm = " ".join(doc.split()).strip().lower()
digest = hashlib.sha1(norm.encode("utf-8")).hexdigest()
if digest in seen:
n_exact_dup += 1
else:
seen.add(digest)
n_unique = len(seen)
except Exception:
n_exact_dup = 0
n_unique = 0
exact_dup_pct = round(n_exact_dup / n_docs * 100, 2) if n_docs > 0 else None
# Casi-duplicados: opcional vía datasketch, degrada solo.
near_dup = _compute_near_dup(valid, near_threshold, sample_max)
return {
"n_docs": n_docs,
"n_exact_dup": n_exact_dup,
"exact_dup_pct": exact_dup_pct,
"n_unique": n_unique,
"near_dup": near_dup,
}
@@ -1,77 +0,0 @@
"""Tests para compute_text_duplicates.
Importa el modulo hoja directamente (`datascience.compute_text_duplicates`)
para no depender de que el paquete reexporte la funcion en su __init__.
datasketch normalmente NO esta instalada en el venv, asi que near_dup
degrada a available=False; los tests no requieren la libreria.
"""
from datascience.compute_text_duplicates import compute_text_duplicates
EXPECTED_KEYS = {"n_docs", "n_exact_dup", "exact_dup_pct", "n_unique", "near_dup"}
def test_duplicados_exactos():
"""3 copias del mismo texto + 2 únicos: n_exact_dup=2, pct>0."""
texts = [
"El gato come pescado",
"El gato come pescado",
"el GATO come pescado", # mismo tras normalizar (espacios + case)
"Un perro ladra",
"La luna brilla",
]
result = compute_text_duplicates(texts)
assert set(result.keys()) == EXPECTED_KEYS
assert result["n_docs"] == 5
# 3 copias del primer texto (2 son repeticion) + 2 textos unicos.
assert result["n_exact_dup"] == 2
assert result["n_unique"] == 3
assert result["exact_dup_pct"] is not None
assert result["exact_dup_pct"] > 0
# 2 / 5 * 100 = 40.0
assert abs(result["exact_dup_pct"] - 40.0) < 1e-9
def test_sin_duplicados():
"""Corpus sin repeticiones: n_exact_dup=0, n_unique==n_docs."""
texts = [
"primero documento distinto",
"segundo documento distinto",
"tercero documento distinto",
]
result = compute_text_duplicates(texts)
assert result["n_docs"] == 3
assert result["n_exact_dup"] == 0
assert result["n_unique"] == 3
assert abs(result["exact_dup_pct"] - 0.0) < 1e-9
def test_vacio():
"""Corpus vacio: n_docs 0, exact_dup_pct None, no lanza."""
result = compute_text_duplicates([])
assert set(result.keys()) == EXPECTED_KEYS
assert result["n_docs"] == 0
assert result["n_exact_dup"] == 0
assert result["exact_dup_pct"] is None
assert result["n_unique"] == 0
assert result["near_dup"]["n_near_dup_docs"] == 0
def test_near_dup_degrada():
"""near_dup expone 'available' (bool) y no lanza aunque falte datasketch."""
texts = ["uno dos tres cuatro", "uno dos tres cuatro cinco", "algo distinto"]
result = compute_text_duplicates(texts)
near = result["near_dup"]
assert "available" in near
assert isinstance(near["available"], bool)
assert "n_near_dup_docs" in near
assert isinstance(near["n_near_dup_docs"], int)
# Tambien tolera None y entradas no-str sin lanzar.
mixed = compute_text_duplicates(["hola", None, 123, "hola"])
assert mixed["n_docs"] == 2
assert mixed["n_exact_dup"] == 1
@@ -1,86 +0,0 @@
---
id: compute_text_length_stats_py_datascience
name: compute_text_length_stats
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: pure
signature: "def compute_text_length_stats(texts, n_bins=20) -> dict"
description: "Profiles the length distribution of a corpus of text documents for EDA: per-document characters, words (unicode \\w+ tokens) and sentences (segments split on .!?… with a minimum of 1 per non-empty doc), each summarized with mean/p50/p90/p99/min/max (nearest-rank percentiles), plus an equal-width histogram of per-document word counts. None and non-str items are discarded. Dict-no-throw: never raises. Stdlib only (re)."
tags: [eda, datascience, text, nlp, length, statistics, pure, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [re, math]
example: |
from datascience.compute_text_length_stats import compute_text_length_stats
result = compute_text_length_stats(["Hola mundo.", "Una frase mas larga aqui."], n_bins=5)
tested: true
tests:
- "test_basico"
- "test_vacio"
- "test_descarta_none"
- "test_un_documento"
test_file_path: "python/functions/datascience/compute_text_length_stats_test.py"
file_path: "python/functions/datascience/compute_text_length_stats.py"
params:
- name: texts
desc: "List of text documents (str). None entries and any non-str items (ints, floats, etc.) are discarded before any computation. An empty string \"\" is kept (chars 0, words 0, sentences 0)."
- name: n_bins
desc: "Number of equal-width bins for the per-document word-count histogram. Default 20. When all docs have the same word count, there are <2 docs, or n_bins < 1, a single covering bin is returned instead."
output: "Dict with keys n_docs (int), chars, words, sentences and word_hist. Each of the three axis sub-dicts has the exact keys mean (float, 2 decimals), p50, p90, p99, min, max (ints). When there are no valid documents, n_docs is 0, every axis statistic is None and word_hist is []. word_hist is a list of {lo: float, hi: float, count: int} bins; the sum of all bin counts equals n_docs."
---
## Ejemplo
```python
from datascience.compute_text_length_stats import compute_text_length_stats
compute_text_length_stats(
[
"Hola mundo.",
"Una frase mas larga con varias palabras aqui.",
"Esto. Tiene. Tres frases distintas!",
],
n_bins=5,
)
# {
# "n_docs": 3,
# "chars": {"mean": 30.33, "p50": 35, "p90": 45, "p99": 45, "min": 11, "max": 45},
# "words": {"mean": 5.0, "p50": 5, "p90": 8, "p99": 8, "min": 2, "max": 8},
# "sentences": {"mean": 1.67, "p50": 1, "p90": 3, "p99": 3, "min": 1, "max": 3},
# "word_hist": [
# {"lo": 2.0, "hi": 3.2, "count": 1},
# {"lo": 3.2, "hi": 4.4, "count": 0},
# {"lo": 4.4, "hi": 5.6, "count": 1},
# {"lo": 5.6, "hi": 6.8, "count": 0},
# {"lo": 6.8, "hi": 8.0, "count": 1},
# ],
# }
```
## Cuando usarla
Úsala al perfilar una columna o corpus de texto libre en un EDA: cuando
necesites saber lo largos que son los documentos (en caracteres, palabras y
frases) y cómo se reparte esa longitud antes de tokenizar, vectorizar o decidir
truncados/ventanas para un modelo. Pásale la lista de strings crudos de la
columna; `None` y valores no-texto se descartan solos. Encaja en el grupo `eda`
como bloque de longitud junto a `summarize_categorical`.
## Gotchas
- Función pura, solo stdlib (`re`). No usa numpy, pandas ni sklearn.
- Percentiles por método **nearest-rank** (devuelven un valor real de la lista,
no interpolan); por eso p50/p90/p99/min/max son enteros y `mean` es el único
float (redondeado a 2 decimales).
- El conteo de frases es una **aproximación** por puntuación (`.!?…`): un texto
sin esa puntuación cuenta como 1 frase si no está vacío; abreviaturas o
ellipsis pueden inflar o reducir el conteo.
- `word_hist` es equal-width entre min y max de palabras: con todos los docs
del mismo tamaño, menos de 2 docs, o `n_bins < 1`, devuelve un único bin.
- Dict-no-throw: ante input inesperado devuelve la forma vacía
(`n_docs` 0, ejes `None`, `word_hist` []) en vez de lanzar.
@@ -1,168 +0,0 @@
"""Pure EDA helper: document length distribution for the `eda` group.
Given a list of text documents, computes the length distribution along three
axes (characters, words and sentences) plus an equal-width histogram of the
per-document word counts. Stdlib only (``re`` + ``statistics`` semantics via a
hand-rolled nearest-rank percentile). No numpy, no sklearn.
The function is dict-no-throw: it never raises. On any unexpected input it
degrades to the empty-shape result.
"""
import math
import re
_WORD_RE = re.compile(r"\w+", re.UNICODE)
_SENT_RE = re.compile(r"[.!?…]+")
def _empty_axis() -> dict:
"""Return an axis sub-dict with every statistic set to ``None``."""
return {"mean": None, "p50": None, "p90": None, "p99": None, "min": None, "max": None}
def _pct(sorted_vals, q):
"""Nearest-rank percentile of an already-sorted list.
Args:
sorted_vals: List of numbers sorted ascending.
q: Percentile in the 0..100 range.
Returns:
The value at the nearest rank, or ``None`` for an empty list.
"""
n = len(sorted_vals)
if n == 0:
return None
if q <= 0:
return sorted_vals[0]
rank = math.ceil(q / 100.0 * n)
if rank < 1:
rank = 1
if rank > n:
rank = n
return sorted_vals[rank - 1]
def _axis_stats(values) -> dict:
"""Compute mean/p50/p90/p99/min/max over a list of integer counts.
``mean`` is rounded to 2 decimals; every other statistic is an integer
(they are counts). Returns an all-``None`` axis for an empty list.
"""
if not values:
return _empty_axis()
sv = sorted(values)
return {
"mean": round(sum(sv) / len(sv), 2),
"p50": int(_pct(sv, 50)),
"p90": int(_pct(sv, 90)),
"p99": int(_pct(sv, 99)),
"min": int(sv[0]),
"max": int(sv[-1]),
}
def _word_hist(word_counts, n_bins) -> list:
"""Equal-width histogram of per-document word counts.
Builds ``n_bins`` bins between ``min`` and ``max`` of the word counts. When
every document has the same number of words, there are fewer than 2
documents, or ``n_bins`` is not at least 1, a single covering bin is
returned. With no documents the result is ``[]``. The sum of bin ``count``
always equals ``len(word_counts)``.
"""
if not word_counts:
return []
wmin = min(word_counts)
wmax = max(word_counts)
if wmax == wmin or len(word_counts) < 2 or n_bins < 1:
return [{"lo": float(wmin), "hi": float(wmax), "count": len(word_counts)}]
width = (wmax - wmin) / n_bins
bins = []
for i in range(n_bins):
lo = wmin + i * width
hi = wmin + (i + 1) * width
bins.append({"lo": float(lo), "hi": float(hi), "count": 0})
# Pin the last upper edge to the real maximum to avoid float drift.
bins[-1]["hi"] = float(wmax)
for wc in word_counts:
if wc >= wmax:
idx = n_bins - 1
else:
idx = int((wc - wmin) / width)
if idx < 0:
idx = 0
elif idx >= n_bins:
idx = n_bins - 1
bins[idx]["count"] += 1
return bins
def compute_text_length_stats(texts, n_bins=20) -> dict:
"""Summarize the length distribution of a corpus of text documents.
For each document three lengths are measured: characters (``len(doc)``),
words (count of ``\\w+`` unicode tokens) and sentences (non-empty segments
after splitting on ``.!?…``, with a minimum of 1 for any non-empty
document). For each axis the mean, p50, p90, p99, min and max are reported,
plus an equal-width histogram of the per-document word counts.
``None`` entries and any non-``str`` items in ``texts`` are discarded.
The function never raises: on empty/``None`` input or any internal error it
returns the empty-shape result (``n_docs`` 0, all-``None`` axes, ``[]``
histogram).
Args:
texts: List of text documents (``str``). ``None`` and non-``str``
items are dropped.
n_bins: Number of equal-width bins for the word-count histogram.
Default 20.
Returns:
Dict with keys ``n_docs``, ``chars``, ``words``, ``sentences`` and
``word_hist``. Each of the three axes is a sub-dict with ``mean``
(float, 2 decimals), ``p50``, ``p90``, ``p99``, ``min`` and ``max``
(ints), all ``None`` when there are no documents. ``word_hist`` is a
list of ``{lo, hi, count}`` bins whose ``count`` sums to ``n_docs``.
"""
empty_axis = _empty_axis()
fallback = {
"n_docs": 0,
"chars": dict(empty_axis),
"words": dict(empty_axis),
"sentences": dict(empty_axis),
"word_hist": [],
}
try:
if not texts:
return fallback
docs = [t for t in texts if isinstance(t, str)]
n_docs = len(docs)
if n_docs == 0:
return fallback
char_counts = [len(d) for d in docs]
word_counts = [len(_WORD_RE.findall(d)) for d in docs]
sent_counts = []
for d in docs:
segments = [s for s in _SENT_RE.split(d) if s.strip()]
n = len(segments)
if d and n == 0:
# Non-empty document with no detectable sentence: count as 1.
n = 1
sent_counts.append(n)
return {
"n_docs": n_docs,
"chars": _axis_stats(char_counts),
"words": _axis_stats(word_counts),
"sentences": _axis_stats(sent_counts),
"word_hist": _word_hist(word_counts, n_bins),
}
except Exception:
return fallback
@@ -1,70 +0,0 @@
"""Tests para compute_text_length_stats.
Inserta `python/functions` en sys.path (relativo a este archivo) para importar
el modulo hoja por su paquete `datascience`, sin depender de que el paquete lo
reexporte en su __init__.
"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from datascience.compute_text_length_stats import compute_text_length_stats
def test_basico():
"""Varios textos de longitudes distintas: stats y histograma coherentes."""
texts = [
"Hola mundo.", # 2 words, 1 sentence
"Una frase mas larga con varias palabras aqui.", # 8 words, 1 sentence
"Corto.", # 1 word, 1 sentence
"Esto. Tiene. Tres frases distintas!", # 5 words, 3 sentences
]
result = compute_text_length_stats(texts)
assert result["n_docs"] == 4
# Diferentes longitudes en palabras -> max estrictamente mayor que min.
assert result["words"]["max"] > result["words"]["min"]
# El histograma de palabras no esta vacio.
assert result["word_hist"] != []
# La suma de counts del histograma cubre todos los documentos.
assert sum(b["count"] for b in result["word_hist"]) == result["n_docs"]
# mean es float redondeado; min/max son enteros.
assert isinstance(result["words"]["mean"], float)
assert isinstance(result["words"]["min"], int)
assert isinstance(result["words"]["max"], int)
# El documento con 3 frases empuja el max de sentences a >= 3.
assert result["sentences"]["max"] >= 3
def test_vacio():
"""Lista vacia: n_docs 0, subdicts None, word_hist []."""
result = compute_text_length_stats([])
assert result["n_docs"] == 0
for axis in ("chars", "words", "sentences"):
for key in ("mean", "p50", "p90", "p99", "min", "max"):
assert result[axis][key] is None
assert result["word_hist"] == []
def test_descarta_none():
"""None y valores no-str se descartan del computo."""
result = compute_text_length_stats(["hello world", None, 123, 4.5, "foo bar baz"])
# Solo dos strings validos.
assert result["n_docs"] == 2
assert result["words"]["min"] == 2 # "hello world"
assert result["words"]["max"] == 3 # "foo bar baz"
assert sum(b["count"] for b in result["word_hist"]) == 2
def test_un_documento():
"""Un solo documento: word_hist tiene exactamente un bin con count 1."""
result = compute_text_length_stats(["solo un documento aqui"])
assert result["n_docs"] == 1
assert len(result["word_hist"]) == 1
assert result["word_hist"][0]["count"] == 1
# Con un unico documento, p50 == min == max == su numero de palabras (4).
assert result["words"]["min"] == 4
assert result["words"]["max"] == 4
assert result["words"]["p50"] == 4
@@ -1,88 +0,0 @@
---
id: compute_text_readability_py_datascience
name: compute_text_readability
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: pure
signature: "def compute_text_readability(texts, sample_max=500) -> dict"
description: "Calcula la legibilidad Flesch Reading Ease de un corpus de texto usando textstat con import perezoso y degradación. Filtra None/no-str/vacíos, muestrea hasta sample_max documentos (los primeros) y agrega los scores Flesch en {mean, p50, min, max}. Si textstat no está instalada devuelve available=False sin lanzar. Estilo dict-no-throw del grupo eda — nunca lanza."
tags: [eda, datascience, text, nlp, readability, flesch, textstat, pure, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [math, textstat]
example: |
from datascience.compute_text_readability import compute_text_readability
out = compute_text_readability(["The cat sat on the mat. It was warm and sunny."])
# {"available": True, "n_scored": 1, "flesch": {"mean": 109.0, "p50": 109.0, "min": 108.96..., "max": 108.96...}}
tested: true
tests:
- "test_prosa_ingles"
- "test_vacio"
- "test_degradacion"
test_file_path: "python/functions/datascience/compute_text_readability_test.py"
file_path: "python/functions/datascience/compute_text_readability.py"
params:
- name: texts
desc: "Lista de str (documentos del corpus). Los elementos None, no-str o vacíos tras strip() se descartan silenciosamente. El orden se respeta: el muestreo toma los primeros documentos válidos."
- name: sample_max
desc: "Número máximo de documentos válidos a puntuar (los primeros). Default 500. Acota el coste en corpus grandes. Valores no convertibles a int caen a 500; negativos se tratan como 0."
output: "Dict con exactamente 3 claves siempre presentes: available (bool: True si textstat se pudo importar), n_scored (int: nº de documentos efectivamente puntuados), flesch (dict con mean, p50, min, max). mean y p50 redondeados a 1 decimal; p50 por nearest-rank sobre los scores ordenados; min/max son los scores extremos sin redondear. Todos los valores de flesch son None cuando n_scored es 0. La función nunca lanza: cualquier excepción global (incluida ImportError de textstat) degrada a available=False, n_scored=0 y flesch todo None."
---
## Ejemplo
```python
from datascience.compute_text_readability import compute_text_readability
textos = [
"The cat sat on the mat. It was a warm and sunny day in the park.",
"Reading is a wonderful habit. Books open doors to new worlds and ideas.",
"He ran quickly to the store to buy some fresh bread and a bottle of milk.",
]
compute_text_readability(textos)
# {
# "available": True,
# "n_scored": 3,
# "flesch": {"mean": 91.4, "p50": 95.4, "min": 70.08..., "max": 108.83...}
# }
# Corpus vacío (textstat presente): available True pero nada que puntuar.
compute_text_readability([])
# {"available": True, "n_scored": 0,
# "flesch": {"mean": None, "p50": None, "min": None, "max": None}}
```
## Cuando usarla
Úsala en un EDA de texto cuando necesites una métrica única y comparable de
**lo fácil que es de leer** un corpus de documentos (descripciones, reviews,
artículos, tickets). Devuelve el resumen Flesch Reading Ease agregado
(`mean`/`p50`/`min`/`max`) listo para un report o un bloque del notebook, sin
tener que iterar `textstat` a mano. Pásale la lista de textos crudos y, si el
corpus es grande, limita el coste con `sample_max`. El estilo dict-no-throw
permite incrustarla en pipelines del grupo `eda` sin envolver en try/except.
## Gotchas
- **`textstat` es una dependencia opcional.** Si no está instalada (o falla al
importar) la función NO lanza: devuelve `available=False`, `n_scored=0` y
`flesch` todo `None`. Comprueba `available` antes de interpretar los números.
- **Flesch Reading Ease está pensado para prosa en inglés.** Aplicado a otros
idiomas o a texto no-prosa (código, listas, tablas, cadenas muy cortas) los
scores no son interpretables, aunque se calculen sin error.
- **Escala Flesch:** valores **altos** = más fácil de leer (≈90100 muy fácil),
valores **bajos** = más difícil (puede ser negativo en texto muy denso). No
se recortan a ningún rango: se reportan tal cual los devuelve `textstat`.
- **`available=True` con `n_scored=0`** significa que `textstat` está presente
pero el corpus no aportó documentos puntuables (vacío, solo None/no-str, o
todos los docs fallaron al puntuar). Es distinto de `available=False`.
- **Muestreo = los primeros `sample_max`**, no aleatorio. Si el orden del corpus
está sesgado, el resumen reflejará ese sesgo.
- **`mean` y `p50` redondean a 1 decimal**; `min`/`max` se devuelven sin
redondear (los scores extremos reales).
@@ -1,121 +0,0 @@
"""Legibilidad Flesch Reading Ease de un corpus de texto.
Función pura del grupo `eda`, estilo dict-no-throw: nunca lanza. Usa la
librería `textstat` con import perezoso y degradación: si `textstat` no está
instalada (o falla al importar), devuelve un resultado con `available=False`
en lugar de propagar el error.
"""
def _percentile_nearest_rank(sorted_values, pct):
"""Percentil por nearest-rank sobre una lista ya ordenada ascendente.
rank = ceil(pct/100 * n); índice 1-based recortado a [1, n].
Devuelve None si la lista está vacía.
"""
n = len(sorted_values)
if n == 0:
return None
import math
rank = math.ceil((pct / 100.0) * n)
if rank < 1:
rank = 1
if rank > n:
rank = n
return sorted_values[rank - 1]
def compute_text_readability(texts, sample_max=500) -> dict:
"""Calcula la legibilidad Flesch Reading Ease de un corpus.
Args:
texts: lista de str. Los elementos None, no-str o vacíos (tras strip)
se descartan. Se muestrean los primeros `sample_max` documentos
válidos.
sample_max: número máximo de documentos a puntuar (los primeros).
Returns:
Dict con la forma exacta::
{"available": bool, "n_scored": int,
"flesch": {"mean": float|None, "p50": float|None,
"min": float|None, "max": float|None}}
`available` es True si `textstat` se pudo importar. La función nunca
lanza: cualquier excepción global degrada a `available=False`.
"""
empty = {
"available": False,
"n_scored": 0,
"flesch": {"mean": None, "p50": None, "min": None, "max": None},
}
try:
# Import perezoso con degradación: textstat es una dependencia opcional.
try:
import textstat
except Exception:
return {
"available": False,
"n_scored": 0,
"flesch": {"mean": None, "p50": None, "min": None, "max": None},
}
# Filtrar y muestrear documentos válidos (los primeros sample_max).
docs = []
if texts is not None:
try:
limit = int(sample_max)
except Exception:
limit = 500
if limit < 0:
limit = 0
for item in texts:
if not isinstance(item, str):
continue
if item.strip() == "":
continue
docs.append(item)
if len(docs) >= limit:
break
scores = []
for doc in docs:
try:
score = textstat.flesch_reading_ease(doc)
except Exception:
continue
try:
score = float(score)
except Exception:
continue
scores.append(score)
n_scored = len(scores)
if n_scored == 0:
# textstat presente pero corpus vacío / sin puntuar.
return {
"available": True,
"n_scored": 0,
"flesch": {"mean": None, "p50": None, "min": None, "max": None},
}
mean_val = round(sum(scores) / n_scored, 1)
sorted_scores = sorted(scores)
p50_raw = _percentile_nearest_rank(sorted_scores, 50)
p50_val = round(p50_raw, 1) if p50_raw is not None else None
min_val = sorted_scores[0]
max_val = sorted_scores[-1]
return {
"available": True,
"n_scored": n_scored,
"flesch": {
"mean": mean_val,
"p50": p50_val,
"min": min_val,
"max": max_val,
},
}
except Exception:
return empty
@@ -1,74 +0,0 @@
"""Tests para compute_text_readability."""
import sys
import os
import builtins
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
from datascience.compute_text_readability import compute_text_readability
EXPECTED_KEYS = {"available", "n_scored", "flesch"}
FLESCH_KEYS = {"mean", "p50", "min", "max"}
def test_prosa_ingles():
"""Varios textos en prosa inglesa: available True, n_scored>0, mean no None."""
texts = [
"The cat sat on the mat. It was a warm and sunny day in the park.",
"She sells sea shells by the sea shore. The shells she sells are surely sea shells.",
"Reading is a wonderful habit. Books open doors to new worlds and ideas.",
"He ran quickly to the store to buy some fresh bread and a bottle of milk.",
]
out = compute_text_readability(texts)
assert set(out.keys()) == EXPECTED_KEYS
assert out["available"] is True
assert out["n_scored"] > 0
assert set(out["flesch"].keys()) == FLESCH_KEYS
assert out["flesch"]["mean"] is not None
assert out["flesch"]["p50"] is not None
assert out["flesch"]["min"] is not None
assert out["flesch"]["max"] is not None
# min <= mean/p50 <= max coherente.
assert out["flesch"]["min"] <= out["flesch"]["max"]
def test_vacio():
"""Corpus vacío con textstat presente: available True, n_scored 0, flesch None."""
out = compute_text_readability([])
assert set(out.keys()) == EXPECTED_KEYS
assert out["available"] is True
assert out["n_scored"] == 0
assert out["flesch"]["mean"] is None
assert out["flesch"]["p50"] is None
assert out["flesch"]["min"] is None
assert out["flesch"]["max"] is None
# Elementos no-str / vacíos también se descartan -> n_scored 0.
out2 = compute_text_readability([None, "", " ", 123])
assert out2["available"] is True
assert out2["n_scored"] == 0
def test_degradacion(monkeypatch):
"""Sin textstat (ImportError forzado): degrada a available False sin lanzar."""
import datascience.compute_text_readability as m
real = builtins.__import__
def fake(name, *a, **k):
if name == "textstat" or name.startswith("textstat."):
raise ImportError("simulado")
return real(name, *a, **k)
monkeypatch.setattr(builtins, "__import__", fake)
out = m.compute_text_readability(["The cat sat on the mat. It was happy and warm."])
assert out["available"] is False
assert out["n_scored"] == 0
assert out["flesch"]["mean"] is None
assert out["flesch"]["p50"] is None
assert out["flesch"]["min"] is None
assert out["flesch"]["max"] is None
@@ -1,103 +0,0 @@
---
id: compute_top_ngrams_py_datascience
name: compute_top_ngrams
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: pure
signature: "def compute_top_ngrams(texts, n=2, top_k=15, remove_stopwords=True) -> dict"
description: "Calcula los n-gramas de palabras más frecuentes de un corpus de texto (n=1 unigramas, 2 bigramas, 3 trigramas...). Tokeniza a minúsculas con re.findall(r'\\w+', ...), descarta tokens numéricos y, si remove_stopwords=True, elimina stopwords ES+EN ANTES de formar los n-gramas (n-gramas contiguos sobre la secuencia de tokens de contenido, sin cruzar documentos). Pura y autocontenida con collections.Counter, sin sklearn. Estilo dict-no-throw del grupo eda: nunca lanza."
tags: [eda, datascience, text, nlp, ngrams, bigrams, trigrams, pure, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [re, collections]
example: |
from datascience.compute_top_ngrams import compute_top_ngrams
texts = ["machine learning rocks", "we love machine learning"]
compute_top_ngrams(texts, n=2, top_k=5)
# {"n": 2, "top": [{"ngram": "machine learning", "count": 2}, ...]}
tested: true
tests:
- "test_bigramas"
- "test_trigramas"
- "test_vacio"
- "test_stopwords"
test_file_path: "python/functions/datascience/compute_top_ngrams_test.py"
file_path: "python/functions/datascience/compute_top_ngrams.py"
params:
- name: texts
desc: "Lista (o tupla) de cadenas. Los elementos None o que no sean str se descartan silenciosamente. Cada documento se tokeniza por separado; los n-gramas no cruzan la frontera entre documentos."
- name: n
desc: "Tamaño del n-grama: 1 unigramas, 2 bigramas, 3 trigramas, etc. Valores < 1 o no enteros producen top vacío (se conserva tal cual en la clave 'n' del retorno)."
- name: top_k
desc: "Número máximo de n-gramas a devolver, ordenados por frecuencia descendente con desempate alfabético determinista. Default 15. Valores negativos se tratan como 0."
- name: remove_stopwords
desc: "Si True (default) elimina las stopwords ES+EN de una lista inline (~130 términos de altísima frecuencia) ANTES de formar los n-gramas, de modo que los n-gramas se construyen sobre la secuencia de tokens de contenido."
output: "Dict con exactamente 2 claves: n (el n recibido, sin normalizar) y top (lista de dicts {'ngram': str, 'count': int} ordenada por count descendente, longitud <= top_k). ngram es la unión de los tokens del n-grama por un espacio. Corpus vacío, tokens insuficientes para formar n-gramas o cualquier excepción interna degradan a {'n': n, 'top': []}. La función nunca lanza."
---
## Ejemplo
```python
from datascience.compute_top_ngrams import compute_top_ngrams
texts = [
"machine learning rocks",
"machine learning is fun",
"we love machine learning",
]
# Bigramas (n=2): "machine learning" aparece en los 3 documentos.
compute_top_ngrams(texts, n=2, top_k=5)
# {
# "n": 2,
# "top": [
# {"ngram": "machine learning", "count": 3},
# {"ngram": "learning fun", "count": 1},
# {"ngram": "learning rocks", "count": 1},
# {"ngram": "love machine", "count": 1},
# ],
# }
# Unigramas con stopwords fuera (default): solo palabras de contenido.
compute_top_ngrams(["the cat sat on the mat"], n=1, top_k=3)
# {"n": 1, "top": [{"ngram": "cat", "count": 1},
# {"ngram": "mat", "count": 1},
# {"ngram": "sat", "count": 1}]}
```
## Cuando usarla
Úsala en la fase de EDA de texto cuando, además del vocabulario suelto, necesites
ver qué **combinaciones de palabras contiguas** dominan un corpus: colocaciones,
frases técnicas recurrentes ("machine learning", "data analyst"), o patrones de
trigramas en titulares/descripciones. Es el complemento natural de un perfil de
vocabulario: pasa de "qué palabras aparecen" a "qué secuencias aparecen". Llámala
con `n=1` para unigramas, `n=2` para bigramas y `n=3` para trigramas, y ajusta
`top_k` al tamaño de la tabla que vas a renderizar. Deja `remove_stopwords=True`
para que los n-gramas reflejen contenido y no conectores gramaticales.
## Gotchas
- **Las stopwords se eliminan ANTES de formar los n-gramas.** Con
`remove_stopwords=True` la frase "data of analysis" produce el bigrama
"data analysis" (el "of" intermedio desaparece y los tokens de contenido se
vuelven contiguos), no "data of" ni "of analysis". Si quieres preservar la
adyacencia literal del texto original, pasa `remove_stopwords=False`.
- **Los n-gramas NO cruzan documentos.** Cada elemento de `texts` se tokeniza y
recorre por separado; el último token de un documento nunca se combina con el
primero del siguiente.
- **Tokens puramente numéricos se descartan** (`tok.isdigit()`), pero los
alfanuméricos mixtos no: "3d" o "covid19" sí cuentan como tokens. Un decimal
como "3.5" se parte en "3" y "5" por `\w+` y ambos se descartan por numéricos.
- **La lista de stopwords es inline ES+EN**, pensada para textos generales en
esos dos idiomas. Para otros idiomas o jerga específica de dominio puede dejar
pasar conectores; en ese caso filtra el corpus aguas arriba o usa
`remove_stopwords=False` y posfiltra.
- **`top` puede tener menos de `top_k` elementos** si el corpus no tiene tantos
n-gramas distintos. El desempate por frecuencia es alfabético (determinista),
no por orden de aparición.
@@ -1,94 +0,0 @@
"""Top n-gramas de palabras más frecuentes de un corpus de texto.
Función pura, autocontenida (solo stdlib: re + collections.Counter). No depende
de scikit-learn ni de ninguna otra librería externa. Estilo dict-no-throw del
grupo `eda`: ante cualquier entrada degenerada o excepción interna devuelve
``{"n": n, "top": []}`` en vez de lanzar.
"""
import re
from collections import Counter
# Lista inline de stopwords ES + EN (~80 términos de altísima frecuencia).
# Se eliminan ANTES de formar los n-gramas: los n-gramas se construyen sobre la
# secuencia de tokens de contenido, no sobre el texto original.
_STOPWORDS = frozenset({
# Español
"de", "la", "que", "el", "en", "y", "a", "los", "del", "se", "las", "por",
"un", "para", "con", "no", "una", "su", "al", "lo", "como", "más", "mas",
"pero", "sus", "le", "ya", "o", "este", "", "si", "porque", "esta",
"entre", "cuando", "muy", "sin", "sobre", "también", "tambien", "me",
"hasta", "hay", "donde", "quien", "desde", "todo", "nos", "durante",
"todos", "uno", "les", "ni", "contra", "otros", "ese", "eso", "ante",
"ellos", "e", "esto", "", "antes", "algunos", "qué", "unos", "yo",
"otro", "otras", "otra", "él", "tanto", "esa", "estos", "mucho", "quienes",
"nada", "muchos", "cual", "poco", "ella", "estar", "estas", "algunas",
"algo", "nosotros",
# Inglés
"the", "of", "and", "to", "in", "is", "it", "for", "on", "with", "as",
"are", "was", "be", "this", "that", "by", "an", "or", "at", "from", "but",
"not", "have", "has", "had", "they", "you", "we", "he", "she", "his",
"her", "their", "its", "i", "my", "me", "our", "us", "do", "does", "did",
"will", "would", "can", "could", "should", "there", "which", "who", "what",
"when", "where", "how", "all", "if", "so", "than", "then", "out", "up",
})
def compute_top_ngrams(texts, n=2, top_k=15, remove_stopwords=True) -> dict:
"""Calcula los n-gramas de palabras más frecuentes de un corpus.
Args:
texts: lista de cadenas. Los elementos ``None`` o que no sean ``str`` se
descartan silenciosamente.
n: tamaño del n-grama (1 = unigramas, 2 = bigramas, 3 = trigramas...).
Valores < 1 o no enteros producen ``top`` vacío.
top_k: número máximo de n-gramas a devolver, ordenados por frecuencia
descendente (con desempate alfabético determinista).
remove_stopwords: si ``True`` elimina las stopwords ES+EN ANTES de
formar los n-gramas, de modo que los n-gramas se construyen sobre la
secuencia de tokens de contenido (no cruzando documentos).
Returns:
``{"n": n, "top": [{"ngram": "w1 w2", "count": int}, ...]}``. Corpus
vacío, sin tokens suficientes o cualquier excepción interna degrada a
``{"n": n, "top": []}``. Nunca lanza.
"""
try:
if not isinstance(n, int) or n < 1:
return {"n": n, "top": []}
try:
limit = int(top_k)
except (TypeError, ValueError):
limit = 0
if limit < 0:
limit = 0
if not isinstance(texts, (list, tuple)):
return {"n": n, "top": []}
counter = Counter()
for doc in texts:
if not isinstance(doc, str):
continue
tokens = [
tok
for tok in re.findall(r"\w+", doc.lower(), re.UNICODE)
if not tok.isdigit()
]
if remove_stopwords:
tokens = [tok for tok in tokens if tok not in _STOPWORDS]
if len(tokens) < n:
continue
for i in range(len(tokens) - n + 1):
ngram = " ".join(tokens[i:i + n])
counter[ngram] += 1
if not counter:
return {"n": n, "top": []}
ordered = sorted(counter.items(), key=lambda kv: (-kv[1], kv[0]))
top = [{"ngram": ngram, "count": count} for ngram, count in ordered[:limit]]
return {"n": n, "top": top}
except Exception:
return {"n": n, "top": []}
@@ -1,65 +0,0 @@
"""Tests para compute_top_ngrams."""
import sys
import os
# sys.path estándar: añade `python/functions/` para importar por paquete raíz.
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
from datascience.compute_top_ngrams import compute_top_ngrams
def test_bigramas():
# "machine learning" se repite en cada documento -> bigrama más frecuente.
texts = [
"machine learning rocks",
"machine learning is fun",
"we love machine learning",
]
result = compute_top_ngrams(texts, n=2, top_k=5)
assert result["n"] == 2
assert result["top"], "esperaba al menos un bigrama"
assert result["top"][0]["ngram"] == "machine learning"
assert result["top"][0]["count"] == 3
# Cada entrada respeta el contrato {"ngram": str, "count": int}.
for item in result["top"]:
assert isinstance(item["ngram"], str)
assert isinstance(item["count"], int)
def test_trigramas():
texts = [
"alpha beta gamma delta",
"alpha beta gamma omega",
]
# Con stopwords desactivadas para no descartar tokens de contenido.
result = compute_top_ngrams(texts, n=3, top_k=5, remove_stopwords=False)
assert result["n"] == 3
ngrams = {item["ngram"]: item["count"] for item in result["top"]}
# "alpha beta gamma" aparece en ambos documentos.
assert ngrams.get("alpha beta gamma") == 2
# Trigramas únicos de cada documento.
assert ngrams.get("beta gamma delta") == 1
assert ngrams.get("beta gamma omega") == 1
def test_vacio():
assert compute_top_ngrams([], n=2) == {"n": 2, "top": []}
# Documentos no-str / None se descartan -> corpus efectivamente vacío.
assert compute_top_ngrams([None, 123, {"a": 1}], n=2) == {"n": 2, "top": []}
def test_stopwords():
# "the cat" debería desaparecer al quitar stopwords ("the" es stopword EN).
texts = ["the cat the cat the cat"]
con = compute_top_ngrams(texts, n=2, top_k=10, remove_stopwords=True)
sin = compute_top_ngrams(texts, n=2, top_k=10, remove_stopwords=False)
con_ngrams = {item["ngram"] for item in con["top"]}
sin_ngrams = {item["ngram"] for item in sin["top"]}
# Sin filtrar, el bigrama dominante es "the cat".
assert "the cat" in sin_ngrams
# Al filtrar stopwords, ya no aparece "the cat" (queda solo "cat cat").
assert "the cat" not in con_ngrams
assert con_ngrams != sin_ngrams
@@ -1,91 +0,0 @@
---
id: compute_vocabulary_stats_py_datascience
name: compute_vocabulary_stats
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: pure
signature: "def compute_vocabulary_stats(texts: list, top_k: int = 20, remove_stopwords: bool = True) -> dict"
description: "Profiles the vocabulary of a text corpus for EDA: tokenises a list of documents, counts term frequencies and derives lexical-richness measures — total tokens, unique types, type-token ratio (TTR), hapax legomena and the top-k most frequent terms. Pure, stdlib only (re + collections.Counter); no nltk, no sklearn. Inline ES+EN stopword list, opt-out via remove_stopwords. Never raises: empty/degenerate input returns the zeroed result."
tags: [eda, datascience, text, nlp, vocabulary, ttr, hapax, pure, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [re, collections]
example: |
from datascience.compute_vocabulary_stats import compute_vocabulary_stats
result = compute_vocabulary_stats(["el gato y el perro", "gato veloz"], top_k=5)
tested: true
tests:
- "test_basico"
- "test_vacio"
- "test_stopwords_quitadas"
- "test_stopwords_conservadas"
test_file_path: "python/functions/datascience/compute_vocabulary_stats_test.py"
file_path: "python/functions/datascience/compute_vocabulary_stats.py"
params:
- name: texts
desc: "List of documents (strings) forming the corpus. Entries that are None or not a str are silently discarded. Tokens are extracted per document with re.findall(r'\\w+', doc.lower(), re.UNICODE); purely numeric tokens (tok.isdigit()) are dropped."
- name: top_k
desc: "Maximum number of most-frequent terms to return in top_terms. Default 20. Does not affect n_tokens/n_types/ttr/hapax — only the length of the top_terms list."
- name: remove_stopwords
desc: "When True (default) common Spanish+English stopwords from the inline _STOPWORDS set (~120 entries) are removed from the token stream before any counting. Set False to keep every word (raw lexical profile)."
output: "Dict with the exact keys n_tokens (int), n_types (int), ttr (float|None, n_types/n_tokens rounded to 4 dp), n_hapax (int, terms occurring exactly once), hapax_pct (float|None, n_hapax/n_types*100 rounded to 2 dp) and top_terms (list of {term, count, pct} sorted by count descending, pct = count/n_tokens*100 rounded to 2 dp). For an empty corpus (no tokens after filtering): n_tokens=0, n_types=0, ttr=None, n_hapax=0, hapax_pct=None, top_terms=[]. Any exception degrades to that same empty result — the function never throws."
---
## Ejemplo
```python
from datascience.compute_vocabulary_stats import compute_vocabulary_stats
compute_vocabulary_stats(
["el gato y el perro", "gato veloz corre", "perro perro perro"],
top_k=5,
)
# {
# "n_tokens": 6, # stopwords (el, y) eliminadas por defecto
# "n_types": 3, # gato, perro, veloz, corre -> tras quitar stopwords
# "ttr": 0.5, # n_types / n_tokens
# "n_hapax": 2, # veloz, corre (1 aparicion cada uno)
# "hapax_pct": 50.0, # n_hapax / n_types * 100
# "top_terms": [
# {"term": "perro", "count": 4, "pct": 44.44},
# {"term": "gato", "count": 2, "pct": 22.22},
# ...
# ],
# }
# Perfil lexico crudo (sin filtrar stopwords):
compute_vocabulary_stats(["the cat and the dog"], remove_stopwords=False)
```
## Cuando usarla
Úsala al perfilar una columna o corpus de texto libre en un EDA del grupo `eda`:
cuando necesites medir la riqueza léxica (cuántos tokens y cuántas palabras
distintas, type-token ratio, porcentaje de palabras que solo aparecen una vez) y
ver qué términos dominan el vocabulario (top-k frecuencias). Pásale la lista de
documentos crudos (filas de la columna); `None` y valores no-string se ignoran
solos. Es el equivalente para texto largo de `summarize_categorical`, que perfila
categorías cortas.
## Gotchas
- Función pura y stdlib-only, pero el resultado depende del **idioma**: la lista
`_STOPWORDS` cubre español e inglés. Para otros idiomas pon
`remove_stopwords=False` o filtra fuera, o el perfil mezclará stopwords no
reconocidas en `top_terms`.
- La tokenización es `\w+` con `re.UNICODE`: separa por puntuación y conserva
acentos/ñ, pero NO hace stemming ni lematización — "gato" y "gatos" cuentan
como tipos distintos. Tampoco hace stripping de acentos, así que "más" (con
tilde) y "mas" son tokens diferentes (ambos están en la stoplist).
- Los tokens **puramente numéricos** (`"123"`) se descartan siempre; un token
alfanumérico mixto (`"covid19"`) se conserva.
- `ttr` baja artificialmente en corpus grandes (más texto, más repetición): no
compares TTR entre corpus de tamaños muy distintos sin normalizar.
- Nunca lanza: entrada vacía, `None`, o cualquier excepción interna devuelven el
resultado con ceros/`None`/`[]`. Comprueba `n_tokens == 0` para detectar el
caso degenerado.
@@ -1,99 +0,0 @@
"""Profile the vocabulary of a text corpus for EDA (pure, stdlib only).
Tokenises a list of documents, counts term frequencies and derives lexical
richness measures (type-token ratio, hapax legomena) plus the top-k terms.
No external NLP dependencies (no nltk, no sklearn) — only ``re`` and
``collections`` from the standard library.
"""
import re
from collections import Counter
# Common Spanish + English stopwords. Inline, lowercase, no accents stripped
# beyond what already appears here. Filtering is opt-in via remove_stopwords.
_STOPWORDS = {
# Spanish
"de", "la", "que", "el", "en", "y", "a", "los", "del", "se", "las", "por",
"un", "para", "con", "no", "una", "su", "al", "es", "lo", "como", "mas",
"más", "pero", "sus", "le", "ya", "o", "este", "si", "", "porque",
"esta", "entre", "cuando", "muy", "sin", "sobre", "tambien", "también",
"me", "hasta", "hay", "donde", "quien", "desde", "todo", "nos", "durante",
"todos", "uno", "les", "ni", "contra", "otros", "ese", "eso", "ante",
"ellos", "e", "esto", "antes", "algunos", "que", "unos", "yo", "otro",
"otras", "otra", "el", "tanto", "esa", "estos", "mucho", "nada", "muchos",
# English
"the", "of", "and", "to", "in", "is", "it", "for", "on", "with", "as",
"was", "but", "are", "this", "that", "an", "be", "by", "or", "not", "at",
"from", "my", "i", "you", "he", "she", "we", "they", "his", "her", "its",
"our", "their", "what", "which", "who", "whom", "has", "have", "had", "do",
"does", "did", "will", "would", "can", "could", "should", "may", "might",
"must", "if", "then", "than", "so", "too", "very", "just", "also", "were",
"been", "being", "there", "here", "all", "any", "some", "more", "most",
"out", "up", "down", "into", "over", "such", "only", "own", "same",
}
def compute_vocabulary_stats(texts, top_k=20, remove_stopwords=True) -> dict:
"""Profile the vocabulary of a corpus of documents.
Args:
texts: List of strings (the corpus). Entries that are None or not a
string are discarded silently.
top_k: Maximum number of most-frequent terms to include in
``top_terms``. Default 20. Does not affect the other measures.
remove_stopwords: When True (default) common ES+EN stopwords are
dropped from the token stream before any counting.
Returns:
A dict with the exact keys ``n_tokens``, ``n_types``, ``ttr``,
``n_hapax``, ``hapax_pct`` and ``top_terms``. For an empty corpus (no
tokens after filtering): n_tokens=0, n_types=0, ttr=None, n_hapax=0,
hapax_pct=None, top_terms=[]. Never raises — any exception degrades to
the empty-corpus result.
"""
empty = {
"n_tokens": 0,
"n_types": 0,
"ttr": None,
"n_hapax": 0,
"hapax_pct": None,
"top_terms": [],
}
try:
tokens = []
for doc in texts or []:
if not isinstance(doc, str):
continue
for tok in re.findall(r"\w+", doc.lower(), re.UNICODE):
if tok.isdigit():
continue
if remove_stopwords and tok in _STOPWORDS:
continue
tokens.append(tok)
n_tokens = len(tokens)
if n_tokens == 0:
return dict(empty)
counts = Counter(tokens)
n_types = len(counts)
ttr = round(n_types / n_tokens, 4)
n_hapax = sum(1 for c in counts.values() if c == 1)
hapax_pct = round(n_hapax / n_types * 100, 2)
top_terms = [
{"term": term, "count": count, "pct": round(count / n_tokens * 100, 2)}
for term, count in counts.most_common(top_k)
]
return {
"n_tokens": n_tokens,
"n_types": n_types,
"ttr": ttr,
"n_hapax": n_hapax,
"hapax_pct": hapax_pct,
"top_terms": top_terms,
}
except Exception:
return dict(empty)
@@ -1,74 +0,0 @@
"""Tests para compute_vocabulary_stats."""
import os
import sys
sys.path.insert(
0, os.path.join(os.path.dirname(__file__), "..", "..", "functions")
)
from datascience.compute_vocabulary_stats import compute_vocabulary_stats
def test_basico():
# Corpus con repeticiones y hapax. Stopwords desactivadas para controlar
# exactamente que tokens entran.
texts = ["gato gato perro", "perro perro raton", "elefante"]
r = compute_vocabulary_stats(texts, top_k=10, remove_stopwords=False)
# n_types < n_tokens cuando hay repeticiones.
assert r["n_types"] < r["n_tokens"]
assert r["n_tokens"] == 7
assert r["n_types"] == 4 # gato, perro, raton, elefante
# ttr en (0, 1].
assert 0 < r["ttr"] <= 1
assert r["ttr"] == round(4 / 7, 4)
# top_terms ordenado por count descendente.
counts = [t["count"] for t in r["top_terms"]]
assert counts == sorted(counts, reverse=True)
assert r["top_terms"][0]["term"] == "perro"
assert r["top_terms"][0]["count"] == 3
# hapax: raton y elefante aparecen exactamente una vez.
assert r["n_hapax"] == 2
assert r["hapax_pct"] == round(2 / 4 * 100, 2)
# pct coherente con count/n_tokens.
assert r["top_terms"][0]["pct"] == round(3 / 7 * 100, 2)
def test_vacio():
# Sin documentos validos -> ceros / None / [].
for arg in ([], None, [None, 123, ""], ["123 456"]):
r = compute_vocabulary_stats(arg)
assert r["n_tokens"] == 0
assert r["n_types"] == 0
assert r["ttr"] is None
assert r["n_hapax"] == 0
assert r["hapax_pct"] is None
assert r["top_terms"] == []
def test_stopwords_quitadas():
texts = ["the gato the perro", "de la casa azul"]
r = compute_vocabulary_stats(texts, remove_stopwords=True)
terms = {t["term"] for t in r["top_terms"]}
# Stopwords ES+EN no deben aparecer.
assert "the" not in terms
assert "de" not in terms
assert "la" not in terms
# Palabras de contenido si.
assert "gato" in terms
assert "casa" in terms
def test_stopwords_conservadas():
texts = ["the gato the perro", "de la casa azul"]
r = compute_vocabulary_stats(texts, remove_stopwords=False)
terms = {t["term"] for t in r["top_terms"]}
# Con el filtro desactivado, las stopwords se conservan.
assert "the" in terms
assert "de" in terms
assert "la" in terms
@@ -1,80 +0,0 @@
---
name: detect_corpus_language
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: pure
signature: "def detect_corpus_language(texts, top_k=10, sample_max=1000) -> dict"
description: "Estima la distribucion de idiomas de un corpus de textos con la libreria langdetect (import perezoso). Funcion pura y defensiva del grupo eda: filtra documentos None/no-str/vacios, muestrea hasta sample_max docs, clasifica cada uno con detect() ignorando los que langdetect no puede resolver (LangDetectException), y devuelve la distribucion top_k por frecuencia mas el idioma dominante. Si langdetect no esta instalada o algo falla, degrada a {available: False, ...} y NUNCA lanza (dict-no-throw). Seed fija (DetectorFactory.seed=0) para deteccion determinista."
tags: [eda, datascience, text, nlp, language-detection, langdetect, pure, python]
params:
- name: texts
desc: "Lista de strings (documentos). Los elementos None, no-str o vacios tras strip se descartan antes de clasificar."
- name: top_k
desc: "Numero maximo de idiomas a devolver en distribution, ordenados por count descendente (desempate por codigo ISO ascendente). Default 10."
- name: sample_max
desc: "Numero maximo de documentos a clasificar (se toman los primeros del corpus) para acotar el coste. Default 1000."
output: >
Dict con forma fija (dict-no-throw, nunca lanza):
{"available": bool, "n_detected": int,
"distribution": [{"lang": str, "count": int, "pct": float}, ...],
"dominant": str|None}.
available=True si langdetect es importable; lang son codigos ISO 639-1 ("es","en","fr",...);
pct = count/n_detected*100 redondeado a 2 decimales; n_detected = docs clasificados con exito;
dominant = idioma mas frecuente (None si no hubo detecciones). Corpus vacio con langdetect
presente -> available True, n_detected 0, distribution [], dominant None. Sin langdetect (o
fallo global) -> available False y el resto de campos a su valor vacio.
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [langdetect]
tested: true
tests: ["test_mixto_es_en", "test_vacio", "test_degradacion"]
test_file_path: "python/functions/datascience/detect_corpus_language_test.py"
file_path: "python/functions/datascience/detect_corpus_language.py"
---
## Ejemplo
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
from datascience.detect_corpus_language import detect_corpus_language
corpus = [
"este es un texto bastante largo en español para detectar el idioma correctamente",
"la inteligencia artificial transforma la manera en que trabajamos cada dia",
"this is a fairly long english text to detect the language correctly without issues",
]
out = detect_corpus_language(corpus)
# {"available": True, "n_detected": 3,
# "distribution": [{"lang": "es", "count": 2, "pct": 66.67},
# {"lang": "en", "count": 1, "pct": 33.33}],
# "dominant": "es"}
```
## Cuando usarla
Cuando perfiles una columna o corpus de texto en un EDA y necesites saber en
que idioma(s) esta escrito antes de elegir tokenizadores, stopwords, modelos
NLP o stemmers. Util tambien como check de calidad: detectar corpus mezclados
o un idioma inesperado. Llamala con la lista de textos crudos; la funcion
limpia, muestrea y resume sola.
## Gotchas
- `langdetect` es **opcional**: si no esta instalada, la funcion no lanza —
devuelve `{"available": False, "n_detected": 0, "distribution": [], "dominant": None}`.
Comprueba `out["available"]` antes de usar la distribucion.
- **Textos cortos** (pocas palabras o sin features lingüisticas) pueden no
detectarse: langdetect lanza `LangDetectException`, que se ignora y el doc no
cuenta en `n_detected`. Pasa frases razonablemente largas para resultados fiables.
- **Determinismo**: se fija `DetectorFactory.seed = 0` en cada llamada para que la
deteccion sea reproducible; sin esa semilla langdetect puede dar resultados
ligeramente distintos entre ejecuciones.
- `distribution` esta truncada a `top_k`; si el corpus tiene mas idiomas que
`top_k`, la suma de los `count` mostrados puede ser menor que `n_detected`
(pero `dominant` siempre refleja el idioma mas frecuente del corpus completo).
@@ -1,91 +0,0 @@
"""Detecta la distribucion de idiomas de un corpus de textos.
Funcion pura y defensiva: el computo es determinista y local (sin I/O de red).
La libreria opcional `langdetect` se importa de forma perezosa dentro de la
funcion; si no esta instalada (o cualquier paso falla), la funcion degrada
limpiamente a `available=False` y NUNCA lanza excepciones.
"""
def detect_corpus_language(texts, top_k=10, sample_max=1000) -> dict:
"""Estima la distribucion de idiomas de un corpus con `langdetect`.
Args:
texts: lista de strings (documentos). Los elementos None, no-str o
vacios tras strip se descartan.
top_k: numero maximo de idiomas a devolver en `distribution`,
ordenados por frecuencia descendente.
sample_max: numero maximo de documentos a clasificar (se toman los
primeros) para acotar el coste.
Returns:
dict con la forma fija (dict-no-throw):
{
"available": bool, # True si langdetect es importable
"n_detected": int, # documentos clasificados con exito
"distribution": [{"lang": str, "count": int, "pct": float}, ...],
"dominant": str | None,
}
"""
degraded = {
"available": False,
"n_detected": 0,
"distribution": [],
"dominant": None,
}
try:
# Import perezoso con degradacion: si langdetect no esta disponible,
# devolvemos el dict degradado sin lanzar.
try:
from langdetect import detect, DetectorFactory
# Semilla fija -> deteccion determinista entre ejecuciones.
DetectorFactory.seed = 0
except Exception:
return dict(degraded)
# Normaliza y filtra el corpus.
docs = []
if texts:
for t in texts:
if isinstance(t, str):
s = t.strip()
if s:
docs.append(s)
# Muestreo de los primeros `sample_max` documentos.
if sample_max is not None and sample_max >= 0:
docs = docs[:sample_max]
# Conteo por idioma; langdetect lanza LangDetectException en textos
# sin features detectables -> se ignora y se sigue.
counts: dict = {}
for doc in docs:
try:
lang = detect(doc)
except Exception:
continue
counts[lang] = counts.get(lang, 0) + 1
n_detected = sum(counts.values())
# Orden estable: por count descendente, desempate por codigo de idioma.
ordered = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
k = top_k if (top_k is not None and top_k >= 0) else len(ordered)
distribution = []
for lang, count in ordered[:k]:
pct = round(count / n_detected * 100, 2) if n_detected else 0.0
distribution.append({"lang": lang, "count": count, "pct": pct})
dominant = ordered[0][0] if ordered else None
return {
"available": True,
"n_detected": n_detected,
"distribution": distribution,
"dominant": dominant,
}
except Exception:
# Cualquier fallo global degrada a available False sin lanzar.
return dict(degraded)
@@ -1,58 +0,0 @@
"""Tests para detect_corpus_language."""
import builtins
import os
import sys
# Anade python/functions a sys.path para importar el paquete `datascience`.
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from datascience.detect_corpus_language import detect_corpus_language
_ES = [
"este es un texto bastante largo en español para detectar el idioma correctamente sin problemas",
"la inteligencia artificial transforma la manera en que trabajamos cada dia en muchos sectores",
]
_EN = [
"this is a fairly long english text to detect the language correctly without any length issues",
"machine learning models can classify documents into many different categories quite reliably",
]
def test_mixto_es_en():
"""Golden: corpus mixto ES+EN claro -> available True, >=2 idiomas, counts coherentes."""
out = detect_corpus_language(_ES + _EN)
assert out["available"] is True
assert out["dominant"] in {"es", "en"}
assert len(out["distribution"]) >= 2
total = sum(item["count"] for item in out["distribution"])
assert total == out["n_detected"]
assert out["n_detected"] == 4
def test_vacio():
"""Edge: lista vacia con langdetect presente -> available True, sin detecciones."""
out = detect_corpus_language([])
assert out["available"] is True
assert out["n_detected"] == 0
assert out["distribution"] == []
assert out["dominant"] is None
def test_degradacion(monkeypatch):
"""Error path: si langdetect no es importable -> degrada a available False sin lanzar."""
import datascience.detect_corpus_language as m
real_import = builtins.__import__
def fake_import(name, *a, **k):
if name == "langdetect" or name.startswith("langdetect."):
raise ImportError("simulado")
return real_import(name, *a, **k)
monkeypatch.setattr(builtins, "__import__", fake_import)
out = m.detect_corpus_language(["hola mundo", "hello world"])
assert out["available"] is False
assert out["n_detected"] == 0
assert out["distribution"] == []
assert out["dominant"] is None
@@ -0,0 +1,97 @@
---
name: extract_null_mask
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: impure
signature: "def extract_null_mask(query_fn, table: str, columns: list, max_rows: int = 5000) -> dict"
description: "Extrae la mascara de nulos (1=falta / 0=presente) de una muestra de filas de una tabla, una lista 0/1 por columna alineada por fila, para alimentar el capitulo de calidad / patron de nulos de AutomaticEDA sin que el capitulo toque la base de datos. Recibe un lector read-only inyectado `query_fn(sql) -> dict` (mismo contrato que duckdb_query_readonly / pg_query / el `_q` de profile_table) y NO abre ninguna conexion por su cuenta. Construye UNA sola query que proyecta por cada columna `CASE WHEN \"col\" IS NULL THEN 1 ELSE 0 END` con identificadores escapados y LIMIT. Devuelve dict dict-no-throw: columns (efectivamente leidas, en orden), mask (lista int 0/1 por columna, misma longitud todas) y n. Una celda None se cuenta defensivamente como 1 (falta)."
tags: [eda, nulls, missing, datascience, automatic-eda, extraction, read-only, duckdb, postgres, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: []
params:
- name: query_fn
desc: "callable lector read-only del backend activo. Recibe un string SQL y devuelve un dict {'status':'ok','rows':[{col:val,...},...]} (mismo contrato que duckdb_query_readonly o el `_q` de profile_table). NO se abre ninguna conexion dentro de la funcion: toda la lectura pasa por query_fn. Si es None -> error."
- name: table
desc: "nombre de la tabla de la que muestrear la mascara de nulos. Se escapa con comillas dobles en la query. Vacio o None -> status error."
- name: columns
desc: "lista de nombres de columna a evaluar. Cada una produce una entrada en `mask` con una lista 0/1 paralela por fila (1=IS NULL, 0=presente). Cada nombre se escapa con comillas dobles. Vacia o None -> status error."
- name: max_rows
desc: "limite de filas a muestrear (clausula LIMIT). Default 5000. Protege frente a tablas enormes; con LIMIT obtienes el primer tramo, no un muestreo uniforme."
output: "dict (nunca lanza). En exito: {'status':'ok','table':str,'columns':[str,...] (en orden),'mask':{col:[int 0/1,...],...} (1=falta/IS NULL, 0=presente; todas las listas con misma longitud = n),'n':int}. En error (sin lanzar): {'status':'error','error':str,'table':str,'columns':[],'mask':{},'n':0}. Errores: query_fn None, table vacia, columns vacia, o query_fn devuelve status!='ok' (se propaga su error)."
tested: true
tests: ["test_golden_mask_alineada", "test_celda_none_cuenta_como_falta", "test_columns_vacia_status_error", "test_query_fn_status_error_propaga", "test_query_fn_none_da_error_sin_reventar", "test_sql_contiene_case_y_limit"]
test_file_path: "python/functions/datascience/extract_null_mask_test.py"
file_path: "python/functions/datascience/extract_null_mask.py"
---
## Ejemplo
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
from datascience.extract_null_mask import extract_null_mask
from infra import duckdb_query_readonly
# El lector read-only se inyecta como closure (igual que el `_q` de profile_table).
db = "data/clientes.duckdb"
def _q(sql):
return duckdb_query_readonly(db, sql)
res = extract_null_mask(_q, "clientes", ["email", "telefono", "edad"])
# res == {
# "status": "ok",
# "table": "clientes",
# "columns": ["email", "telefono", "edad"],
# "mask": {
# "email": [0, 0, 1, 0, ...], # fila 2 sin email
# "telefono": [1, 0, 1, 0, ...],
# "edad": [0, 0, 0, 1, ...],
# },
# "n": 5000,
# }
# % de nulos por columna a partir de la muestra:
pct = {c: 100 * sum(bits) / max(res["n"], 1) for c, bits in res["mask"].items()}
# Se entrega al capitulo de calidad sin que este toque la BD:
ctx = {"null_mask": res}
```
## Cuando usarla
Cuando el capitulo de calidad / patron de nulos de AutomaticEDA necesita saber
DONDE faltan los valores (no solo cuantos) y NO debe abrir la base de datos por
su cuenta: extraes aqui la mascara 0/1 por columna alineada por fila y se la pasas
en `ctx['null_mask']`. Usala siempre que quieras detectar co-ocurrencia de nulos
(filas que fallan en varias columnas a la vez), calcular el % de nulos sobre una
muestra, o pintar un heatmap de missingness reutilizando un unico lector read-only
inyectado, en vez de hacer N `COUNT(*) WHERE col IS NULL` por separado.
## Gotchas
- **Impura**: lee de la base de datos a traves de `query_fn`. No abre conexiones
por su cuenta — depende por completo del lector inyectado. Sigue el estilo
dict-no-throw del grupo `eda`: nunca lanza; ante cualquier fallo devuelve
`{"status":"error","error":...}` con `columns=[]`, `mask={}`, `n=0`.
- **`error_type` en el frontmatter es `error_go_core` por convencion del registry**
(toda funcion impura debe declararlo y el indexer lo exige), pero el codigo
NO lanza esa excepcion: degrada al dict de error. Es metadata, no comportamiento.
- **Muestra, no censo**: con `LIMIT max_rows` obtienes el primer tramo de filas que
devuelva el backend, no un muestreo uniforme ni la tabla entera. El % de nulos
derivado es una estimacion sobre esa muestra; para el conteo exacto usa un
agregado `COUNT(*)`/`COUNT(col)` aparte.
- **Alineacion por fila**: `mask[col][i]` corresponde a la misma fila `i` que
`mask[otra_col][i]`. Todas las listas tienen longitud `n`, asi que puedes cruzar
columnas por indice (co-ocurrencia de nulos) sin re-alinear.
- **Defensa None -> 1**: el SQL ya devuelve 0/1, pero si una celda llega como `None`
(CASE no aplicado, columna ausente en la fila, backend que nulifica) se cuenta
como 1 (falta). Un valor inesperado no convertible a int se trata como presente (0).
- **No loguear los datos crudos**: aunque `mask` es solo 0/1, los nombres de columna
pueden revelar el esquema. En trazas usa `n` y el numero de columnas, no el dict
completo.
@@ -0,0 +1,101 @@
"""extract_null_mask — extrae la mascara de nulos (1=falta / 0=presente) de una tabla.
Lector read-only inyectado: recibe `query_fn(sql) -> dict` con el mismo contrato
que duckdb_query_readonly / pg_query (y que el `_q` de profile_table):
`{"status": "ok", "rows": [{col: val, ...}, ...]}`. Esta funcion NO abre ninguna
conexion por su cuenta — solo usa `query_fn`. Construye UNA sola query que, por
cada columna pedida, evalua `CASE WHEN "col" IS NULL THEN 1 ELSE 0 END` y devuelve
una muestra de filas con esos bits. El resultado es un dict `mask` con una lista
0/1 por columna, alineada por fila (1 = el valor falta / IS NULL, 0 = presente),
listo para alimentar el capitulo de calidad / patron de nulos de AutomaticEDA sin
que el capitulo toque la base de datos.
Estilo dict-no-throw del grupo `eda`: nunca lanza; captura cualquier excepcion y
degrada a `{"status": "error", "error": str, ...}`.
"""
def _to_bit(value):
"""Coacciona el valor 0/1 del CASE a int de forma defensiva.
El SQL ya devuelve 0 (presente) o 1 (falta). Por si una celda llega como None
(el CASE no se aplico o el backend la nulifico), se cuenta como 1 (falta). El
resto se reduce a int: un entero distinto de 0 cuenta como 1 (falta), 0 como
presente. Un valor no convertible se trata como presente (0) — nunca lanza.
"""
if value is None:
return 1
try:
return 1 if int(value) != 0 else 0
except (TypeError, ValueError):
return 0
def extract_null_mask(query_fn, table, columns, max_rows=5000):
"""Extrae la mascara de nulos (1=falta / 0=presente) de una muestra de la tabla.
Args:
query_fn: callable lector read-only del backend activo. Recibe un string
SQL y devuelve un dict {"status": "ok", "rows": [{col: val, ...}]}
(mismo contrato que duckdb_query_readonly / el `_q` de profile_table).
No se abre ninguna conexion aqui: toda la lectura pasa por query_fn.
table: nombre de la tabla. Se escapa con comillas dobles en la query.
columns: lista de nombres de columna a evaluar. Cada una produce una
entrada en `mask` con una lista 0/1 paralela por fila. Vacia o None ->
status error.
max_rows: limite de filas a muestrear (clausula LIMIT). Default 5000.
Returns:
dict (nunca lanza):
{
"status": "ok" | "error",
"error": str, # solo si status == "error"
"table": str,
"columns": [str, ...], # columnas efectivamente leidas, en orden
"mask": {col: [int 0/1, ...], ...}, # alineada por fila, 1=falta, 0=presente
"n": int # nº de filas muestreadas
}
Todas las listas de `mask` tienen la misma longitud (= n).
"""
base = {"status": "ok", "table": table, "columns": [], "mask": {}, "n": 0}
try:
if query_fn is None:
return {**base, "status": "error", "error": "query_fn es None"}
if not table:
return {**base, "status": "error", "error": "table es obligatorio"}
if not columns:
return {**base, "status": "error", "error": "columns vacío"}
# Identificadores escapados con comillas dobles (como hace profile_table)
# para tolerar nombres con mayusculas/espacios/palabras reservadas. Cada
# columna se proyecta como su propio bit IS NULL conservando el alias.
select_sql = ", ".join(
f'(CASE WHEN "{c}" IS NULL THEN 1 ELSE 0 END) AS "{c}"' for c in columns
)
sql = f'SELECT {select_sql} FROM "{table}" LIMIT {int(max_rows)}'
q = query_fn(sql)
if not isinstance(q, dict) or q.get("status") != "ok":
err = (
q.get("error", "query_fn fallo")
if isinstance(q, dict)
else "query_fn no devolvio un dict"
)
return {**base, "status": "error", "error": err}
rows = q.get("rows", []) or []
mask = {c: [] for c in columns}
for row in rows:
for c in columns:
# row.get tolera filas que no traigan la columna (None -> falta).
mask[c].append(_to_bit(row.get(c) if isinstance(row, dict) else None))
return {
"status": "ok",
"table": table,
"columns": list(columns),
"mask": mask,
"n": len(rows),
}
except Exception as e: # noqa: BLE001 - dict-no-throw: degradar, nunca lanzar
return {**base, "status": "error", "error": str(e)}
@@ -0,0 +1,116 @@
"""Tests para extract_null_mask.
No usa DuckDB real: inyecta un query_fn FAKE (closure) que devuelve filas
predefinidas (simulando el SELECT de bits 0/1) y, opcionalmente, captura el SQL
recibido para verificar la query generada (CASE WHEN ... IS NULL + LIMIT). Asi el
test es autocontenido y no depende de ningun backend.
"""
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
from extract_null_mask import extract_null_mask
def _fake_query(rows, captured=None, status="ok", error=None):
"""Crea un query_fn FAKE.
`captured` (lista opcional) recibe el SQL ejecutado para poder inspeccionarlo.
`status`/`error` permiten simular un fallo del backend.
"""
def _q(sql):
if captured is not None:
captured.append(sql)
if status != "ok":
return {"status": "error", "error": error or "boom"}
return {"status": "ok", "rows": rows}
return _q
def test_golden_mask_alineada():
"""Golden: mask 0/1 por columna alineada por fila, n correcto, status ok."""
# Cada fila simula el SELECT (CASE WHEN col IS NULL THEN 1 ELSE 0 END) AS col.
rows = [
{"email": 0, "telefono": 1, "edad": 0},
{"email": 0, "telefono": 0, "edad": 1},
{"email": 1, "telefono": 1, "edad": 0},
]
res = extract_null_mask(_fake_query(rows), "clientes", ["email", "telefono", "edad"])
assert res["status"] == "ok"
assert res["table"] == "clientes"
assert res["columns"] == ["email", "telefono", "edad"]
assert res["n"] == 3
assert res["mask"]["email"] == [0, 0, 1]
assert res["mask"]["telefono"] == [1, 0, 1]
assert res["mask"]["edad"] == [0, 1, 0]
# Todas las listas con la misma longitud.
assert all(len(v) == res["n"] for v in res["mask"].values())
def test_celda_none_cuenta_como_falta():
"""Una celda None se cuenta defensivamente como 1 (falta)."""
rows = [
{"email": 0, "telefono": None},
{"email": None, "telefono": 1},
{"email": 1, "telefono": 0},
]
res = extract_null_mask(_fake_query(rows), "clientes", ["email", "telefono"])
assert res["status"] == "ok"
assert res["mask"]["email"] == [0, 1, 1]
assert res["mask"]["telefono"] == [1, 1, 0]
assert res["n"] == 3
def test_columns_vacia_status_error():
"""columns vacia -> status error con columns/mask/n vacios."""
res = extract_null_mask(_fake_query([]), "clientes", [])
assert res["status"] == "error"
assert "columns" in res["error"]
assert res["table"] == "clientes"
assert res["columns"] == []
assert res["mask"] == {}
assert res["n"] == 0
def test_query_fn_status_error_propaga():
"""query_fn que devuelve status != ok -> se propaga como error, mask {}."""
res = extract_null_mask(
_fake_query([], status="error", error="db locked"),
"clientes",
["email"],
)
assert res["status"] == "error"
assert "db locked" in res["error"]
assert res["mask"] == {}
assert res["n"] == 0
def test_query_fn_none_da_error_sin_reventar():
"""query_fn None -> error degradado, sin excepcion."""
res = extract_null_mask(None, "clientes", ["email"])
assert res["status"] == "error"
assert res["columns"] == []
assert res["mask"] == {}
assert res["n"] == 0
def test_sql_contiene_case_y_limit():
"""La query genera un CASE WHEN IS NULL por columna escapada + LIMIT sobre la tabla."""
captured = []
rows = [{"email": 0}]
extract_null_mask(
_fake_query(rows, captured),
"clientes_tbl",
["email"],
max_rows=123,
)
assert len(captured) == 1
sql = captured[0]
assert 'CASE WHEN "email" IS NULL THEN 1 ELSE 0 END' in sql
assert 'AS "email"' in sql
assert 'FROM "clientes_tbl"' in sql
assert "LIMIT 123" in sql
@@ -1,102 +0,0 @@
---
name: extract_text_sample
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: impure
signature: "def extract_text_sample(db_path: str, table: str, columns: list, backend: str = 'duckdb', sample: int = 2000) -> dict"
description: "Muestrea columnas de texto de una tabla DuckDB/Postgres con push-down SQL (LIMIT sample), SIN traer la tabla entera a RAM. Funcion impura del grupo de capacidad `eda`: la usan los capitulos de texto/NLP del AutomaticEDA que necesitan valores crudos de texto (longitudes, tokens, ejemplos) sobre una muestra acotada. Construye el lector read-only query_fn(sql)->dict igual que build_eda_render_ctx (closure sobre duckdb_query_readonly / pg_query importados perezosamente desde infra). Escapa los identificadores con comillas dobles y lanza una sola query SELECT \"c1\", \"c2\" FROM \"table\" LIMIT n. Por columna, la lista de strings solo contiene valores NO None y NO vacios: cada celda no nula se convierte con str(...) y se descarta si queda cadena vacia. Estilo dict-no-throw del grupo eda: NUNCA lanza; ante cualquier fallo (query, conversion, backend desconocido) devuelve {status:'error', error:str, columns:{}, n:0}. La clave n reporta el numero de FILAS leidas por la query (antes de filtrar None/vacios)."
tags: [eda, datascience, text, nlp, extraction, read-only, duckdb, postgres, python]
uses_functions: [duckdb_query_readonly_py_infra, pg_query_py_infra]
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: []
params:
- name: db_path
desc: "ruta al archivo DuckDB, o DSN PostgreSQL si backend='postgres'. Se inyecta en el closure query_fn. No se valida aqui: si la base no existe o el DSN es invalido, la query devuelve status error y el resultado es {status:'error', ...} (no lanza)."
- name: table
desc: "nombre de la tabla. Se escapa con comillas dobles en la query (SELECT ... FROM \"table\")."
- name: columns
desc: "lista de nombres de columna de texto a muestrear. Se filtra a las entradas que sean str no vacio; cada nombre se escapa con comillas dobles. Si tras filtrar queda vacia -> {status:'ok', columns:{}, n:0} sin tocar la base."
- name: backend
desc: "'duckdb' (default) o 'postgres'. Selecciona el lector read-only del registry (duckdb_query_readonly / pg_query). Cualquier otro valor -> {status:'error', error:'backend desconocido: <valor>', columns:{}, n:0}."
- name: sample
desc: "maximo de filas a muestrear (clausula LIMIT). Default 2000. Acota memoria y tiempo: con tablas grandes obtienes el primer tramo por orden fisico (sin ORDER BY), no un muestreo uniforme."
output: "dict dict-no-throw (NUNCA lanza): {status:'ok'|'error', columns:{col_name:[str,...]}, n:int, error:str}. En exito (status='ok') columns mapea cada columna pedida a la lista de sus valores de texto NO None y NO vacios (cada celda convertida con str(...)); n es el numero de FILAS leidas por la query (antes de filtrar None/vacios). columns vacio -> {status:'ok', columns:{}, n:0}. En error (backend desconocido, query con status!='ok', o cualquier excepcion) -> {status:'error', error:str, columns:{}, n:0}; la clave error solo aparece en este caso."
tested: true
tests: ["test_extract_basic", "test_backend_desconocido", "test_columns_vacio", "test_sample_limit"]
test_file_path: "python/functions/datascience/extract_text_sample_test.py"
file_path: "python/functions/datascience/extract_text_sample.py"
---
## Ejemplo
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
# Import directo del submodulo (no requiere export en datascience/__init__.py).
from datascience.extract_text_sample import extract_text_sample
# Muestrea hasta 2000 filas de dos columnas de texto de una tabla DuckDB.
res = extract_text_sample(
"data/reviews.duckdb", "reviews", ["title", "body"],
backend="duckdb", sample=2000,
)
# res == {
# "status": "ok",
# "columns": {
# "title": ["Gran producto", "No funciona", ...], # solo no-None, no-""
# "body": ["Lo uso a diario...", ...],
# },
# "n": 2000, # filas leidas por la query (antes de filtrar None/vacios)
# }
# Postgres: db_path es el DSN.
res_pg = extract_text_sample(
"postgresql://user:pass@localhost:5433/trends", "comentarios", ["texto"],
backend="postgres", sample=500,
)
```
## Cuando usarla
Cuando necesites valores CRUDOS de texto de una o varias columnas para analisis
NLP/texto (distribucion de longitudes, conteo de tokens, ejemplos representativos,
deteccion de idioma) pero NO quieras cargar la tabla entera en memoria. Es el
muestreador de texto del grupo `eda`: una sola llamada con push-down `LIMIT`
devuelve listas de strings por columna, limpias de None y vacios, listas para
alimentar un capitulo de texto del AutomaticEDA o cualquier rutina de tokenizado.
Usala junto a `profile_table` / `build_eda_render_ctx` cuando el perfil agregado
no basta y hace falta el texto real.
## Gotchas
- **Impura**: lee de la base de datos a traves de `query_fn` (closure sobre
`duckdb_query_readonly` / `pg_query`). No abre conexiones fuera de esos wrappers
del registry. Estilo dict-no-throw del grupo `eda`: NUNCA lanza; ante cualquier
fallo devuelve `{status:'error', error:str, columns:{}, n:0}`.
- **`error_type` en el frontmatter es `error_go_core` por convencion del registry**
(toda funcion impura debe declararlo y el indexer lo exige), pero el codigo NO
lanza esa excepcion: degrada al dict de error. Es metadata, no comportamiento.
- **Backend desconocido**: con un `backend` que no sea `duckdb` ni `postgres`
devuelve `{status:'error', error:'backend desconocido: <valor>', columns:{},
n:0}` sin tocar la base.
- **Las listas NO incluyen None ni cadenas vacias**: cada celda no nula se pasa
por `str(...)` y se descarta si queda `""`. Por eso `len(columns[col])` puede ser
menor que `n` (que cuenta las filas leidas). Si necesitas alineacion por fila
(una entrada por fila aunque sea None), usa `build_eda_render_ctx` (raw_numeric),
no esta funcion.
- **`LIMIT sample` sin `ORDER BY`**: con tablas grandes obtienes el primer tramo
por orden fisico del backend, no un muestreo uniforme ni reproducible. Sube
`sample` para mas cobertura, o pre-ordena/aleatoriza la tabla si necesitas
representatividad.
- **DuckDB en sandbox por defecto**: `duckdb_query_readonly` abre la conexion con
`enable_external_access=False`, asi que la query solo puede leer la propia base
(no `read_csv`/`httpfs`/`ATTACH` a paths externos). Lee tablas ya existentes en
el archivo DuckDB sin problema.
- **No loguear los datos crudos**: las listas de `columns` pueden contener texto
sensible (reviews, comentarios, PII). En trazas usa solo conteos (`n`,
`len(columns[col])`) y nombres de columna, no el dict completo.
@@ -1,112 +0,0 @@
"""extract_text_sample — muestrea columnas de texto de una tabla sin cargarla en RAM.
Funcion impura (lee de la base de datos) del grupo de capacidad `eda`. Dado un
``db_path`` + ``table`` (DuckDB o PostgreSQL) y una lista de ``columns`` de texto,
trae una MUESTRA de esas columnas con push-down SQL (``LIMIT sample``), nunca la
tabla entera. La usan los capitulos de texto/NLP del AutomaticEDA que necesitan
valores crudos de texto (longitudes, tokens, ejemplos) sin materializar millones
de filas en memoria.
El lector read-only ``query_fn(sql) -> dict`` se construye igual que en
``build_eda_render_ctx`` / ``profile_table``: un closure sobre el wrapper del
registry (``duckdb_query_readonly`` / ``pg_query``), importado perezosamente
dentro de la funcion para no crear ciclos al cargar el ``__init__`` del paquete
``datascience``. Nunca abre conexiones fuera de esos wrappers.
Estilo dict-no-throw del grupo `eda`: la funcion NUNCA lanza. Captura cualquier
excepcion (query, conversion) y devuelve ``{"status":"error", "error":str(e),
"columns":{}, "n":0}``. Si la query subyacente devuelve ``status != "ok"``, se
propaga como error con el mensaje del wrapper.
Por columna, la lista de strings solo contiene valores NO nulos y NO vacios:
cada celda no-None se convierte con ``str(...)`` y se descarta si queda ``""``.
La clave ``n`` reporta el numero de FILAS leidas por la query (antes de filtrar
los None/vacios), util para saber cuanto se muestreo realmente.
"""
def extract_text_sample(db_path, table, columns, backend="duckdb", sample=2000):
"""Muestrea columnas de texto de una tabla DuckDB/Postgres con push-down SQL.
Args:
db_path: ruta al archivo DuckDB, o DSN PostgreSQL si backend="postgres".
Se inyecta en el closure query_fn. No se valida aqui: si la base no
existe o el DSN es invalido, la query devuelve status error y el
resultado es {status:'error', ...} (no lanza).
table: nombre de la tabla. Se escapa con comillas dobles en la query.
columns: lista de nombres de columna de texto a muestrear. Se filtra a las
entradas que sean str no vacio; cada nombre se escapa con comillas
dobles. Si tras filtrar queda vacia -> {status:'ok', columns:{}, n:0}.
backend: "duckdb" (default) o "postgres". Selecciona el lector read-only
del registry (duckdb_query_readonly / pg_query). Cualquier otro valor
-> {status:'error', error:'backend desconocido: ...', columns:{}, n:0}.
sample: maximo de filas a muestrear (clausula LIMIT). Default 2000. Acota
memoria y tiempo: con tablas grandes obtienes el primer tramo por
orden fisico, no un muestreo uniforme.
Returns:
dict (dict-no-throw, NUNCA lanza):
{"status": "ok"|"error",
"columns": {col_name: [str, str, ...], ...}, # solo no-None, no-""
"n": int, # nº de filas leidas por la query (antes de filtrar)
"error": str} # solo presente si status == "error"
"""
try:
# 1) Lector read-only del backend activo, construido como en
# build_eda_render_ctx (closure sobre el wrapper del registry). Imports
# perezosos: este modulo vive en el paquete `datascience`, importar a
# `infra` a nivel de modulo crearia un ciclo al cargar el __init__.
if backend == "duckdb":
from infra import duckdb_query_readonly
def query_fn(sql):
return duckdb_query_readonly(db_path, sql)
elif backend == "postgres":
from infra import pg_query
def query_fn(sql):
return pg_query(db_path, sql)
else:
return {
"status": "error",
"error": f"backend desconocido: {backend}",
"columns": {},
"n": 0,
}
# 2) Columnas validas (str no vacio). Si no queda ninguna, nada que
# muestrear: ok con columns vacio.
cols = []
if isinstance(columns, (list, tuple)):
cols = [c for c in columns if isinstance(c, str) and c != ""]
if not cols:
return {"status": "ok", "columns": {}, "n": 0}
# 3) Push-down: una sola query con LIMIT. Identificadores escapados con
# comillas dobles, igual que build_eda_render_ctx.
cols_sql = ", ".join(f'"{c}"' for c in cols)
sql = f'SELECT {cols_sql} FROM "{table}" LIMIT {int(sample)}'
q = query_fn(sql)
if not isinstance(q, dict) or q.get("status") != "ok":
err = q.get("error") if isinstance(q, dict) else "query sin resultado"
return {"status": "error", "error": str(err), "columns": {}, "n": 0}
rows = q.get("rows") or []
out = {c: [] for c in cols}
for row in rows:
if not isinstance(row, dict):
continue
for c in cols:
value = row.get(c)
if value is None:
continue
s = str(value)
if s == "":
continue
out[c].append(s)
return {"status": "ok", "columns": out, "n": len(rows)}
except Exception as exc: # noqa: BLE001 - dict-no-throw del grupo eda
return {"status": "error", "error": str(exc), "columns": {}, "n": 0}
@@ -1,83 +0,0 @@
"""Tests para extract_text_sample.
Self-contained: crea un DuckDB temporal pequeño con una columna de texto (algunas
filas con NULL) y una numerica, y verifica que la muestra de texto trae solo los
valores no nulos, que el backend desconocido y la lista de columnas vacia se
manejan dict-no-throw, y que sample acota el numero de filas leidas.
"""
import os
import sys
_HERE = os.path.dirname(os.path.abspath(__file__))
_FUNCTIONS = os.path.abspath(os.path.join(_HERE, "..")) # python/functions
if _FUNCTIONS not in sys.path:
sys.path.insert(0, _FUNCTIONS)
import duckdb # noqa: E402
from datascience.extract_text_sample import extract_text_sample # noqa: E402
_TABLE = "t"
# 6 filas: txt VARCHAR con dos NULL, other INT siempre presente.
_ROWS = [
("alpha", 1),
("beta", 2),
(None, 3),
("gamma", 4),
(None, 5),
("delta", 6),
]
_TXT_NON_NULL = {"alpha", "beta", "gamma", "delta"}
def _make_db(tmp_path):
"""Crea un DuckDB temporal con la tabla de prueba y devuelve su ruta."""
db_path = os.path.join(str(tmp_path), "text_sample.duckdb")
con = duckdb.connect(db_path)
try:
con.execute(f'CREATE TABLE "{_TABLE}" (txt VARCHAR, other INTEGER)')
con.executemany(f'INSERT INTO "{_TABLE}" VALUES (?, ?)', _ROWS)
finally:
con.close()
return db_path
def test_extract_basic(tmp_path):
db_path = _make_db(tmp_path)
res = extract_text_sample(db_path, _TABLE, ["txt"])
assert res["status"] == "ok"
# n = filas leidas por la query (6), antes de filtrar None.
assert res["n"] == len(_ROWS)
# columns["txt"] trae solo los strings no nulos (los dos NULL fuera).
assert "txt" in res["columns"]
assert set(res["columns"]["txt"]) == _TXT_NON_NULL
assert len(res["columns"]["txt"]) == len(_TXT_NON_NULL)
# No se pidio "other", no debe aparecer.
assert "other" not in res["columns"]
def test_backend_desconocido(tmp_path):
db_path = _make_db(tmp_path)
res = extract_text_sample(db_path, _TABLE, ["txt"], backend="mysql")
assert res["status"] == "error"
assert "backend desconocido" in res["error"]
assert res["columns"] == {}
assert res["n"] == 0
def test_columns_vacio(tmp_path):
db_path = _make_db(tmp_path)
res = extract_text_sample(db_path, _TABLE, [])
assert res["status"] == "ok"
assert res["columns"] == {}
assert res["n"] == 0
def test_sample_limit(tmp_path):
db_path = _make_db(tmp_path)
res = extract_text_sample(db_path, _TABLE, ["txt"], sample=2)
assert res["status"] == "ok"
# sample=2 -> la query lee como mucho 2 filas.
assert res["n"] == 2
assert len(res["columns"]["txt"]) <= 2
@@ -0,0 +1,103 @@
---
id: missingness_corr_heatmap_figure_py_datascience
name: missingness_corr_heatmap_figure
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: impure
signature: "def missingness_corr_heatmap_figure(matrix, labels, title=\"Co-ocurrencia de ausencias\") -> \"matplotlib.figure.Figure\""
description: "Construye una figura matplotlib (heatmap) de la matriz NxN de correlación de ausencias entre columnas: +1 = dos columnas suelen ser nulas a la vez, -1 = cuando una falta la otra está presente, 0 = ausencias independientes. Usa ax.imshow con coolwarm fijado a [-1,1], ticks con los labels truncados (X rotados 45º), colorbar y anota el valor de cada celda si N<=12. Devuelve un matplotlib.figure.Figure listo para rasterizar por el renderer del informe EDA (capítulo de datos faltantes). Backend Agg sin pyplot global; defensivo ante matrix/labels vacíos o celdas no numéricas (nunca lanza)."
tags: [eda, missing, missingness, correlation, heatmap, matplotlib, figure, visualization, datascience, impure]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [matplotlib]
example: |
from datascience.missingness_corr_heatmap_figure import missingness_corr_heatmap_figure
matrix = [
[1.0, 0.82, -0.10],
[0.82, 1.0, 0.05],
[-0.10, 0.05, 1.0],
]
labels = ["telefono", "movil", "email"]
fig = missingness_corr_heatmap_figure(matrix, labels, title="Co-ocurrencia de ausencias")
tested: true
tests:
- "test_returns_figure_with_axes"
- "test_empty_matrix_does_not_raise_and_returns_figure"
- "test_empty_labels_returns_message_figure"
- "test_large_matrix_omits_annotations"
- "test_ragged_and_non_numeric_cells_are_handled"
test_file_path: "python/functions/datascience/missingness_corr_heatmap_figure_test.py"
file_path: "python/functions/datascience/missingness_corr_heatmap_figure.py"
params:
- name: matrix
desc: "Lista de listas (NxN) de floats en [-1,1]: la correlación de ausencias por pares de columnas. Puede venir vacía. Filas de longitud desigual se toleran (se rellenan/recortan a N); celdas None, NaN o no numéricas se coercen a 0.0. No se muta el original."
- name: labels
desc: "Lista de N nombres de columna, paralela a matrix. Puede venir vacía (devuelve figura \"sin columnas con ausencia variable\"). Se truncan a ~14 chars con elipsis para los ticks; los originales no se mutan."
- name: title
desc: "Título de la figura. Se trunca a ~60 chars con elipsis si es muy largo. Default \"Co-ocurrencia de ausencias\"."
output: "Un matplotlib.figure.Figure (figsize 6.4x5.2, dpi 150) con un Axes heatmap (imshow vmin=-1, vmax=1, cmap coolwarm) más una colorbar etiquetada \"correlación de ausencias\". Ticks en ambos ejes con los labels truncados (X rotados 45º). Si N<=12 cada celda lleva su valor numérico anotado (texto blanco sobre celdas saturadas, oscuro sobre pálidas); con N grande se omiten las anotaciones para no saturar. Si matrix o labels vienen vacíos devuelve una Figure con texto centrado \"sin columnas con ausencia variable\"; cualquier error inesperado se captura y devuelve una Figure con el mensaje de error (nunca lanza). El caller rasteriza/cierra la figura; la función no la muestra ni la guarda."
---
## Ejemplo
```python
from datascience.missingness_corr_heatmap_figure import missingness_corr_heatmap_figure
# Correlación de ausencias entre 3 columnas de contacto:
# telefono y movil tienden a faltar juntos (0.82); email es casi independiente.
matrix = [
[1.00, 0.82, -0.10],
[0.82, 1.00, 0.05],
[-0.10, 0.05, 1.00],
]
labels = ["telefono", "movil", "email"]
fig = missingness_corr_heatmap_figure(
matrix,
labels,
title="Co-ocurrencia de ausencias",
)
# El renderer del informe lo rasteriza; aquí solo persistimos para inspección.
fig.savefig("/tmp/missingness_heatmap.png")
```
## Cuando usarla
Úsala en el capítulo de datos faltantes de un informe EDA cuando quieras ver de
un vistazo qué columnas faltan juntas (mismo formulario sin rellenar, mismo
proceso roto) frente a columnas cuyas ausencias son independientes. Pásale la
matriz de correlación de ausencias (calculada sobre la máscara de nulos, p. ej.
`df.isnull().corr()`) restringida a las columnas que de verdad tienen ausencia
variable, junto con sus nombres. Es la pareja "estructura" del ranking de % de
nulos: las barras dicen *cuánto* falta cada columna, este heatmap dice *si las
ausencias están relacionadas* entre columnas.
## Gotchas
- **Impura por matplotlib.** Toca la maquinaria de render. Usa el backend `Agg`
y la API orientada a objetos `Figure`/`add_subplot` — NUNCA `pyplot.*` aquí,
para no tocar el estado global ni filtrar figuras entre llamadas. `pyplot` NO
es thread-safe; esta función evita ese riesgo construyendo el `Figure`
directamente, así que es segura de llamar en bucle desde el renderer.
- **El caller cierra la figura.** Devuelve el `Figure` pero no lo muestra ni lo
guarda. Quien la consume debe rasterizarla y luego liberarla
(`matplotlib.pyplot.close(fig)`) para no acumular memoria en lotes grandes.
- **Escala de color fija en [-1, 1].** `vmin=-1`, `vmax=1` están fijados a
propósito para que el color sea comparable entre informes y entre columnas. No
se autoescala al rango real de la matriz; valores fuera de `[-1, 1]` se
saturan al extremo del colormap.
- **Anotaciones solo con N<=12.** Por encima de 12 columnas el grid de números
se vuelve ilegible y se omite; queda solo el color + la colorbar. Filtra a las
columnas con ausencia variable antes de llamar para no llegar a matrices
enormes.
- **Defensiva, nunca lanza.** `matrix=[]`, `labels=[]`, filas cortas, celdas
`None`/`NaN`/no numéricas o cualquier error inesperado se manejan sin propagar:
en el peor caso devuelve una `Figure` con "sin columnas con ausencia variable"
o con el texto del error. No envuelvas la llamada en try/except por miedo a un
raise — no lo hay.
@@ -0,0 +1,158 @@
"""Impure EDA helper: heatmap of missingness co-occurrence (`eda` group).
Builds a matplotlib heatmap of the pairwise missingness correlation matrix of a
dataset: a value near ``+1`` means two columns tend to be null together, near
``-1`` means when one is null the other tends to be present, and ``0`` means
their absences are independent. Returns a ready-to-rasterize
``matplotlib.figure.Figure``; it never shows nor saves it.
Impure because it touches matplotlib's rendering machinery. It uses the headless
Agg backend and the object-oriented ``Figure`` API (no ``pyplot``) so it leaks no
global state and is safe to call repeatedly from a report renderer.
"""
import matplotlib
matplotlib.use("Agg")
from matplotlib.figure import Figure # noqa: E402
# Muted gray for secondary text (no-data / fallback messages).
_MUTED_TEXT = "#5f6b7a"
# Soft red for the error fallback message (kept readable, not alarming).
_ERROR_TEXT = "#b00020"
def _truncate(text, width: int = 14) -> str:
"""Truncate ``text`` to ``width`` chars, appending an ellipsis if cut."""
s = "" if text is None else str(text)
if len(s) <= width:
return s
if width <= 1:
return s[:width]
return s[: width - 1] + ""
def _message_figure(message: str, color: str = _MUTED_TEXT) -> "Figure":
"""Return a fallback ``Figure`` carrying a single centered message."""
fig = Figure(figsize=(6.4, 4.0), dpi=150)
ax = fig.add_subplot(111)
ax.axis("off")
ax.text(
0.5,
0.5,
message,
ha="center",
va="center",
fontsize=12,
color=color,
wrap=True,
transform=ax.transAxes,
)
fig.tight_layout()
return fig
def missingness_corr_heatmap_figure(
matrix,
labels,
title: str = "Co-ocurrencia de ausencias",
) -> "matplotlib.figure.Figure":
"""Build a heatmap figure of a missingness correlation matrix.
Renders an ``NxN`` matrix of missingness correlations in ``[-1, 1]`` with a
diverging ``coolwarm`` colormap (fixed ``vmin=-1``, ``vmax=1`` so the color
scale is comparable across reports). Both axes are tick-labelled with the
column names (truncated to ~14 chars; the X labels rotated 45°). A colorbar
is attached. When the matrix is small (``N <= 12``) each cell is annotated
with its numeric value; for larger matrices the annotations are omitted to
avoid an unreadable grid.
The function is fully defensive: empty/ragged/non-numeric input never raises.
When there is nothing valid to draw it returns a ``Figure`` carrying a
centered "sin columnas con ausencia variable" message, and any unexpected
error is caught and turned into a fallback ``Figure`` carrying the error text.
Args:
matrix: List of lists (``NxN``) of floats in ``[-1, 1]`` — the pairwise
missingness correlation. May be empty; rows of unequal length are
tolerated by treating the matrix as invalid only when it is empty or
its label count does not match. Non-numeric/``None`` cells are
coerced to ``0.0``.
labels: List of ``N`` column names, parallel to ``matrix``. May be empty.
Truncated for display; the originals are not mutated.
title: Figure title. Default "Co-ocurrencia de ausencias".
Returns:
A ``matplotlib.figure.Figure`` with a single heatmap Axes plus a
colorbar. The caller is responsible for rasterizing/closing it.
"""
try:
# --- Validate shape: need a non-empty square-ish matrix with labels.
if (
not isinstance(matrix, (list, tuple))
or not isinstance(labels, (list, tuple))
or len(matrix) == 0
or len(labels) == 0
):
return _message_figure("sin columnas con ausencia variable")
n = len(labels)
# Build a clean NxN grid: coerce each cell to float, default 0.0, pad/clip
# rows so a ragged input never crashes imshow.
grid = []
for i in range(n):
row_src = matrix[i] if i < len(matrix) else []
if not isinstance(row_src, (list, tuple)):
row_src = []
row = []
for j in range(n):
cell = row_src[j] if j < len(row_src) else 0.0
try:
val = float(cell)
except (TypeError, ValueError):
val = 0.0
if val != val: # NaN guard.
val = 0.0
row.append(val)
grid.append(row)
fig = Figure(figsize=(6.4, 5.2), dpi=150)
ax = fig.add_subplot(111)
im = ax.imshow(grid, vmin=-1, vmax=1, cmap="coolwarm", aspect="equal")
short = [_truncate(lab, 14) for lab in labels]
ax.set_xticks(range(n))
ax.set_yticks(range(n))
ax.set_xticklabels(short, rotation=45, ha="right", fontsize=8)
ax.set_yticklabels(short, fontsize=8)
# Annotate each cell only when the grid is small enough to stay legible.
if n <= 12:
for i in range(n):
for j in range(n):
val = grid[i][j]
# White text over saturated (dark) cells, dark over pale.
txt_color = "white" if abs(val) >= 0.55 else "#202020"
ax.text(
j,
i,
f"{val:.2f}",
ha="center",
va="center",
fontsize=7,
color=txt_color,
)
cbar = fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04)
cbar.ax.tick_params(labelsize=8)
cbar.set_label("correlación de ausencias", fontsize=8)
if title:
ax.set_title(_truncate(title, 60), fontsize=12, loc="center", pad=10)
fig.tight_layout()
return fig
except Exception as exc: # noqa: BLE001 — never raise from a figure builder.
return _message_figure(f"error al dibujar heatmap: {exc}", color=_ERROR_TEXT)
@@ -0,0 +1,62 @@
"""Tests para missingness_corr_heatmap_figure (heatmap de ausencias, grupo eda).
Usa el backend Agg sin pyplot; no muestra ni guarda figuras. Cada test cierra
explícitamente la Figure construida (matplotlib.pyplot.close) para no acumular
estado entre tests.
"""
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt # noqa: E402
from matplotlib.figure import Figure # noqa: E402
from missingness_corr_heatmap_figure import missingness_corr_heatmap_figure
def _identity_matrix(n):
"""Matriz NxN con diagonal 1.0 y resto 0.0 (correlación de ausencias)."""
return [[1.0 if i == j else 0.0 for j in range(n)] for i in range(n)]
def test_returns_figure_with_axes():
matrix = [[1.0, 0.3, -0.2], [0.3, 1.0, 0.5], [-0.2, 0.5, 1.0]]
labels = ["edad", "ingresos", "ciudad"]
fig = missingness_corr_heatmap_figure(matrix, labels, title="ausencias")
assert isinstance(fig, Figure)
# Heatmap (>=1 axes) + colorbar añade su propio Axes -> al menos 1.
assert len(fig.axes) >= 1
plt.close(fig)
def test_empty_matrix_does_not_raise_and_returns_figure():
fig = missingness_corr_heatmap_figure([], [], title="vacía")
assert isinstance(fig, Figure)
assert len(fig.axes) >= 1
plt.close(fig)
def test_empty_labels_returns_message_figure():
fig = missingness_corr_heatmap_figure([[1.0]], [], title="sin labels")
assert isinstance(fig, Figure)
plt.close(fig)
def test_large_matrix_omits_annotations():
n = 16
fig = missingness_corr_heatmap_figure(
_identity_matrix(n), [f"col_{i}" for i in range(n)]
)
assert isinstance(fig, Figure)
assert len(fig.axes) >= 1
plt.close(fig)
def test_ragged_and_non_numeric_cells_are_handled():
# Fila corta + celda None + celda string -> se rellenan/coercen sin lanzar.
matrix = [[1.0, None], ["x", 1.0, 0.5]]
labels = ["a", "b"]
fig = missingness_corr_heatmap_figure(matrix, labels)
assert isinstance(fig, Figure)
plt.close(fig)
@@ -0,0 +1,68 @@
---
name: missingness_correlation
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: pure
signature: "def missingness_correlation(null_mask: dict, top_k: int = 20) -> dict"
description: "Co-ocurrencia de ausencias: nucleo del capitulo de missingness del grupo eda. Recibe la mascara binaria de nulos de una tabla (1 = falta, 0 = presente, alineada por fila) y mide hasta que punto las columnas faltan juntas. Calcula la matriz de correlacion de Pearson entre los vectores binarios de ausencia de las columnas con varianza (al menos un 1 y un 0), mas las cifras de solapamiento de conjuntos por par (co-missing, either-missing, Jaccard). Excluye las columnas constantes en su ausencia (correlacion indefinida) y reporta cuantas. Compone la funcion atomica pearson del registry; no la reimplementa. Lectura defensiva; NUNCA lanza."
tags: [eda, missingness, correlation, pearson, co-occurrence, jaccard, datascience]
params:
- name: null_mask
desc: "dict {col: [int 0/1, ...]} con la mascara de ausencias de la tabla, alineada por fila: 1 = el valor falta en esa fila, 0 = presente. Todas las listas se asumen de la misma longitud (numero de filas). Valores truthy distintos de 0 se tratan como ausencia; entradas no-lista se ignoran sin romper."
- name: top_k
desc: "Numero maximo de pares a devolver en `pairs`, ordenados por valor absoluto de correlacion descendente. Default 20. Solo limita la lista de pares; la matriz cubre siempre todas las columnas con varianza."
output: "dict con: columns (columnas con varianza en la ausencia, en orden de entrada); matrix (len(columns) x len(columns) de correlacion de Pearson entre las mascaras binarias, diagonal 1.0); pairs (hasta top_k pares i<j ordenados por |corr| desc, cada uno {a, b, corr, co_missing, either_missing, jaccard} donde co_missing = filas en que ambas faltan, either_missing = filas en que al menos una falta, jaccard = co_missing/either_missing o 0.0 si either_missing=0); n_excluded (nº de columnas con algun nulo pero sin varianza, constantes en la ausencia); excluded_cols (esas columnas en orden de entrada). Si hay <2 columnas con varianza, columns/matrix/pairs van vacios pero n_excluded/excluded_cols se rellenan. NUNCA lanza."
uses_functions: [pearson_py_datascience]
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
tested: true
tests: ["test_co_ocurrencia_fuerte_corr_uno_jaccard_uno", "test_ausencias_disjuntas_corr_negativa_jaccard_cero", "test_columna_sin_varianza_se_excluye", "test_menos_de_dos_columnas_con_varianza_vacio_pero_cuenta_excluidas", "test_mask_vacio_todo_vacio", "test_top_k_limita_pares", "test_no_lanza_con_entradas_raras"]
test_file_path: "python/functions/datascience/missingness_correlation_test.py"
file_path: "python/functions/datascience/missingness_correlation.py"
---
## Ejemplo
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
from datascience.missingness_correlation import missingness_correlation
# Mascara de ausencias de 6 filas. 1 = falta, 0 = presente.
mask = {
"ingresos": [1, 0, 1, 0, 1, 0], # falta junto a "deducciones"
"deducciones": [1, 0, 1, 0, 1, 0], # mismas filas que "ingresos"
"telefono": [0, 0, 0, 1, 0, 0], # casi siempre presente
"verificado": [1, 1, 1, 1, 1, 1], # siempre ausente -> constante, excluida
}
out = missingness_correlation(mask, top_k=10)
print(out["columns"]) # ['ingresos', 'deducciones', 'telefono']
print(out["n_excluded"]) # 1
print(out["excluded_cols"]) # ['verificado']
# El par mas fuerte: ingresos y deducciones faltan siempre juntas.
top = out["pairs"][0]
print(top["a"], top["b"], round(top["corr"], 3)) # ingresos deducciones 1.0
print(top["co_missing"], top["either_missing"], top["jaccard"]) # 3 3 1.0
```
## Cuando usarla
- Usala en el capitulo de **missingness** de `AutomaticEDA` cuando ya tengas la mascara binaria de nulos por columna y quieras detectar **patrones de ausencia conjunta**: que columnas faltan siempre juntas (posible misma fuente/proceso roto) y cuales faltan de forma independiente.
- Cuando necesites ordenar los pares de columnas por fuerza de co-ocurrencia (|corr|) para priorizar que bloques de ausencia investigar o imputar juntos.
- Cuando quieras la cifra de solapamiento de conjuntos (Jaccard, co-missing) ademas de la correlacion lineal, para distinguir "faltan juntas" de "estan presentes juntas".
- Antes de elegir una estrategia de imputacion: dos columnas con corr de ausencia ~1.0 no aportan informacion independiente sobre por que falta la otra.
## Gotchas
- Funcion pura, sin I/O y determinista. Lectura defensiva: entradas no-dict, columnas no-lista o vacias se ignoran sin lanzar.
- Solo entran al calculo las columnas con **varianza en la ausencia** (al menos un 1 y al menos un 0). Una columna siempre-presente (todo 0) no aporta ausencia y **no** se cuenta como excluida; una columna siempre-ausente o constante con nulos (todo 1) tiene correlacion indefinida y se excluye, sumando a `n_excluded` / `excluded_cols`.
- Con menos de 2 columnas con varianza, `columns`/`matrix`/`pairs` quedan vacios pero `n_excluded`/`excluded_cols` se rellenan igual — el caller debe contemplar el caso "sin pares".
- La correlacion es la de Pearson sobre vectores binarios (equivale al coeficiente phi). El signo importa: corr negativa = las ausencias tienden a ser **complementarias** (cuando una falta, la otra suele estar presente).
- Asume todas las listas alineadas por fila y de la misma longitud. Si vienen de longitudes distintas, `pearson` opera sobre el solapamiento que permita `zip` y degrada a 0.0 cuando no hay varianza efectiva; alinea la mascara antes de llamar.
@@ -0,0 +1,120 @@
"""Co-ocurrencia de ausencias: matriz de correlacion de Pearson entre mascaras de nulos.
Funcion pura del grupo eda, nucleo del capitulo de missingness. Recibe la mascara
binaria de ausencias de una tabla (1 = falta, 0 = presente, alineada por fila) y
mide hasta que punto las columnas faltan juntas. Para cada par de columnas con
varianza en su ausencia calcula la correlacion de Pearson entre los vectores
binarios, mas las cifras de solapamiento de conjuntos (co-missing, either-missing,
Jaccard). Compone la funcion atomica `pearson` del registry; no reimplementa la
correlacion. Lectura defensiva; NUNCA lanza.
"""
from datascience import pearson
def missingness_correlation(null_mask, top_k=20) -> dict:
"""Correlacion de co-ocurrencia de ausencias entre columnas.
Args:
null_mask: dict {col: [int 0/1, ...]} alineado por fila (1 = el valor
falta en esa fila). Todas las listas se asumen de la misma longitud.
top_k: numero maximo de pares a devolver, ordenados por |corr| desc.
Returns:
dict con:
- columns: columnas con varianza en la ausencia (al menos un 1 y al
menos un 0), en orden de entrada.
- matrix: matriz len(columns) x len(columns) de correlacion de Pearson
entre las mascaras binarias, diagonal 1.0.
- pairs: lista de hasta top_k pares (i<j) ordenados por |corr| desc.
Cada par: {a, b, corr, co_missing, either_missing, jaccard}.
- n_excluded: numero de columnas con algun nulo pero sin varianza
(constantes en la ausencia: siempre presentes o siempre ausentes).
- excluded_cols: lista de esas columnas (en orden de entrada).
Si hay menos de 2 columnas con varianza, columns/matrix/pairs van vacios
pero n_excluded/excluded_cols se rellenan igualmente. NUNCA lanza.
"""
# Salida base, defensiva ante entradas no-dict.
result = {
"columns": [],
"matrix": [],
"pairs": [],
"n_excluded": 0,
"excluded_cols": [],
}
if not isinstance(null_mask, dict) or not null_mask:
return result
varying = [] # columnas con varianza en la ausencia
varying_vecs = [] # sus vectores binarios saneados (floats 0.0/1.0)
excluded_cols = [] # columnas con nulos pero sin varianza (constantes)
for col, raw in null_mask.items():
if not isinstance(raw, (list, tuple)):
continue
# Sanea a 0/1: cualquier valor truthy distinto de 0 cuenta como ausencia.
vec = [1 if bool(v) else 0 for v in raw]
if not vec:
continue
ones = sum(vec)
zeros = len(vec) - ones
if ones > 0 and zeros > 0:
varying.append(col)
varying_vecs.append([float(v) for v in vec])
elif ones > 0:
# Tiene nulos pero todos (constante en la ausencia): sin varianza.
excluded_cols.append(col)
# ones == 0 -> columna siempre presente, sin nulos: no se cuenta como
# excluida (no aporta ausencia al analisis de co-ocurrencia).
result["n_excluded"] = len(excluded_cols)
result["excluded_cols"] = excluded_cols
n = len(varying)
if n < 2:
return result
result["columns"] = list(varying)
# Matriz de correlacion de Pearson, diagonal 1.0.
matrix = [[0.0] * n for _ in range(n)]
for i in range(n):
matrix[i][i] = 1.0
for i in range(n):
for j in range(i + 1, n):
r = pearson(varying_vecs[i], varying_vecs[j])
matrix[i][j] = r
matrix[j][i] = r
result["matrix"] = matrix
# Pares con cifras de solapamiento de conjuntos.
pairs = []
for i in range(n):
vi = varying_vecs[i]
for j in range(i + 1, n):
vj = varying_vecs[j]
co_missing = 0
either_missing = 0
for a, b in zip(vi, vj):
a_miss = a != 0.0
b_miss = b != 0.0
if a_miss and b_miss:
co_missing += 1
if a_miss or b_miss:
either_missing += 1
jaccard = co_missing / either_missing if either_missing > 0 else 0.0
pairs.append({
"a": varying[i],
"b": varying[j],
"corr": matrix[i][j],
"co_missing": co_missing,
"either_missing": either_missing,
"jaccard": jaccard,
})
pairs.sort(key=lambda p: abs(p["corr"]), reverse=True)
result["pairs"] = pairs[:top_k] if top_k is not None and top_k >= 0 else pairs
return result
@@ -0,0 +1,115 @@
"""Tests para missingness_correlation."""
from datascience.missingness_correlation import missingness_correlation
def test_co_ocurrencia_fuerte_corr_uno_jaccard_uno():
# a y b faltan EXACTAMENTE en las mismas filas -> corr 1.0, jaccard 1.0.
mask = {
"a": [1, 0, 1, 0, 1, 0],
"b": [1, 0, 1, 0, 1, 0],
}
out = missingness_correlation(mask)
assert out["columns"] == ["a", "b"]
assert out["n_excluded"] == 0
# Diagonal 1.0, off-diagonal ~1.0.
assert out["matrix"][0][0] == 1.0
assert out["matrix"][1][1] == 1.0
assert abs(out["matrix"][0][1] - 1.0) < 1e-9
assert len(out["pairs"]) == 1
pair = out["pairs"][0]
assert {pair["a"], pair["b"]} == {"a", "b"}
assert abs(pair["corr"] - 1.0) < 1e-9
assert pair["co_missing"] == 3 # filas 0,2,4
assert pair["either_missing"] == 3 # mismas filas
assert abs(pair["jaccard"] - 1.0) < 1e-9
def test_ausencias_disjuntas_corr_negativa_jaccard_cero():
# a y b nunca faltan en la misma fila -> co_missing 0, jaccard 0, corr <= 0.
mask = {
"a": [1, 1, 0, 0],
"b": [0, 0, 1, 1],
}
out = missingness_correlation(mask)
assert out["columns"] == ["a", "b"]
pair = out["pairs"][0]
assert pair["co_missing"] == 0
assert pair["either_missing"] == 4
assert pair["jaccard"] == 0.0
# Solapamiento nulo + ausencias complementarias -> correlacion negativa.
assert pair["corr"] < 0.0
assert abs(pair["corr"] - out["matrix"][0][1]) < 1e-12
def test_columna_sin_varianza_se_excluye():
# c esta siempre presente (todo 0): no aporta ausencia -> no entra ni como
# excluida. d esta siempre ausente (todo 1): tiene nulos pero sin varianza
# -> excluida y n_excluded incrementa. a y b tienen varianza.
mask = {
"a": [1, 0, 1, 0],
"b": [1, 0, 0, 0],
"c": [0, 0, 0, 0], # siempre presente
"d": [1, 1, 1, 1], # siempre ausente, constante
}
out = missingness_correlation(mask)
assert out["columns"] == ["a", "b"]
assert "d" in out["excluded_cols"]
assert "c" not in out["excluded_cols"]
assert out["n_excluded"] == 1
# Matriz solo de las columnas con varianza.
assert len(out["matrix"]) == 2
assert len(out["matrix"][0]) == 2
def test_menos_de_dos_columnas_con_varianza_vacio_pero_cuenta_excluidas():
# Solo una columna con varianza (a) + una constante-ausente (d).
mask = {
"a": [1, 0, 1, 0],
"d": [1, 1, 1, 1],
}
out = missingness_correlation(mask)
assert out["columns"] == []
assert out["matrix"] == []
assert out["pairs"] == []
assert out["n_excluded"] == 1
assert out["excluded_cols"] == ["d"]
def test_mask_vacio_todo_vacio():
out = missingness_correlation({})
assert out == {
"columns": [],
"matrix": [],
"pairs": [],
"n_excluded": 0,
"excluded_cols": [],
}
def test_top_k_limita_pares():
# 4 columnas con varianza -> 6 pares; top_k=2 deja 2.
mask = {
"a": [1, 0, 1, 0, 0],
"b": [1, 0, 0, 1, 0],
"c": [0, 1, 1, 0, 1],
"d": [1, 1, 0, 0, 1],
}
out = missingness_correlation(mask, top_k=2)
assert len(out["columns"]) == 4
assert len(out["pairs"]) == 2
# Ordenados por |corr| desc.
assert abs(out["pairs"][0]["corr"]) >= abs(out["pairs"][1]["corr"])
def test_no_lanza_con_entradas_raras():
# Valores no-lista y no-dict no deben romper.
assert missingness_correlation(None)["columns"] == []
mask = {
"a": [1, 0, 1, 0],
"b": [1, 0, 1, 0],
"bad": "not a list",
"empty": [],
}
out = missingness_correlation(mask)
assert out["columns"] == ["a", "b"]
@@ -0,0 +1,99 @@
---
id: missingness_overview_py_datascience
name: missingness_overview
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: pure
signature: "def missingness_overview(null_mask) -> dict"
description: "Resumen de ausencias a nivel de dataset a partir de una máscara de nulos 0/1 por columna ({col: [1=falta, 0=presente]} alineada por fila). Calcula celdas y porcentaje de datos faltantes, cuántas columnas tienen algún nulo y cuántas filas son completas vs. incompletas. Estilo dict-no-throw del grupo eda: nunca lanza. Lectura defensiva — no-dict o dict vacío devuelve todo a 0; columnas no-lista se tratan como vacías; listas de longitud distinta se alinean a la longitud máxima rellenando la cola corta como presente (0); valores None/no-int cuentan como presente; sin ZeroDivisionError."
tags: [eda, missing, missingness, nulls, profiling, datascience, pure]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
example: |
from datascience.missingness_overview import missingness_overview
mask = {
"a": [1, 0, 0, 0, 1],
"b": [1, 0, 1, 0, 0],
"c": [0, 0, 0, 0, 1],
}
missingness_overview(mask)
# n_missing_cells=5, missing_cell_pct≈33.33, complete_rows=2, incomplete_rows=3
tested: true
tests:
- "test_cooccurrence_three_cols_exact"
- "test_empty_dict_all_zero"
- "test_output_keys_contract"
- "test_not_a_dict_returns_zero"
- "test_no_nulls_all_complete"
- "test_none_values_treated_as_present"
- "test_unequal_lengths_pad_with_max"
- "test_columns_present_but_no_rows"
- "test_never_raises_on_garbage"
test_file_path: "python/functions/datascience/missingness_overview_test.py"
file_path: "python/functions/datascience/missingness_overview.py"
params:
- name: null_mask
desc: "Dict {col_name: [int 0/1, ...]} con la máscara de nulos por columna, alineada por fila (1 = el valor falta, 0 = el valor está presente). Normalmente todas las listas tienen la misma longitud = nº de filas. Lectura defensiva: si no es dict o está vacío se devuelve todo a 0; columnas cuyo valor no es lista/tupla se tratan como vacías; listas de longitud distinta se alinean a la longitud máxima (las posiciones inexistentes de las columnas más cortas cuentan como presentes, 0); valores None o no enteros cuentan como presentes."
output: "Dict con exactamente 9 claves, todas siempre presentes (la función nunca lanza): n_rows (longitud de fila = longitud máxima entre columnas, 0 si vacío), n_cols (nº de columnas), n_cols_with_null (columnas con >=1 falta), n_missing_cells (suma total de 1s), missing_cell_pct (0-100 = n_missing_cells / (n_rows*n_cols) * 100), complete_rows (filas sin ninguna falta), incomplete_rows (filas con >=1 falta), complete_pct (0-100), incomplete_pct (0-100). Los porcentajes son 0.0 cuando el denominador es 0 (sin ZeroDivisionError)."
---
## Ejemplo
```python
from datascience.missingness_overview import missingness_overview
# Máscara de nulos por columna: 1 = falta, 0 = presente, alineada por fila.
mask = {
"a": [1, 0, 0, 0, 1],
"b": [1, 0, 1, 0, 0],
"c": [0, 0, 0, 0, 1],
}
missingness_overview(mask)
# {
# "n_rows": 5,
# "n_cols": 3,
# "n_cols_with_null": 3, # a, b y c tienen al menos una falta
# "n_missing_cells": 5, # 2 (a) + 2 (b) + 1 (c)
# "missing_cell_pct": 33.33, # 5 / (5*3) * 100
# "complete_rows": 2, # filas 1 y 3 sin ninguna falta
# "incomplete_rows": 3, # filas 0 (a&b), 2 (b), 4 (a&c)
# "complete_pct": 40.0, # 2 / 5 * 100
# "incomplete_pct": 60.0, # 3 / 5 * 100
# }
missingness_overview({})
# Todo a 0: {"n_rows": 0, "n_cols": 0, "n_cols_with_null": 0,
# "n_missing_cells": 0, "missing_cell_pct": 0.0,
# "complete_rows": 0, "incomplete_rows": 0,
# "complete_pct": 0.0, "incomplete_pct": 0.0}
```
## Cuando usarla
Úsala al perfilar un dataset cuando ya tienes una máscara de nulos 0/1 por
columna (p. ej. derivada del paso de carga/perfilado del EDA) y quieres la foto
global de ausencias en una llamada: cuánta proporción de celdas falta, cuántas
columnas están afectadas y, sobre todo, cuántas filas quedan completas vs.
incompletas. Es el bloque resumen del capítulo de calidad/missingness de un EDA,
y la base para decidir estrategias de imputación o de borrado de filas. Como es
pura y dict-no-throw, puedes alimentarla con la máscara tal cual sin validarla
antes: entradas malformadas degradan a ceros en vez de romper el pipeline.
## Gotchas
- **`n_rows` es la longitud máxima entre columnas.** Con listas de longitud
desigual, las posiciones que faltan en las columnas más cortas se cuentan como
presentes (`0`); no se descartan filas. En el caso normal (todas las listas de
igual longitud) `n_rows` es simplemente esa longitud.
- **Solo el valor exacto `1` cuenta como falta.** `None`, `0`, cadenas y
cualquier otro valor se tratan como presentes. `True` (== 1) también cuenta
como falta por la igualdad.
- **Porcentajes en escala 0-100**, no fracciones. División por cero protegida:
con `n_rows*n_cols == 0` los porcentajes salen `0.0`.
@@ -0,0 +1,116 @@
"""Pure EDA helper: dataset-level missingness overview from a 0/1 null mask.
Part of the `eda` capability group. Consumes a per-column null mask
(``{col_name: [int 0/1, ...]}`` aligned by row, ``1`` = value is missing,
``0`` = value is present) and derives dataset-wide missingness metrics: cell
count and percentage of missing data, how many columns carry any null, and how
many rows are complete vs. incomplete.
Dict-no-throw style of the `eda` group: it NEVER raises. A non-dict, an empty
dict, malformed columns, ragged lists or non-int cell values all degrade
gracefully to the zero/contract output. Stdlib only.
Ragged-length policy: columns are allowed to have different lengths. ``n_rows``
is the **maximum** column length; positions that don't exist in a shorter
column are treated as present (``0``). This keeps the ``n_rows * n_cols`` cell
grid well defined without dropping rows.
"""
def _is_missing(value) -> int:
"""Return ``1`` iff ``value`` denotes a missing cell, else ``0``.
Only an exact equality to ``1`` (covers ``int`` ``1`` and ``float`` ``1.0``)
counts as missing. ``None``, ``0``, strings and any other value are treated
as present. The comparison cannot raise for standard inputs.
"""
try:
return 1 if value == 1 else 0
except Exception:
return 0
def missingness_overview(null_mask) -> dict:
"""Summarize dataset-level missingness from a 0/1 null mask.
Args:
null_mask: Dict ``{col_name: [int 0/1, ...]}`` where each list is aligned
by row (``1`` = missing, ``0`` = present). Lists are normally all the
same length (= number of rows). Defensive: a non-dict or empty dict
returns the all-zero contract; non-list columns are treated as empty;
ragged lists are aligned to the maximum length, padding the missing
tail of shorter columns as present (``0``); ``None`` / non-int cells
count as present.
Returns:
Dict with exactly these keys, all always present (the function never
raises): ``n_rows``, ``n_cols``, ``n_cols_with_null``,
``n_missing_cells``, ``missing_cell_pct`` (0-100), ``complete_rows``,
``incomplete_rows``, ``complete_pct`` (0-100), ``incomplete_pct``
(0-100). Percentages are ``0.0`` when the denominator is zero (no
``ZeroDivisionError``).
"""
zero = {
"n_rows": 0,
"n_cols": 0,
"n_cols_with_null": 0,
"n_missing_cells": 0,
"missing_cell_pct": 0.0,
"complete_rows": 0,
"incomplete_rows": 0,
"complete_pct": 0.0,
"incomplete_pct": 0.0,
}
if not isinstance(null_mask, dict) or not null_mask:
return dict(zero)
# Normalize every column to a list; non-list columns become empty.
cols = {}
for name, seq in null_mask.items():
cols[name] = seq if isinstance(seq, (list, tuple)) else []
n_cols = len(cols)
lengths = [len(seq) for seq in cols.values()]
n_rows = max(lengths) if lengths else 0
if n_rows == 0:
# Columns exist but carry no rows: everything zero except n_cols.
out = dict(zero)
out["n_cols"] = n_cols
return out
n_missing_cells = 0
n_cols_with_null = 0
row_has_missing = [False] * n_rows
for seq in cols.values():
col_len = len(seq)
col_has_null = False
for r in range(n_rows):
if r < col_len and _is_missing(seq[r]):
n_missing_cells += 1
row_has_missing[r] = True
col_has_null = True
if col_has_null:
n_cols_with_null += 1
incomplete_rows = sum(1 for flag in row_has_missing if flag)
complete_rows = n_rows - incomplete_rows
total_cells = n_rows * n_cols
missing_cell_pct = (n_missing_cells / total_cells * 100.0) if total_cells else 0.0
complete_pct = complete_rows / n_rows * 100.0
incomplete_pct = incomplete_rows / n_rows * 100.0
return {
"n_rows": n_rows,
"n_cols": n_cols,
"n_cols_with_null": n_cols_with_null,
"n_missing_cells": n_missing_cells,
"missing_cell_pct": missing_cell_pct,
"complete_rows": complete_rows,
"incomplete_rows": incomplete_rows,
"complete_pct": complete_pct,
"incomplete_pct": incomplete_pct,
}
@@ -0,0 +1,146 @@
"""Tests para missingness_overview."""
import sys
import os
import pytest
sys.path.insert(0, os.path.dirname(__file__))
from missingness_overview import missingness_overview
# Output contract: every call returns exactly these 9 keys.
EXPECTED_KEYS = {
"n_rows",
"n_cols",
"n_cols_with_null",
"n_missing_cells",
"missing_cell_pct",
"complete_rows",
"incomplete_rows",
"complete_pct",
"incomplete_pct",
}
def test_cooccurrence_three_cols_exact():
# 3 columns, 5 rows. Hand-computed expectations:
# col a missing at rows 0, 4 -> 2
# col b missing at rows 0, 2 -> 2
# col c missing at row 4 -> 1
# n_missing_cells = 5, total_cells = 5*3 = 15 -> 33.333...%
# row 0 (a&b co-occur) -> incomplete
# row 1 (all present) -> complete
# row 2 (b only) -> incomplete
# row 3 (all present) -> complete
# row 4 (a&c co-occur) -> incomplete
mask = {
"a": [1, 0, 0, 0, 1],
"b": [1, 0, 1, 0, 0],
"c": [0, 0, 0, 0, 1],
}
out = missingness_overview(mask)
assert out["n_rows"] == 5
assert out["n_cols"] == 3
assert out["n_cols_with_null"] == 3
assert out["n_missing_cells"] == 5
assert out["missing_cell_pct"] == pytest.approx(33.33333333, abs=1e-6)
assert out["complete_rows"] == 2
assert out["incomplete_rows"] == 3
assert out["complete_pct"] == pytest.approx(40.0)
assert out["incomplete_pct"] == pytest.approx(60.0)
def test_empty_dict_all_zero():
out = missingness_overview({})
assert out == {
"n_rows": 0,
"n_cols": 0,
"n_cols_with_null": 0,
"n_missing_cells": 0,
"missing_cell_pct": 0.0,
"complete_rows": 0,
"incomplete_rows": 0,
"complete_pct": 0.0,
"incomplete_pct": 0.0,
}
def test_output_keys_contract():
# The 9-key contract holds even for the garbage/zero path.
assert set(missingness_overview({}).keys()) == EXPECTED_KEYS
assert set(missingness_overview({"a": [1, 0]}).keys()) == EXPECTED_KEYS
def test_not_a_dict_returns_zero():
for bad in (None, [1, 0, 1], 42, "nope", 3.14):
out = missingness_overview(bad)
assert out["n_rows"] == 0
assert out["n_cols"] == 0
assert out["n_missing_cells"] == 0
assert out["missing_cell_pct"] == 0.0
def test_no_nulls_all_complete():
mask = {"a": [0, 0, 0], "b": [0, 0, 0]}
out = missingness_overview(mask)
assert out["n_rows"] == 3
assert out["n_cols"] == 2
assert out["n_cols_with_null"] == 0
assert out["n_missing_cells"] == 0
assert out["missing_cell_pct"] == 0.0
assert out["complete_rows"] == 3
assert out["incomplete_rows"] == 0
assert out["complete_pct"] == pytest.approx(100.0)
assert out["incomplete_pct"] == pytest.approx(0.0)
def test_none_values_treated_as_present():
# None and other non-1 values count as present (0).
mask = {"a": [None, 1, None, "x", 0]}
out = missingness_overview(mask)
assert out["n_rows"] == 5
assert out["n_cols"] == 1
assert out["n_missing_cells"] == 1 # only the explicit 1 at row 1
assert out["n_cols_with_null"] == 1
assert out["complete_rows"] == 4
assert out["incomplete_rows"] == 1
def test_unequal_lengths_pad_with_max():
# Ragged lists: n_rows = max length; shorter column padded as present.
# a = [1, 1] -> missing at rows 0, 1
# b = [0] -> row 1 padded to present
# n_rows = 2, n_cols = 2, total_cells = 4, n_missing_cells = 2 -> 50%
mask = {"a": [1, 1], "b": [0]}
out = missingness_overview(mask)
assert out["n_rows"] == 2
assert out["n_cols"] == 2
assert out["n_cols_with_null"] == 1
assert out["n_missing_cells"] == 2
assert out["missing_cell_pct"] == pytest.approx(50.0)
assert out["complete_rows"] == 0
assert out["incomplete_rows"] == 2
assert out["incomplete_pct"] == pytest.approx(100.0)
def test_columns_present_but_no_rows():
# Columns exist but all empty -> zero metrics, n_cols preserved.
out = missingness_overview({"a": [], "b": []})
assert out["n_rows"] == 0
assert out["n_cols"] == 2
assert out["n_missing_cells"] == 0
assert out["missing_cell_pct"] == 0.0
assert out["complete_pct"] == 0.0
def test_never_raises_on_garbage():
# Non-list column values, mixed junk -> must not raise.
mask = {"a": "not a list", "b": 123, "c": [1, 0, 1]}
out = missingness_overview(mask)
assert set(out.keys()) == EXPECTED_KEYS
assert out["n_rows"] == 3
assert out["n_cols"] == 3
assert out["n_missing_cells"] == 2 # only col c contributes
assert out["n_cols_with_null"] == 1
@@ -0,0 +1,93 @@
---
id: missingness_rank_bar_figure_py_datascience
name: missingness_rank_bar_figure
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: impure
signature: "def missingness_rank_bar_figure(names, pcts, title=\"% de valores faltantes por columna\") -> \"matplotlib.figure.Figure\""
description: "Construye una figura matplotlib de barras horizontales que ordena las columnas de un dataset por su porcentaje de valores faltantes (0-100), la mayor arriba, etiquetando cada barra con su NN.N% al final. Usa ax.barh, eje X fijo 0-100 y labels truncados a ~22 chars. Devuelve un matplotlib.figure.Figure listo para rasterizar por el renderer del informe EDA (capítulo de datos faltantes). Backend Agg sin pyplot global; defensivo ante listas vacías, longitudes desiguales o valores no numéricos (nunca lanza)."
tags: [eda, missing, missingness, ranking, bar, barh, matplotlib, figure, visualization, datascience, impure]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [matplotlib]
example: |
from datascience.missingness_rank_bar_figure import missingness_rank_bar_figure
names = ["edad", "ingresos", "ciudad", "email"]
pcts = [12.5, 40.0, 3.2, 0.0]
fig = missingness_rank_bar_figure(names, pcts, title="% de valores faltantes por columna")
tested: true
tests:
- "test_returns_figure_with_axes"
- "test_sorted_descending_largest_on_top"
- "test_empty_lists_do_not_raise_and_returns_figure"
- "test_xlim_is_zero_to_hundred"
- "test_length_mismatch_and_non_numeric_are_handled"
test_file_path: "python/functions/datascience/missingness_rank_bar_figure_test.py"
file_path: "python/functions/datascience/missingness_rank_bar_figure.py"
params:
- name: names
desc: "Lista de nombres de columna. Puede venir vacía (devuelve figura \"sin datos faltantes\"). Los items se convierten a str y se truncan a ~22 chars con elipsis para las etiquetas del eje Y; los originales no se mutan."
- name: pcts
desc: "Lista paralela a names con el % de nulos en [0,100]. Valores None, NaN o no numéricos se coercen a 0.0 y los negativos se recortan a 0. Si len(names) != len(pcts) se recorta al menor de ambos para no romper."
- name: title
desc: "Título de la figura. Se trunca a ~60 chars con elipsis si es muy largo. Default \"% de valores faltantes por columna\"."
output: "Un matplotlib.figure.Figure (figsize 6.4 x alto adaptativo según nº de barras, dpi 150) con un Axes de barras horizontales (ax.barh) ordenadas por % descendente, la mayor arriba. Eje X fijado a [0,100] con label \"% faltante\", etiquetas del eje Y truncadas a ~22 chars, y cada barra anotada con su NN.N% al final. Si names o pcts vienen vacíos devuelve una Figure con texto centrado \"sin datos faltantes\"; cualquier error inesperado se captura y devuelve una Figure con el mensaje de error (nunca lanza). El caller rasteriza/cierra la figura; la función no la muestra ni la guarda."
---
## Ejemplo
```python
from datascience.missingness_rank_bar_figure import missingness_rank_bar_figure
# % de nulos por columna (p. ej. (df.isnull().mean() * 100).
names = ["edad", "ingresos", "ciudad", "email"]
pcts = [12.5, 40.0, 3.2, 0.0]
fig = missingness_rank_bar_figure(
names,
pcts,
title="% de valores faltantes por columna",
)
# ingresos (40.0%) queda arriba; email (0.0%) abajo.
# El renderer del informe lo rasteriza; aquí solo persistimos para inspección.
fig.savefig("/tmp/missingness_rank.png")
```
## Cuando usarla
Úsala al abrir el capítulo de datos faltantes de un informe EDA para responder
"¿qué columnas están más incompletas?" de un vistazo. Pásale los nombres de
columna y el % de nulos de cada una (`(df.isnull().mean() * 100).round(1)`); la
función se encarga de ordenar de mayor a menor y poner la peor arriba. Es la
pareja "magnitud" del heatmap de co-ocurrencia: las barras dicen *cuánto* falta
en cada columna, el heatmap dice *si esas ausencias están relacionadas* entre
columnas.
## Gotchas
- **Impura por matplotlib.** Toca la maquinaria de render. Usa el backend `Agg`
y la API orientada a objetos `Figure`/`add_subplot` — NUNCA `pyplot.*` aquí,
para no tocar el estado global ni filtrar figuras entre llamadas. `pyplot` NO
es thread-safe; esta función evita ese riesgo construyendo el `Figure`
directamente, así que es segura de llamar en bucle desde el renderer.
- **El caller cierra la figura.** Devuelve el `Figure` pero no lo muestra ni lo
guarda. Quien la consume debe rasterizarla y luego liberarla
(`matplotlib.pyplot.close(fig)`) para no acumular memoria en lotes grandes.
- **Espera porcentajes 0-100, no fracciones 0-1.** El eje X está fijado a
`[0, 100]`. Si pasas fracciones (`0.4` en vez de `40.0`) las barras saldrán
pegadas al origen. Multiplica por 100 antes de llamar.
- **Alto adaptativo.** La altura de la figura crece con el número de barras
(hasta un tope) para que reports con muchas columnas sigan legibles; aun así,
conviene filtrar a las columnas con algún nulo antes de llamar para no listar
decenas de barras a 0%.
- **Defensiva, nunca lanza.** Listas vacías, longitudes desiguales, valores
`None`/`NaN`/no numéricos o cualquier error inesperado se manejan sin propagar:
en el peor caso devuelve una `Figure` con "sin datos faltantes" o con el texto
del error. No envuelvas la llamada en try/except por miedo a un raise — no lo
hay.
@@ -0,0 +1,150 @@
"""Impure EDA helper: ranked bar figure of missing-value share (`eda` group).
Builds a horizontal bar chart ranking the columns of a dataset by their
percentage of missing values (0-100), largest at the top, each bar labelled with
its ``NN.N%`` at the end. Returns a ready-to-rasterize
``matplotlib.figure.Figure``; it never shows nor saves it.
Impure because it touches matplotlib's rendering machinery. It uses the headless
Agg backend and the object-oriented ``Figure`` API (no ``pyplot``) so it leaks no
global state and is safe to call repeatedly from a report renderer.
"""
import matplotlib
matplotlib.use("Agg")
from matplotlib.figure import Figure # noqa: E402
# Muted gray for secondary text (no-data / fallback messages).
_MUTED_TEXT = "#5f6b7a"
# Soft red for the error fallback message.
_ERROR_TEXT = "#b00020"
# Bar fill — a calm blue that reads well on white at report size.
_BAR_COLOR = "#4C72B0"
def _truncate(text, width: int = 22) -> str:
"""Truncate ``text`` to ``width`` chars, appending an ellipsis if cut."""
s = "" if text is None else str(text)
if len(s) <= width:
return s
if width <= 1:
return s[:width]
return s[: width - 1] + ""
def _message_figure(message: str, color: str = _MUTED_TEXT) -> "Figure":
"""Return a fallback ``Figure`` carrying a single centered message."""
fig = Figure(figsize=(6.4, 4.0), dpi=150)
ax = fig.add_subplot(111)
ax.axis("off")
ax.text(
0.5,
0.5,
message,
ha="center",
va="center",
fontsize=12,
color=color,
wrap=True,
transform=ax.transAxes,
)
fig.tight_layout()
return fig
def missingness_rank_bar_figure(
names,
pcts,
title: str = "% de valores faltantes por columna",
) -> "matplotlib.figure.Figure":
"""Build a horizontal ranked bar figure of missing-value share per column.
Pairs each column name with its missing percentage, sorts by percentage
descending and draws horizontal bars with the largest at the top. The X axis
is pinned to ``[0, 100]`` so bars are comparable across reports, each bar is
annotated with its ``NN.N%`` at the end, and the Y tick labels are truncated
to ~22 chars.
The function is fully defensive: empty/mismatched/non-numeric input never
raises. When there is nothing valid to draw it returns a ``Figure`` carrying
a centered "sin datos faltantes" message, and any unexpected error is caught
and turned into a fallback ``Figure`` carrying the error text.
Args:
names: List of column names. May be empty. Items are stringified and
truncated for display; the originals are not mutated.
pcts: List parallel to ``names`` of missing-value percentages in
``[0, 100]``. Non-numeric/``None`` values are coerced to ``0.0`` and
negatives are clamped to ``0``. The list is truncated to
``min(len(names), len(pcts))`` so a length mismatch never crashes.
title: Figure title. Default "% de valores faltantes por columna".
Returns:
A ``matplotlib.figure.Figure`` with a single horizontal-bar Axes. The
caller is responsible for rasterizing/closing it.
"""
try:
if (
not isinstance(names, (list, tuple))
or not isinstance(pcts, (list, tuple))
or len(names) == 0
or len(pcts) == 0
):
return _message_figure("sin datos faltantes")
# --- Pair names with coerced percentages, tolerating length mismatch.
pairs = []
for name, pct in zip(names, pcts):
try:
val = float(pct)
except (TypeError, ValueError):
val = 0.0
if val != val: # NaN guard.
val = 0.0
val = max(0.0, val)
pairs.append((name, val))
if not pairs:
return _message_figure("sin datos faltantes")
# Sort by percentage descending; barh draws bottom-up, so the largest
# ends at the top when we reverse the order before plotting.
pairs.sort(key=lambda p: p[1], reverse=True)
ordered = list(reversed(pairs)) # smallest first -> largest on top.
labels = [_truncate(name, 22) for name, _ in ordered]
values = [val for _, val in ordered]
y_pos = range(len(ordered))
# Height scales with the number of bars so dense reports stay readable.
height = max(2.4, min(0.4 * len(ordered) + 1.2, 14.0))
fig = Figure(figsize=(6.4, height), dpi=150)
ax = fig.add_subplot(111)
ax.barh(list(y_pos), values, color=_BAR_COLOR, edgecolor="white")
ax.set_yticks(list(y_pos))
ax.set_yticklabels(labels, fontsize=8)
ax.set_xlim(0, 100)
ax.set_xlabel("% faltante", fontsize=9)
# Annotate each bar with its percentage at the end of the bar.
for y, val in zip(y_pos, values):
ax.text(
min(val + 1.5, 99.0),
y,
f"{val:.1f}%",
va="center",
ha="left" if val < 90 else "right",
fontsize=7,
color="#202020",
)
if title:
ax.set_title(_truncate(title, 60), fontsize=12, loc="left", pad=10)
fig.tight_layout()
return fig
except Exception as exc: # noqa: BLE001 — never raise from a figure builder.
return _message_figure(f"error al dibujar barras: {exc}", color=_ERROR_TEXT)
@@ -0,0 +1,64 @@
"""Tests para missingness_rank_bar_figure (barras de % faltante, grupo eda).
Usa el backend Agg sin pyplot; no muestra ni guarda figuras. Cada test cierra
explícitamente la Figure construida (matplotlib.pyplot.close) para no acumular
estado entre tests.
"""
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt # noqa: E402
from matplotlib.figure import Figure # noqa: E402
from missingness_rank_bar_figure import missingness_rank_bar_figure
def test_returns_figure_with_axes():
names = ["edad", "ingresos", "ciudad"]
pcts = [12.5, 40.0, 3.2]
fig = missingness_rank_bar_figure(names, pcts, title="faltantes")
assert isinstance(fig, Figure)
assert len(fig.axes) >= 1
plt.close(fig)
def test_sorted_descending_largest_on_top():
names = ["a", "b", "c"]
pcts = [10.0, 50.0, 25.0]
fig = missingness_rank_bar_figure(names, pcts)
ax = fig.axes[0]
# barh dibuja de abajo arriba; la mayor (50, "b") debe quedar arriba (mayor y).
bars = ax.patches
# El último parche (mayor índice y) corresponde a la barra superior.
widths = [b.get_width() for b in bars]
assert max(widths) == 50.0
# La barra con la mayor anchura es la de mayor coordenada y (arriba).
top_bar = max(bars, key=lambda b: b.get_y())
assert top_bar.get_width() == 50.0
plt.close(fig)
def test_empty_lists_do_not_raise_and_returns_figure():
fig = missingness_rank_bar_figure([], [], title="vacía")
assert isinstance(fig, Figure)
assert len(fig.axes) >= 1
plt.close(fig)
def test_xlim_is_zero_to_hundred():
fig = missingness_rank_bar_figure(["a"], [42.0])
ax = fig.axes[0]
assert ax.get_xlim() == (0.0, 100.0)
plt.close(fig)
def test_length_mismatch_and_non_numeric_are_handled():
# Más names que pcts + un pct None -> zip recorta y None se coacciona a 0.
names = ["a", "b", "c"]
pcts = [None, 30.0]
fig = missingness_rank_bar_figure(names, pcts)
assert isinstance(fig, Figure)
assert len(fig.axes) >= 1
plt.close(fig)
@@ -0,0 +1,65 @@
---
name: missingness_row_patterns
kind: function
lang: py
domain: datascience
version: "1.0.0"
purity: pure
signature: "def missingness_row_patterns(null_mask, top_n=10) -> dict"
description: "Agrupa las filas de un dataset por su patron de ausencias (estilo matriz de missingno): para cada fila, el patron es la tupla ORDENADA de columnas que faltan en esa fila (las que tienen 1 en el null_mask). Cuenta la frecuencia de cada patron distinto, incluido el patron vacio (fila completa). Devuelve el top_n por frecuencia con su pct sobre el total. Pura, lectura defensiva, NUNCA lanza; {} -> n_rows 0."
tags: [eda, missingness, missingno, patterns, profiling, datascience, data-quality]
params:
- name: null_mask
desc: "Dict {col: [0/1, ...]} alineado por fila, donde 1 = la celda falta en esa fila y 0 = presente. Todas las columnas deberian tener la misma longitud (una entrada por fila); si difieren, n_rows es la lista mas larga y las celdas fuera de rango cuentan como presentes. Las claves se ordenan por str(col) para canonizar el patron. {} (o no-dict) -> n_rows 0."
- name: top_n
desc: "Maximo de patrones devueltos en `patterns`, rankeados por n_rows desc (desempate: menos columnas primero, luego nombres de columna). El recuento total de patrones distintos siempre se reporta en `n_patterns`, no se trunca. Default 10. Valores negativos -> 0; no-int -> 10."
output: "Dict {n_rows: int (filas totales), n_patterns: int (patrones distintos, incluye el patron vacio = fila completa), complete_rows: int (filas con patron vacio, nada falta), patterns: lista del top_n ordenada por n_rows desc con [{missing_cols: [col,...] (vacio = fila completa), n_rows: int, pct: float 0-100 sobre n_rows total, redondeado a 2 decimales}]}. Para {} devuelve n_rows 0 y patterns []. NUNCA lanza."
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
tested: true
tests: ["test_patron_dominante_completas_singleton", "test_mask_vacio", "test_top_n_trunca_pero_cuenta_todos"]
test_file_path: "python/functions/datascience/missingness_row_patterns_test.py"
file_path: "python/functions/datascience/missingness_row_patterns.py"
---
## Ejemplo
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
from datascience.missingness_row_patterns import missingness_row_patterns
# null_mask alineado por fila: 1 = la celda falta en esa fila.
null_mask = {
"A": [1, 1, 1, 1, 0, 0, 0, 0, 0, 0],
"B": [1, 1, 1, 1, 0, 0, 0, 0, 0, 0],
"C": [0, 0, 0, 0, 0, 0, 0, 0, 0, 1],
}
out = missingness_row_patterns(null_mask, top_n=10)
print(out["n_rows"], out["n_patterns"], out["complete_rows"]) # 10 3 5
for p in out["patterns"]:
label = p["missing_cols"] or "(fila completa)"
print(label, p["n_rows"], p["pct"])
# (fila completa) 5 50.0
# ['A', 'B'] 4 40.0
# ['C'] 1 10.0
```
## Cuando usarla
- Usala en el capitulo de calidad/ausencias de `AutomaticEDA` para mostrar la "matriz de patrones de missingno": en vez de pintar celda a celda, resume que combinaciones de columnas se quedan en blanco juntas y con que frecuencia.
- Cuando ya tengas el null_mask por columna (1=falta) y quieras detectar co-ausencia estructural ("A y B siempre faltan juntas") antes de decidir una imputacion o un drop conjunto de columnas.
- Cuando necesites una tabla compacta "patron -> nº filas -> pct" para un report o un grafico de barras de los patrones de ausencia mas comunes, separando ademas cuantas filas estan completas (`complete_rows`).
## Gotchas
- Funcion pura, sin I/O y determinista. Lectura defensiva: `{}` o un no-dict devuelven `n_rows` 0 con `patterns` []. NUNCA lanza.
- El patron vacio (fila completa, `missing_cols=[]`) SI cuenta como patron: aparece en `n_patterns` y puede aparecer en `patterns`. El consumidor lo etiqueta como "(fila completa)".
- `pct` es sobre `n_rows` total (0-100), redondeado a 2 decimales. La suma de los `pct` de TODOS los patrones es 100; si `top_n` trunca, los `pct` mostrados sumaran menos.
- Las columnas se ordenan por `str(col)` para canonizar cada patron, asi `{A,B}` y `{B,A}` colapsan al mismo patron `["A", "B"]`.
- Una celda cuenta como ausente solo si vale 1 (`int(cell) == 1`); 0, None y valores no numericos se tratan como presentes.
- Si las listas de columnas tienen longitudes distintas, `n_rows` es la mas larga y las posiciones fuera de rango de una columna corta cuentan como presentes (0).
@@ -0,0 +1,107 @@
"""missingness_row_patterns — distinct per-row missingness patterns (missingno matrix style).
Pure function: no I/O, deterministic, NEVER raises. Given a per-column null mask
aligned by row ({col: [0/1, ...]}, 1 = missing), it groups rows by their missing
"pattern" — the sorted tuple of column names that are missing in that row — and
counts how often each distinct pattern occurs.
This mirrors the missingno matrix idea: instead of plotting per-cell nullity, it
collapses each row to the SET of columns it lacks, surfacing co-missing structure
(e.g. "A and B always go missing together"). The empty pattern (a fully complete
row) is a first-class pattern and may appear in the result with missing_cols=[];
the caller labels it "(fila completa)".
"""
def _is_missing(cell) -> bool:
"""A cell counts as missing when it equals 1 (truthy 0/1 mask).
None / 0 / non-numeric are treated as present. Defensive: never raises.
"""
try:
return int(cell) == 1
except (TypeError, ValueError):
return bool(cell)
def missingness_row_patterns(null_mask, top_n=10) -> dict:
"""Count distinct per-row missingness patterns from a column null mask.
For each row, its pattern is the sorted tuple of column names missing in that
row (the columns whose value is 1). The frequency of each distinct pattern is
counted, including the empty pattern (a complete row with nothing missing).
Args:
null_mask: Dict {col: [0/1, ...]} aligned by row, where 1 means the cell
is missing in that row. Read defensively; columns with differing
lengths are tolerated (n_rows is the longest list; out-of-range cells
count as present). Empty dict -> n_rows 0.
top_n: Maximum number of patterns returned in `patterns`, ranked by
n_rows desc (tiebreak: fewer columns first, then column names). The
full count of distinct patterns is always reported in `n_patterns`.
Returns:
Dict:
{
"n_rows": int, # total rows
"n_patterns": int, # distinct patterns (incl. the empty pattern)
"complete_rows": int, # rows with the empty pattern (nothing missing)
"patterns": [ # top_n patterns, n_rows desc
{"missing_cols": [col, ...], "n_rows": int, "pct": float} # [] = complete row
],
}
For {} (or a non-dict) returns n_rows 0 and patterns []. NEVER raises.
"""
empty = {"n_rows": 0, "n_patterns": 0, "complete_rows": 0, "patterns": []}
if not isinstance(null_mask, dict) or not null_mask:
return empty
# Stable, canonical column order so each row's pattern tuple is sorted.
items = sorted(null_mask.items(), key=lambda kv: str(kv[0]))
names = [str(k) for k, _ in items]
lists = [v if isinstance(v, (list, tuple)) else [] for _, v in items]
n_rows = max((len(lst) for lst in lists), default=0)
if n_rows == 0:
return empty
# Defensive parsing of top_n.
try:
limit = int(top_n)
except (TypeError, ValueError):
limit = 10
if limit < 0:
limit = 0
counts: dict = {}
n_cols = len(names)
for r in range(n_rows):
# names is sorted, so iterating in order yields an already-sorted tuple.
pattern = tuple(
names[c]
for c in range(n_cols)
if r < len(lists[c]) and _is_missing(lists[c][r])
)
counts[pattern] = counts.get(pattern, 0) + 1
complete_rows = counts.get((), 0)
n_patterns = len(counts)
# Rank: n_rows desc, then fewer columns first, then column names (deterministic).
ordered = sorted(counts.items(), key=lambda kv: (-kv[1], len(kv[0]), kv[0]))
patterns = [
{
"missing_cols": list(pat),
"n_rows": cnt,
"pct": round(100.0 * cnt / n_rows, 2),
}
for pat, cnt in ordered[:limit]
]
return {
"n_rows": n_rows,
"n_patterns": n_patterns,
"complete_rows": complete_rows,
"patterns": patterns,
}
@@ -0,0 +1,87 @@
"""Tests para missingness_row_patterns."""
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
from missingness_row_patterns import missingness_row_patterns
_EXPECTED_KEYS = {"n_rows", "n_patterns", "complete_rows", "patterns"}
def test_patron_dominante_completas_singleton():
"""Golden: {A,B} co-faltan en 4 filas + 5 filas completas + 1 singleton {C}."""
# 10 filas. A y B faltan juntas en las filas 0-3; filas 4-8 completas;
# la fila 9 solo le falta C.
null_mask = {
"A": [1, 1, 1, 1, 0, 0, 0, 0, 0, 0],
"B": [1, 1, 1, 1, 0, 0, 0, 0, 0, 0],
"C": [0, 0, 0, 0, 0, 0, 0, 0, 0, 1],
}
out = missingness_row_patterns(null_mask)
assert set(out.keys()) == _EXPECTED_KEYS
assert out["n_rows"] == 10
# 3 patrones distintos: (A,B), () y (C,).
assert out["n_patterns"] == 3
# 5 filas completas (filas 4-8).
assert out["complete_rows"] == 5
# Orden: n_rows desc; desempate menos columnas primero.
# () tiene 5 filas, (A,B) 4, (C,) 1.
pats = out["patterns"]
assert len(pats) == 3
assert pats[0]["missing_cols"] == []
assert pats[0]["n_rows"] == 5
assert pats[0]["pct"] == 50.0
assert pats[1]["missing_cols"] == ["A", "B"]
assert pats[1]["n_rows"] == 4
assert pats[1]["pct"] == 40.0
assert pats[2]["missing_cols"] == ["C"]
assert pats[2]["n_rows"] == 1
assert pats[2]["pct"] == 10.0
# Tipos de salida.
assert isinstance(out["n_rows"], int)
assert isinstance(pats[0]["pct"], float)
def test_mask_vacio():
"""{} -> n_rows 0, sin patrones, nunca lanza."""
out = missingness_row_patterns({})
assert out == {
"n_rows": 0,
"n_patterns": 0,
"complete_rows": 0,
"patterns": [],
}
# No dict / None tambien degradan a vacio sin lanzar.
assert missingness_row_patterns(None)["n_rows"] == 0
# Columnas presentes pero listas vacias -> n_rows 0.
assert missingness_row_patterns({"A": [], "B": []})["patterns"] == []
def test_top_n_trunca_pero_cuenta_todos():
"""top_n limita `patterns`, pero n_patterns reporta TODOS los distintos."""
null_mask = {
"A": [0, 1, 1, 0, 1],
"B": [0, 0, 0, 1, 1],
"C": [0, 0, 0, 0, 1],
}
# Filas: () (A,) (A,) (B,) (A,B,C)
out = missingness_row_patterns(null_mask, top_n=2)
assert out["n_rows"] == 5
assert out["n_patterns"] == 4 # (), (A,), (B,), (A,B,C)
assert out["complete_rows"] == 1
# Solo 2 patrones devueltos pese a haber 4.
assert len(out["patterns"]) == 2
# (A,) domina con 2 filas; desempate del 2o entre los de 1 fila -> () (0 cols).
assert out["patterns"][0]["missing_cols"] == ["A"]
assert out["patterns"][0]["n_rows"] == 2
assert out["patterns"][1]["missing_cols"] == []
assert out["patterns"][1]["n_rows"] == 1
-2
View File
@@ -18,7 +18,6 @@ dependencies = [
"google-cloud-bigquery-storage>=2.27",
"google-cloud-storage>=3.10.1",
"httpx",
"langdetect>=1.0.9",
"matplotlib>=3.10.9",
"opencv-contrib-python-headless>=4.13.0.92",
"openpyxl>=3.1.5",
@@ -41,7 +40,6 @@ dependencies = [
"seaborn>=0.13.2",
"shapely>=2.1.2",
"statsmodels>=0.14.6",
"textstat>=0.7.13",
"trimesh>=4.12.2",
"xlrd>=2.0.2",
]
-96
View File
@@ -899,7 +899,6 @@ dependencies = [
{ name = "google-cloud-bigquery-storage" },
{ name = "google-cloud-storage" },
{ name = "httpx" },
{ name = "langdetect" },
{ name = "matplotlib" },
{ name = "opencv-contrib-python-headless" },
{ name = "openpyxl" },
@@ -907,11 +906,9 @@ dependencies = [
{ name = "polars" },
{ name = "pymeshlab" },
{ name = "pymssql" },
{ name = "pymupdf" },
{ name = "pypdf" },
{ name = "pyproj" },
{ name = "python-docx" },
{ name = "python-pptx" },
{ name = "pyyaml" },
{ name = "qrcode", extra = ["pil"] },
{ name = "rapidfuzz" },
@@ -922,7 +919,6 @@ dependencies = [
{ name = "seaborn" },
{ name = "shapely" },
{ name = "statsmodels" },
{ name = "textstat" },
{ name = "trimesh" },
{ name = "xlrd" },
]
@@ -963,7 +959,6 @@ requires-dist = [
{ name = "jupyter-collaboration", marker = "extra == 'jupyter'", specifier = ">=2.0" },
{ name = "jupyter-mcp-server", marker = "extra == 'jupyter'" },
{ name = "jupyterlab", marker = "extra == 'jupyter'", specifier = ">=4.0" },
{ name = "langdetect", specifier = ">=1.0.9" },
{ name = "matplotlib", specifier = ">=3.10.9" },
{ name = "opencv-contrib-python-headless", specifier = ">=4.13.0.92" },
{ name = "openpyxl", specifier = ">=3.1.5" },
@@ -971,11 +966,9 @@ requires-dist = [
{ name = "polars", specifier = ">=1.40.1" },
{ name = "pymeshlab", specifier = ">=2025.7.post1" },
{ name = "pymssql", specifier = ">=2.3.13" },
{ name = "pymupdf", specifier = ">=1.28.0" },
{ name = "pypdf", specifier = ">=6.10.0" },
{ name = "pyproj", specifier = ">=3.7.2" },
{ name = "python-docx", specifier = ">=1.2.0" },
{ name = "python-pptx", specifier = ">=1.0.2" },
{ name = "pyyaml", specifier = ">=6.0.3" },
{ name = "qrcode", extras = ["pil"], specifier = ">=8.2" },
{ name = "rapidfuzz", specifier = ">=3.14.5" },
@@ -986,7 +979,6 @@ requires-dist = [
{ name = "seaborn", specifier = ">=0.13.2" },
{ name = "shapely", specifier = ">=2.1.2" },
{ name = "statsmodels", specifier = ">=0.14.6" },
{ name = "textstat", specifier = ">=0.7.13" },
{ name = "trimesh", specifier = ">=4.12.2" },
{ name = "xlrd", specifier = ">=2.0.2" },
]
@@ -2206,15 +2198,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b5/91/53255615acd2a1eaca307ede3c90eb550bae9c94581f8c00081b6b1c8f44/kiwisolver-1.5.0-graalpy312-graalpy250_312_native-win_amd64.whl", hash = "sha256:1f1489f769582498610e015a8ef2d36f28f505ab3096d0e16b4858a9ec214f57", size = 75987, upload-time = "2026-03-09T13:15:39.65Z" },
]
[[package]]
name = "langdetect"
version = "1.0.9"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "six" },
]
sdist = { url = "https://files.pythonhosted.org/packages/0e/72/a3add0e4eec4eb9e2569554f7c70f4a3c27712f40e3284d483e88094cc0e/langdetect-1.0.9.tar.gz", hash = "sha256:cbc1fef89f8d062739774bd51eda3da3274006b3661d199c2655f6b3f6d605a0", size = 981474, upload-time = "2021-05-07T07:54:13.562Z" }
[[package]]
name = "lark"
version = "1.3.1"
@@ -2716,21 +2699,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/9e/c9/b2622292ea83fbb4ec318f5b9ab867d0a28ab43c5717bb85b0a5f6b3b0a4/networkx-3.6.1-py3-none-any.whl", hash = "sha256:d47fbf302e7d9cbbb9e2555a0d267983d2aa476bac30e90dfbe5669bd57f3762", size = 2068504, upload-time = "2025-12-08T17:02:38.159Z" },
]
[[package]]
name = "nltk"
version = "3.9.4"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "click" },
{ name = "joblib" },
{ name = "regex" },
{ name = "tqdm" },
]
sdist = { url = "https://files.pythonhosted.org/packages/74/a1/b3b4adf15585a5bc4c357adde150c01ebeeb642173ded4d871e89468767c/nltk-3.9.4.tar.gz", hash = "sha256:ed03bc098a40481310320808b2db712d95d13ca65b27372f8a403949c8b523d0", size = 2946864, upload-time = "2026-03-24T06:13:40.641Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/9d/91/04e965f8e717ba0ab4bdca5c112deeab11c9e750d94c4d4602f050295d39/nltk-3.9.4-py3-none-any.whl", hash = "sha256:f2fa301c3a12718ce4a0e9305c5675299da5ad9e26068218b69d692fda84828f", size = 1552087, upload-time = "2026-03-24T06:13:38.47Z" },
]
[[package]]
name = "notebook-shim"
version = "0.2.4"
@@ -3782,23 +3750,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/25/50/4be9bd9cf4b43208a7175117a533ece200cfe4131a39f9909bdc7560ddeb/pymssql-2.3.13-cp314-cp314-win_amd64.whl", hash = "sha256:7d7037d2b5b907acc7906d0479924db2935a70c720450c41339146a4ada2b93d", size = 2049139, upload-time = "2026-02-14T05:00:23.951Z" },
]
[[package]]
name = "pymupdf"
version = "1.28.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/8e/e9/6d6c5d6c0a3551bffd47681a6240caf941727f195b45593cf20ab36f018f/pymupdf-1.28.0.tar.gz", hash = "sha256:e53f3567403a92da15caa9e7ae0164327fff48817e9f40175367fb9de524258d", size = 87637751, upload-time = "2026-06-29T09:08:47.547Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c8/b7/88043e38cc7529de070f0c9bd267fa258035cca0b4ad5260536b994594a7/pymupdf-1.28.0-cp310-abi3-macosx_10_15_x86_64.whl", hash = "sha256:892b89ba88e8f98b53133b62877a9dc9b5e7dc6a4aeb837b612db56a8d2e03ac", size = 24597385, upload-time = "2026-06-29T09:03:30.608Z" },
{ url = "https://files.pythonhosted.org/packages/33/f4/23775bbda0781b61fc398cc75079a2b0e64696d8fcf93271748883e9627e/pymupdf-1.28.0-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:4d692dcf44d3566ae96bc6f6346c6ad432274a29ba617bf7a9fe18009e24adb4", size = 23828292, upload-time = "2026-06-29T09:03:46.129Z" },
{ url = "https://files.pythonhosted.org/packages/1c/f5/bf75fc7a415722f8b33662054f82d88520c0cbfd4c36d0e08aeaec605e49/pymupdf-1.28.0-cp310-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:47a5c29ed4eb0744de9c4e37bb49b1259b18d4d75fcc8a7c130f7c9fa15956f6", size = 25045507, upload-time = "2026-06-29T09:04:03.86Z" },
{ url = "https://files.pythonhosted.org/packages/58/69/5d12c9f1f2d76f28383d6110a069c79fbfced5a4f97bb1ee6e8354f52bb7/pymupdf-1.28.0-cp310-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:44f0973f5e5edbaec95bc34b64e71d1959d4ee90b1328de1b4f4f5b4fa78673f", size = 25716599, upload-time = "2026-06-29T09:04:19.367Z" },
{ url = "https://files.pythonhosted.org/packages/4d/b4/ec0e017bc42857cc86bd651441dbc41cc18be48d4698ecd27aac491e0c9a/pymupdf-1.28.0-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:4d61ec323a706e153a12e262e51febfb43eeaa20977785ace135d18d48bcdc83", size = 25940489, upload-time = "2026-06-29T09:04:36.624Z" },
{ url = "https://files.pythonhosted.org/packages/06/86/f831fef09013f33b3c9c09fb3923f2ff53e1e437f6ace14b8ae46392f558/pymupdf-1.28.0-cp310-abi3-win32.whl", hash = "sha256:caea2b3b67347fd79e5d15ed7929b0e886aac594ea228073b6d39de0078189da", size = 18489703, upload-time = "2026-06-29T20:50:30.599Z" },
{ url = "https://files.pythonhosted.org/packages/2e/5d/1a03f53eb0449900469335fcfc742ca28e3ba159b7d650e0921d50b8b308/pymupdf-1.28.0-cp310-abi3-win_amd64.whl", hash = "sha256:e01e90fd86abfeb37ceb921eddb951f988a11d45ff6ce6b7664f2039849068ec", size = 19773102, upload-time = "2026-06-29T09:04:49.773Z" },
{ url = "https://files.pythonhosted.org/packages/72/f6/1e52ce243ca792254f6223b4017c5667194c146ce9b88baf37bc5eb3d1c9/pymupdf-1.28.0-cp313-abi3-pyemscripten_2025_0_wasm32.whl", hash = "sha256:74c6d00ba2a9aad3a635db73b07c15db462b480741d831a34a75a56535ebc22b", size = 18357011, upload-time = "2026-06-29T20:50:50.353Z" },
{ url = "https://files.pythonhosted.org/packages/62/b1/46b5b3d8ef3cc71114667cf10c4d8b33f39af97253af32e9a0986775b638/pymupdf-1.28.0-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:b3e1399c7a64c6914239116a369efcdaac4cfb9e838bde2656d7accc4a85c72d", size = 25753599, upload-time = "2026-06-29T09:05:09.398Z" },
]
[[package]]
name = "pyogrio"
version = "0.12.1"
@@ -3860,15 +3811,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/55/f2/7ebe366f633f30a6ad105f650f44f24f98cb1335c4157d21ae47138b3482/pypdf-6.10.0-py3-none-any.whl", hash = "sha256:90005e959e1596c6e6c84c8b0ad383285b3e17011751cedd17f2ce8fcdfc86de", size = 334459, upload-time = "2026-04-10T09:34:54.966Z" },
]
[[package]]
name = "pyphen"
version = "0.17.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/69/56/e4d7e1bd70d997713649c5ce530b2d15a5fc2245a74ca820fc2d51d89d4d/pyphen-0.17.2.tar.gz", hash = "sha256:f60647a9c9b30ec6c59910097af82bc5dd2d36576b918e44148d8b07ef3b4aa3", size = 2079470, upload-time = "2025-01-20T13:18:36.296Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/7b/1f/c2142d2edf833a90728e5cdeb10bdbdc094dde8dbac078cee0cf33f5e11b/pyphen-0.17.2-py3-none-any.whl", hash = "sha256:3a07fb017cb2341e1d9ff31b8634efb1ae4dc4b130468c7c39dd3d32e7c3affd", size = 2079358, upload-time = "2025-01-20T13:18:29.629Z" },
]
[[package]]
name = "pyproj"
version = "3.7.2"
@@ -3993,21 +3935,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/1c/fd/0318007beb234790993d3ec5afd051d1dbceb733e81e3afe2b981ece3f37/python_multipart-0.0.30-py3-none-any.whl", hash = "sha256:830964def8c90607ac5daa00514e3987815865713ade8d20febc9177ac0c3c5b", size = 29730, upload-time = "2026-05-31T19:24:53.814Z" },
]
[[package]]
name = "python-pptx"
version = "1.0.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "lxml" },
{ name = "pillow" },
{ name = "typing-extensions" },
{ name = "xlsxwriter" },
]
sdist = { url = "https://files.pythonhosted.org/packages/52/a9/0c0db8d37b2b8a645666f7fd8accea4c6224e013c42b1d5c17c93590cd06/python_pptx-1.0.2.tar.gz", hash = "sha256:479a8af0eaf0f0d76b6f00b0887732874ad2e3188230315290cd1f9dd9cc7095", size = 10109297, upload-time = "2024-08-07T17:33:37.772Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d9/4f/00be2196329ebbff56ce564aa94efb0fbc828d00de250b1980de1a34ab49/python_pptx-1.0.2-py3-none-any.whl", hash = "sha256:160838e0b8565a8b1f67947675886e9fea18aa5e795db7ae531606d68e785cba", size = 472788, upload-time = "2024-08-07T17:33:28.192Z" },
]
[[package]]
name = "pywin32"
version = "311"
@@ -5009,20 +4936,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/6a/9e/2064975477fdc887e47ad42157e214526dcad8f317a948dee17e1659a62f/terminado-0.18.1-py3-none-any.whl", hash = "sha256:a4468e1b37bb318f8a86514f65814e1afc977cf29b3992a4500d9dd305dcceb0", size = 14154, upload-time = "2024-03-12T14:34:36.569Z" },
]
[[package]]
name = "textstat"
version = "0.7.13"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "nltk" },
{ name = "pyphen" },
{ name = "setuptools" },
]
sdist = { url = "https://files.pythonhosted.org/packages/8c/0f/b673fcec5ad6e976b2e8368ef3651fe0fea3348a1191bacfcd41a17ddec6/textstat-0.7.13.tar.gz", hash = "sha256:a88d1da76287cd27ca4ce7bcba1ebaf2890544a5f0bb6a5758fa84cef3bceccb", size = 138932, upload-time = "2026-02-18T21:07:39.525Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ca/31/0eb4cc5bb021b4ceaaa602c59ba16ce99256b9dd30981bef3f3a53d8555f/textstat-0.7.13-py3-none-any.whl", hash = "sha256:04b1ec995d1e8b2e628759497e6b23204a9ec91dcd652447d8cbba9478f25471", size = 177050, upload-time = "2026-02-18T21:07:38.163Z" },
]
[[package]]
name = "threadpoolctl"
version = "3.6.0"
@@ -5399,15 +5312,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/1a/62/c8d562e7766786ba6587d09c5a8ba9f718ed3fa8af7f4553e8f91c36f302/xlrd-2.0.2-py2.py3-none-any.whl", hash = "sha256:ea762c3d29f4cca48d82df517b6d89fbce4db3107f9d78713e48cd321d5c9aa9", size = 96555, upload-time = "2025-06-14T08:46:37.766Z" },
]
[[package]]
name = "xlsxwriter"
version = "3.2.9"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/46/2c/c06ef49dc36e7954e55b802a8b231770d286a9758b3d936bd1e04ce5ba88/xlsxwriter-3.2.9.tar.gz", hash = "sha256:254b1c37a368c444eac6e2f867405cc9e461b0ed97a3233b2ac1e574efb4140c", size = 215940, upload-time = "2025-09-16T00:16:21.63Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3a/0c/3662f4a66880196a590b202f0db82d919dd2f89e99a27fadef91c4a33d41/xlsxwriter-3.2.9-py3-none-any.whl", hash = "sha256:9a5db42bc5dff014806c58a20b9eae7322a134abb6fce3c92c181bfb275ec5b3", size = 175315, upload-time = "2025-09-16T00:16:20.108Z" },
]
[[package]]
name = "xxhash"
version = "3.7.0"