Source code for etna.transforms.math.lags

from typing import Dict
from typing import List
from typing import Literal
from typing import Optional
from typing import Union

import pandas as pd

from etna.datasets import TSDataset
from etna.models.utils import determine_num_steps
from etna.transforms.base import IrreversibleTransform

[docs]class LagTransform(IrreversibleTransform): """Generates series of lags from given dataframe.""" def __init__(self, in_column: str, lags: Union[List[int], int], out_column: Optional[str] = None): """Create instance of LagTransform. Parameters ---------- in_column: name of processed column lags: int value or list of values for lags computation; if int, generate range of lags from 1 to given value out_column: base for the name of created columns; * if set the final name is '{out_column}_{lag_number}'; * if don't set, name will be ``transform.__repr__()``, repr will be made for transform that creates exactly this column Raises ------ ValueError: if lags value contains non-positive values """ super().__init__(required_features=[in_column]) if isinstance(lags, int): if lags < 1: raise ValueError(f"{type(self).__name__} works only with positive lags values, {lags} given") self.lags = list(range(1, lags + 1)) else: if any(lag_value < 1 for lag_value in lags): raise ValueError(f"{type(self).__name__} works only with positive lags values") self.lags = lags self.in_column = in_column self.out_column = out_column def _get_column_name(self, lag: int) -> str: if self.out_column is None: temp_transform = LagTransform(in_column=self.in_column, out_column=self.out_column, lags=[lag]) return repr(temp_transform) else: return f"{self.out_column}_{lag}" def _fit(self, df: pd.DataFrame) -> "LagTransform": """Fit method does nothing and is kept for compatibility. Parameters ---------- df: dataframe with data. Returns ------- result: LagTransform """ return self def _transform(self, df: pd.DataFrame) -> pd.DataFrame: """Add lags to the dataset. Parameters ---------- df: dataframe with data to transform. Returns ------- result: pd.Dataframe transformed dataframe """ result = df segments = sorted(set(df.columns.get_level_values("segment"))) all_transformed_features = [] features = df.loc[:, pd.IndexSlice[:, self.in_column]] for lag in self.lags: column_name = self._get_column_name(lag) transformed_features = features.shift(lag) transformed_features.columns = pd.MultiIndex.from_product([segments, [column_name]]) all_transformed_features.append(transformed_features) result = pd.concat([result] + all_transformed_features, axis=1) result = result.sort_index(axis=1) return result
[docs] def get_regressors_info(self) -> List[str]: """Return the list with regressors created by the transform.""" return [self._get_column_name(lag) for lag in self.lags]
[docs]class ExogShiftTransform(IrreversibleTransform): """Shifts exogenous variables from a given dataframe.""" def __init__(self, lag: Union[int, Literal["auto"]], horizon: Optional[int] = None): """Create instance of ExogShiftTransform. Parameters ---------- lag: value for shift estimation * if set to `int` all exogenous variables will be shifted `lag` steps forward; * if set to `auto` minimal shift will be estimated for each variable based on the prediction horizon and available timeline horizon: prediction horizon. Mandatory when set to `lag="auto"`, ignored otherwise """ super().__init__(required_features="all") self.lag: Optional[int] = None self.horizon: Optional[int] = None self._auto = False self._freq: Optional[str] = None self._created_regressors: Optional[List[str]] = None self._exog_shifts: Optional[Dict[str, int]] = None self._exog_last_date: Optional[Dict[str, pd.Timestamp]] = None self._filter_out_columns = {"target"} if isinstance(lag, int): if lag <= 0: raise ValueError(f"{self.__class__.__name__} works only with positive lags values, {lag} given") self.lag = lag else: if horizon is None: raise ValueError("Value of `horizon` should be specified when using `auto`!") if horizon < 1: raise ValueError(f"{self.__class__.__name__} works only with positive horizon values, {horizon} given") self.horizon = horizon self._auto = True def _save_exog_last_date(self, df_exog: Optional[pd.DataFrame] = None): """Save last available date of each exogenous variable.""" self._exog_last_date = {} if df_exog is not None: exog_names = set(df_exog.columns.get_level_values("feature")) for name in exog_names: feature = df_exog.loc[:, pd.IndexSlice[:, name]] na_mask = pd.isna(feature).any(axis=1) last_date = feature.index[~na_mask].max() self._exog_last_date[name] = last_date
[docs] def fit(self, ts: TSDataset) -> "ExogShiftTransform": """Fit the transform. Parameters ---------- ts: Dataset to fit the transform on. Returns ------- : The fitted transform instance. """ self._freq = ts.freq self._save_exog_last_date(df_exog=ts.df_exog) super().fit(ts=ts) return self
def _fit(self, df: pd.DataFrame) -> "ExogShiftTransform": """Estimate shifts for exogenous variables. Parameters ---------- df: dataframe with data. Returns ------- : Fitted `ExogShiftTransform` instance. """ feature_names = self._get_feature_names(df=df) self._exog_shifts = dict() self._created_regressors = [] for feature_name in feature_names: shift = self._estimate_shift(df=df, feature_name=feature_name) self._exog_shifts[feature_name] = shift if shift > 0: self._created_regressors.append(f"{feature_name}_shift_{shift}") return self def _get_feature_names(self, df: pd.DataFrame) -> List[str]: """Return the names of exogenous variables.""" feature_names = [] if self._exog_last_date is not None: feature_names = list(self._exog_last_date.keys()) df_columns = df.columns.get_level_values("feature") for name in feature_names: if name not in df_columns: raise ValueError(f"Feature `{name}` is expected to be in the dataframe!") return feature_names def _estimate_shift(self, df: pd.DataFrame, feature_name: str) -> int: """Estimate shift value for exogenous variable.""" if not self._auto: return self.lag # type: ignore if self._exog_last_date is None or self._freq is None: raise ValueError("Call `fit()` method before estimating exog shifts!") last_date = df.index.max() last_feature_date = self._exog_last_date[feature_name] if last_feature_date > last_date: delta = -determine_num_steps(start_timestamp=last_date, end_timestamp=last_feature_date, freq=self._freq) elif last_feature_date < last_date: delta = determine_num_steps(start_timestamp=last_feature_date, end_timestamp=last_date, freq=self._freq) else: delta = 0 shift = max(0, delta + self.horizon) # type: ignore return shift def _transform(self, df: pd.DataFrame) -> pd.DataFrame: """Shift exogenous variables. Parameters ---------- df: dataframe with data to transform. Returns ------- : Transformed dataframe. """ if self._exog_shifts is None: raise ValueError("Transform is not fitted!") result = df segments = sorted(set(df.columns.get_level_values("segment"))) feature_names = self._get_feature_names(df=df) shifted_features = [] features_to_remove = [] for feature_name in feature_names: shift = self._exog_shifts[feature_name] feature = df.loc[:, pd.IndexSlice[:, feature_name]] if shift > 0: shifted_feature = feature.shift(shift, freq=self._freq) column_name = f"{feature_name}_shift_{shift}" shifted_feature.columns = pd.MultiIndex.from_product([segments, [column_name]]) shifted_features.append(shifted_feature) features_to_remove.append(feature_name) if len(features_to_remove) > 0: result = result.drop(columns=pd.MultiIndex.from_product([segments, features_to_remove])) result = pd.concat([result] + shifted_features, axis=1) result.sort_index(axis=1, inplace=True) return result
[docs] def get_regressors_info(self) -> List[str]: """Return the list with regressors created by the transform.""" if self._created_regressors is None: raise ValueError("Fit the transform to get the regressors info!") return self._created_regressors