OrdinalDecomposition.py

"""OrdinalDecomposition ensemble."""

import numpy as np
from sklearn.base import BaseEstimator, ClassifierMixin, _fit_context
from sklearn.utils._param_validation import StrOptions
from sklearn.utils.validation import check_array, check_is_fitted, check_X_y

from orca_python.model_selection import load_classifier


class OrdinalDecomposition(BaseEstimator, ClassifierMixin):
    """OrdinalDecomposition ensemble classifier.

    This class implements an ensemble model where an ordinal problem is decomposed into
    several binary subproblems, each one of which will generate a different (binary)
    model, though all will share the same base classifier and parameters.

    There are 4 different ways to decompose the original problem based on how the
    coding matrix is built.

    Parameters
    ----------
    dtype : str
        Type of decomposition to be performed by classifier. May be one of 4 different
        types: 'ordered_partitions', 'one_vs_next', 'one_vs_followers' or
        'one_vs_previous'.

        The coding matrix generated by each method, for a problem with 5 classes will
        be as follows:

        ordered_partitions	one_vs_next	one_vs_followers	one_vs_previous

        -, -, -, -;		-,  ,  ,  ;		-,  ,  ,  ;		+, +, +, +;
        +, -, -, -;		+, -,  ,  ;		+, -,  ,  ;		+, +, +, -;
        +, +, -, -;		 , +, -,  ;		+, +, -,  ;		+, +, -,  ;
        +, +, +, -;		 ,  , +, -;		+, +, +, -;		+, -,  ,  ;
        +, +, +, +;		 ,  ,  , +;		+, +, +, +;		-,  ,  ,  ;

        where rows represent classes and columns represent base classifiers. Plus signs
        indicate that for that classifier, the label will be part of the positive
        class, on the other hand, a minus sign places that class into the negative one
        for that binary problem. If there is no sign, then those samples will not be
        used when building the model.

    decision_method : str
        Decision method that transforms the predictions of the n different base
        classifiers to produce the final label (one among the real ordinal classes).

    base_classifier : str
        Base classifier used to build a model for each binary subproblem. The base
        classifier need to be a classifier of orca-python framework or any classifier
        available in sklearn. Other classifiers implemented in sklearn's API can be
        used here.

    parameters : dict
        This dictionary will store the parameters used to build the base classifier.
        Only one value per parameter is allowed.

    Attributes
    ----------
    classes_ : list
        List that contains all different class labels found in the original dataset.

    coding_matrix_ : array-like, shape (n_targets, n_targets-1)
        Matrix that defines which classes will be used to build the model of each
        subproblem, and in which binary class they belong inside those new models.
        Further explained previously.

    classifiers_ : list of classifiers
        Initially empty, will include all fitted models for each subproblem once the fit
        function for this class is called successfully.

    X_ : array-like, shape (n_samples, n_features)
        Training patterns array, where n_samples is the number of samples and
        n_features is the number of features.

    y_ : array-like, shape (n_samples,)
        Target vector relative to X.

    References
    ----------
    .. [1] P.A. Gutierrez, M. Perez-Ortiz, J. Sanchez-Monedero, F. Fernandez-Navarro
           and C. Hervas-Martinez, "Ordinal regression methods: survey and
           experimental study", IEEE Transactions on Knowledge and Data Engineering,
           Vol. 28. Issue 1, 2016, https://doi.org/10.1109/TKDE.2015.2457911

    """

    _parameter_constraints: dict = {
        "dtype": [
            StrOptions(
                {
                    "ordered_partitions",
                    "one_vs_next",
                    "one_vs_followers",
                    "one_vs_previous",
                }
            )
        ],
        "decision_method": [
            StrOptions(
                {"exponential_loss", "hinge_loss", "logarithmic_loss", "frank_hall"}
            )
        ],
        "base_classifier": [str],
        "parameters": [dict],
    }

    def __init__(
        self,
        dtype="ordered_partitions",
        decision_method="frank_hall",
        base_classifier="LogisticRegression",
        parameters={},
    ):
        self.dtype = dtype
        self.decision_method = decision_method
        self.base_classifier = base_classifier
        self.parameters = parameters

    @_fit_context(prefer_skip_nested_validation=True)
    def fit(self, X, y):
        """Fit the model with the training data.

        Parameters
        ----------
        X : {array-like, sparse matrix} of shape (n_samples, n_features)
            Training patterns array, where n_samples is the number of samples and
            n_features is the number of features.

        y : array-like of shape (n_samples,)
            Target vector relative to X.

        Returns
        -------
        self : object
            Fitted estimator.

        Raises
        ------
        ValueError
            If parameters are invalid or data has wrong format.

        """
        X, y = check_X_y(X, y)

        self.X_ = X
        self.y_ = y

        # Get list of different labels of the dataset
        self.classes_ = np.unique(y)

        # Give each train input its corresponding output label
        # for each binary classifier
        self.coding_matrix_ = self._coding_matrix(
            self.dtype.lower(), len(self.classes_)
        )
        class_labels = self.coding_matrix_[(np.digitize(y, self.classes_) - 1), :]

        self.classifiers_ = []
        # Fitting n_targets - 1 classifiers
        for n in range(len(class_labels[0, :])):

            estimator = load_classifier(
                self.base_classifier, param_grid=self.parameters
            )
            estimator.fit(
                X[np.where(class_labels[:, n] != 0)],
                np.ravel(class_labels[np.where(class_labels[:, n] != 0), n].T),
            )

            self.classifiers_.append(estimator)

        return self

    def predict(self, X):
        """Perform classification on samples in X.

        Parameters
        ----------
        X : {array-like, sparse matrix} of shape (n_samples, n_features)
            Test patterns array, where n_samples is the number of samples and
            n_features is the number of features.

        Returns
        -------
        y_pred : array, shape (n_samples,)
            Class labels for samples in X.

        Raises
        ------
        NotFittedError
            If the model is not fitted yet.

        ValueError
            If input is invalid.

        AttributeError
            If the specified loss method is not implemented.

        """
        check_is_fitted(self, ["X_", "y_"])
        X = check_array(X)

        # Getting predicted labels for dataset from each classifier
        predictions = self._get_predictions(X)

        decision_method = self.decision_method.lower()
        if decision_method == "exponential_loss":

            # Scaling predictions from [0,1] range to [-1,1]
            predictions = predictions * 2 - 1

            # Transforming from binary problems to the original problem
            losses = self._exponential_loss(predictions)
            y_pred = self.classes_[np.argmin(losses, axis=1)]

        elif decision_method == "hinge_loss":

            # Scaling predictions from [0,1] range to [-1,1]
            predictions = predictions * 2 - 1

            # Transforming from binary problems to the original problem
            losses = self._hinge_loss(predictions)
            y_pred = self.classes_[np.argmin(losses, axis=1)]

        elif decision_method == "logarithmic_loss":

            # Scaling predictions from [0,1] range to [-1,1]
            predictions = predictions * 2 - 1

            # Transforming from binary problems to the original problem
            losses = self._logarithmic_loss(predictions)
            y_pred = self.classes_[np.argmin(losses, axis=1)]

        elif decision_method == "frank_hall":

            # Transforming from binary problems to the original problem
            y_proba = self._frank_hall_method(predictions)
            y_pred = self.classes_[np.argmax(y_proba, axis=1)]

        else:
            raise AttributeError(
                'The specified loss method "%s" is not implemented' % decision_method
            )

        return y_pred

    def predict_proba(self, X):
        """Return the probability of the sample for each class in the model.

        Parameters
        ----------
        X : {array-like, sparse matrix} of shape (n_samples, n_features)
            Test patterns array, where n_samples is the number of samples and
            n_features is the number of features.

        Returns
        -------
        y_proba : ndarray of shape (n_samples,)
            The probability of the sample for each class in the model, where classes are
            ordered as they are in self.classes_.

        Raises
        ------
        NotFittedError
            If the model is not fitted yet.

        ValueError
            If input is invalid.

        AttributeError
            If the specified loss method is not implemented.

        """
        check_is_fitted(self, ["X_", "y_"])
        X = check_array(X)

        # Getting predicted labels for dataset from each classifier
        predictions = self._get_predictions(X)

        decision_method = self.decision_method.lower()
        if decision_method == "exponential_loss":

            # Scaling predictions from [0,1] range to [-1,1]
            predictions = predictions * 2 - 1

            # Transforming from binary problems to the original problem
            losses = self._exponential_loss(predictions)
            losses = 1 / losses.astype(float)
            y_proba = []
            for losse in losses:
                y_proba.append((np.exp(losse) / np.sum(np.exp(losse))))
            y_proba = np.array(y_proba)

        elif decision_method == "hinge_loss":

            # Scaling predictions from [0,1] range to [-1,1]
            predictions = predictions * 2 - 1

            # Transforming from binary problems to the original problem
            losses = self._hinge_loss(predictions)
            losses = 1 / losses.astype(float)
            y_proba = []
            for losse in losses:
                y_proba.append((np.exp(losse) / np.sum(np.exp(losse))))
            y_proba = np.array(y_proba)

        elif decision_method == "logarithmic_loss":

            # Scaling predictions from [0,1] range to [-1,1]
            predictions = predictions * 2 - 1

            # Transforming from binary problems to the original problem
            losses = self._logarithmic_loss(predictions)
            losses = 1 / losses.astype(float)
            y_proba = []
            for losse in losses:
                y_proba.append((np.exp(losse) / np.sum(np.exp(losse))))
            y_proba = np.array(y_proba)

        elif decision_method == "frank_hall":

            # Transforming from binary problems to the original problem
            y_proba = self._frank_hall_method(predictions)

        else:
            raise AttributeError(
                'The specified loss method "%s" is not implemented' % decision_method
            )

        return y_proba

    def _coding_matrix(self, dtype, n_classes):
        """Return the coding matrix for a given dataset.

        Parameters
        ----------
        dtype : str
            Type of decomposition to be performed by classifier.

        n_classes : int
            Number of different classes in actual dataset.

        Returns
        -------
        coding_matrix: array-like, shape (n_targets, n_targets-1)
            Each value must be in range {-1, 1, 0}, whether that class will belong to
            negative class, positive class or will not be used for that particular
            binary classifier.

        Raises
        ------
        ValueError
            If the decomposition type does not exist.

        """
        if dtype == "ordered_partitions":

            coding_matrix = np.triu((-2 * np.ones(n_classes - 1))) + 1
            coding_matrix = np.vstack([coding_matrix, np.ones((1, n_classes - 1))])

        elif dtype == "one_vs_next":

            plus_ones = np.diagflat(np.ones((1, n_classes - 1), dtype=int), -1)
            minus_ones = -(np.eye(n_classes, n_classes - 1, dtype=int))
            coding_matrix = minus_ones + plus_ones[:, :-1]

        elif dtype == "one_vs_followers":

            minus_ones = np.diagflat(-np.ones((1, n_classes), dtype=int))
            plus_ones = np.tril(np.ones(n_classes), -1)
            coding_matrix = (plus_ones + minus_ones)[:, :-1]

        elif dtype == "one_vs_previous":

            plusones = np.triu(np.ones(n_classes))
            minusones = -np.diagflat(np.ones((1, n_classes - 1)), -1)
            coding_matrix = np.flip((plusones + minusones)[:, :-1], axis=1)

        else:

            raise ValueError("Decomposition type %s does not exist" % dtype)

        return coding_matrix.astype(int)

    def _get_predictions(self, X):
        """Return the probability of positive class membership.

        For each pattern inside the dataset X, this method returns the probability for
        that pattern to belong to the positive class. There will be as many predictions
        (columns) as different binary classifiers have been fitted previously.

        Parameters
        ----------
        X : {array-like, sparse matrix} of shape (n_samples, n_features)
            Test patterns array, where n_samples is the number of samples and
            n_features is the number of features.

        Returns
        -------
        predictions : array, shape (n_samples, n_targets-1)
            Probability estimates or binary classification outcomes.

        """
        predictions = np.array(
            list(map(lambda c: c.predict_proba(X)[:, 1], self.classifiers_))
        ).T

        return predictions

    def _exponential_loss(self, predictions):
        """Compute the exponential losses for each label.

        Computation of the exponential losses for each label of the original ordinal
        multinomial problem. Transforms from n-1 binary subproblems to the original
        ordinal problem with n targets.

        Parameters
        ----------
        predictions : array, shape (n_samples, n_targets-1)
            Probability estimates or binary classification outcomes.

        Returns
        -------
        e_losses : array, shape (n_samples, n_targets)
            Exponential losses for each sample of dataset X. One different value for
            each class label.

        """
        # Computing exponential losses
        e_losses = np.zeros((predictions.shape[0], (predictions.shape[1] + 1)))
        for i in range(predictions.shape[1] + 1):

            e_losses[:, i] = np.sum(
                np.exp(
                    -predictions
                    * np.tile(self.coding_matrix_[i, :], (predictions.shape[0], 1))
                ),
                axis=1,
            )

        return e_losses

    def _hinge_loss(self, predictions):
        """Compute the Hinge losses for each label.

        Computation of the Hinge losses for each label of the original ordinal
        multinomial problem. Transforms from n-1 binary subproblems to the original
        ordinal problem with n targets.

        Parameters
        ----------
        predictions : array, shape (n_samples, n_targets-1)
            Probability estimates or binary classification outcomes.

        Returns
        -------
        h_losses : array, shape (n_samples, n_targets)
            Hinge losses for each sample of dataset X. One different value for each
            class label.

        """
        # Computing Hinge losses
        h_losses = np.zeros((predictions.shape[0], (predictions.shape[1] + 1)))
        for i in range(predictions.shape[1] + 1):

            h_losses[:, i] = np.sum(
                np.maximum(
                    0,
                    (
                        1
                        - np.tile(self.coding_matrix_[i, :], (predictions.shape[0], 1))
                        * predictions
                    ),
                ),
                axis=1,
            )

        return h_losses

    def _logarithmic_loss(self, predictions):
        """Compute the logarithmic losses for each label.

        Computation of the logarithmic losses for each label of the original ordinal
        multinomial problem. Transforms from n-1 binary subproblems to the original
        ordinal problem with n targets.

        Parameters
        ----------
        predictions : array, shape (n_samples, n_targets-1)
            Probability estimates or binary classification outcomes.

        Returns
        -------
        l_losses : array, shape (n_samples, n_targets)
            Logarithmic losses for each sample of dataset X. One different value for
            each class label.

        """
        # Computing logarithmic losses
        l_losses = np.zeros((predictions.shape[0], (predictions.shape[1] + 1)))
        for i in range(predictions.shape[1] + 1):

            l_losses[:, i] = np.sum(
                np.log(
                    1
                    + np.exp(
                        -2
                        * np.tile(self.coding_matrix_[i, :], (predictions.shape[0], 1))
                        * predictions
                    )
                ),
                axis=1,
            )

        return l_losses

    def _frank_hall_method(self, predictions):
        """Calculate probability of each pattern belonging to each target.

        Returns the probability for each pattern of dataset to belong to each one of
        the original targets. Transforms from n-1 subproblems to the original ordinal
        problem with n targets.

        Parameters
        ----------
        predictions : array, shape (n_samples, n_targets-1)
            Probability estimates or binary classification outcomes.

        Returns
        -------
        y_proba : array, shape (n_samples, n_targets)
            Class labels predicted for samples in dataset X.

        Raises
        ------
        AttributeError
            If the decomposition type is not ordered_partitions.

        """
        if self.dtype.lower() != "ordered_partitions":
            raise AttributeError(
                "When using Frank and Hall decision method,\
								ordered_partitions must be used"
            )

        y_proba = np.empty([(predictions.shape[0]), (predictions.shape[1] + 1)])

        # Probabilities of each set to belong to the first ordinal class
        y_proba[:, 0] = 1 - predictions[:, 0]

        # Probabilities for the central classes
        y_proba[:, 1:-1] = predictions[:, :-1] - predictions[:, 1:]

        # Probabilities of each set to belong to the last class
        y_proba[:, -1] = predictions[:, -1]

        return y_proba