1+ # pampaneira_imputation/evaluation.py
2+ import numpy as np
3+ import pandas as pd
4+ from pypots .nn .functional import calc_mae , calc_mse , calc_rmse , calc_mre
5+ from typing import Dict , Tuple , List
6+ from . import config
7+
8+ def calculate_imputation_metrics (y_true : np .ndarray ,
9+ y_pred : np .ndarray ,
10+ indicating_mask : np .ndarray ) -> Dict [str , float ]:
11+ """
12+ Calcula MAE, MSE, RMSE, MRE para valores imputados donde la máscara es 1.
13+
14+ Args:
15+ y_true (np.ndarray): Datos verdaderos (potencialmente con NaNs donde faltaban originalmente).
16+ y_pred (np.ndarray): Datos imputados.
17+ indicating_mask (np.ndarray): Máscara donde 1 indica un valor faltante que fue imputado,
18+ 0 indica un valor observado.
19+
20+ Returns:
21+ Dict[str, float]: Diccionario que contiene 'mae', 'mse', 'rmse', 'mre'.
22+ """
23+ # Asegura que las entradas sean arrays numpy
24+ y_true = np .asarray (y_true )
25+ y_pred = np .asarray (y_pred )
26+ indicating_mask = np .asarray (indicating_mask )
27+
28+ # Reemplaza NaNs en la verdad fundamental con 0 para el cálculo donde la máscara es 1
29+ # Esto es necesario porque las funciones pypots esperan una verdad fundamental sin NaN
30+ # Solo evaluamos donde indicating_mask es 1, por lo que este reemplazo es seguro.
31+ y_true_filled = np .nan_to_num (y_true , nan = 0.0 )
32+
33+ if y_true .shape != y_pred .shape or y_true .shape != indicating_mask .shape :
34+ raise ValueError (f"Desajuste de forma: y_true={ y_true .shape }
35+ f"y_pred={ y_pred .shape } { indicating_mask .shape } )
36+
37+ # Verifica si la suma de la máscara es cero (no hay valores para evaluar)
38+ if indicating_mask .sum () == 0 :
39+ print ("Advertencia: La suma de la máscara indicadora es 0. No hay valores imputados para evaluar." )
40+ return {'mae' : np .nan , 'mse' : np .nan , 'rmse' : np .nan , 'mre' : np .nan }
41+
42+ try :
43+ mae = calc_mae (y_pred , y_true_filled , indicating_mask )
44+ mse = calc_mse (y_pred , y_true_filled , indicating_mask )
45+ rmse = calc_rmse (y_pred , y_true_filled , indicating_mask )
46+ mre = calc_mre (y_pred , y_true_filled , indicating_mask ) # Precaución con MRE si los valores verdaderos están cerca de cero
47+
48+ return {'mae' : mae , 'mse' : mse , 'rmse' : rmse , 'mre' : mre }
49+ except Exception as e :
50+ print (f"Error durante el cálculo de métricas: { e } )
51+ # Añade más información de depuración si es necesario
52+ print (f"Formas: y_pred={ y_pred .shape } { y_true_filled .shape } { indicating_mask .shape } )
53+ print (f"Suma de máscara: { indicating_mask .sum ()} )
54+ print (f"Conteo de NaN: pred={ np .isnan (y_pred ).sum ()} { np .isnan (y_true_filled ).sum ()} { np .isnan (indicating_mask ).sum ()} )
55+ # Considera verificar también por infinitos
56+ return {'mae' : np .nan , 'mse' : np .nan , 'rmse' : np .nan , 'mre' : np .nan }
57+
58+
59+ def evaluate_all_methods (preprocessed_data : Dict ,
60+ imputed_results : Dict [str , np .ndarray ],
61+ methods_to_evaluate : list = ['median' , 'mean' , 'linear' , 'ffill_bfill' , 'bfill_ffill' , 'saits' ]) -> pd .DataFrame :
62+ """
63+ Evalúa múltiples métodos de imputación usando los resultados del conjunto de prueba.
64+
65+ Args:
66+ preprocessed_data (Dict): Diccionario de preprocess_for_imputation.
67+ imputed_results (Dict[str, np.ndarray]): Diccionario que mapea nombres de métodos a arrays NumPy imputados (conjunto de prueba).
68+ methods_to_evaluate (list, optional): Lista de claves en imputed_results para evaluar.
69+ (por defecto: ['median', 'mean', 'linear', 'ffill_bfill', 'bfill_ffill', 'saits'])
70+
71+ Returns:
72+ pd.DataFrame: DataFrame de Pandas que resume MAE, MSE, RMSE, MRE para cada método.
73+ """
74+ results = []
75+ y_true = preprocessed_data ['test_X_ori' ]
76+ indicating_mask = preprocessed_data ['test_indicating_mask' ]
77+
78+ # Maneja la posible eliminación de columnas para ffill/bfill si es necesario
79+ # Esta lógica asume que WS/WD se eliminaron *antes* de la imputación para ffill/bfill
80+ cols_to_drop_indices = [config .FEATURE_COLUMNS .index (col ) for col in config .COLS_TO_DROP_FOR_BASELINE if col in config .FEATURE_COLUMNS ]
81+
82+ for method_name in methods_to_evaluate :
83+ if method_name not in imputed_results :
84+ print (f"Advertencia: No se encontraron resultados imputados para el método '{ method_name } )
85+ continue
86+
87+ y_pred = imputed_results [method_name ]
88+ current_y_true = y_true
89+ current_mask = indicating_mask
90+
91+ # Manejo específico para métodos donde las columnas podrían haberse eliminado
92+ if method_name in ['ffill_bfill' , 'bfill_ffill' ] and cols_to_drop_indices :
93+ print (f"Ajustando datos verdaderos y máscara para { method_name } )
94+ current_y_true = np .delete (y_true , cols_to_drop_indices , axis = 2 )
95+ current_mask = np .delete (indicating_mask , cols_to_drop_indices , axis = 2 )
96+ # y_pred para estos métodos ya debería tener las columnas eliminadas
97+
98+ print (f"\n Calculando métricas para: { method_name } )
99+ metrics = calculate_imputation_metrics (current_y_true , y_pred , current_mask )
100+
101+ results .append ({
102+ "Method" : method_name .replace ('_' , ' ' ).title (), # Nombre más bonito
103+ "RMSE" : metrics .get ('rmse' , np .nan ),
104+ "MSE" : metrics .get ('mse' , np .nan ),
105+ "MAE" : metrics .get ('mae' , np .nan ),
106+ "MRE" : metrics .get ('mre' , np .nan )
107+ })
108+
109+ error_table = pd .DataFrame .from_records (results )
110+ error_table = error_table .set_index ("Method" ).round (4 )
111+ return error_table
0 commit comments