Spaces:

taseebali
/

verdict

Sleeping

verdict / src /explain /explainability.py

Taseeb Ali

new feature - decision helper

f58e709 10 months ago

12.3 kB

"""Explainability module using SHAP and permutation importance."""

from __future__ import annotations

import warnings
from typing import Dict, Tuple, List, Any, Optional

import numpy as np
import pandas as pd
import shap
from sklearn.inspection import permutation_importance
import matplotlib.pyplot as plt

warnings.filterwarnings("ignore")


class ExplainabilityAnalyzer:
    """Analyzes model predictions and feature importance."""

    def __init__(self, model, X_train, X_test, feature_names: List[str]):
        """
        Args:
            model: Trained model
            X_train: Training features (numpy / pandas / sparse)
            X_test: Test features (numpy / pandas / sparse)
            feature_names: Feature names aligned with model input space
        """
        self.model = model
        self.feature_names = feature_names

        self.X_train = self._to_dense_2d(X_train)
        self.X_test = self._to_dense_2d(X_test)

        self.explainer: Optional[Any] = None
        self.use_fallback: bool = False

        # lazily computed cache (optional)
        self._global_shap_cached: bool = False
        self._global_shap_values: Optional[Any] = None  # can be array/list/Explanation

        self._init_explainer()

    # -------------------------
    # Utilities
    # -------------------------
    def _to_dense_2d(self, X) -> np.ndarray:
        """Convert DataFrame / numpy / sparse matrix to a dense 2D numpy array."""
        if isinstance(X, pd.DataFrame):
            X = X.to_numpy()
        elif hasattr(X, "toarray"):  # scipy sparse
            X = X.toarray()
        else:
            X = np.asarray(X)

        if X.ndim == 1:
            X = X.reshape(-1, 1)
        return X

    def _is_tree_model(self) -> bool:
        return hasattr(self.model, "estimators_") or hasattr(self.model, "tree_")

    def _is_linear_model(self) -> bool:
        return hasattr(self.model, "coef_")

    def _background_sample(self, X: np.ndarray, n: int = 100) -> np.ndarray:
        """Sample background rows for Kernel/Linear explainers."""
        if X.shape[0] <= n:
            return X
        idx = np.random.choice(X.shape[0], n, replace=False)
        return X[idx]

    def _predict_fn(self):
        """
        For classification: explain P(class=1) if possible.
        For regression: explain predicted value.
        """
        if hasattr(self.model, "predict_proba"):
            return lambda X: self.model.predict_proba(X)[:, 1]
        return lambda X: self.model.predict(X)

    def _init_explainer(self) -> None:
        """Initialize best available SHAP explainer for this model."""
        try:
            if self._is_tree_model():
                # Fast + stable for tree-based models
                self.explainer = shap.TreeExplainer(self.model)
                return

            if self._is_linear_model():
                # Good for linear/logistic
                bg = self._background_sample(self.X_train, n=200)
                # For some SHAP versions, feature_names parameter here may not exist; keep safe
                try:
                    self.explainer = shap.LinearExplainer(self.model, bg, feature_names=self.feature_names)
                except TypeError:
                    self.explainer = shap.LinearExplainer(self.model, bg)
                return

            # Fallback: KernelExplainer (slow)
            bg = self._background_sample(self.X_train, n=50)
            self.explainer = shap.KernelExplainer(self._predict_fn(), bg)

        except Exception as e:
            print(
                f"Warning: Could not create SHAP explainer ({type(e).__name__}: {e}). "
                "Using permutation importance fallback."
            )
            self.explainer = None
            self.use_fallback = True

    # -------------------------
    # Permutation importance
    # -------------------------
    def get_feature_importance_permutation(
        self, y_true: np.ndarray, scoring: str = "accuracy"
    ) -> Dict[str, float]:
        """Calculate permutation importance."""
        try:
            result = permutation_importance(
                self.model,
                self.X_test,
                y_true,
                n_repeats=10,
                random_state=42,
                scoring=scoring,
                n_jobs=-1,
            )
            importance_dict = dict(zip(self.feature_names, result.importances_mean))
            return dict(sorted(importance_dict.items(), key=lambda x: x[1], reverse=True))
        except Exception as e:
            print(f"Error calculating permutation importance: {e}")
            return {}

    # -------------------------
    # SHAP helpers
    # -------------------------
    def _compute_shap_for_row(self, x_row: np.ndarray) -> Tuple[Optional[np.ndarray], Optional[float]]:
        """
        Returns:
          shap_values_vector: shape (n_features,)
          base_value: float or None
        """
        if self.explainer is None:
            return None, None

        # Try modern SHAP API: explainer(X) -> shap.Explanation
        try:
            exp = self.explainer(x_row)
            if hasattr(exp, "values"):
                values = exp.values
                base = getattr(exp, "base_values", None)

                # values can be (1, n_features) or (1, n_features, classes)
                values = np.asarray(values)

                # base can be scalar or array-like
                base_val = None
                if base is not None:
                    base_arr = np.asarray(base)
                    # could be shape (1,) or (1, classes)
                    base_val = float(base_arr.flatten()[0])

                # reduce to 1D
                if values.ndim == 3:
                    # (1, n_features, classes) -> take class 1 if exists else class 0
                    cls_idx = 1 if values.shape[2] > 1 else 0
                    v = values[0, :, cls_idx]
                else:
                    v = values.reshape(-1)

                return v, base_val
        except Exception:
            pass

        # Compatibility / older explainers
        try:
            sv = self.explainer.shap_values(x_row)

            # sv can be:
            # - array (1, n_features)
            # - list of arrays for multiclass: [ (1,n_features), ...]
            if isinstance(sv, list):
                # choose class 1 if exists; else class 0
                cls_idx = 1 if len(sv) > 1 else 0
                v = np.asarray(sv[cls_idx]).reshape(-1)
            else:
                v = np.asarray(sv).reshape(-1)

            base_val = None
            ev = getattr(self.explainer, "expected_value", None)
            if ev is not None:
                ev_arr = np.asarray(ev)
                base_val = float(ev_arr.flatten()[0])

            return v, base_val
        except Exception as e:
            print(f"Error computing SHAP values: {e}")
            return None, None

    def get_shap_values(self) -> Tuple[Optional[Any], Optional[Any]]:
        """
        Optional: compute global SHAP values for the full test set (can be slow).
        Returns cached values + explainer.
        """
        if self.explainer is None:
            return None, None
        try:
            # Try calling explainer(X) first
            try:
                exp = self.explainer(self.X_test)
                self._global_shap_values = exp
                self._global_shap_cached = True
                return exp, self.explainer
            except Exception:
                pass

            # Fallback to shap_values API
            sv = self.explainer.shap_values(self.X_test)
            self._global_shap_values = sv
            self._global_shap_cached = True
            return sv, self.explainer
        except Exception as e:
            print(f"Error computing global SHAP values: {e}")
            return None, None

    # -------------------------
    # Prediction + explanation API
    # -------------------------
    def predict_with_confidence(self, X) -> Tuple[np.ndarray, Optional[np.ndarray]]:
        """Get predictions with confidence estimates."""
        Xd = self._to_dense_2d(X)
        predictions = self.model.predict(Xd)

        confidence = None
        if hasattr(self.model, "predict_proba"):
            proba = self.model.predict_proba(Xd)
            confidence = np.max(proba, axis=1)

        return predictions, confidence

    def explain_prediction(self, sample_idx: int) -> Dict[str, Any]:
        """
        Explain a single prediction using SHAP (preferred).
        If SHAP isn't available, returns a fallback explanation (permutation importance not per-row).
        """
        x_row = self.X_test[sample_idx : sample_idx + 1]
        pred = self.model.predict(x_row)[0]

        # SHAP path
        if self.explainer is not None and not self.use_fallback:
            shap_vec, base_val = self._compute_shap_for_row(x_row)

            if shap_vec is None:
                return {"error": "Could not compute SHAP values for this sample", "prediction": pred}

            # map contributions
            contrib = dict(zip(self.feature_names, shap_vec))
            sorted_contrib = dict(sorted(contrib.items(), key=lambda kv: abs(kv[1]), reverse=True))

            return {
                "prediction": pred,
                "feature_contributions": sorted_contrib,
                "base_value": base_val,
                "sample_features": dict(zip(self.feature_names, x_row.flatten())),
                "method": "shap",
            }

        # Fallback (no SHAP): provide zeros + sample values (keeps UI alive)
        return {
            "prediction": pred,
            "feature_contributions": {k: 0.0 for k in self.feature_names},
            "base_value": None,
            "sample_features": dict(zip(self.feature_names, x_row.flatten())),
            "method": "fallback_no_shap",
        }

    # -------------------------
    # Plotting helpers
    # -------------------------
    def plot_shap_summary(self) -> Optional[plt.Figure]:
        """Create SHAP summary plot (global)."""
        if self.explainer is None:
            return None

        # Ensure global shap exists (compute if needed)
        if not self._global_shap_cached:
            self.get_shap_values()

        if self._global_shap_values is None:
            return None

        try:
            fig, ax = plt.subplots(figsize=(10, 6))

            # If we cached a shap.Explanation, use its .values
            if hasattr(self._global_shap_values, "values"):
                vals = self._global_shap_values.values
                vals = np.asarray(vals)
                # reduce multiclass if needed
                if vals.ndim == 3:
                    cls_idx = 1 if vals.shape[2] > 1 else 0
                    vals = vals[:, :, cls_idx]
                shap.summary_plot(vals, self.X_test, feature_names=self.feature_names, show=False)
            else:
                # list/array style
                sv = self._global_shap_values
                if isinstance(sv, list):
                    sv = sv[1] if len(sv) > 1 else sv[0]
                shap.summary_plot(sv, self.X_test, feature_names=self.feature_names, show=False)

            plt.tight_layout()
            return fig
        except Exception as e:
            print(f"Error creating SHAP summary plot: {e}")
            return None

    def plot_shap_force(self, sample_idx: int):
        """Create SHAP force plot object for a single prediction (works best in notebooks)."""
        if self.explainer is None:
            return None

        x_row = self.X_test[sample_idx : sample_idx + 1]
        shap_vec, base_val = self._compute_shap_for_row(x_row)
        if shap_vec is None:
            return None

        try:
            # force_plot returns an object that can be rendered in some environments
            return shap.force_plot(
                base_val if base_val is not None else 0.0,
                shap_vec,
                x_row.flatten(),
                feature_names=self.feature_names,
                show=False,
            )
        except Exception as e:
            print(f"Error creating SHAP force plot: {e}")
            return None

	"""Explainability module using SHAP and permutation importance."""

	from __future__ import annotations

	import warnings
	from typing import Dict, Tuple, List, Any, Optional

	import numpy as np
	import pandas as pd
	import shap
	from sklearn.inspection import permutation_importance
	import matplotlib.pyplot as plt

	warnings.filterwarnings("ignore")


	class ExplainabilityAnalyzer:
	"""Analyzes model predictions and feature importance."""

	def __init__(self, model, X_train, X_test, feature_names: List[str]):
	"""
	Args:
	model: Trained model
	X_train: Training features (numpy / pandas / sparse)
	X_test: Test features (numpy / pandas / sparse)
	feature_names: Feature names aligned with model input space
	"""
	self.model = model
	self.feature_names = feature_names

	self.X_train = self._to_dense_2d(X_train)
	self.X_test = self._to_dense_2d(X_test)

	self.explainer: Optional[Any] = None
	self.use_fallback: bool = False

	# lazily computed cache (optional)
	self._global_shap_cached: bool = False
	self._global_shap_values: Optional[Any] = None # can be array/list/Explanation

	self._init_explainer()

	# -------------------------
	# Utilities
	# -------------------------
	def _to_dense_2d(self, X) -> np.ndarray:
	"""Convert DataFrame / numpy / sparse matrix to a dense 2D numpy array."""
	if isinstance(X, pd.DataFrame):
	X = X.to_numpy()
	elif hasattr(X, "toarray"): # scipy sparse
	X = X.toarray()
	else:
	X = np.asarray(X)

	if X.ndim == 1:
	X = X.reshape(-1, 1)
	return X

	def _is_tree_model(self) -> bool:
	return hasattr(self.model, "estimators_") or hasattr(self.model, "tree_")

	def _is_linear_model(self) -> bool:
	return hasattr(self.model, "coef_")

	def _background_sample(self, X: np.ndarray, n: int = 100) -> np.ndarray:
	"""Sample background rows for Kernel/Linear explainers."""
	if X.shape[0] <= n:
	return X
	idx = np.random.choice(X.shape[0], n, replace=False)
	return X[idx]

	def _predict_fn(self):
	"""
	For classification: explain P(class=1) if possible.
	For regression: explain predicted value.
	"""
	if hasattr(self.model, "predict_proba"):
	return lambda X: self.model.predict_proba(X)[:, 1]
	return lambda X: self.model.predict(X)

	def _init_explainer(self) -> None:
	"""Initialize best available SHAP explainer for this model."""
	try:
	if self._is_tree_model():
	# Fast + stable for tree-based models
	self.explainer = shap.TreeExplainer(self.model)
	return

	if self._is_linear_model():
	# Good for linear/logistic
	bg = self._background_sample(self.X_train, n=200)
	# For some SHAP versions, feature_names parameter here may not exist; keep safe
	try:
	self.explainer = shap.LinearExplainer(self.model, bg, feature_names=self.feature_names)
	except TypeError:
	self.explainer = shap.LinearExplainer(self.model, bg)
	return

	# Fallback: KernelExplainer (slow)
	bg = self._background_sample(self.X_train, n=50)
	self.explainer = shap.KernelExplainer(self._predict_fn(), bg)

	except Exception as e:
	print(
	f"Warning: Could not create SHAP explainer ({type(e).__name__}: {e}). "
	"Using permutation importance fallback."
	)
	self.explainer = None
	self.use_fallback = True

	# -------------------------
	# Permutation importance
	# -------------------------
	def get_feature_importance_permutation(
	self, y_true: np.ndarray, scoring: str = "accuracy"
	) -> Dict[str, float]:
	"""Calculate permutation importance."""
	try:
	result = permutation_importance(
	self.model,
	self.X_test,
	y_true,
	n_repeats=10,
	random_state=42,
	scoring=scoring,
	n_jobs=-1,
	)
	importance_dict = dict(zip(self.feature_names, result.importances_mean))
	return dict(sorted(importance_dict.items(), key=lambda x: x[1], reverse=True))
	except Exception as e:
	print(f"Error calculating permutation importance: {e}")
	return {}

	# -------------------------
	# SHAP helpers
	# -------------------------
	def _compute_shap_for_row(self, x_row: np.ndarray) -> Tuple[Optional[np.ndarray], Optional[float]]:
	"""
	Returns:
	shap_values_vector: shape (n_features,)
	base_value: float or None
	"""
	if self.explainer is None:
	return None, None

	# Try modern SHAP API: explainer(X) -> shap.Explanation
	try:
	exp = self.explainer(x_row)
	if hasattr(exp, "values"):
	values = exp.values
	base = getattr(exp, "base_values", None)

	# values can be (1, n_features) or (1, n_features, classes)
	values = np.asarray(values)

	# base can be scalar or array-like
	base_val = None
	if base is not None:
	base_arr = np.asarray(base)
	# could be shape (1,) or (1, classes)
	base_val = float(base_arr.flatten()[0])

	# reduce to 1D
	if values.ndim == 3:
	# (1, n_features, classes) -> take class 1 if exists else class 0
	cls_idx = 1 if values.shape[2] > 1 else 0
	v = values[0, :, cls_idx]
	else:
	v = values.reshape(-1)

	return v, base_val
	except Exception:
	pass

	# Compatibility / older explainers
	try:
	sv = self.explainer.shap_values(x_row)

	# sv can be:
	# - array (1, n_features)
	# - list of arrays for multiclass: [ (1,n_features), ...]
	if isinstance(sv, list):
	# choose class 1 if exists; else class 0
	cls_idx = 1 if len(sv) > 1 else 0
	v = np.asarray(sv[cls_idx]).reshape(-1)
	else:
	v = np.asarray(sv).reshape(-1)

	base_val = None
	ev = getattr(self.explainer, "expected_value", None)
	if ev is not None:
	ev_arr = np.asarray(ev)
	base_val = float(ev_arr.flatten()[0])

	return v, base_val
	except Exception as e:
	print(f"Error computing SHAP values: {e}")
	return None, None

	def get_shap_values(self) -> Tuple[Optional[Any], Optional[Any]]:
	"""
	Optional: compute global SHAP values for the full test set (can be slow).
	Returns cached values + explainer.
	"""
	if self.explainer is None:
	return None, None
	try:
	# Try calling explainer(X) first
	try:
	exp = self.explainer(self.X_test)
	self._global_shap_values = exp
	self._global_shap_cached = True
	return exp, self.explainer
	except Exception:
	pass

	# Fallback to shap_values API
	sv = self.explainer.shap_values(self.X_test)
	self._global_shap_values = sv
	self._global_shap_cached = True
	return sv, self.explainer
	except Exception as e:
	print(f"Error computing global SHAP values: {e}")
	return None, None

	# -------------------------
	# Prediction + explanation API
	# -------------------------
	def predict_with_confidence(self, X) -> Tuple[np.ndarray, Optional[np.ndarray]]:
	"""Get predictions with confidence estimates."""
	Xd = self._to_dense_2d(X)
	predictions = self.model.predict(Xd)

	confidence = None
	if hasattr(self.model, "predict_proba"):
	proba = self.model.predict_proba(Xd)
	confidence = np.max(proba, axis=1)

	return predictions, confidence

	def explain_prediction(self, sample_idx: int) -> Dict[str, Any]:
	"""
	Explain a single prediction using SHAP (preferred).
	If SHAP isn't available, returns a fallback explanation (permutation importance not per-row).
	"""
	x_row = self.X_test[sample_idx : sample_idx + 1]
	pred = self.model.predict(x_row)[0]

	# SHAP path
	if self.explainer is not None and not self.use_fallback:
	shap_vec, base_val = self._compute_shap_for_row(x_row)

	if shap_vec is None:
	return {"error": "Could not compute SHAP values for this sample", "prediction": pred}

	# map contributions
	contrib = dict(zip(self.feature_names, shap_vec))
	sorted_contrib = dict(sorted(contrib.items(), key=lambda kv: abs(kv[1]), reverse=True))

	return {
	"prediction": pred,
	"feature_contributions": sorted_contrib,
	"base_value": base_val,
	"sample_features": dict(zip(self.feature_names, x_row.flatten())),
	"method": "shap",
	}

	# Fallback (no SHAP): provide zeros + sample values (keeps UI alive)
	return {
	"prediction": pred,
	"feature_contributions": {k: 0.0 for k in self.feature_names},
	"base_value": None,
	"sample_features": dict(zip(self.feature_names, x_row.flatten())),
	"method": "fallback_no_shap",
	}

	# -------------------------
	# Plotting helpers
	# -------------------------
	def plot_shap_summary(self) -> Optional[plt.Figure]:
	"""Create SHAP summary plot (global)."""
	if self.explainer is None:
	return None

	# Ensure global shap exists (compute if needed)
	if not self._global_shap_cached:
	self.get_shap_values()

	if self._global_shap_values is None:
	return None

	try:
	fig, ax = plt.subplots(figsize=(10, 6))

	# If we cached a shap.Explanation, use its .values
	if hasattr(self._global_shap_values, "values"):
	vals = self._global_shap_values.values
	vals = np.asarray(vals)
	# reduce multiclass if needed
	if vals.ndim == 3:
	cls_idx = 1 if vals.shape[2] > 1 else 0
	vals = vals[:, :, cls_idx]
	shap.summary_plot(vals, self.X_test, feature_names=self.feature_names, show=False)
	else:
	# list/array style
	sv = self._global_shap_values
	if isinstance(sv, list):
	sv = sv[1] if len(sv) > 1 else sv[0]
	shap.summary_plot(sv, self.X_test, feature_names=self.feature_names, show=False)

	plt.tight_layout()
	return fig
	except Exception as e:
	print(f"Error creating SHAP summary plot: {e}")
	return None

	def plot_shap_force(self, sample_idx: int):
	"""Create SHAP force plot object for a single prediction (works best in notebooks)."""
	if self.explainer is None:
	return None

	x_row = self.X_test[sample_idx : sample_idx + 1]
	shap_vec, base_val = self._compute_shap_for_row(x_row)
	if shap_vec is None:
	return None

	try:
	# force_plot returns an object that can be rendered in some environments
	return shap.force_plot(
	base_val if base_val is not None else 0.0,
	shap_vec,
	x_row.flatten(),
	feature_names=self.feature_names,
	show=False,
	)
	except Exception as e:
	print(f"Error creating SHAP force plot: {e}")
	return None