# utils/shap_explainer.py import streamlit as st import pandas as pd import numpy as np import shap import plotly.express as px import plotly.graph_objects as go from typing import Dict, Any, Union, Optional from sklearn.pipeline import Pipeline import h2o import streamlit as st import pandas as pd import numpy as np import shap import plotly.express as px import plotly.graph_objects as go from typing import Dict, Any, Optional, Union from sklearn.pipeline import Pipeline class SHAPExplainer: """ Clase para realizar explicaciones de modelos usando SHAP (SHapley Additive exPlanations) """ def __init__(self, model, X: pd.DataFrame, problem_type: str = 'classification', explanation_method: str = 'auto'): """ Inicializar el explicador SHAP Args: model: Modelo de machine learning entrenado X (pd.DataFrame): Datos de entrada para el modelo problem_type (str): Tipo de problema ('classification' o 'regression') explanation_method (str): Método de explicación ('auto', 'tree', 'linear', 'kernel') """ self.model = model self.X = X self.problem_type = problem_type self.explanation_method = explanation_method self.explainer = self._create_explainer() self.X_sample = None # Inicializar X_sample def _create_explainer(self): """ Crear el explicador SHAP apropiado según el tipo de modelo y método seleccionado, manejando correctamente los Pipelines y modelos de H2O. Returns: Explainer de SHAP """ try: # Si el modelo es un Pipeline, extraer el estimador final if isinstance(self.model, Pipeline): estimator = self.model.steps[-1][1] else: estimator = self.model # Verificar si el modelo es de H2O if isinstance(estimator, h2o.estimators.H2OEstimator): # Usar KernelExplainer para modelos de H2O # Obtener función de predicción compatible con SHAP def predict_function(x): h2o_frame = h2o.H2OFrame(x) preds = estimator.predict(h2o_frame) return preds.as_data_frame()['predict'].values return shap.KernelExplainer(predict_function, shap.sample(self.X, 100)) # Crear el explicador usando el método seleccionado if self.explanation_method.lower() == 'tree': return shap.TreeExplainer(estimator) elif self.explanation_method.lower() == 'linear': return shap.LinearExplainer(estimator, self.X, feature_dependence="independent") elif self.explanation_method.lower() == 'kernel': return shap.KernelExplainer(estimator.predict, shap.sample(self.X, 100)) else: # 'auto' o cualquier otro valor: usar shap.Explainer que selecciona automáticamente return shap.Explainer(estimator, self.X) except Exception as e: st.error(f"Error al crear explicador SHAP: {str(e)}") return None def compute_shap_values(self, X_sample: Optional[pd.DataFrame] = None, max_samples: int = 100): """ Calcular valores SHAP Args: X_sample (pd.DataFrame, opcional): Muestra de datos para calcular SHAP max_samples (int): Número máximo de muestras a procesar Returns: Valores SHAP """ try: # Usar muestra si no se proporciona if X_sample is None: X_sample = self.X.sample(n=min(max_samples, len(self.X)), random_state=42) # Almacenar el subconjunto de datos utilizado self.X_sample = X_sample # Asegurarse de que X_sample es 2D if X_sample.ndim != 2: raise ValueError(f"Debe pasar una entrada 2D a SHAP. La forma actual es {X_sample.shape}") # Calcular valores SHAP usando el explicador shap_values = self.explainer.shap_values(X_sample) # Para clasificación multiclase, SHAP devuelve una lista de arrays if isinstance(shap_values, list): # Promediar las contribuciones de todas las clases shap_values = np.mean(np.abs(shap_values), axis=0) else: shap_values = np.abs(shap_values) return shap_values except Exception as e: st.error(f"Error al calcular valores SHAP: {str(e)}") return None def plot_summary(self, shap_values, title: str = "SHAP Summary Plot"): """ Generar gráfico de resumen de valores SHAP Args: shap_values: Valores SHAP calculados title (str): Título del gráfico Returns: Figura de Plotly """ try: feature_names = self.X.columns.tolist() # Calcular importancia de características feature_importance = np.mean(shap_values, axis=0) importance_df = pd.DataFrame({ 'feature': feature_names, 'importance': feature_importance }).sort_values('importance', ascending=False) # Gráfico de barras de importancia fig = px.bar( importance_df, x='importance', y='feature', orientation='h', title=title, labels={'importance': 'Importancia SHAP', 'feature': 'Características'} ) return fig except Exception as e: st.error(f"Error al generar gráfico de resumen: {str(e)}") return None def plot_dependence(self, shap_values, feature_name: str): """ Generar gráfico de dependencia para una característica Args: shap_values: Valores SHAP calculados feature_name (str): Nombre de la característica Returns: Figura de Plotly """ try: feature_idx = self.X.columns.get_loc(feature_name) # Preparar datos usando el mismo subconjunto de datos utilizado para SHAP if self.X_sample is not None: x = self.X_sample.iloc[:, feature_idx] else: x = self.X.iloc[:, feature_idx] y = shap_values[:, feature_idx] # Verificar que las longitudes coincidan if len(x) != len(y): raise ValueError(f"Longitud de 'x' ({len(x)}) y 'y' ({len(y)}) no coinciden.") # Crear scatter plot fig = px.scatter( x=x, y=y, title=f'SHAP Dependence Plot - {feature_name}', labels={'x': feature_name, 'y': 'SHAP Value'} ) return fig except Exception as e: st.error(f"Error al generar gráfico de dependencia: {str(e)}") return None def generate_feature_importance_report(self, shap_values) -> Dict[str, Any]: """ Generar un informe detallado de importancia de características Args: shap_values: Valores SHAP calculados Returns: Diccionario con información de importancia de características """ try: # Calcular importancia feature_importance = np.mean(shap_values, axis=0) # Crear DataFrame de importancia importance_df = pd.DataFrame({ 'feature': self.X.columns, 'importance': feature_importance }).sort_values('importance', ascending=False) # Generar informe report = { 'top_features': importance_df.head(5).to_dict('records'), 'bottom_features': importance_df.tail(5).to_dict('records'), 'total_features': len(importance_df), 'max_importance': importance_df['importance'].max(), 'min_importance': importance_df['importance'].min() } return report except Exception as e: st.error(f"Error al generar informe de importancia: {str(e)}") return {} def create_shap_analysis_dashboard(model, X: pd.DataFrame, problem_type: str = 'classification'): """ Crear un dashboard de análisis SHAP en Streamlit Args: model: Modelo de machine learning X (pd.DataFrame): Datos de entrada problem_type (str): Tipo de problema """ st.title("🔍 Análisis de Explicabilidad SHAP") # Inicializar los valores SHAP en session_state si no existen if 'shap_explainer' not in st.session_state: # Parámetros por defecto explanation_method = 'auto' max_samples = 100 # Crear y almacenar el explicador SHAP st.session_state.shap_explainer = SHAPExplainer( model=model, X=X, problem_type=problem_type, explanation_method=explanation_method ) # Calcular y almacenar los valores SHAP st.session_state.shap_values = st.session_state.shap_explainer.compute_shap_values( max_samples=max_samples ) shap_explainer = st.session_state.shap_explainer shap_values = st.session_state.shap_values if shap_values is None: st.error("No se pudieron calcular los valores SHAP") return # Pestañas para diferentes visualizaciones tab1, tab2, tab3, tab4 = st.tabs([ "Resumen de Importancia", "Dependencia de Características", "Informe Detallado", "Configuración Avanzada" ]) with tab1: st.header("Resumen de Importancia de Características") # Gráfico de resumen summary_fig = shap_explainer.plot_summary(shap_values) if summary_fig: st.plotly_chart(summary_fig, use_container_width=True) # Selector de características para análisis detallado selected_feature = st.selectbox( "Seleccionar característica para análisis detallado", X.columns.tolist() ) # Gráfico de dependencia para la característica seleccionada dependence_fig = shap_explainer.plot_dependence(shap_values, selected_feature) if dependence_fig: st.plotly_chart(dependence_fig, use_container_width=True) with tab2: st.header("Análisis de Dependencia de Características") # Matriz de correlación de valores SHAP shap_correlation = pd.DataFrame(shap_values).corr() # Heatmap de correlación de valores SHAP fig_corr = px.imshow( shap_correlation, title="Correlación entre Valores SHAP de Características", labels=dict(x="Características", y="Características", color="Correlación") ) st.plotly_chart(fig_corr, use_container_width=True) with tab3: st.header("Informe Detallado de Importancia") # Generar informe de importancia de características importance_report = shap_explainer.generate_feature_importance_report(shap_values) # Mostrar características más importantes st.subheader("Top 5 Características Más Importantes") top_features_df = pd.DataFrame(importance_report.get('top_features', [])) st.dataframe(top_features_df) # Visualización de características más importantes fig_top_features = px.bar( top_features_df, x='importance', y='feature', orientation='h', title="Top 5 Características por Importancia SHAP" ) st.plotly_chart(fig_top_features, use_container_width=True) # Métricas de resumen col1, col2, col3 = st.columns(3) with col1: st.metric("Total de Características", importance_report.get('total_features', 'N/A')) with col2: st.metric("Máxima Importancia", f"{importance_report.get('max_importance', 'N/A'):.4f}") with col3: st.metric("Mínima Importancia", f"{importance_report.get('min_importance', 'N/A'):.4f}") with tab4: st.header("Configuración Avanzada") # Controles de configuración st.subheader("Parámetros de Explicación") # Selector de método de explicación explanation_method = st.selectbox( "Método de Explicación", ["auto", "tree", "linear", "kernel"] ) # Número de muestras para cálculo num_samples = st.slider( "Número de Muestras para Análisis", min_value=10, max_value=min(1000, len(X)), value=min(100, len(X)) ) # Botón para recalcular con nuevos parámetros if st.button("Recalcular SHAP"): with st.spinner("Recalculando valores SHAP..."): try: # Crear y actualizar el explicador SHAP con nuevos parámetros shap_explainer = SHAPExplainer( model=model, X=X, problem_type=problem_type, explanation_method=explanation_method ) st.session_state.shap_explainer = shap_explainer # Calcular y actualizar los valores SHAP shap_values = shap_explainer.compute_shap_values( max_samples=num_samples ) st.session_state.shap_values = shap_values st.success("Valores SHAP recalculados correctamente.") except Exception as e: st.error(f"Error al recalcular SHAP: {str(e)}") def validate_shap_compatibility(model): """ Validar si un modelo es compatible con SHAP Args: model: Modelo de machine learning Returns: bool: True si es compatible, False en caso contrario """ compatible_types = [ 'RandomForestClassifier', 'RandomForestRegressor', 'GradientBoostingClassifier', 'GradientBoostingRegressor', 'XGBClassifier', 'XGBRegressor', 'DecisionTreeClassifier', 'DecisionTreeRegressor', 'LogisticRegression', 'LinearRegression' ] return any( comp_type in str(type(model).__name__) for comp_type in compatible_types ) def generate_shap_documentation(): """ Generar documentación sobre el uso de SHAP Returns: str: Documentación en formato markdown """ documentation = """ ## 🔍 Explicabilidad de Modelos con SHAP ### ¿Qué es SHAP? SHAP (SHapley Additive exPlanations) es una metodología para explicar las predicciones de modelos de machine learning basada en la teoría de juegos. ### Características Principales - Interpretación global y local de modelos - Calcula la contribución de cada característica a la predicción - Funciona con diferentes tipos de modelos ### Tipos de Visualizaciones 1. **Summary Plot**: Importancia general de características 2. **Dependence Plot**: Relación entre características y predicciones 3. **Force Plot**: Contribución individual de características ### Limitaciones - Computacionalmente intensivo para grandes datasets - Puede ser lento con modelos complejos - Requiere comprensión estadística para interpretación precisa ### Mejores Prácticas - Usar como complemento, no como única fuente de verdad - Combinar con otras técnicas de explicabilidad - Interpretar en contexto del problema de negocio """ return documentation # Punto de entrada principal para pruebas def main(): import streamlit as st from sklearn.ensemble import RandomForestClassifier from sklearn.datasets import load_iris # Cargar datos de ejemplo iris = load_iris() X = pd.DataFrame(iris.data, columns=iris.feature_names) y = iris.target # Entrenar modelo de ejemplo model = RandomForestClassifier(n_estimators=100, random_state=42) model.fit(X, y) # Crear dashboard de análisis SHAP create_shap_analysis_dashboard(model, X) if __name__ == "__main__": main()