ETSC models

.github/workflows/poetry_unit_test.yml

@@ -39,7 +39,7 @@ jobs:
         run: poetry install
 
       - name: Bump up FEDOT to a stable revision (temporary)
-        run: poetry add git+https://github.com/aimclub/FEDOT.git@e0b4ee7
+        run: poetry add git+https://github.com/aimclub/FEDOT.git@master
 
       - name: Run tests with pytest
         run: poetry run pytest --cov=fedot_ind --cov-report xml:coverage.xml tests/unit

fedot_ind/core/architecture/abstraction/decorators.py

@@ -11,9 +11,10 @@
 
 def fedot_data_type(func):
     def decorated_func(self, *args):
-        if not isinstance(args[0], InputData):
-            args[0] = DataConverter(data=args[0])
-        features = args[0].features
+        data, *rest_args = args
+        if not isinstance(data, InputData):
+            data = DataConverter(data=data)
+        features = data.features
 
         if len(features.shape) < 4:
             try:
@@ -22,7 +23,7 @@ def decorated_func(self, *args):
                 input_data_squeezed = np.squeeze(features)
         else:
             input_data_squeezed = features
-        return func(self, input_data_squeezed, args[1])
+        return func(self, input_data_squeezed, *rest_args)
 
     return decorated_func
 
@@ -42,13 +43,14 @@ def decorated_func(self, *args):
 
 def convert_to_3d_torch_array(func):
     def decorated_func(self, *args):
-        init_data = args[0]
+        init_data, *args = args
         data = DataConverter(data=init_data).convert_to_torch_format()
         if isinstance(init_data, InputData):
             init_data.features = data
         else:
             init_data = data
-        return func(self, init_data, *args[1:])
+        return func(self, init_data, *args)
+
     return decorated_func
 
 

fedot_ind/core/metrics/metrics_implementation.py

@@ -1,6 +1,7 @@
 from typing import Optional
 from typing import Union
 
+import matplotlib.pyplot as plt
 import numpy as np
 import pandas as pd
 from fedot.core.data.data import InputData
@@ -221,6 +222,10 @@ def smape(a, f, _=None):
                                (np.abs(a) + np.abs(f)) * 100)
 
 
+def rmse(y_true, y_pred):
+    return mean_squared_error(y_true, y_pred, squared=False)
+
+
 def mape(A, F):
     return mean_absolute_percentage_error(A, F)
 
@@ -232,9 +237,6 @@ def calculate_regression_metric(target,
                                 **kwargs):
     target = target.astype(float)
 
-    def rmse(y_true, y_pred):
-        return np.sqrt(mean_squared_error(y_true, y_pred))
-
     metric_dict = {'r2': r2_score,
                    'mse': mean_squared_error,
                    'rmse': rmse,
@@ -261,9 +263,6 @@ def calculate_forecasting_metric(target,
                                  **kwargs):
     target = target.astype(float)
 
-    def rmse(y_true, y_pred):
-        return np.sqrt(mean_squared_error(y_true, y_pred))
-
     metric_dict = {
         'rmse': rmse,
         'mae': mean_absolute_error,
@@ -347,8 +346,102 @@ def kl_divergence(solution: pd.DataFrame,
         return np.average(solution.mean())
 
 
-class AnomalyMetric(QualityMetric):
+class ETSCPareto(QualityMetric, ParetoMetrics):
+    def __init__(self,
+                 target,
+                 predicted_labels,
+                 predicted_probs=None,
+                 weigths: tuple = None,
+                 mode: str = 'robust',
+                 reduce: bool = True,
+                 metric_list: tuple = (
+                     'f1', 'roc_auc', 'accuracy', 'logloss', 'precision'),
+                 default_value: float = 0.0):
+        self.target = target.flatten()
+        self.predicted_labels = predicted_labels
+        self.predicted_probs = predicted_probs
+        self.metric_list = metric_list
+        self.default_value = default_value
+        self.weights = weigths
+        self.mode = mode
+        self.columns = ['robustness'] if self.mode == 'robust' else []
+        self.columns.extend(metric_list)
+        self.reduce = reduce
+
+    def metric(self) -> float:
+        if len(self.predicted_labels.shape) == 1:
+            self.predicted_labels = self.predicted_labels[None, ...]
+            self.predicted_probs = self.predicted_probs[None, ...]
+        print(f'''
+        target shape {self.target.shape}
+        prediction {self.predicted_labels.shape}
+        predicted_probs (scores) {self.predicted_probs.shape}
+        ''')
+        n_metrics = len(self.metric_list) + (self.mode == 'robust')
+        n_est = self.predicted_labels.shape[0]
+        result = np.zeros((n_est, n_metrics))
+        print(result.shape)
+        if self.mode == 'robust':
+            mask = self.predicted_probs >= 0
+            print('mask', mask.shape)
+            if not mask.any():
+                return result
+            robustness = mask.sum(-1) / self.predicted_probs.shape[-1]
+            print('rob', robustness.shape)
+            result[:, 0] = robustness.flatten()
+        else:
+            mask = np.ones_like(self.predicted_probs, dtype=bool)
+
+        for est in range(n_est):
+            for i, metric in enumerate(self.metric_list, 1):
+                assert metric in CLASSIFICATION_METRIC_DICT, f'{metric} is not found in available metrics'
+                metric_value = CLASSIFICATION_METRIC_DICT[metric](self.target[mask[est]],
+                                                                  self.predicted_labels[est][mask[est]])
+                result[est, i] = metric_value
+
+        if self.weights is None:
+            if self.reduce:
+                self.weights = np.empty(n_metrics)
+                self.weights.fill(1 / len(self.weights))
+            else:
+                self.weights = np.eye(n_metrics)
+        else:
+            assert self.weights.shape[-1] == self.metrics.shape[-1], 'Metrics and weights size mismatch!'
+            self.weights /= self.weights.sum()
 
+        result = result @ self.weights.T
+        result[np.isnan(result)] = self.default_value
+        if not self.reduce:
+            return pd.DataFrame(result, columns=self.columns)
+        else:
+            return result
+
+    def plot_bicrit_metric(self, metrics, select=None, metrics_names=None):
+        if not metrics_names:
+            metrics_names = ('Robustness', 'Accuracy')
+        plt.figure(figsize=(10, 10))
+        assert metrics.shape[-1] == 2, 'only 2 metrics can be plotted'
+        for i, metric in enumerate(metrics):
+            selection = metric[select]
+            sizes = ((np.arange(selection.shape[0]) * 2)[::-1]) ** 1.5 + 10
+            plt.scatter(*(metric[select]).T,
+                        s=sizes,
+                        label=i)
+        plt.legend(loc="upper right", bbox_to_anchor=(1.5, 1))
+        plt.ylabel(metrics_names[1])
+        plt.xlabel(metrics_names[0])
+        plt.xlim((-0.05, 1.05))
+        plt.ylim((-0.05, 1.05))
+        plt.xticks(np.linspace(0, 1, 11))
+        plt.yticks(np.linspace(0, 1, 11))
+        plt.grid(True)
+
+    def select_pareto_front(self, metrics, maximize=True):
+        pareto_mask = self.pareto_metric_list(metrics, maximise=maximize)
+        return metrics[pareto_mask]
+
+
+class AnomalyMetric(QualityMetric):
     def __init__(self,
                  target,
                  predicted_labels,
@@ -617,3 +710,29 @@ def calculate_detection_metric(
         target=target,
         predicted_labels=labels).metric()
     return metric_dict
+
+
+REGRESSION_METRIC_DICT = {'r2': r2_score,
+                          'mse': mean_squared_error,
+                          'rmse': rmse,
+                          'mae': mean_absolute_error,
+                          'msle': mean_squared_log_error,
+                          'mape': mean_absolute_percentage_error,
+                          'median_absolute_error': median_absolute_error,
+                          'explained_variance_score': explained_variance_score,
+                          'max_error': max_error,
+                          'd2_absolute_error_score': d2_absolute_error_score}
+
+CLASSIFICATION_METRIC_DICT = {'accuracy': accuracy_score,
+                              'f1': f1_score,
+                              'roc_auc': roc_auc_score,
+                              'precision': precision_score,
+                              'logloss': log_loss}
+
+FORECASTING_METRICS_DICT = {
+    'rmse': rmse,
+    'mae': mean_absolute_error,
+    'median_absolute_error': median_absolute_error,
+    'smape': smape,
+    'mase': mase
+}

fedot_ind/core/models/early_tc/base_early_tc.py

@@ -0,0 +1,165 @@
+from typing import Optional, List
+from fedot.core.operations.operation_parameters import OperationParameters
+from sklearn.preprocessing import StandardScaler
+from sklearn.base import ClassifierMixin, BaseEstimator
+from sktime.classification.dictionary_based import WEASEL
+from fedot_ind.core.architecture.settings.computational import backend_methods as np
+
+
+class EarlyTSClassifier(ClassifierMixin, BaseEstimator):
+    """
+    Base class for Early Time Series Classification models
+    which implement prefix-wise predictions via traiing multiple slave estimators.
+
+    Args:
+        ``interval_percentage (float in (1, 100])``: define how much points should be between prediction points.
+        ``consecutive_predictions (int)``: how many last subsequent estimators should classify object equally.
+        ``accuracy_importance (float in [0, 1])``: trade-off coefficient between earliness and accuracy.
+        ``prediction_mode (str in ['last_available', 'best_by_metrics_mean', 'all'])``:
+            - if 'last_available', returns the latest estimator prediction allowed by prefix length;
+            - if 'best_by_metrics_mean', returns the best of estimators estimated
+              with weighted average of accuracy and earliness
+            - if 'all', returns all estiamtors predictions
+        ``transform_score (bool)``: whether or not to scale scores to [-1, 1] interval
+        ``min_ts_step (int)``: minimal difference between to subsequent prefix' lengths
+    """
+
+    def __init__(self, params: Optional[OperationParameters] = {}):
+        super().__init__()
+        self.interval_percentage = params.get('interval_percentage', 10)
+        self.consecutive_predictions = params.get('consecutive_predictions', 1)
+        self.accuracy_importance = params.get('accuracy_importance', 1.)
+        self.min_ts_length = params.get('min_ts_step', 3)
+        self.random_state = params.get('random_state', None)
+
+        self.prediction_mode = params.get('prediction_mode', 'last_available')
+        self.transform_score = params.get('transform_score', True)
+        self.weasel_params = {}
+
+    def _init_model(self, X, y):
+        max_data_length = X.shape[-1]
+        self.prediction_idx = self._compute_prediction_points(max_data_length)
+        self.n_pred = len(self.prediction_idx)
+        self.slave_estimators = [
+            WEASEL(random_state=self.random_state, support_probabilities=True, **self.weasel_params)
+            for _ in range(self.n_pred)]
+        self.scalers = [StandardScaler() for _ in range(self.n_pred)]
+        self._chosen_estimator_idx = -1
+        self.classes_ = [np.unique(y)]
+        self._estimator_for_predict = [-1]
+
+    @property
+    def required_length(self):
+        if not hasattr(self, '_chosen_estimator_idx'):
+            return None
+        return self.prediction_idx[self._chosen_estimator_idx]
+
+    @property
+    def n_classes(self):
+        return len(self.classes_[0])
+
+    def fit(self, X, y=None):
+        assert y is not None, 'Pass y'
+        y = np.array(y).flatten()
+        self._init_model(X, y)
+        for i in range(self.n_pred):
+            self._fit_one_interval(X, y, i)
+
+    def _fit_one_interval(self, X, y, i):
+        X_part = X[..., :self.prediction_idx[i] + 1]
+        X_part = self.scalers[i].fit_transform(X_part)
+        probas = self.slave_estimators[i].fit_predict_proba(X_part, y)
+        return probas
+
+    def _predict_one_slave(self, X, i, offset=0):
+        X_part = X[..., max(0, offset - 1):self.prediction_idx[i] + 1]
+        X_part = self.scalers[i].transform(X_part)
+        probas = self.slave_estimators[i].predict_proba(X_part)
+        return probas, np.argmax(probas, axis=-1)
+
+    def _compute_prediction_points(self, n_idx):
+        interval_length = max(int(n_idx * self.interval_percentage / 100), self.min_ts_length)
+        prediction_idx = np.arange(n_idx - 1, -1, -interval_length)[::-1][1:]
+        self.earliness = 1 - prediction_idx / n_idx  # /n_idx because else the last hm score is always 0
+        return prediction_idx
+
+    def _select_estimators(self, X, training=False):
+        offset = 0
+        if not training and self.prediction_mode == 'best_by_metrics_mean':
+            estimator_indices = [self._chosen_estimator_idx]
+        elif not training and self.prediction_mode == 'last_available':
+            last_idx, offset = self._get_applicable_index(X.shape[-1] - 1)
+            estimator_indices = [last_idx]
+        elif training or self.prediction_mode == 'all':
+            last_idx, offset = self._get_applicable_index(X.shape[-1] - 1)
+            estimator_indices = np.arange(last_idx + 1)
+        else:
+            raise ValueError('Unknown prediction mode')
+        return estimator_indices, offset
+
+    def _predict(self, X, training=True):
+        estimator_indices, offset = self._select_estimators(X, training)
+        if not training:
+            self._estimator_for_predict = estimator_indices
+        prediction = (np.stack(array_list) for array_list in zip(
+            *[self._predict_one_slave(X, i, offset) for i in estimator_indices]  # check boundary
+        ))
+        return prediction  # see the output in _predict_one_slave
+
+    def _consecutive_count(self, predicted_labels: List[np.array]):
+        n = len(predicted_labels[0])
+        prediction_points = len(predicted_labels)
+        consecutive_labels = np.ones((prediction_points, n))
+        for i in range(1, prediction_points):
+            equal = predicted_labels[i - 1] == predicted_labels[i]
+            consecutive_labels[i, equal] = consecutive_labels[i - 1, equal] + 1
+        return consecutive_labels  # prediction_points x n_instances
+
+    def predict_proba(self, *args):
+        """
+        Args:
+            X (np.array): input features
+        Returns:
+            predictions as a numpy array of shape (2, n_selected_estimators, n_instances, n_classes)
+            where first subarray stands for probas, and second for scores
+        """
+        predicted_probas, scores, *_ = args
+        if self.transform_score:
+            scores = self._transform_score(scores)
+        scores = np.tile(scores[..., None], (1, 1, self.n_classes))
+        prediction = np.stack([predicted_probas, scores], axis=0)
+        if prediction.shape[1] == 1:
+            prediction = prediction.squeeze(1)
+        return prediction
+
+    def predict(self, X):
+        """
+        Args:
+            X (np.array): input features
+        Returns:
+            predictions as a numpy array of shape (2, n_selected_estimators, n_instances)
+            where first subarray stands for labels, and second for scores
+        """
+        prediction = self.predict_proba(X)
+        labels = prediction[0:1].argmax(-1)
+        scores = prediction[1:2, ..., 0]
+        prediction = np.stack([labels, scores], 0)
+        if prediction.shape[1] == 1:
+            prediction = prediction.squeeze(1)
+        return prediction
+
+    def _score(self, X, y, accuracy_importance=None, training=True):
+        y = np.array(y).flatten()
+        accuracy_importance = accuracy_importance or self.accuracy_importance
+        predictions = self._predict(X, training)[0]
+        prediction_points = predictions.shape[0]
+        accuracies = (predictions == np.tile(y, (prediction_points, 1))).sum(axis=1) / len(y)
+        return (1 - accuracy_importance) * self.earliness[:prediction_points] + accuracy_importance * accuracies
+
+    def _get_applicable_index(self, last_available_idx):
+        idx = np.searchsorted(self.prediction_idx, last_available_idx, side='right')
+        if idx == 0:
+            raise RuntimeError('Too few points for prediction!')
+        idx -= 1
+        offset = last_available_idx - self.prediction_idx[idx]
+        return idx, offset