added support and tests for pipeline explainer

Author: jarverha

powershap/shap_wrappers/shap_explainer.py

@@ -12,6 +12,7 @@
 from numpy.random import RandomState
 from sklearn.model_selection import train_test_split
 from sklearn.utils.validation import validate_data
+import ShapExplainerFactory
 
 from sklearn.utils._tags import (
     ClassifierTags,
@@ -367,6 +368,62 @@ def _fit_get_shap(self, X_train, Y_train, X_val, Y_val, random_seed, **kwargs) -
         C_explainer = shap.explainers.Linear(PowerShap_model, X_train)
         return C_explainer.shap_values(X_val)
 
+# This support an Sklearn Pipeline Explainer, which will be a wrapper around a ShapExplainer
+class PipelineExplainer(ShapExplainer):
+
+    def __init__(self, model: Any):
+        """Create a Powershap explainer instance.
+
+        Parameters
+        ----------
+        model: Any
+            The  model from which powershap will use its shap values to perform feature
+            selection.
+
+        """
+        assert self.supports_model(model)
+        self.shap_explainer = ShapExplainerFactory.get_explainer(model=ShapExplainer(model.steps[-1][1]))
+
+    @staticmethod
+    def supports_model(model) -> bool:
+        from sklearn.pipeline import Pipeline
+
+        return isinstance(model, Pipeline)
+
+    def _fit_get_shap(self, X_train, Y_train, X_val, Y_val, random_seed, **kwargs) -> np.array:
+        from sklearn.base import clone
+        from sklearn.pipeline import Pipeline
+        
+        # Because the ShapExplainer behavior is different for each model, we extract the model and only keep the preprocessing pipeline
+        powershap_pipeline = clone(Pipeline(self.model.steps[:-1]))
+
+        # 2. Build the parameter dictionary to set the random states to the random seed
+        params_to_set = {}
+        for step_name, step_estimator in powershap_pipeline.steps:
+            if 'random_state' in step_estimator.get_params():
+                # Format: 'step_name__parameter_name'
+                params_to_set[f'{step_name}__random_state'] = random_seed
+
+        # 3. Apply the parameters to the cloned pipeline
+        powershap_pipeline.set_params(**params_to_set)
+                
+        # We fit the pipeline here to be used to transform the data
+        powershap_pipeline.fit(X_train, Y_train, **kwargs)
+
+        # Get the transformed data from all the preceding steps
+        transformed_X_train = powershap_pipeline.transform(X_train)
+        transformed_X_val = powershap_pipeline.transform(X_val)
+
+        return self.shap_explainer._fit_get_shap(transformed_X_train, Y_train, transformed_X_val, Y_val, random_seed, **kwargs)
+
+
+    def validate_data(self, _estimator, X, y, **kwargs):
+        # The assumption here is that the used model is the limiting factor for validation of the data
+        self.shap_explainer.validate_data(_estimator, X, y, **kwargs)
+    
+    def _get_more_tags(self):
+        return self.shap_explainer._get_more_tags()
+
 
 ### DEEP LEARNING
 

powershap/shap_wrappers/shap_explainer_factory.py

@@ -10,6 +10,7 @@
     LinearExplainer,
     ShapExplainer,
     XGBoostExplainer,
+    PipelineExplainer
 )
 
 
@@ -23,6 +24,7 @@ class ShapExplainerFactory:
         EnsembleExplainer,
         LinearExplainer,
         DeepLearningExplainer,
+        PipelineExplainer
     ]
 
     @classmethod

tests/test_pipeline_powershap.py

@@ -0,0 +1,204 @@
+__author__ = "Jeroen Van Der Donckt, Jarne Verhaeghe"
+
+import numpy as np
+import pandas as pd
+from catboost import CatBoostClassifier, CatBoostRegressor
+from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
+
+from powershap import PowerShap
+
+from .conftest import dummy_classification, dummy_regression
+from sklearn.pipeline import Pipeline
+from sklearn.pipeline import make_pipeline
+from sklearn.preprocessing import FunctionTransformer
+
+def test_pipeline_catboost_class_powershap(dummy_classification):
+    X, y = dummy_classification
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    DummyScaler = FunctionTransformer(lambda x: x)
+
+    selector = PowerShap(
+        model=make_pipeline(DummyScaler, CatBoostClassifier(n_estimators=250, verbose=0)), power_iterations=15, automatic=False
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) == n_informative
+    assert all([c.startswith("informative") for c in selected_feats.columns])
+
+
+def test_pipeline_catboost_regr_powershap(dummy_regression):
+    X, y = dummy_regression
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    DummyScaler = FunctionTransformer(lambda x: x)
+
+    selector = PowerShap(
+        model=make_pipeline(DummyScaler, CatBoostRegressor(n_estimators=250, verbose=0)), power_iterations=15, automatic=False
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) == n_informative
+    assert all([c.startswith("informative") for c in selected_feats.columns])
+
+
+def test_pipeline_catboost_handle_nans(dummy_classification):
+    X, y = dummy_classification
+    X.iloc[:5] = None
+    X["nan_col"] = None
+    assert np.any(pd.isna(X))
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    DummyScaler = FunctionTransformer(lambda x: x)
+
+    selector = PowerShap(
+        model=make_pipeline(DummyScaler, CatBoostClassifier(n_estimators=10, verbose=0)), power_iterations=15
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) == n_informative
+    assert all([c.startswith("informative") for c in selected_feats.columns])
+
+
+def test_pipeline_catboost_handle_infs(dummy_classification):
+    X, y = dummy_classification
+    X.iloc[:5] = np.Inf
+    X["inf_col"] = np.Inf
+    assert np.any(X.isin([np.inf, -np.inf]))
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    DummyScaler = FunctionTransformer(lambda x: x)
+
+    selector = PowerShap(
+        model=make_pipeline(DummyScaler, CatBoostClassifier(n_estimators=10, verbose=0)), power_iterations=15
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) == n_informative
+    assert all([c.startswith("informative") for c in selected_feats.columns])
+
+
+def test_pipeline_catboost_handle_infs_nans(dummy_classification):
+    X, y = dummy_classification
+    X.iloc[:5] = np.Inf
+    X.iloc[5:10] = None
+    X["inf_col"] = np.Inf
+    X["nan_col"] = None
+    assert np.any(X.isin([np.inf, -np.inf]))
+    assert np.any(pd.isna(X))
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    DummyScaler = FunctionTransformer(lambda x: x)
+
+    selector = PowerShap(
+        model=make_pipeline(DummyScaler, CatBoostClassifier(n_estimators=10, verbose=0)), power_iterations=15
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) == n_informative
+    assert all([c.startswith("informative") for c in selected_feats.columns])
+
+
+def test_pipeline_catboost_handle_strings(dummy_classification):
+    X, y = dummy_classification
+    X["cat"] = "miauw"
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    DummyScaler = FunctionTransformer(lambda x: x)
+
+    selector = PowerShap(
+        model=make_pipeline(DummyScaler,CatBoostClassifier(n_estimators=30, verbose=0, cat_features=[X.shape[1] - 1])), 
+        power_iterations=15
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) == n_informative
+    assert all([c.startswith("informative") for c in selected_feats.columns])
+
+
+def test_pipeline_ensemble_class_powershap(dummy_classification):
+    X, y = dummy_classification
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    DummyScaler = FunctionTransformer(lambda x: x)
+    
+    selector = PowerShap(
+        model=make_pipeline(DummyScaler,RandomForestClassifier(n_estimators=25)), power_iterations=15, automatic=False
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) >= n_informative
+    assert sum([c.startswith("informative") for c in selected_feats.columns]) == n_informative
+
+
+def test_pipeline_ensemble_regr_powershap(dummy_regression):
+    X, y = dummy_regression
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    DummyScaler = FunctionTransformer(lambda x: x)
+    
+    selector = PowerShap(
+        model=make_pipeline(DummyScaler,RandomForestRegressor(n_estimators=25)), power_iterations=15, automatic=False
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) >= n_informative
+    assert sum([c.startswith("informative") for c in selected_feats.columns]) == n_informative
+
+def test_pipeline_catboost_class_standardscaler_powershap(dummy_classification):
+    from sklearn.preprocessing import StandardScaler
+
+    X, y = dummy_classification
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    selector = PowerShap(
+        model=make_pipeline(StandardScaler, CatBoostClassifier(n_estimators=250, verbose=0)), power_iterations=15, automatic=False
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) == n_informative
+    assert all([c.startswith("informative") for c in selected_feats.columns])
+
+def test_pipeline_catboost_class_maxabsscalerr_robustscaler_powershap(dummy_classification):
+    from sklearn.preprocessing import MaxAbsScaler, RobustScaler
+
+    X, y = dummy_classification
+    n_informative = sum([c.startswith("informative") for c in X.columns])
+    assert n_informative > 0, "No informative columns in the dummy data!"
+
+    selector = PowerShap(
+        model=make_pipeline(MaxAbsScaler, RobustScaler, CatBoostClassifier(n_estimators=250, verbose=0)), power_iterations=15, automatic=False
+    )
+
+    selector.fit(X, y)
+    selected_feats = selector.transform(X)
+
+    assert len(selected_feats.columns) == n_informative
+    assert all([c.startswith("informative") for c in selected_feats.columns])

tests/test_shap_explainer.py

@@ -10,6 +10,7 @@
     LGBMExplainer,
     LinearExplainer,
     XGBoostExplainer,
+    PipelineExplainer,
 )
 
 
@@ -110,6 +111,69 @@ def test_get_ensemble_explainer():
         explainer = ShapExplainerFactory.get_explainer(model_class())
         assert isinstance(explainer, EnsembleExplainer)
 
+def test_get_pipeline_explainer():
+    from sklearn.linear_model import (
+        LinearRegression,
+        LogisticRegression,
+        LogisticRegressionCV,
+        PassiveAggressiveClassifier,
+        Perceptron,
+        Ridge,
+        RidgeClassifier,
+        RidgeClassifierCV,
+        RidgeCV,
+        SGDClassifier,
+        SGDRegressor,
+    )
+    from sklearn.ensemble import (
+        ExtraTreesClassifier,
+        ExtraTreesRegressor,
+        GradientBoostingClassifier,
+        GradientBoostingRegressor,
+        RandomForestClassifier,
+        RandomForestRegressor,
+    )
+    from catboost import CatBoostClassifier, CatBoostRegressor
+    from lightgbm import LGBMClassifier, LGBMRegressor
+    from xgboost import XGBClassifier, XGBRegressor
+
+    model_classes = [
+        LogisticRegression,
+        LogisticRegressionCV,
+        PassiveAggressiveClassifier,
+        Perceptron,
+        RidgeClassifier,
+        RidgeClassifierCV,
+        SGDClassifier,
+        LinearRegression,
+        Ridge,
+        RidgeCV,
+        SGDRegressor,
+        RandomForestClassifier,
+        GradientBoostingClassifier,
+        ExtraTreesClassifier,
+        RandomForestRegressor,
+        GradientBoostingRegressor,
+        ExtraTreesRegressor,
+        XGBClassifier, XGBRegressor,
+        LGBMClassifier, LGBMRegressor,
+        CatBoostClassifier, CatBoostRegressor,
+    ]
+    from sklearn.pipeline import Pipeline
+    from sklearn.pipeline import make_pipeline
+    from sklearn.preprocessing import FunctionTransformer
+
+
+
+    for model_class in model_classes:
+        DummyScaler = FunctionTransformer(lambda x: x)
+
+        make_pipeline(DummyScaler, model_class)
+
+        explainer = ShapExplainerFactory.get_explainer(make_pipeline)
+        assert isinstance(explainer, PipelineExplainer)
+
+
 
 # def test_get_deep_learning_explainer():
 #     import tensorflow as tf