Source code for evalml.pipelines.time_series_regression_pipeline

import pandas as pd

from evalml.pipelines.pipeline_meta import TimeSeriesPipelineBaseMeta
from evalml.pipelines.regression_pipeline import RegressionPipeline
from evalml.problem_types import ProblemTypes
from evalml.utils import (
    drop_rows_with_nans,
    infer_feature_types,
    pad_with_nans,
)


[docs]class TimeSeriesRegressionPipeline( RegressionPipeline, metaclass=TimeSeriesPipelineBaseMeta ): """Pipeline base class for time series regression problems. Arguments: component_graph (list or dict): List of components in order. Accepts strings or ComponentBase subclasses in the list. Note that when duplicate components are specified in a list, the duplicate component names will be modified with the component's index in the list. For example, the component graph [Imputer, One Hot Encoder, Imputer, Logistic Regression Classifier] will have names ["Imputer", "One Hot Encoder", "Imputer_2", "Logistic Regression Classifier"] parameters (dict): Dictionary with component names as keys and dictionary of that component's parameters as values. An empty dictionary {} implies using all default values for component parameters. Pipeline-level parameters such as date_index, gap, and max_delay must be specified with the "pipeline" key. For example: Pipeline(parameters={"pipeline": {"date_index": "Date", "max_delay": 4, "gap": 2}}). random_seed (int): Seed for the random number generator. Defaults to 0. """ problem_type = ProblemTypes.TIME_SERIES_REGRESSION """ProblemTypes.TIME_SERIES_REGRESSIO""" def __init__( self, component_graph, parameters=None, custom_name=None, random_seed=0, ): if not parameters or "pipeline" not in parameters: raise ValueError( "date_index, gap, and max_delay parameters cannot be omitted from the parameters dict. " "Please specify them as a dictionary with the key 'pipeline'." ) pipeline_params = parameters["pipeline"] self.date_index = pipeline_params["date_index"] self.gap = pipeline_params["gap"] self.max_delay = pipeline_params["max_delay"] super().__init__( component_graph, custom_name=custom_name, parameters=parameters, random_seed=random_seed, )
[docs] def fit(self, X, y): """Fit a time series regression pipeline. Arguments: X (pd.DataFrame or np.ndarray): The input training data of shape [n_samples, n_features] y (pd.Series, np.ndarray): The target training targets of length [n_samples] Returns: self """ if X is None: X = pd.DataFrame() X = infer_feature_types(X) y = infer_feature_types(y) self.input_target_name = y.name X_t = self.component_graph.fit_features(X, y) y_shifted = y.shift(-self.gap) X_t, y_shifted = drop_rows_with_nans(X_t, y_shifted) self.estimator.fit(X_t, y_shifted) self.input_feature_names = self.component_graph.input_feature_names return self
[docs] def predict(self, X, y=None, objective=None): """Make predictions using selected features. Arguments: X (pd.DataFrame, or np.ndarray): Data of shape [n_samples, n_features] y (pd.Series, np.ndarray, None): The target training targets of length [n_samples] objective (Object or string): The objective to use to make predictions Returns: pd.Series: Predicted values. """ if X is None: X = pd.DataFrame() X = infer_feature_types(X) y = infer_feature_types(y) features = self.compute_estimator_features(X, y) features_no_nan, y = drop_rows_with_nans(features, y) y_arg = None if self.estimator.predict_uses_y: y_arg = y predictions = self.estimator.predict(features_no_nan, y_arg) predictions.index = y.index predictions = self.inverse_transform(predictions) predictions = predictions.rename(self.input_target_name) padded = pad_with_nans( predictions, max(0, features.shape[0] - predictions.shape[0]) ) return infer_feature_types(padded)
[docs] def score(self, X, y, objectives): """Evaluate model performance on current and additional objectives. Arguments: X (pd.DataFrame or np.ndarray): Data of shape [n_samples, n_features] y (pd.Series): True labels of length [n_samples] objectives (list): Non-empty list of objectives to score on Returns: dict: Ordered dictionary of objective scores """ # Only converting X for the call to _score_all_objectives if X is None: X = pd.DataFrame() X = infer_feature_types(X) y = infer_feature_types(y) y_predicted = self.predict(X, y) y_shifted = y.shift(-self.gap) objectives = self.create_objectives(objectives) y_shifted, y_predicted = drop_rows_with_nans(y_shifted, y_predicted) return self._score_all_objectives( X, y_shifted, y_predicted, y_pred_proba=None, objectives=objectives )