Skip to content
Draft
Show file tree
Hide file tree
Changes from 10 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,7 @@ minversion = "8.0"
pythonpath = "src"
testpaths = "test"
addopts = [
"-v", # verbose
"-rA", # short test summary info
"--import-mode=importlib", # better control of importing packages
"--showlocals", # show local variable inn trackbacks
Expand Down
29 changes: 16 additions & 13 deletions src/hidimstat/conditional_feature_importance.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from collections.abc import Callable

import numpy as np
from joblib import Parallel, delayed
from sklearn.base import BaseEstimator, check_is_fitted, clone
Expand Down Expand Up @@ -64,7 +66,7 @@ def __init__(
self,
estimator,
method: str = "predict",
loss: callable = mean_squared_error,
loss: Callable = mean_squared_error,
n_permutations: int = 50,
imputation_model_continuous=RidgeCV(),
imputation_model_categorical=LogisticRegressionCV(),
Expand All @@ -86,14 +88,6 @@ def __init__(
random_state=random_state,
)

# check the validity of the inputs
assert imputation_model_continuous is None or issubclass(
imputation_model_continuous.__class__, BaseEstimator
), "Continuous imputation model invalid"
assert imputation_model_categorical is None or issubclass(
imputation_model_categorical.__class__, BaseEstimator
), "Categorial imputation model invalid"

self.feature_types = feature_types
self._list_imputation_models = []
self.categorical_max_cardinality = categorical_max_cardinality
Expand All @@ -116,23 +110,32 @@ def fit(self, X, y=None):
self : object
Returns the instance itself.
"""
del y
super().fit(X, None)
# check the validity of the inputs
assert self.imputation_model_continuous is None or issubclass(
self.imputation_model_continuous.__class__, BaseEstimator
), "Continuous imputation model invalid"
assert self.imputation_model_categorical is None or issubclass(
self.imputation_model_categorical.__class__, BaseEstimator
), "Categorial imputation model invalid"

super().fit(X, y)

# check the feature type
if isinstance(self.feature_types, str):
if self.feature_types in ["auto", "continuous", "categorical"]:
self.feature_types = [
self.feature_types_ = [
self.feature_types for _ in range(self.n_features_groups_)
]
else:
raise ValueError(
"feature_types support only the string 'auto', 'continuous', 'categorical'"
)
else:
self.feature_types_ = self.feature_types

self._list_imputation_models = [
ConditionalSampler(
data_type=self.feature_types[features_group_id],
data_type=self.feature_types_[features_group_id],
model_regression=(
None
if self.imputation_model_continuous is None
Expand Down
3 changes: 3 additions & 0 deletions test/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,10 @@ def rng():


def fitted_linear_regression():
Comment thread
Remi-Gau marked this conversation as resolved.
<<<<<<< sklearn/CFI
=======
Comment thread
Remi-Gau marked this conversation as resolved.
Outdated
"""Return a fitted linear regression model."""
>>>>>>> main
Comment thread
Remi-Gau marked this conversation as resolved.
Outdated
X = _rng().integers(0, 2, size=(100, 2, 1))
estimator = LinearRegression()
estimator.fit(X[:, 0], X[:, 1])
Expand Down
15 changes: 3 additions & 12 deletions test/test_base_perturbation.py
Original file line number Diff line number Diff line change
@@ -1,27 +1,18 @@
import numpy as np
import pytest
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import KFold
from sklearn.utils.estimator_checks import parametrize_with_checks

from hidimstat.base_perturbation import BasePerturbation, BasePerturbationCV

from .conftest import SKLEARN_LT_1_6, _rng, check_estimator


def _fitted_linear_regression():
X = _rng().integers(0, 2, size=(100, 2, 1))
estimator = LinearRegression()
estimator.fit(X[:, 0], X[:, 1])
return estimator

from .conftest import SKLEARN_LT_1_6, check_estimator, fitted_linear_regression

ESTIMATORS_TO_CHECK = [
BasePerturbation(estimator=_fitted_linear_regression()),
BasePerturbation(estimator=fitted_linear_regression()),
BasePerturbation(estimator=LinearRegression()),
BasePerturbationCV(estimators=LinearRegression(), cv=KFold(n_splits=2)),
BasePerturbationCV(
estimators=[_fitted_linear_regression(), _fitted_linear_regression()],
estimators=[fitted_linear_regression(), fitted_linear_regression()],
cv=KFold(n_splits=2),
),
]
Expand Down
158 changes: 145 additions & 13 deletions test/test_conditional_feature_importance.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@
import pytest
from scipy.stats import ttest_1samp
from sklearn.base import clone
from sklearn.exceptions import NotFittedError
from sklearn.linear_model import (
LinearRegression,
LogisticRegression,
Expand All @@ -16,14 +15,145 @@
)
from sklearn.metrics import log_loss, mean_squared_error
from sklearn.model_selection import KFold, train_test_split
from sklearn.utils.estimator_checks import parametrize_with_checks

from hidimstat import CFI, cfi_importance
from hidimstat._utils.exception import InternalError
from hidimstat._utils.scenario import multivariate_simulation
from hidimstat.base_perturbation import BasePerturbation
from hidimstat.conditional_feature_importance import CFICV
from hidimstat.conditional_feature_importance import CFI, CFICV, cfi_importance
from hidimstat.statistical_tools.multiple_testing import fdp_power

from .conftest import SKLEARN_LT_1_6, check_estimator, fitted_linear_regression


def fitted_ridged_cv():
Comment thread
Remi-Gau marked this conversation as resolved.
Outdated
X, y, _, _ = multivariate_simulation(
n_samples=500,
n_features=50,
)
return RidgeCV(alphas=np.logspace(-3, 3, 13)).fit(X, y)


def list_fitted_ridge_cv():
X, y, _, _ = multivariate_simulation(
n_samples=500,
n_features=50,
)
model = RidgeCV(alphas=np.logspace(-3, 3, 13))
cv = KFold(n_splits=2, shuffle=True, random_state=0)
return [
clone(model.fit(X[train_index], y[train_index]))
for train_index, _ in cv.split(X)
]


ESTIMATORS_TO_CHECK = [
CFI(
estimator=fitted_linear_regression(),
imputation_model_continuous=LinearRegression(),
),
CFI(
estimator=LinearRegression(),
imputation_model_continuous=LinearRegression(),
),
CFICV(
estimators=RidgeCV(),
cv=KFold(n_splits=2),
),
CFICV(
estimators=fitted_ridged_cv(),
cv=KFold(n_splits=2),
),
CFICV(
estimators=list_fitted_ridge_cv(),
cv=KFold(n_splits=2, shuffle=True, random_state=0),
),
]


def expected_failed_checks(estimator):
if isinstance(estimator, CFI):
return {
"check_estimator_sparse_array": "TODO",
"check_estimator_sparse_matrix": "TODO",
"check_estimator_sparse_tag": "TODO",
"check_estimators_nan_inf": "TODO",
"check_fit2d_1feature": "TODO",
"check_fit2d_1sample": "TODO",
"check_parameters_default_constructible": "TODO",
}
elif isinstance(estimator, CFICV):
failed_checks = {
"check_fit2d_1feature": "TODO",
"check_parameters_default_constructible": "TODO",
"check_dict_unchanged": "TODO",
"check_dont_overwrite_parameters": "TODO",
"check_estimator_sparse_tag": "TODO",
"check_estimator_sparse_array": "TODO",
"check_estimator_sparse_matrix": "TODO",
"check_estimators_dtypes": "TODO",
"check_estimators_nan_inf": "TODO",
"check_estimators_fit_returns_self": "TODO",
"check_estimators_overwrite_params": "TODO",
"check_f_contiguous_array_estimator": "TODO",
"check_fit_check_is_fitted": "TODO",
"check_fit2d_predict1d": "TODO",
"check_n_features_in": "TODO",
"check_n_features_in_after_fitting": "TODO",
"check_methods_sample_order_invariance": "TODO",
"check_methods_subset_invariance": "TODO",
"check_readonly_memmap_input": "TODO",
}
if isinstance(estimator.estimators, list):
failed_checks |= {
"check_complex_data": "TODO",
"check_dtype_object": "TODO",
"check_estimators_empty_data_messages": "TODO",
"check_estimators_pickle": "TODO",
"check_fit2d_1sample": "TODO",
"check_fit_idempotent": "TODO",
"check_fit_score_takes_y": "TODO",
"check_pipeline_consistency": "TODO",
"check_positive_only_tag_during_fit": "TODO",
}
return failed_checks


if SKLEARN_LT_1_6:

@pytest.mark.parametrize(
"estimator, check, name",
check_estimator(
estimators=ESTIMATORS_TO_CHECK,
return_expected_failed_checks=expected_failed_checks,
),
)
def test_check_estimator_sklearn_valid(estimator, check, name): # noqa: ARG001
"""Check compliance with sklearn estimators."""
check(estimator)

@pytest.mark.xfail(reason="invalid checks should fail")
@pytest.mark.parametrize(
"estimator, check, name",
check_estimator(
estimators=ESTIMATORS_TO_CHECK,
valid=False,
return_expected_failed_checks=expected_failed_checks,
),
)
def test_check_estimator_sklearn_invalid(estimator, check, name): # noqa: ARG001
"""Check compliance with sklearn estimators."""
check(estimator)

else:

@parametrize_with_checks(
estimators=ESTIMATORS_TO_CHECK,
expected_failed_checks=expected_failed_checks,
)
def test_check_estimator_sklearn(estimator, check):
check(estimator)


def run_cfi(X, y, n_permutation, seed):
"""
Expand Down Expand Up @@ -586,23 +716,25 @@ def test_incompatible_imputer(self, data_generator):
X, y, _, _ = data_generator
fitted_model = LinearRegression().fit(X, y)

cfi = CFI(
estimator=fitted_model,
imputation_model_continuous="invalid_imputer",
method="predict",
)
with pytest.raises(
AssertionError, match="Continuous imputation model invalid"
):
cfi = CFI(
estimator=fitted_model,
imputation_model_continuous="invalid_imputer",
method="predict",
)
cfi.fit(X)

cfi = CFI(
estimator=fitted_model,
imputation_model_categorical="invalid_imputer",
method="predict",
)
with pytest.raises(
AssertionError, match="Categorial imputation model invalid"
):
cfi = CFI(
estimator=fitted_model,
imputation_model_categorical="invalid_imputer",
method="predict",
)
cfi.fit(X)

def test_invalid_groups_format(self, data_generator):
"""Test when groups are provided in invalid format"""
Expand Down
Loading