[ENH] forecasting horizon conditional parameter wrapper

[fkiraly]

A common situation in forecasting multiple forecasting horizons is that one wants to do something different per horizon element.

For instance, if fh=[1, 2, 3], I may like to lag my exogeneous data by 1, when forecasting 1 step ahead, lag my exogeneous data by 2, when forecasting 2 ahead, and lag my exogeneous data by 3, when forecasting 3 ahead. Or, I might like to lag only some exogeneous variables, this way, so do a FeatureUnion with ColumnSelect with Lag

This is done internally, in the "reduction" transformers, but that is not very flexible.
We should have generic functionality for this, it could look like as follows:

class FhConditionalForecast(BaseForecaster):
    """Uses different parameters by forecasting horizon element.

    When provided with forecasting horizon [f1, f2, ..., fn],
    will fit forecaster with fh=f1 and parameters fh_params[f1] to forecast f1,
    forecaster with fh=f2 and parameters fh_params[f1] to forecast f2, etc.

    Parameters
    ----------
    forecaster : sktime compatible forecaster
    fh_params : dict, with keys = fh elements, values = param dict for forecaster
    """

Example usage:

forecaster = ForecastingPipeline([Lag(), ARIMA()])
fc_by_lag = FhConditionalForecast(
    forecaster,
    fh_params = {
        1: {"Lag__lags": 1}
        2: {"Lag__lags": 2}
    }
)

fc_by_lag.fit(y, X, fh=[1, 2])
# internally, calls:
# forecaster.set_params(Lag__lags=1).fit(y, X, fh=1)
# forecaster.set_params(Lag__lags=2).fit(y, X, fh=2)

#then, we can do this, do not need X_pred!
fc_by_lag.predict()

[aiwalter]

proposal to do this with a transformer:

class XpredTransformer(BaseTransformer):
    # fh is int for simplification here, but we can do this probably with fh.to_indexer() method
    #  there can be also a param "column: Union[int, str, List] to select columns."
    def __init__(self,transformer, fh: int=3):
        self.transformer = transformer
        self.fh = fh

    def _fit(self, X, y=None):
        X_lagged = Lag().fit_transform(X=X, y=y)
        self._X_fit = X_lagged.iloc[:-self.fh]
        self._X_pred = X_lagged.iloc[:self.fh]
        self.transformer_ = self.transformer.clone().fit(X=self.X_fit)
        return self

    def _transform(self, X, y=None):
        # we have to assert here that if X is not None, X has same index as self._X_pred
        if X is not None:
            assert X.index == self._X_pred.index # or use index.equals()?
        # If X is not None, Xt will be added to the columns
        Xt = self.transformer_.transform(X=self._X_pred)
        return Xt

[fkiraly]

is the XpredTransformer(my_trafo, lag) not the same as Lag(fh) * my_trafo?

[fkiraly]

@aiwalter said: it's not the same, because XpredTransformer remembers the lagged X and produces it.

[fkiraly]

So, I think the transformer is doing sth slightly different - I think both estimators would be useful to have.

The XpredTransformer would probably solve the problem I had here!
#2681