Merge pull request #201 from basf/feat/splines

AnFreTh · web-flow · commit b398d13001e9 · 2025-01-10T13:08:58.000+01:00
RBF and Sigmoid, with scaling strategy
diff --git a/mambular/models/sklearn_base_classifier.py b/mambular/models/sklearn_base_classifier.py
@@ -27,8 +27,12 @@ def __init__(self, model, config, **kwargs):
             "task",
             "cat_cutoff",
             "treat_all_integers_as_numerical",
-            "knots",
             "degree",
+            "scaling_strategy",
+            "n_knots",
+            "use_decision_tree_knots",
+            "knots_strategy",
+            "spline_implementation",
         ]
 
         self.config_kwargs = {
@@ -146,7 +150,7 @@ def build_model(
             The validation target values. Required if `X_val` is provided.
         random_state : int, default=101
             Controls the shuffling applied to the data before applying the split.
-        batch_size : int, default=64
+        batch_size : int, default=128
             Number of samples per gradient update.
         shuffle : bool, default=True
             Whether to shuffle the training data before each epoch.
diff --git a/mambular/models/sklearn_base_lss.py b/mambular/models/sklearn_base_lss.py
@@ -48,8 +48,12 @@ def __init__(self, model, config, **kwargs):
             "task",
             "cat_cutoff",
             "treat_all_integers_as_numerical",
-            "knots",
             "degree",
+            "scaling_strategy",
+            "n_knots",
+            "use_decision_tree_knots",
+            "knots_strategy",
+            "spline_implementation",
         ]
 
         self.config_kwargs = {
diff --git a/mambular/models/sklearn_base_regressor.py b/mambular/models/sklearn_base_regressor.py
@@ -26,6 +26,7 @@ def __init__(self, model, config, **kwargs):
             "cat_cutoff",
             "treat_all_integers_as_numerical",
             "degree",
+            "scaling_strategy",
             "n_knots",
             "use_decision_tree_knots",
             "knots_strategy",
diff --git a/mambular/preprocessing/basis_expansion.py b/mambular/preprocessing/basis_expansion.py
@@ -1,8 +1,10 @@
 import numpy as np
 from scipy.interpolate import BSpline
 from sklearn.base import BaseEstimator, TransformerMixin
+from sklearn.metrics import pairwise_distances
 from sklearn.preprocessing import SplineTransformer
 from sklearn.tree import DecisionTreeClassifier, DecisionTreeRegressor
+from sklearn.utils.validation import check_array
 
 
 class SplineExpansion(BaseEstimator, TransformerMixin):
@@ -41,6 +43,24 @@ def __init__(
         if spline_implementation not in ["scipy", "sklearn"]:
             raise ValueError("Invalid spline implementation. Choose 'scipy' or 'sklearn'.")
 
+        self.fitted = False
+
+    @staticmethod
+    def knot_identification_using_decision_tree(X, y, task="regression", n_knots=5):
+        # Use DecisionTreeClassifier for classification tasks
+        knots = []
+        if task == "classification":
+            tree = DecisionTreeClassifier(max_leaf_nodes=n_knots + 1)
+        elif task == "regression":
+            tree = DecisionTreeRegressor(max_leaf_nodes=n_knots + 1)
+        else:
+            raise ValueError("Invalid task type. Choose 'regression' or 'classification'.")
+        tree.fit(X, y)
+        # Extract thresholds from the decision tree
+        thresholds = tree.tree_.threshold[tree.tree_.threshold != -2]  # type: ignore
+        knots.append(np.sort(thresholds))
+        return knots
+
     def fit(self, X, y=None):
         """
         Fit the preprocessor by determining the knot positions.
@@ -52,43 +72,57 @@ def fit(self, X, y=None):
         Returns:
         - self: Fitted preprocessor.
         """
-        X = np.asarray(X)
+        if self.use_decision_tree and y is None:
+            raise ValueError("Target variable 'y' must be provided when use_decision_tree=True.")
 
-        if self.use_decision_tree:
-            if y is None:
-                raise ValueError("Target variable 'y' must be provided when use_decision_tree=True.")
-            y = np.asarray(y)
-
-            self.knots = []
-            for i in range(X.shape[1]):
-                x_col = X[:, i].reshape(-1, 1)
-
-                # Use DecisionTreeClassifier for classification tasks
-                if self.task == "classification":
-                    tree = DecisionTreeClassifier(max_leaf_nodes=self.n_knots + 1)
-                elif self.task == "regression":
-                    tree = DecisionTreeRegressor(max_leaf_nodes=self.n_knots + 1)
-                else:
-                    raise ValueError("Invalid task type. Choose 'regression' or 'classification'.")
-
-                tree.fit(x_col, y)
-
-                # Extract thresholds from the decision tree
-                thresholds = tree.tree_.threshold[tree.tree_.threshold != -2]  # type: ignore
-                self.knots.append(np.sort(thresholds))
-        else:
-            # Compute knots based on uniform spacing or quantile
-            self.knots = []
-            for i in range(X.shape[1]):
-                if self.strategy == "quantile":
-                    # Use quantile to determine knot locations
-                    quantiles = np.linspace(0, 1, self.n_knots + 2)[1:-1]
-                    knots = np.quantile(X[:, i], quantiles)
-                    self.knots.append(knots)
-                elif self.strategy == "uniform":
-                    # Use uniform spacing within the range of the feature
-                    knots = np.linspace(np.min(X[:, i]), np.max(X[:, i]), self.n_knots + 2)[1:-1]
-                    self.knots.append(knots)
+        self.knots = []
+
+        if self.use_decision_tree and self.spline_implementation == "scipy":
+            self.knots = self.knot_identification_using_decision_tree(X, y, self.task, self.n_knots)
+            self.fitted = True
+
+        elif self.spline_implementation == "scipy" and not self.use_decision_tree:
+            if self.strategy == "quantile":
+                # Use quantile to determine knot locations
+                quantiles = np.linspace(0, 1, self.n_knots + 2)[1:-1]
+                knots = np.quantile(X, quantiles)
+                self.knots.append(knots)
+                self.fitted = True
+                # print("Scipy spline implementation using quantile works in fit phase")
+            elif self.strategy == "uniform":
+                # Use uniform spacing within the range of the feature
+                knots = np.linspace(np.min(X), np.max(X), self.n_knots + 2)[1:-1]
+                self.knots.append(knots)
+                self.fitted = True
+                # print("Scipy spline implementation using uniform works in fit phase")
+
+        elif self.use_decision_tree and self.spline_implementation == "sklearn":
+            self.knots = self.knot_identification_using_decision_tree(X, y, self.task, self.n_knots)
+            knots = np.vstack(self.knots).T
+            self.transformer = SplineTransformer(
+                n_knots=self.n_knots, degree=self.degree, include_bias=False, knots=knots
+            )
+            self.transformer.fit(X)
+            self.fitted = True
+
+        elif self.spline_implementation == "sklearn" and not self.use_decision_tree:
+            if self.strategy == "quantile":
+                # print("Using sklearn spline transformer using quantile")
+                # print()
+                self.transformer = SplineTransformer(
+                    n_knots=self.n_knots, degree=self.degree, include_bias=False, knots="quantile"
+                )
+                self.fitted = True
+                self.transformer.fit(X)
+
+            elif self.strategy == "uniform":
+                # print("Using sklearn spline transformer using uniform")
+                # print()
+                self.transformer = SplineTransformer(
+                    n_knots=self.n_knots, degree=self.degree, include_bias=False, knots="uniform"
+                )
+                self.fitted = True
+                self.transformer.fit(X)
 
         return self
 
@@ -105,43 +139,148 @@ def transform(self, X):
         if self.knots is None:
             raise ValueError("Knots have not been initialized. Please fit the preprocessor first.")
 
-        X = np.asarray(X)
         transformed_features = []
 
+        if self.fitted is False:
+            raise ValueError("Model has not been fitted. Please fit the model first.")
+
         if self.spline_implementation == "scipy":
-            for i in range(X.shape[1]):
-                x_col = X[:, i]
-                knots = self.knots[i]  # type: ignore
+            # Extend the knots for boundary conditions
+            t = np.concatenate(([self.knots[0]] * self.degree, self.knots, [self.knots[-1]] * self.degree))
+
+            # Create spline basis functions for this feature
+            spline_basis = [
+                BSpline.basis_element(t[j : j + self.degree + 2])(X) for j in range(len(t) - self.degree - 1)
+            ]
+            # Stack and append transformed features
+            transformed_features.append(np.vstack(spline_basis).T)
+            # Concatenate all transformed features
+            return np.hstack(transformed_features)
+        elif self.spline_implementation == "sklearn":
+            return self.transformer.transform(X)
 
-                # Extend the knots for boundary conditions
-                t = np.concatenate(([knots[0]] * self.degree, knots, [knots[-1]] * self.degree))
 
-                # Create spline basis functions for this feature
-                spline_basis = [
-                    BSpline.basis_element(t[j : j + self.degree + 2])(x_col) for j in range(len(t) - self.degree - 1)
-                ]
+def center_identification_using_decision_tree(X, y, task="regression", n_centers=5):
+    # Use DecisionTreeClassifier for classification tasks
+    centers = []
+    if task == "classification":
+        tree = DecisionTreeClassifier(max_leaf_nodes=n_centers + 1)
+    elif task == "regression":
+        tree = DecisionTreeRegressor(max_leaf_nodes=n_centers + 1)
+    else:
+        raise ValueError("Invalid task type. Choose 'regression' or 'classification'.")
+    tree.fit(X, y)
+    # Extract thresholds from the decision tree
+    thresholds = tree.tree_.threshold[tree.tree_.threshold != -2]  # type: ignore
+    centers.append(np.sort(thresholds))
+    return centers
 
-                # Stack and append transformed features
-                transformed_features.append(np.vstack(spline_basis).T)
 
-                # Concatenate all transformed features
-            return np.hstack(transformed_features)
+class RBFExpansion(BaseEstimator, TransformerMixin):
+    def __init__(
+        self, n_centers=10, gamma: float = 1.0, use_decision_tree=True, task: str = "regression", strategy="uniform"
+    ):
+        """
+        Radial Basis Function Expansion.
+
+        Parameters:
+        - n_centers: Number of RBF centers.
+        - gamma: Width of the RBF kernel.
+        - use_decision_tree: If True, use a decision tree to determine RBF centers.
+        - task: Task type, 'regression' or 'classification'.
+        - strategy: If 'uniform', centers are uniformly spaced. If 'quantile', centers are
+                    determined by data quantile.
+        """
+        self.n_centers = n_centers
+        self.gamma = gamma
+        self.use_decision_tree = use_decision_tree
+        self.strategy = strategy
+        self.task = task
+
+        if self.strategy not in ["uniform", "quantile"]:
+            raise ValueError("Invalid strategy. Choose 'uniform' or 'quantile'.")
+
+    def fit(self, X, y=None):
+        X = check_array(X)
+
+        if self.use_decision_tree and y is None:
+            raise ValueError("Target variable 'y' must be provided when use_decision_tree=True.")
+
+        if self.use_decision_tree:
+            self.centers_ = center_identification_using_decision_tree(X, y, self.task, self.n_centers)
+            self.centers_ = np.vstack(self.centers_)
         else:
-            if self.use_decision_tree:
-                knots = np.vstack(self.knots).T
-                transformer = SplineTransformer(
-                    n_knots=self.n_knots, degree=self.degree, include_bias=False, knots=knots
-                )
-            else:
-                if self.strategy == "quantile":
-                    transformer = SplineTransformer(
-                        n_knots=self.n_knots, degree=self.degree, include_bias=False, knots="quantile"
-                    )
-                elif self.strategy == "uniform":
-                    transformer = SplineTransformer(
-                        n_knots=self.n_knots, degree=self.degree, include_bias=False, knots="uniform"
-                    )
-                else:
-                    raise ValueError("Invalid strategy for knot location calculation. Choose 'quantile' or 'uniform'.")
-
-            return transformer.fit_transform(X)
+            # Compute centers
+            if self.strategy == "quantile":
+                self.centers_ = np.percentile(X, np.linspace(0, 100, self.n_centers), axis=0)
+            elif self.strategy == "uniform":
+                self.centers_ = np.linspace(X.min(axis=0), X.max(axis=0), self.n_centers)
+
+        # Compute gamma if not provided
+        # if self.gamma is None:
+        #     dists = pairwise_distances(self.centers_)
+        #     self.gamma = 1 / (2 * np.mean(dists[dists > 0]) ** 2)  # Mean pairwise distance
+        return self
+
+    def transform(self, X):
+        X = check_array(X)
+        transformed = []
+        self.centers_ = np.array(self.centers_)
+        for center in self.centers_.T:
+            rbf_features = np.exp(-self.gamma * (X - center) ** 2)  # type: ignore
+            transformed.append(rbf_features)
+        return np.hstack(transformed)
+
+
+class SigmoidExpansion(BaseEstimator, TransformerMixin):
+    def __init__(
+        self, n_centers=10, scale: float = 1.0, use_decision_tree=True, task: str = "regression", strategy="uniform"
+    ):
+        """
+        Sigmoid Basis Expansion.
+
+        Parameters:
+        - n_centers: Number of sigmoid centers.
+        - scale: Scale parameter for sigmoid function.
+        - use_decision_tree: If True, use a decision tree to determine sigmoid centers.
+        - task: Task type, 'regression' or 'classification'.
+        - strategy: If 'uniform', centers are uniformly spaced. If 'quantile', centers are
+                    determined by data quantile.
+        """
+        self.n_centers = n_centers
+        self.scale = scale
+        self.use_decision_tree = use_decision_tree
+        self.strategy = strategy
+        self.task = task
+
+    def fit(self, X, y=None):
+        X = check_array(X)
+
+        if self.use_decision_tree and y is None:
+            raise ValueError("Target variable 'y' must be provided when use_decision_tree=True.")
+
+        if self.use_decision_tree:
+            self.centers_ = center_identification_using_decision_tree(X, y, self.task, self.n_centers)
+            self.centers_ = np.vstack(self.centers_)
+        else:
+            # Compute centers
+            if self.strategy == "quantile":
+                self.centers_ = np.percentile(X, np.linspace(0, 100, self.n_centers), axis=0)
+            elif self.strategy == "uniform":
+                self.centers_ = np.linspace(X.min(axis=0), X.max(axis=0), self.n_centers)
+
+        # Compute gamma if not provided
+        # if self.gamma is None:
+        #     dists = pairwise_distances(self.centers_)
+        #     self.gamma = 1 / (2 * np.mean(dists[dists > 0]) ** 2)  # Mean pairwise distance
+        return self
+
+    def transform(self, X):
+        X = check_array(X)
+        transformed = []
+
+        self.centers_ = np.array(self.centers_)
+        for center in self.centers_.T:
+            sigmoid_features = 1 / (1 + np.exp(-(X - center) / self.scale))
+            transformed.append(sigmoid_features)
+        return np.hstack(transformed)
diff --git a/mambular/preprocessing/preprocessor.py b/mambular/preprocessing/preprocessor.py
