28934299777

Committed 08 Jul 2026 10:02AM UTC coverage: 94.432% (+0.1%) from 94.323%

Build # 28934299777

Build Type

push

github

Committed by

web-flow

Commit Message

sklearn PLS.fit_transform returns (X-scores, Y-scores); parameter constraints & type hints (6.1.0) (#49)

* IKPLS 6.1.0: sklearn PLS.fit_transform returns (X-scores, Y-scores)

ikpls.sklearn.PLS previously inherited TransformerMixin.fit_transform, which
returns only the X-scores of transform(X). Override it to fit (through fit, so a
subclass overriding fit still runs) and return transform(X, y) -- the
(x_scores, y_scores) tuple, matching sklearn.cross_decomposition.PLSRegression
and ikpls.numpy.PLS.fit_transform. This is a behavior change for callers that
relied on the previous X-scores-only return.

It also introduces two expected failures in scikit-learn's check_estimator suite
(check_transformer_general, check_transformer_data_not_an_array): those compare
fit_transform(X, y) against transform(X), and scikit-learn only special-cases the
tuple return for a hard-coded set of class names (CROSS_DECOMPOSITION), of which
"PLS" is not one. scikit-learn's own PLSRegression has identical tuple behavior
and passes only by virtue of its name, so these are not real conformance
defects; the conformance test marks them expected-fail.

Bumps the version to 6.1.0.

* Add type hints and _parameter_constraints to ikpls.sklearn.PLS

Annotate the public method signatures of the scikit-learn wrapper (fit,
fit_transform, transform, inverse_transform, predict, predict_all_components,
score, y_rotations_, __sklearn_tags__), matching the NumPy backend's
numpy.typing conventions. The constructor parameters are deliberately left
unannotated: the sklearn contract stores them verbatim and validates at fit, and
annotating them would make the typeguard-instrumented test suite reject
non-conforming values at construction instead of yielding the clean fit-time
error.

Replace the ad-hoc _validate_n_components with scikit-learn's
_parameter_constraints / _validate_params: n_components is Interval(Integral, 1,
None), algorithm is Options({1, 2}), the centering/scaling flags are bool... (continued)

Coverage Stats

1238 of 1311 relevant lines covered (94.43%)

11.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.27

ikpls/numpy.py

"""
This file contains the PLS Class which implements partial least-squares regression
using Improved Kernel PLS by Dayal and MacGregor:
https://doi.org/10.1002/(SICI)1099-128X(199701)11:1%3C73::AID-CEM435%3E3.0.CO;2-%23

This class is written using NumPy. Its native API is optimized for fast
cross-validation: ``fit`` takes the number of components ``A`` directly, and
``predict`` can return predictions for every number of components ``1..A`` at
once. For use inside the scikit-learn ecosystem (e.g. ``cross_validate``,
``Pipeline``, ``GridSearchCV``), a thin scikit-learn-conformant wrapper,
``ikpls.sklearn.PLS``, subclasses ``BaseEstimator`` and delegates to this
class.

Author: Ole-Christian Galbo Engstrøm
E-mail: ocge@foss.dk
"""

from collections.abc import Callable, Hashable
from typing import Any, Iterable, Optional, Tuple, Union

import joblib
import numpy as np
import numpy.typing as npt
from joblib import Parallel, delayed
from sklearn.exceptions import NotFittedError

from ikpls._impl.pls_steps import improved_kernel_pls_inner_loop
from ikpls._impl.r_y_mapping import _R_Y_Mapping


class PLS:
    """
    Implements partial least-squares regression using Improved Kernel PLS by Dayal and
    MacGregor:
    https://doi.org/10.1002/(SICI)1099-128X(199701)11:1%3C73::AID-CEM435%3E3.0.CO;2-%23

    Parameters
    ----------
    algorithm : int, default=1
        Whether to use Improved Kernel PLS Algorithm #1 or #2.

    center_X : bool, default=True
        Whether to center `X` before fitting by subtracting its row of
        column-wise means from each row.

    center_Y : bool, default=True
        Whether to center `Y` before fitting by subtracting its row of
        column-wise means from each row.

    scale_X : bool, default=True
        Whether to scale `X` before fitting by dividing each row with the row of `X`'s
        column-wise standard deviations.

    scale_Y : bool, default=True
        Whether to scale `Y` before fitting by dividing each row with the row of `Y`'s
        column-wise standard deviations.

    ddof : int, default=0
        The delta degrees of freedom to use when computing the sample standard
        deviation. A value of 0 corresponds to the biased estimate of the sample
        standard deviation, while a value of 1 corresponds to Bessel's correction for
        the sample standard deviation.

    copy : bool, default=True
            Whether to copy `X` and `Y` in before potentially applying centering and
            scaling. If True, then the data is copied before fitting. If False, and `dtype`
            matches the type of `X` and `Y`, then centering and scaling is done inplace,
            modifying both arrays.

    dtype : type[np.floating], default=numpy.float64
        The float datatype to use in computation of the PLS algorithm. This should be
        numpy.float32 or numpy.float64. Using a lower precision than float64
        will yield significantly worse results when using an increasing number of
        components due to propagation of numerical errors.

    Raises
    ------
    ValueError
        If `algorithm` is not 1 or 2.

    Notes
    -----
    Any centering and scaling is undone before returning predictions to ensure that
    predictions are on the original scale. If both centering and scaling are True, then
    the data is first centered and then scaled.
    """

    def __init__(
        self,
        algorithm: int = 1,
        center_X: bool = True,
        center_Y: bool = True,
        scale_X: bool = True,
        scale_Y: bool = True,
        ddof: int = 0,
        copy: bool = True,
        dtype: type[np.floating] = np.float64,
    ) -> None:
        self.algorithm = algorithm
        self.center_X = center_X
        self.center_Y = center_Y
        self.scale_X = scale_X
        self.scale_Y = scale_Y
        self.ddof = ddof
        self.copy = copy
        self.dtype = dtype
        self.eps = np.finfo(dtype).eps
        self.name = f"Improved Kernel PLS Algorithm #{algorithm}"
        if self.algorithm not in [1, 2]:
            raise ValueError(
                f"Invalid algorithm: {self.algorithm}. Algorithm must be 1 or 2."
            )
        self.A = None
        self.N = None
        self.K = None
        self.M = None
        self.B = None
        self.W = None
        self.P = None
        self.Q = None
        self.C = None  # Y-weights; aliased to Q in fit (see the C attribute docs)
        self.R = None
        self.R_Y = None
        self.T = None
        self.X_mean = None
        self.Y_mean = None
        self.X_std = None
        self.Y_std = None
        self.max_stable_components = None
        self.fitted_ = False

    def _convert_input_to_array(
        self, arr: Optional[npt.ArrayLike]
    ) -> Optional[npt.NDArray[np.floating]]:
        """
        Converts input array to numpy array of type self.dtype. Inserts a second axis
        if the input array is 1-dimensional.

        Parameters
        ----------
        arr : ArrayLike or None
            Input array.

        Returns
        -------
        array_converted : Array of shape of input array or None
            Converted input array.
        """
        if arr is not None:
            arr = np.asarray(arr, dtype=self.dtype)
            if arr.ndim == 1:
                arr = arr.reshape(-1, 1)
        return arr

    def _copy_arrays(
        self,
        copy_X: bool,
        copy_Y: bool,
        X: Optional[npt.NDArray[np.floating]] = None,
        Y: Optional[npt.NDArray[np.floating]] = None,
    ) -> Tuple[Optional[npt.NDArray[np.floating]], Optional[npt.NDArray[np.floating]]]:
        """
        Copies `X and `Y` so that downstream centering and scaling do not modify the
        input directly but instead work on copies thereof.

        Parameters
        ----------
        copy_X: bool
            Whether to copy `X`.

        copy_Y : bool
            Whether to copy `Y`.

        X : Array of shape (N, K) or None
            Predictor variables.

        Y : Array of shape (N, M) or None
            Response variables.

        Returns
        -------
        X_copy : Array of shape (N, K) or None
            A copy of `X` if it is provided and the following conditions are met:
            `copy_X` is True, and `X` will be either centered or scaled.

        Y_copy : Array of shape (N, M) or None
            A copy of `Y` if it is provided and the following conditions are met:
            `copy_Y` is True, and `Y` will be either centered or scaled.
        """
        if copy_X and X is not None and (self.center_X or self.scale_X):
            X = X.copy()
        if copy_Y and Y is not None and (self.center_Y or self.scale_Y):
            Y = Y.copy()
        return X, Y

    def _get_input(
        self,
        copy: bool,
        X: Optional[npt.ArrayLike] = None,
        Y: Optional[npt.ArrayLike] = None,
        sample_weight: Optional[npt.ArrayLike] = None,
    ) -> Tuple[
        Optional[npt.ArrayLike], Optional[npt.ArrayLike], Optional[npt.ArrayLike]
    ]:
        """
        Applies `self._convert_input_to_array` on `X`, `Y`, and `sample_weight` and then
        applies `self._copy_arrays` to `X` and `Y`. See those two functions for a
        description of parameters and return values.

        Returns
        -------
        X, Y, sample_weight
            Converted inputs for safe downstream processing.
        """

        new_X = self._convert_input_to_array(arr=X)
        new_Y = self._convert_input_to_array(arr=Y)
        copy_X = copy and (np.may_share_memory(new_X, X))
        copy_Y = copy and (np.may_share_memory(new_Y, Y))
        sample_weight = self._convert_input_to_array(arr=sample_weight)
        if sample_weight is not None:
            self._check_nonnegative_weights(sample_weight=sample_weight)
        new_X, new_Y = self._copy_arrays(copy_X=copy_X, copy_Y=copy_Y, X=new_X, Y=new_Y)
        return new_X, new_Y, sample_weight

    def _center_scale_X_Y(
        self,
        X: Optional[npt.NDArray[np.floating]] = None,
        Y: Optional[npt.NDArray[np.floating]] = None,
    ) -> Tuple[Optional[npt.NDArray[np.floating]], Optional[npt.NDArray[np.floating]]]:
        """
        Centers and scales the input array based on the model's parameters.

        Parameters
        ----------
        X : Array of shape (N, K) or None
            Predictor variables.

        Y : Array of shape (N, M) or None
            Response variables.

        Returns
        -------
        X_centered_scaled : Array of shape (N, K) or None
            Potentially centered and scaled `X`.

        Y_centered_scaled : Array of shape (N, M) or None
            Potentially centered and scaled `Y`.
        """
        if X is not None:
            if self.center_X:
                X -= self.X_mean if self.X_mean is not None else X
            if self.scale_X:
                X /= self.X_std if self.X_std is not None else X
        if Y is not None:
            if self.center_Y:
                Y -= self.Y_mean if self.Y_mean is not None else Y
            if self.scale_Y:
                Y /= self.Y_std if self.Y_std is not None else Y
        return (X, Y)

    def _un_center_scale_X_Y(
        self,
        X: Optional[npt.NDArray[np.floating]] = None,
        Y: Optional[npt.NDArray[np.floating]] = None,
    ) -> Tuple[Optional[npt.NDArray[np.floating]], Optional[npt.NDArray[np.floating]]]:
        """
        Restores `X` and `Y`to their original scale and location by undoing any
        centering and scaling.

        Parameters
        ----------
        X : Array of shape (N, K) or None
            Predictor variables.

        Y : Array of shape (N, M) or None
            Response variables.

        Returns
        -------
        X_restored : Array of shape (N, K) or None
            `X` with any centering and scaling undone.

        Y_restored : Array of shape (N, M) or None
            `Y` with any centering and scaling undone.
        """
        if X is not None:
            if self.scale_X:
                X *= self.X_std
            if self.center_X:
                X += self.X_mean
        if Y is not None:
            if self.scale_Y:
                Y *= self.Y_std
            if self.center_Y:
                Y += self.Y_mean
        return (X, Y)

    def _check_nonnegative_weights(self, sample_weight: npt.NDArray[np.floating]) -> None:
        """
        Checks that weights are non-negative.

        Parameters
        ----------
        sample_weight : Array of shape (N, 1)
            Sample weights used to fit PLS.

        Returns
        -------
        None

        Raises
        ------
        ValueError
            If weights are not all non-negative.
        """
        if np.any(sample_weight < 0):
            raise ValueError("Weights must be non-negative.")

    def _check_num_nonzero_weights_greater_than_ddof(
        self, num_nonzero_weights: Union[int, np.integer]
    ) -> None:
        """
        Checks that the number of nonzero weights are greater than `self.ddof` if
        either of `self.scale_X` or `self.scale_Y` is True so we must compute the
        corresponding weighted standard deviation.

        Parameters
        ----------
        num_nonzero_weights : int
            The number of nonzero weights. ``np.count_nonzero`` returns a builtin
            ``int`` on NumPy < 2.4 and an ``np.intp`` on newer NumPy, so both are
            admitted."

        Returns
        -------
        None

        Raises
        ------
        ValueError
            If the number of nonzero weights is not greater than `self.ddof`
        """
        if self.scale_X or self.scale_Y:
            if not num_nonzero_weights > self.ddof:
                raise ValueError(
                    f"The number of nonzero weights: {num_nonzero_weights} must be greater than ddof: {self.ddof}"
                )

    def _compute_mean_and_std(
        self,
        X: npt.NDArray[np.floating],
        Y: npt.NDArray[np.floating],
        sample_weight: Optional[npt.NDArray[np.floating]],
    ) -> None:
        """
        Computes the weighted means and standard deviations for `X` and `Y`.

        Parameters
        ----------
        X : Array of shape (N, K)
            Predictor variables.

        Y : Array of shape (N, M):
            Response variables.

        sample_weight : Array of shape (N, 1) or None
            Sample weights.

        Attributes
        ----------
        X_mean : Array of shape (1, K) or None
            Mean of X. If centering is not performed, this is None. If weights are
            used, then this is the weighted mean.

        Y_mean : Array of shape (1, M) or None
            Mean of Y. If centering is not performed, this is None. If weights are
            used, then this is the weighted mean.

        X_std : Array of shape (1, K) or None
            Sample standard deviation of X. If scaling is not performed, this is None.
            If weights are used, then this is the weighted standard deviation.

        Y_std : Array of shape (1, M) or None
            Sample standard deviation of Y. If scaling is not performed, this is None.
            If weights are used, then this is the weighted standard deviation.

        Returns
        -------
        None

        Raises
        ------
        ValueError
            If `sample_weight` are provided and scaling of `X` or `Y` is required and the
            number of non-negative weights is less or equal to `self.ddof`.
        """
        if not (self.center_X or self.center_Y or self.scale_X or self.scale_Y):
            return
        if sample_weight is not None:
            flattened_weights = sample_weight.flatten()
            if self.scale_X or self.scale_Y:
                num_non_zero_weights = np.count_nonzero(sample_weight)
                self._check_num_nonzero_weights_greater_than_ddof(
                    num_nonzero_weights=num_non_zero_weights
                )
                scale_dof = num_non_zero_weights - self.ddof
                avg_non_zero_weights = np.sum(sample_weight) / num_non_zero_weights
        else:
            flattened_weights = None

        if self.center_X or self.scale_X:
            self.X_mean = np.asarray(
                np.average(X, axis=0, weights=flattened_weights, keepdims=True),
                dtype=self.dtype,
            )

        if self.center_Y or self.scale_Y:
            self.Y_mean = np.asarray(
                np.average(Y, axis=0, weights=flattened_weights, keepdims=True),
                dtype=self.dtype,
            )

        if self.scale_X:
            if sample_weight is None:
                self.X_std = X.std(
                    axis=0,
                    ddof=self.ddof,
                    dtype=self.dtype,
                    keepdims=True,
                    mean=self.X_mean,
                )
            else:
                self.X_std = np.sqrt(
                    np.sum(sample_weight * (X - self.X_mean) ** 2, axis=0, keepdims=True)
                    / (scale_dof * avg_non_zero_weights)
                )
            self.X_std[self.X_std <= self.eps] = 1

        if self.scale_Y:
            if sample_weight is None:
                self.Y_std = Y.std(
                    axis=0,
                    ddof=self.ddof,
                    dtype=self.dtype,
                    keepdims=True,
                    mean=self.Y_mean,
                )
            else:
                self.Y_std = np.sqrt(
                    np.sum(sample_weight * (Y - self.Y_mean) ** 2, axis=0, keepdims=True)
                    / (scale_dof * avg_non_zero_weights)
                )
            self.Y_std[self.Y_std <= self.eps] = 1

    def fit(
        self,
        X: npt.ArrayLike,
        Y: npt.ArrayLike,
        A: int,
        sample_weight: Optional[npt.ArrayLike] = None,
    ) -> "PLS":
        """
        Fits Improved Kernel PLS Algorithm #1 on `X` and `Y` using `A` components.

        Parameters
        ----------
        X : Array of shape (N, K)
            Predictor variables.

        Y : Array of shape (N, M) or (N,)
            Response variables.

        A : int
            Number of components in the PLS model.

        sample_weight : Array of shape (N,) or None, optional, default=None
            Weights for each observation. If None, then all observations are weighted
            equally.

        Attributes
        ----------
        A : int
            Number of components in the PLS model.

        max_stable_components : int
            Number of leading, numerically stable components. Equals `A` unless a
            component is detected as unstable, in which case it is the index of the
            first such component. A component is unstable if its X-weight norm
            underflows below machine epsilon (the X-Y cross-covariance is exhausted)
            or if its score norm ``t^T t`` collapses relative to the largest score
            norm seen so far (a null-space direction past the numerical rank of X).
            Such components carry no predictive information: the regression
            coefficients are carried forward, so `predict` with more than
            `max_stable_components` components returns the same result as with exactly
            `max_stable_components`. Their scores/loadings (`W`, `P`, `Q`, `R`, `T`),
            however, are not meaningful -- zeroed here in the NumPy backend, left
            un-normalized in the JAX backend -- so slice to `max_stable_components`
            before comparing them across backends.

        B : Array of shape (A, K, M)
            PLS regression coefficients tensor.

        W : Array of shape (K, A)
            PLS weights matrix for X.

        P : Array of shape (K, A)
            PLS loadings matrix for X.

        Q : Array of shape (M, A)
            PLS Loadings matrix for Y.

        R : Array of shape (K, A)
            PLS weights matrix to compute scores T directly from original X.

        R_Y : Mapping[int, Array]
            Mapping from number of components to PLS weights matrix to compute scores U
            directly from original Y. Keys range from 1 to A. Values are arrays of
            shape (M, n_components) where n_components is the key. Values are computed
            lazily and cached upon first access. See Notes for more information.

        C : Array of shape (M, A)
            PLS Y-weights. In Improved Kernel PLS these equal the Y-loadings Q
            (matching scikit-learn's PLSRegression, whose Y-weights equal its
            Y-loadings). Exposed as an alias of Q under the conventional Y-weights
            name.

        T : Array of shape (N, A)
            PLS scores matrix of X. Only assigned for Improved Kernel PLS Algorithm #1.
            IMPORTANT: If weights are provided, these are NOT the scores of X but
            instead weighted scores. In this case, scores can be computerd using
            transform.

        X_mean : Array of shape (1, K) or None
            Mean of X. If centering is not performed, this is None. If weights are
            used, then this is the weighted mean.

        Y_mean : Array of shape (1, M) or None
            Mean of Y. If centering is not performed, this is None. If weights are
            used, then this is the weighted mean.

        X_std : Array of shape (1, K) or None
            Sample standard deviation of X. If scaling is not performed, this is None.
            If weights are used, then this is the weighted standard deviation.

        Y_std : Array of shape (1, M) or None
            Sample standard deviation of Y. If scaling is not performed, this is None.
            If weights are used, then this is the weighted standard deviation.

        Returns
        -------
        self : PLS
            Fitted model.

        Raises
        ------
        ValueError
            If `sample_weight` are provided and not all weights are non-negative.

        ValueError
            If `sample_weight` are provided, and `scale_X` or `scale_Y` is True, and the
            number of non-zero weights is not greater than `ddof`.

        Warns
        -----
        UserWarning.
            If at any point during iteration over the number of components `A`, the
            residual goes below machine precision for np.float64.

        See Also
        --------
        transform : Transforms `X` and `Y` to their respective scores.
        fit_transform : Fits the model and returns the scores of `X` and `Y`.

        Notes
        -----
        `R_Y` is provided for convenience only as it is not required to derive `B`.
        Therefore, every value in `R_Y` is computed lazily and only actually evaluated
        when accessed by its key for the first time after a call to `fit` - either by
        the user or because it is needed by `transform`. After a value is computed, it
        is cached for fast future retrieval. R_Y is implemented as a concrete Mapping.
        """
        self.fitted_ = True
        X, Y, sample_weight = self._get_input(copy=self.copy, X=X, Y=Y, sample_weight=sample_weight)
        self._compute_mean_and_std(X=X, Y=Y, sample_weight=sample_weight)
        X, Y = self._center_scale_X_Y(X=X, Y=Y)

        N, K = X.shape
        M = Y.shape[1]

        self.A = A
        self.N = N
        self.K = K
        self.M = M

        # Step 1
        if sample_weight is not None:
            if self.algorithm == 2 or K < M:
                XTW = (X * sample_weight).T
                XTY = XTW @ Y
            else:
                XTY = X.T @ (Y * sample_weight)
        else:
            XTY = X.T @ Y

        # Used for algorithm #2
        if self.algorithm == 2:
            if sample_weight is not None:
                XTX = XTW @ X
            else:
                XTX = X.T @ X
        else:
            if sample_weight is not None:
                X = np.sqrt(sample_weight) * X

        # Execute Improved Kernel PLS steps 2-5 (shared with fast cross-validation).
        # The inner loop reports how many leading components are numerically stable
        # (see _stability_warning); it emits the underflow/collapse warnings itself.
        self.B, W, P, Q, R, T, max_stable = improved_kernel_pls_inner_loop(
            self.algorithm,
            A,
            K,
            M,
            XTY,
            X=X if self.algorithm == 1 else None,
            XTX=XTX if self.algorithm == 2 else None,
            dtype=self.dtype,
            eps=self.eps,
        )
        self.max_stable_components = max_stable
        self.W = W
        self.P = P
        self.Q = Q
        self.C = self.Q  # Y-weights == Y-loadings (see the C attribute docs)
        self.R = R
        self.R_Y = _R_Y_Mapping(self.Q, backend="numpy")
        if self.algorithm == 1:
            self.T = T

        return self

    def predict(
        self, X: npt.ArrayLike, n_components: Optional[int] = None
    ) -> npt.NDArray[np.floating]:
        """
        Predicts on `X` with `B` using `n_components` components. If `n_components` is
        None, then predictions are returned for each individual number of components.

        Parameters
        ----------
        X : Array of shape (N, K)
            Predictor variables.

        n_components : int or None, optional, default=None.
            Number of components in the PLS model. If None, then each individual number
            of components is used.

        Returns
        -------
        Y_pred : Array of shape (N, M) or (A, N, M)
            If `n_components` is an int, then an array of shape (N, M) with the
            predictions for that specific number of components is used. If
            `n_components` is None, returns a prediction for each number of components
            up to `A`.

        Raises
        ------
        NotFittedError
            If the model has not been fitted before calling `predict()`.
        """
        if not self.fitted_:
            raise NotFittedError(
                "This model is not fitted yet, call 'fit' with approriate arguments before 'predict'."
            )
        X, _, _ = self._get_input(copy=True, X=X)
        X, _ = self._center_scale_X_Y(X=X)

        if n_components is None:
            Y_pred = X @ self.B
        else:
            Y_pred = X @ self.B[n_components - 1]

        _, Y_pred = self._un_center_scale_X_Y(Y=Y_pred)
        return Y_pred

    def transform(
        self,
        X: Optional[npt.ArrayLike] = None,
        Y: Optional[npt.ArrayLike] = None,
        n_components: Optional[int] = None,
    ) -> Union[npt.ArrayLike, Tuple[npt.ArrayLike, npt.ArrayLike], None]:
        """
        Transforms `X` and `Y` to their respective scores using `n_components` components.
        If `n_components` is None, then scores for all components up to `A` are returned.

        Parameters
        ----------
        X : Array of shape (N, K) or None, optional, default=None
            Predictor variables.

        Y : Array of shape (N, M) or None, optional, default=None
            Response variables.

        n_components : int or None, optional, default=None.
            Number of components in the PLS model. If None, then scores for all
            components up to `A` are returned.

        Returns
        -------
        T : Array of shape (N, n_components) or (N, A) or None
            X scores of `X`. If `n_components` is an int, then the scores up to
            `n_components` is used. If `n_components` is None, returns scores for all
            components up to `A`.

        U : Array of shape (N, n_components) or (N, A) or None
            Y scores of `Y`. If `n_components` is an int, then the scores up to
            `n_components` is used. If `n_components` is None, returns scores for all
            components up to `A`.

        Raises
        ------
        NotFittedError
            If the model has not been fitted before calling `transform()`.

        See Also
        --------
        fit_transform : Fits the model and returns the scores of `X` and `Y`.
        inverse_transform : Reconstructs `X` and `Y` from their respective scores.

        Notes
        -----
        If multiple calls to `transform` are made with Y provided and a previously seen
        value of `n_components`, then self.R_Y[n_components] is only computed once and
        stored for future use.

        Any centering and scaling of `X`and `Y` is carried out before computation of
        the scores and is undone afterwards.
        """
        if not self.fitted_:
            raise NotFittedError(
                "This model is not fitted yet, call 'fit' with approriate arguments before 'transform'."
            )
        if n_components is None:
            n_components = self.A
        X, Y, _ = self._get_input(copy=True, X=X, Y=Y)
        X, Y = self._center_scale_X_Y(X=X, Y=Y)
        if X is not None:
            T = X @ self.R[:, :n_components]
        if Y is not None:
            U = Y @ self.R_Y[n_components]
        if X is not None and Y is not None:
            return T, U
        if X is not None:
            return T
        if Y is not None:
            return U
        return None

    def fit_transform(
        self,
        X: npt.ArrayLike,
        Y: npt.ArrayLike,
        A: int,
        sample_weight: Optional[npt.ArrayLike] = None,
    ) -> Tuple[npt.NDArray[np.floating], npt.NDArray[np.floating]]:
        """
        Fits Improved Kernel PLS Algorithm #1 on `X` and `Y` using `A` components and
        returns T and U which are the scores of `X` and `Y`, respectively.

        Parameters
        ----------
        X : Array of shape (N, K)
            Predictor variables.

        Y : Array of shape (N, M) or (N,)
            Response variables.

        A : int
            Number of components in the PLS model.

        sample_weight : Array of shape (N,) or None, optional, default=None
            Weights for each observation. If None, then all observations are weighted
            equally.

        Returns
        -------
        T : Array of shape (N, A)
            PLS scores matrix of X.

        U : Array of shape (N, A)
            PLS scores matrix of Y.

        See Also
        --------
        transform : Transforms `X` and `Y` to their respective scores.
        inverse_transform : Reconstructs `X` and `Y` from their respective scores.
        """
        self.fit(X=X, Y=Y, A=A, sample_weight=sample_weight)
        if self.algorithm == 1 and sample_weight is None:
            return np.copy(self.T), self.transform(Y=Y)
        return self.transform(X=X, Y=Y)

    def inverse_transform(
        self,
        T: Optional[npt.ArrayLike] = None,
        U: Optional[npt.ArrayLike] = None,
    ) -> Union[npt.ArrayLike, Tuple[npt.ArrayLike, npt.ArrayLike], None]:
        """
        Reconstructs `X` and `Y` from their respective scores.

        Parameters
        ----------
        T : Array of shape (N, n_X_components) or (N, A) or None, optional, default=None
            Scores of predictor variables (X).
        U : Array of shape (N, n_Y_components) or (N, A) or None, optional, default=None
            Scores of response variables (Y).

        Returns
        -------
        X_reconstructed : Array of shape (N, K)
            If `T` is not None, returns the reconstructed `X`.
        Y_reconstructed : Array of shape (N, M)
            If `U` is not None, returns the reconstructed `Y`.

        Raises
        ------
        NotFittedError
            If the model has not been fitted before calling `inverse_transform()`.

        See Also
        --------
        transform : Transforms `X` and `Y` to their respective scores.
        fit_transform : Fits the model and returns the scores of `X` and `Y`.
        """
        if not self.fitted_:
            raise NotFittedError(
                "This model is not fitted yet, call 'fit' with approriate arguments before 'inverse_transform'."
            )
        X_reconstructed = None
        Y_reconstructed = None
        T = self._convert_input_to_array(arr=T)
        U = self._convert_input_to_array(arr=U)

        if T is not None:
            X_components = T.shape[1]
            X_reconstructed = T @ self.P[:, :X_components].T

        if U is not None:
            Y_components = U.shape[1]
            Y_reconstructed = U @ self.Q[:, :Y_components].T

        X_reconstructed, Y_reconstructed = self._un_center_scale_X_Y(
            X=X_reconstructed, Y=Y_reconstructed
        )

        if X_reconstructed is not None and Y_reconstructed is not None:
            return X_reconstructed, Y_reconstructed
        elif X_reconstructed is not None:
            return X_reconstructed
        elif Y_reconstructed is not None:
            return Y_reconstructed
        else:
            return None

    def cross_validate(
        self,
        X: npt.ArrayLike,
        Y: npt.ArrayLike,
        A: int,
        folds: Iterable[Hashable],
        metric_function: Union[
            Callable[
                [
                    npt.NDArray[np.floating],
                    npt.NDArray[np.floating],
                    npt.NDArray[np.floating],
                ],
                Any,
            ],
            Callable[
                [
                    npt.NDArray[np.floating],
                    npt.NDArray[np.floating],
                ],
                Any,
            ],
        ],
        preprocessing_function: Union[
            None,
            Union[
                Callable[
                    [
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                    ],
                    Tuple[
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                    ],
                ],
                Callable[
                    [
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                    ],
                    Tuple[
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                        npt.NDArray[np.floating],
                    ],
                ],
            ],
        ] = None,
        sample_weight: Optional[npt.ArrayLike] = None,
        n_jobs: int = -1,
        verbose: int = 10,
    ) -> dict[Hashable, Any]:
        """
        Performs cross-validation for the Partial Least-Squares (PLS) model on given
        data. `preprocessing_function` will be applied before any potential centering
        and scaling as determined by `self.center_X`, `self.center_Y`, `self.scale_X`,
        and `self.scale_Y`. Any such potential centering and scaling is applied for
        each split using training set statistics to avoid data leakage from the
        validation set. If `sample_weight` are provided, then these will be provided in
        `preprocessing_function` and `metric_function` and used for any centering and
        scaling in the cross-validation procedure.

        Parameters
        ----------
        X : Array of shape (N, K)
            Predictor variables.

        Y : Array of shape (N, M) or (N,)
            Response variables.

        A : int
            Number of components in the PLS model.

        folds : Iterable of Hashable with N elements
            An iterable defining cross-validation splits. Each unique value in
            `folds` corresponds to a different fold.

        metric_function : Callable receiving arrays `Y_val` (N_val, M), `Y_pred`\
        (A, N_val, M), and, if `sample_weight` is not None, also, `weights_val` (N_val,),\
        and returning Any.
            Computes a metric based on true values `Y_val` and predicted values
            `Y_pred`. `Y_pred` contains a prediction for all `A` components.

        preprocessing_function : Callable or None, optional, default=None,
            A function that preprocesses the training and validation data for each
            fold. It should return preprocessed arrays for `X_train`, `Y_train`,
            `X_val`, and `Y_val`. If Callable, it should receive arrays `X_train`,
            `Y_train`, `X_val`, `Y_val`, and, if `sample_weight` is not None, also
            `weights_train`, and `weights_val`, and returning a Tuple of preprocessed
            `X_train`, `Y_train`, `X_val`, and `Y_val`.

        sample_weight : Array of shape (N,) or None, optional, default=None
            Weights for each observation. If None, then all observations are weighted
            equally.

        n_jobs : int, optional default=-1
            Number of parallel jobs to use. A value of -1 will use the minimum of all
            available cores and the number of unique values in `folds`.

        verbose : int, optional default=10
            Controls verbosity of parallel jobs.

        Returns
        -------
        metrics : dict of Hashable to Any
            A dictionary mapping each unique value in `folds` to the result of
            evaluating `metric_function` on the validation set corresponding to that
            value.

        Raises
        ------
        ValueError
            If `sample_weight` are provided and not all weights are non-negative.

        Notes
        -----
        This method is used to perform cross-validation on the PLS model with different
        data splits and evaluate its performance using user-defined metrics.
        """
        X, Y, sample_weight = self._get_input(copy=False, X=X, Y=Y, sample_weight=sample_weight)

        if sample_weight is not None:
            self._check_nonnegative_weights(sample_weight=sample_weight)
            flattened_weights = sample_weight.squeeze()
        else:
            flattened_weights = None

        folds_dict = self._init_folds_dict(folds)
        num_splits = len(folds_dict)
        all_indices = np.arange(X.shape[0])

        if n_jobs == -1:
            n_jobs = min(num_splits, joblib.cpu_count())
        else:
            n_jobs = min(num_splits, n_jobs)

        def worker(val_indices: npt.NDArray[np.int_]) -> Any:
            train_indices = np.setdiff1d(all_indices, val_indices, assume_unique=True)
            X_train = X[train_indices]
            Y_train = Y[train_indices]
            X_val = X[val_indices]
            Y_val = Y[val_indices]
            if flattened_weights is not None:
                weights_train = flattened_weights[train_indices]
                weights_val = flattened_weights[val_indices]
            else:
                weights_train = None
                weights_val = None

            if preprocessing_function is not None:
                if flattened_weights is not None:
                    X_train, Y_train, X_val, Y_val = preprocessing_function(
                        X_train, Y_train, X_val, Y_val, weights_train, weights_val
                    )
                else:
                    X_train, Y_train, X_val, Y_val = preprocessing_function(
                        X_train, Y_train, X_val, Y_val
                    )

            self.fit(X_train, Y_train, A, weights_train)
            Y_pred = self.predict(X_val, n_components=None)
            if flattened_weights is not None:
                return metric_function(Y_val, Y_pred, weights_val)
            return metric_function(Y_val, Y_pred)

        metrics_list = Parallel(n_jobs=n_jobs, verbose=verbose)(
            delayed(worker)(val_indices) for val_indices in folds_dict.values()
        )
        metrics_dict = dict(zip(folds_dict, metrics_list))
        return metrics_dict

    def _init_folds_dict(
        self, folds: Iterable[Hashable]
    ) -> dict[Hashable, npt.NDArray[np.int_]]:
        """
        Generates a list of validation indices for each fold in `folds`.

        Parameters
        ----------
        folds : Iterable of Hashable with N elements
            An iterable defining cross-validation splits. Each unique value in
            `folds` corresponds to a different fold.

        Returns
        -------
        index_dict : dict of Hashable to Array
            A dictionary mapping each unique value in `folds` to an array of
            validation indices.
        """
        index_dict = {}
        for i, fold in enumerate(folds):
            try:
                index_dict[fold].append(i)
            except KeyError:
                index_dict[fold] = [i]
        for fold in index_dict:
            index_dict[fold] = np.asarray(index_dict[fold], dtype=int)
        return index_dict

sm00thix / ikpls / 28934299777

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous