How to specify dynamic pipelines with sequential pipelines and graphical pipelines (and dunder methods)

[yarnabrina]

Hi, here's an example of a dynamic pipeline creation function:

from sktime.forecasting.arima import ARIMA
from sktime.forecasting.base import BaseForecaster, ForecastingHorizon
from sktime.forecasting.compose import (
    ForecastingPipeline,
    ForecastX,
    TransformedTargetForecaster,
)
from sktime.forecasting.exp_smoothing import ExponentialSmoothing
from sktime.transformations.base import BaseTransformer
from sktime.transformations.compose import FeatureUnion, TransformerPipeline
from sktime.transformations.series.boxcox import BoxCoxTransformer
from sktime.transformations.series.detrend import Detrender
from sktime.transformations.series.difference import Differencer
from sktime.transformations.series.subset import ColumnSelect


def create_pipeline(
    endogenous_forecaster: BaseForecaster,
    prediction_horizon: ForecastingHorizon,
    future_known_exogenous: list[str] | None = None,
    future_unknown_exogenous: list[str] | None = None,
    future_unknown_forecaster: BaseForecaster | None = None,
    endogenous_transformation: list[BaseTransformer] | None = None,
    future_known_transformation: list[BaseTransformer] | None = None,
    future_unknown_transformation: list[BaseTransformer] | None = None,
) -> BaseForecaster:
    if not endogenous_transformation:
        y_pipe = endogenous_forecaster
    else:
        y_pipe = TransformedTargetForecaster(
            [TransformerPipeline(endogenous_transformation), endogenous_forecaster]
        )

    if not future_known_exogenous:
        known_X_pipe = None
    elif not future_known_transformation:
        known_X_pipe = future_known_transformation
    else:
        known_X_pipe = TransformerPipeline(
            [ColumnSelect(columns=future_known_exogenous), *future_known_transformation]
        )

    if not future_unknown_exogenous:
        unknown_X_trafo_pipe = None
    elif not future_unknown_transformation:
        unknown_X_trafo_pipe = None
    else:
        unknown_X_trafo_pipe = TransformerPipeline(
            [ColumnSelect(columns=future_unknown_exogenous), *future_unknown_transformation]
        )

    if not future_unknown_exogenous:
        unknown_X_fcst_pipe = None
    elif not future_unknown_forecaster:
        unknown_X_fcst_pipe = endogenous_forecaster
    else:
        unknown_X_fcst_pipe = future_unknown_forecaster

    e2e_trafo_pipe_steps: list[BaseTransformer] = []

    if known_X_pipe and not unknown_X_trafo_pipe:
        e2e_trafo_pipe_steps.append(known_X_pipe)
    elif not known_X_pipe and unknown_X_trafo_pipe:
        e2e_trafo_pipe_steps.append(unknown_X_trafo_pipe)
    elif known_X_pipe and unknown_X_trafo_pipe:
        e2e_trafo_pipe_steps.append(FeatureUnion([known_X_pipe, unknown_X_trafo_pipe]))

    if e2e_trafo_pipe_steps:
        e2e_trafo_pipe = TransformerPipeline(e2e_trafo_pipe_steps)
    else:
        e2e_trafo_pipe = None

    if not unknown_X_fcst_pipe:
        e2e_fcst_pipe = y_pipe
    else:
        e2e_fcst_pipe = ForecastX(y_pipe, unknown_X_fcst_pipe, fh_X=prediction_horizon)

    if e2e_trafo_pipe:
        e2e_pipe = ForecastingPipeline([e2e_trafo_pipe, e2e_fcst_pipe])
    else:
        e2e_pipe = e2e_fcst_pipe

    return e2e_pipe

This is untested in the sense I have tried to do fit/predict/update with this. This is also incomplete as error handling stuff is missing, but the objective of the function is to support all of these (plus more) cases:

1. only y with no transformation - simplest case

create_pipeline(ARIMA(), ForecastingHorizon(values=range(1, 4)))

2. y with transformation(s)

create_pipeline(
    ARIMA(), ForecastingHorizon(values=range(1, 4)), endogenous_transformation=[Differencer()]
)

3. y with X which are known for future

create_pipeline(
    ARIMA(), ForecastingHorizon(values=range(1, 4)), future_known_exogenous=["alpha", "beta"]
)

4. y with known X with transformations for X

create_pipeline(
    ARIMA(),
    ForecastingHorizon(values=range(1, 4)),
    future_known_exogenous=["alpha", "beta"],
    future_known_transformation=[BoxCoxTransformer()],
)

5. y with X which are unknown for future with no specification for separate forecaster

create_pipeline(
    ARIMA(),
    ForecastingHorizon(values=range(1, 4)),
    future_unknown_exogenous=["gamma", "delta"],
)

6. y with unknown X with dedicated forecaster

create_pipeline(
    ARIMA(),
    ForecastingHorizon(values=range(1, 4)),
    future_unknown_exogenous=["gamma", "delta"],
    future_unknown_forecaster=ExponentialSmoothing(),
)

7. y with unknown X with transformations for X

create_pipeline(
    ARIMA(),
    ForecastingHorizon(values=range(1, 4)),
    future_unknown_exogenous=["gamma", "delta"],
    future_unknown_forecaster=ExponentialSmoothing(),
    future_unknown_transformation=[Detrender()],
)

8. y with both known and unknown X with all types of transformations and specific forecaster for unknown X

create_pipeline(
    ARIMA(),
    ForecastingHorizon(values=range(1, 4)),
    future_known_exogenous=["alpha", "beta"],
    future_unknown_exogenous=["gamma", "delta"],
    future_unknown_forecaster=ExponentialSmoothing(),
    endogenous_transformation=[Differencer()],
    future_known_transformation=[BoxCoxTransformer()],
    future_unknown_transformation=[Detrender()],
)

This can be further expanded logically, e.g. graphical pipeline should able to support a distinction of "future-unknown" features which need other exogenous features and which do not.

@benHeid if you can share how will you convert the above function for graphical version, with extra modifications to support the above case, that will be very much appreciated. Since not all components exist all the time I had a hard time figuring out how to convert even for more or less sequential flow, and had to give up for more generalised dynamic DAG.

[yarnabrina]

FYI @fkiraly. I am hoping we can use this thread to create few different specifications of same pipeline in sequential/dunder/graphical, and can use for both documentation purpose and also to highlight to users the capability and simplification of dunder+graphical approaches.

(This is from point of view of a developer of non-techinical user faced applications on top of sktime, not from sktime contributor view.)

[fkiraly]

It occurs to me that we had an issue like that already: #4413

Although it seems I did not action it.

[yarnabrina]

Do we already have such things documented for dunder ones? I went through the ones in main tutorial notebooks, but *, **, + etc. get's too complicated for me to track in a dynamic function like above. If the above thing can be simplified with dunder, that'll be a nice thing to learn.

(I'll probably still hesitate to use in our production work though, a new developer may not find that readable if they are not already familiar with sktime syntax.)

[fkiraly]

not in this form. If I were to write the code with dunders, it's a bit tedious due to iteration over lists:

y_pipe = endogenous_forecaster
if endogenous_transformation:
    for t in endogenous_transformation:
        endogenous_forecaster = t * endogenous_forecaster

if future_known_transformation:
    known_X_pipe = future_known_transformation[0]
    for t in future_known_transformation[1:]:
        known_X_pipe *= t
if future_known_exogenous:
    known_X_pipe = known_X_pipe[future_known_exogenous]

if future_unknown_transformation:
    unknown_X_pipe = future_unnknown_transformation[0]
    for t in future_unknown_transformation[1:]:
        unknown_X_pipe *= t
if future_unknown_exogenous:
    unknown_X_pipe = unknown_X_pipe[future_unknown_exogenous]

(this is halfway but should show the issue with lists)

[yarnabrina]

Yes, I understand. Apart from tedious, known_X_pipe = known_X_pipe[future_known_exogenous] is also potentially very confusing to new users, so let me stick to direct class use for now. But I am looking for cases where I'd want to have bit more control but less complicated way, so I'm hoping @benHeid can give some options in graphical one below. Currently it seems very difficult to keep track of node identifiers, for me anyway.

[benHeid]

That is a complicated example and I haven't tested the following code, so probably it will not run directly. However, I hope it helps to get a feeling how it might look like.

from sktime.forecasting.arima import ARIMA
from sktime.forecasting.base import BaseForecaster, ForecastingHorizon
from sktime.forecasting.compose import (
    ForecastingPipeline,
    ForecastX,
    TransformedTargetForecaster,
)
from sktime.forecasting.exp_smoothing import ExponentialSmoothing
from sktime.transformations.base import BaseTransformer
from sktime.transformations.compose import FeatureUnion, TransformerPipeline
from sktime.transformations.series.boxcox import BoxCoxTransformer
from sktime.transformations.series.detrend import Detrender
from sktime.transformations.series.difference import Differencer
from sktime.transformations.series.subset import ColumnSelect

from sktime.pipeline import Pipeline



def create_pipeline(
    endogenous_forecaster: BaseForecaster,
    prediction_horizon: ForecastingHorizon,
    future_known_exogenous: list[str] | None = None,
    future_unknown_exogenous: list[str] | None = None,
    future_unknown_forecaster: BaseForecaster | None = None,
    endogenous_transformation: list[BaseTransformer] | None = None,
    future_known_transformation: list[BaseTransformer] | None = None,
    future_unknown_transformation: list[BaseTransformer] | None = None,
) -> BaseForecaster:
    
    pipe = Pipeline()

    # Add all the endogenous transformation to the pipelien
    tranfo_input = "y"
    for endogenous_trafo in endogenous_transformation:
        pipe = pipe.add_step(endogenous_trafo, endogenous_trafo_name, {"X", tranfo_input})
        tranfo_input = endogenous_trafo_name

    # Add all transformation for the known x values to the pipeline
    known_xs = []
    for col in future_known_exogenous:
        ex_trafo_input_known = f"X__{col}"
        for future_ex_trafo in future_known_transformation:
            pipe = pipe.add_step(future_ex_trafo, future_ex_trafo_name, {"X": ex_trafo_input_known})
            ex_trafo_input_known = future_ex_trafo_name
        known_xs.append(future_ex_trafo_name)
    
    # For all unknown x values add the transformation and the corresponding forecaster for the x values
    unknown_xs = []
    for col in future_unknown_exogenous:
        ex_trafo_input = f"X__{col}"
        for future_ex_trafo in future_known_transformation:
            pipe = pipe.add_step(future_ex_trafo, future_ex_trafo_name, {"X": ex_trafo_input})
            ex_trafo_input = future_ex_trafo_name
        unknown_xs.append(future_ex_trafo_name)

        # Add the forecaster for the unknown x values
        future_forecater = future_unknown_forecaster if future_unknown_forecaster else endogenous_forecaster
        pipe = pipe.add_step(future_forecater, f"future_forecater_{col}", {"y": unknown_xs})
        # Add the forecasted to the known x values
        known_xs.append("future_forecater_{col}")

    # Add the endogenous forecaster to the pipeline with x values if available
    y_forecast_input = {"y": tranfo_input} if len(known_xs) == 0 else {"y": tranfo_input, "X": known_xs}
    pipe = pipe.add(endogenous_forecaster, "endogenous_forecaster", y_forecast_input)


    # Add the inverse transformation to the pipeline
    tranfo_input = "endogenous_forecaster"
    for endogenous_trafo in endogenous_transformation:
        if endogenous_forecaster.is_inverse:
            pipe = pipe.add_step(endogenous_trafo, endogenous_trafo_name, {"X", tranfo_input}, method="inverse_transform")
        tranfo_input = endogenous_trafo_name

    return pipe

To admit, this code looks also a bit complicated. My first idea of creating a blueprint and filling it out dynamically, seems to be even more complicated.
And for such use-cases we need definitely a better visualization/drawing of the pipelines.

[yarnabrina]

Thanks a lot for sharing. It does not work as you said, but I learnt quite a few things:

1. name is mandatory, which is a bit tricky in automated sort of situations as most of the times they are not provided. May be we can generate UUID to not bother much about the name.

2. almost all compose objects are unnecessary. I was thinking that I still have to use ForecastX and etc. and that's what I was trying.

3. it does not handle inverse transform, and we have to add that manually.

Now a few doubts and questions:

1. Is there a possibility of names being optional? Can we somehow smartly generate default names for sktime native estimators?

2. In my example, user passes a already initialised transformer (or forecaster) instance. I noted in the tutorial link you shared before that graphical pipeline keep track of objects. Does this mean same transformer (and forecaster) will be used for for variables? Or will it understand to use a clone for each one?

3. I saw you added forecasters for unknown features per feature. Does it mean that it does not support multivariate forecasters? What I mean is I was expecting unknown-X's will go though their transformations, and together the transformed values will be passed to a forecaster, irrespective of whether it's truly multivariate or not. Is that not possible?

4. Can you please explain what do you mean by "better visualization/drawing of the pipelines"? Do you mean something like this (https://elyra.readthedocs.io/en/stable/getting_started/overview.html), or something completely different?

[yarnabrina]

This is my modification of your code to make it work with the 8 situations I already shared in my question.

from sktime.forecasting.arima import ARIMA
from sktime.forecasting.base import BaseForecaster, ForecastingHorizon
from sktime.forecasting.compose import (
    ForecastingPipeline,
    ForecastX,
    TransformedTargetForecaster,
)
from sktime.forecasting.exp_smoothing import ExponentialSmoothing
from sktime.pipeline import Pipeline
from sktime.transformations.base import BaseTransformer
from sktime.transformations.compose import FeatureUnion, TransformerPipeline
from sktime.transformations.series.boxcox import BoxCoxTransformer
from sktime.transformations.series.detrend import Detrender
from sktime.transformations.series.difference import Differencer
from sktime.transformations.series.subset import ColumnSelect


def create_graphical_pipeline(
    endogenous_forecaster: BaseForecaster,
    future_known_exogenous: list[str] | None = None,
    future_unknown_exogenous: list[str] | None = None,
    future_unknown_forecaster: BaseForecaster | None = None,
    endogenous_transformation: list[tuple[BaseTransformer, str]] | None = None,
    future_known_transformation: list[tuple[BaseTransformer, str]] | None = None,
    future_unknown_transformation: list[tuple[BaseTransformer, str]] | None = None,
) -> BaseForecaster:
    pipe = Pipeline()

    # Add all the endogenous transformation to the pipeline
    y_trafo_input = "y"

    if endogenous_transformation:
        for endogenous_trafo, endogenous_trafo_name in endogenous_transformation:
            pipe = pipe.add_step(endogenous_trafo, endogenous_trafo_name, {"X", y_trafo_input})
            y_trafo_input = endogenous_trafo_name

    # Add all transformation for the known X values to the pipeline
    known_X_trafos = []

    if future_known_exogenous:
        for col in future_known_exogenous:
            known_X_trafo_input = f"X__{col}"

            if future_known_transformation:
                for known_X_trafo, known_X_trafo_name in future_known_transformation:
                    pipe = pipe.add_step(
                        known_X_trafo, known_X_trafo_name, {"X": known_X_trafo_input}
                    )
                    known_X_trafo_input = known_X_trafo_name

            known_X_trafos.append(known_X_trafo_input)

    # For all unknown X values add the transformation and corresponding forecaster for the X values
    if future_unknown_exogenous:
        for col in future_unknown_exogenous:
            unknown_X_trafo_input = f"X__{col}"

            if future_unknown_transformation:
                for unknown_X_trafo, unknown_X_trafo_name in future_unknown_transformation:
                    pipe = pipe.add_step(
                        unknown_X_trafo, unknown_X_trafo_name, {"X": unknown_X_trafo_input}
                    )
                    unknown_X_trafo_input = unknown_X_trafo_name

            # Add the forecaster for the unknown X values
            future_forecaster = (
                future_unknown_forecaster if future_unknown_forecaster else endogenous_forecaster
            )
            pipe = pipe.add_step(
                future_forecaster, f"future_forecaster_{col}", {"y": unknown_X_trafo_input}
            )

            # Add the inverse transformation to the pipeline
            unknown_X_trafo_input = f"future_forecaster_{col}"

            if future_unknown_transformation:
                for unknown_X_trafo, unknown_X_trafo_name in future_unknown_transformation:
                    if unknown_X_trafo.get_tag("capability:inverse_transform", True):
                        pipe = pipe.add_step(
                            unknown_X_trafo,
                            unknown_X_trafo_name,
                            {"X", unknown_X_trafo_input},
                            method="inverse_transform",
                        )
                    unknown_X_trafo_input = unknown_X_trafo_name

            # Add the forecasted to the known X values
            known_X_trafos.append(unknown_X_trafo_input)

    # Add the endogenous forecaster to the pipeline with X values if available
    y_forecast_input = (
        {"y": y_trafo_input} if not known_X_trafos else {"y": y_trafo_input, "X": known_X_trafos}
    )
    pipe = pipe.add_step(endogenous_forecaster, "endogenous_forecaster", y_forecast_input)

    # Add the inverse transformation to the pipeline
    if endogenous_transformation:
        y_trafo_input = "endogenous_forecaster"
        for endogenous_trafo, endogenous_trafo_name in endogenous_transformation:
            if endogenous_trafo.get_tag("capability:inverse_transform", True):
                pipe = pipe.add_step(
                    endogenous_trafo,
                    endogenous_trafo_name,
                    {"X", y_trafo_input},
                    method="inverse_transform",
                )
            y_trafo_input = endogenous_trafo_name

    return pipe

I am quite nervous about the changes I did for unknown-X's, so if you can review it once, that'd be great. I still haven't tested this with a dataset though, only verified that I can run the function for all of those cases, e.g. here's one example for 8th one:

create_graphical_pipeline(
    ARIMA(),
    future_known_exogenous=["alpha", "beta"],
    future_unknown_exogenous=["gamma", "delta"],
    future_unknown_forecaster=ExponentialSmoothing(),
    endogenous_transformation=[(Differencer(), "differencer")],
    future_known_transformation=[(BoxCoxTransformer(), "box_cox")],
    future_unknown_transformation=[(Detrender(), "detrend")],
)