122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
"""PCA rapido sobre columnas numericas para revelar estructura latente.
|
|
|
|
Estandariza las columnas (z-score), descarta filas con valores faltantes y
|
|
ajusta un PCA determinista para ver cuanta varianza concentran pocos
|
|
componentes. Pensado para exploracion de datos (EDA) barata.
|
|
"""
|
|
|
|
import math
|
|
|
|
|
|
def pca_explained(columns: dict, n_components: int = 2) -> dict:
|
|
"""Ejecuta PCA sobre columnas numericas y resume la varianza explicada.
|
|
|
|
Args:
|
|
columns: mapa {nombre_columna: [valores numericos]}. Las listas estan
|
|
alineadas por fila (misma longitud). Las columnas no numericas o
|
|
con menos de dos valores distintos se descartan.
|
|
n_components: numero maximo de componentes principales a calcular.
|
|
Se acota a min(n_features, n_filas_validas).
|
|
|
|
Returns:
|
|
dict con:
|
|
n_components: numero de componentes realmente calculados.
|
|
n_rows_used: filas validas usadas (sin None/NaN).
|
|
n_features: columnas numericas usadas.
|
|
explained_variance_ratio: varianza explicada por componente.
|
|
cumulative: varianza acumulada componente a componente.
|
|
top_loadings: cargas mas grandes (en valor absoluto) por componente.
|
|
projection: proyeccion de las filas (cap a 1000 filas).
|
|
Si hay menos de 2 columnas numericas o menos de 3 filas validas,
|
|
devuelve {n_components: 0, explained_variance_ratio: [],
|
|
note: "datos insuficientes"} sin lanzar excepcion.
|
|
"""
|
|
import numpy as np
|
|
from sklearn.decomposition import PCA
|
|
from sklearn.preprocessing import StandardScaler
|
|
|
|
insufficient = {
|
|
"n_components": 0,
|
|
"explained_variance_ratio": [],
|
|
"note": "datos insuficientes",
|
|
}
|
|
|
|
if not isinstance(columns, dict) or not columns:
|
|
return insufficient
|
|
|
|
# Quedarnos solo con columnas que se puedan interpretar como numericas.
|
|
numeric_cols: dict[str, list] = {}
|
|
for name, values in columns.items():
|
|
if not isinstance(values, (list, tuple)):
|
|
continue
|
|
coerced = []
|
|
usable = True
|
|
for v in values:
|
|
if v is None:
|
|
coerced.append(math.nan)
|
|
continue
|
|
try:
|
|
f = float(v)
|
|
except (TypeError, ValueError):
|
|
usable = False
|
|
break
|
|
coerced.append(f)
|
|
if usable:
|
|
numeric_cols[name] = coerced
|
|
|
|
if len(numeric_cols) < 2:
|
|
return insufficient
|
|
|
|
feature_names = list(numeric_cols.keys())
|
|
matrix = np.array([numeric_cols[n] for n in feature_names], dtype=float).T
|
|
|
|
# Descartar filas con cualquier NaN (incluye los None convertidos).
|
|
valid_mask = ~np.isnan(matrix).any(axis=1)
|
|
data = matrix[valid_mask]
|
|
|
|
if data.shape[0] < 3:
|
|
return insufficient
|
|
|
|
n_rows_used = int(data.shape[0])
|
|
n_features = int(data.shape[1])
|
|
|
|
k = min(n_components, n_features, n_rows_used)
|
|
if k < 1:
|
|
return insufficient
|
|
|
|
scaled = StandardScaler().fit_transform(data)
|
|
pca = PCA(n_components=k, random_state=0)
|
|
proj = pca.fit_transform(scaled)
|
|
|
|
evr = [float(x) for x in pca.explained_variance_ratio_]
|
|
cumulative = []
|
|
running = 0.0
|
|
for x in evr:
|
|
running += x
|
|
cumulative.append(float(running))
|
|
|
|
# Cargas: una fila por componente, una columna por feature.
|
|
top_loadings = []
|
|
for comp_idx, comp in enumerate(pca.components_):
|
|
order = np.argsort(np.abs(comp))[::-1]
|
|
for feat_idx in order:
|
|
top_loadings.append(
|
|
{
|
|
"component": int(comp_idx),
|
|
"feature": feature_names[int(feat_idx)],
|
|
"loading": float(comp[int(feat_idx)]),
|
|
}
|
|
)
|
|
|
|
projection = [[float(v) for v in row] for row in proj[:1000]]
|
|
|
|
return {
|
|
"n_components": int(k),
|
|
"n_rows_used": n_rows_used,
|
|
"n_features": n_features,
|
|
"explained_variance_ratio": evr,
|
|
"cumulative": cumulative,
|
|
"top_loadings": top_loadings,
|
|
"projection": projection,
|
|
}
|