Repository: KIT-IAI/pyWATTS

Issue:
[Online Learning] Fill buffer before online execution starts
**What is your feature request related to?**
* Problem related

**Please describe the module/problem**

If you simulate the online behaviour of modules and you use retraining, then it might happen that the relearning is triggered at the beginning of the online execution/simulation. If the module which should be refitted needs a big refitting_batch, then not enough data is available. 

**Describe the solution you'd like**
The test method should get an online_start argument for specifying a date when the online simulation should be started.


**What type of Feature Request**

* Core enhancement


Discussion:

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 30be6e8..8235e3f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,10 +7,13 @@
 * Metric can be calculated on cutouts ([#149](https://github.com/KIT-IAI/pyWATTS/issues/149))
 * Add custom scaler module for pre-processing ([#167](https://github.com/KIT-IAI/pyWATTS/issues/167))
 * Add a MASE Summary ([#148](https://github.com/KIT-IAI/pyWATTS/issues/148))
+* Online Start Feature ([#189](https://github.com/KIT-IAI/pyWATTS/issues/189))
 
 
 ### Changed
 * Retraining is triggered after all steps are transformed ([#117](https://github.com/KIT-IAI/pyWATTS/issues/117))
+* The sklean-wrapper includes now methods based on the SelectorMixin ([#185](https://github.com/KIT-IAI/pyWATTS/issues/185))
+
 
 ### Fixed
 * Try to handle non matching shapes in metrics. If not possible raise an exception.  ([#154](https://github.com/KIT-IAI/pyWATTS/issues/154))
diff --git a/README.md b/README.md
index 143db0f..88902b7 100644
--- a/README.md
+++ b/README.md
@@ -258,12 +258,13 @@ For a common understanding of the various terms used in this framework, see the
 
 # Funding
 
-This project is supported by the Helmholtz Association under the Joint Initiative "Energy System 2050 - A Contribution
-of the Research Field Energy" and the Helmholtz AI Corporation Unit Energy as well as by the German Research Foundation
-(DFG) Research Training Group 2153 "Energy Status Data: Informatics Methods for its Collection, Analysis and
-Exploitation".
+This project is supported by the Helmholtz Association under the Program “Energy System Design”, by the Helmholtz Association’s Initiative and Networking Fund through Helmholtz AI, by the Helmholtz Association under the Joint Initiative "Energy System 2050 - A Contribution of the Research Field Energy", and by the German Research Foundation (DFG) Research Training Group 2153 "Energy Status Data: Informatics Methods for its Collection, Analysis and Exploitation".
 
 # Citation
 If you use this framework in a scientific publication please cite the corresponding paper:
 
->Heidrich, Benedikt, A. Bartschat, Marian Turowski, Oliver Neumann, Kaleb Phipps, Stefan Meisenbacher, Kai Schmieder, Nicole Ludwig, R. Mikut and V. Hagenmeyer. “pyWATTS: Python Workflow Automation Tool for Time Series.” (2021). ). arXiv:2106.10157. http://arxiv.org/abs/2106.10157
+>Benedikt Heidrich, Andreas Bartschat, Marian Turowski, Oliver Neumann, Kaleb Phipps, Stefan Meisenbacher, Kai Schmieder, Nicole Ludwig, Ralf Mikut, Veit Hagenmeyer. “pyWATTS: Python Workflow Automation Tool for Time Series.” (2021). ). arXiv:2106.10157. http://arxiv.org/abs/2106.10157
+
+# Contact
+If you have any questions and want to talk to the pyWATTS Team directly, feel free to [contact us](mailto:pywatts-team@iai.kit.edu).
+For more information on pyWATTSvisit the [project website](https://www.iai.kit.edu/english/1266_4162.php).
diff --git a/docs/source/about_us.rst b/docs/source/about_us.rst
index a58fab3..6ef9e3c 100644
--- a/docs/source/about_us.rst
+++ b/docs/source/about_us.rst
@@ -6,6 +6,8 @@ History
 This project was started in 2019 by the Research Group "Machine Learning for Time Series and Images" of the Institute for
 Automation and Applied Informatics at the Karlsruhe Institute of Technology.
 
+For more information on pyWATTS you can visit https://www.iai.kit.edu/english/1266_4162.php
+
 Authors
 -------
 The following people are currently involved in pyWATTS' core development and maintenance:
@@ -13,6 +15,7 @@ The following people are currently involved in pyWATTS' core development and mai
 * Andreas Bartschat
 * Benedikt Heidrich
 * Nicole Ludwig
+* Stefan Meisenbacher
 * Oliver Neumann
 * Kaleb Phipps
 * Marian Turowski
@@ -30,11 +33,10 @@ The following people have been active contributors in the past, but are no longe
 
 Citing pyWATTS
 --------------
-tbd
+If you use this framework in a scientific publication please cite the corresponding paper:
+
+Benedikt Heidrich, Andreas Bartschat, Marian Turowski, Oliver Neumann, Kaleb Phipps, Stefan Meisenbacher, Kai Schmieder, Nicole Ludwig, Ralf Mikut, Veit Hagenmeyer. “pyWATTS: Python Workflow Automation Tool for Time Series.” (2021).). arXiv:2106.10157. http://arxiv.org/abs/2106.10157
 
 Funding
 -------
-This project is supported by the Helmholtz Association under the Joint Initiative "Energy System 2050 - A Contribution
-of the Research Field Energy" and the Helmholtz AI Corporation Unit Energy as well as by the German Research Foundation
-(DFG) Research Training Group 2153 "Energy Status Data: Informatics Methods for its Collection, Analysis and
-Exploitation".
+This project is supported by the Helmholtz Association under the Program “Energy System Design”, by the Helmholtz Association’s Initiative and Networking Fund through Helmholtz AI, by the Helmholtz Association under the Joint Initiative "Energy System 2050 - A Contribution of the Research Field Energy", and by the German Research Foundation (DFG) Research Training Group 2153 "Energy Status Data: Informatics Methods for its Collection, Analysis and Exploitation".
diff --git a/examples/example.py b/examples/example.py
index 2a6c1bc..eb2ce99 100644
--- a/examples/example.py
+++ b/examples/example.py
@@ -7,13 +7,14 @@
 import pandas as pd
 from sklearn.linear_model import LinearRegression
 from sklearn.preprocessing import StandardScaler
+from sklearn.feature_selection import SelectKBest, f_regression
 
 # From pyWATTS the pipeline is imported
 from pywatts.callbacks import LinePlotCallback
 from pywatts.core.computation_mode import ComputationMode
 from pywatts.core.pipeline import Pipeline
 # All modules required for the pipeline are imported
-from pywatts.modules import CalendarExtraction, CalendarFeature, ClockShift, LinearInterpolater, SKLearnWrapper
+from pywatts.modules import CalendarExtraction, CalendarFeature, ClockShift, LinearInterpolater, SKLearnWrapper, Sampler
 from pywatts.summaries import RMSE
 
 # The main function is where the pipeline is created and run
@@ -21,7 +22,7 @@ if __name__ == "__main__":
     # Create a pipeline
     pipeline = Pipeline(path="../results")
 
-    # Extract dummy calender features, using holidays from Germany
+    # Extract dummy calendar features, using holidays from Germany
     # NOTE: CalendarExtraction can't return multiple features.
     calendar = CalendarExtraction(continent="Europe", country="Germany", features=[CalendarFeature.month,
                                                                                    CalendarFeature.weekday,
@@ -43,28 +44,38 @@ if __name__ == "__main__":
     shift_power_statistics2 = ClockShift(lag=2, name="ClockShift_Lag2"
                                          )(x=scale_power_statistics)
 
+    target_multiple_output = Sampler(24, name="sampled_data")(x=scale_power_statistics)
+
+    # Select features based on F-statistic
+    selected_features = SKLearnWrapper(
+        module=SelectKBest(score_func=f_regression, k=2)
+    )(
+        power_lag1=shift_power_statistics,
+        power_lag2=shift_power_statistics2,
+        calendar=calendar,
+        target=scale_power_statistics,
+    )
+
     # Create a linear regression that uses the lagged values to predict the current value
     # NOTE: SKLearnWrapper has to collect all **kwargs itself and fit it against target.
     #       It is also possible to implement a join/collect class
     regressor_power_statistics = SKLearnWrapper(
         module=LinearRegression(fit_intercept=True)
     )(
-        power_lag1=shift_power_statistics,
-        power_lag2=shift_power_statistics2,
-        calendar=calendar,
-        target=scale_power_statistics,
-        callbacks=[LinePlotCallback('linear_regression')],
+        features=selected_features,
+        target=target_multiple_output,
+        callbacks=[LinePlotCallback("linear_regression")],
     )
 
     # Rescale the predictions to be on the original time scale
     inverse_power_scale = power_scaler(
         x=regressor_power_statistics, computation_mode=ComputationMode.Transform,
-        use_inverse_transform=True, callbacks=[LinePlotCallback('rescale')]
+        use_inverse_transform=True, callbacks=[LinePlotCallback("rescale")]
     )
 
     # Calculate the root mean squared error (RMSE) between the linear regression and the true values
     # save it as csv file
-    rmse = RMSE()(y_hat=inverse_power_scale, y=pipeline["load_power_statistics"])
+    rmse = RMSE()(y_hat=inverse_power_scale, y=target_multiple_output)
 
     # Now, the pipeline is complete so we can run it and explore the results
     # Start the pipeline
diff --git a/examples/example_batch_execution.py b/examples/example_batch_execution.py
index e439c4b..2587e32 100644
--- a/examples/example_batch_execution.py
+++ b/examples/example_batch_execution.py
@@ -104,5 +104,5 @@ if __name__ == "__main__":
     # Now, the pipeline is complete so we can run it and explore the results
     # Start the pipeline
     print("Start testing")
-    result = pipeline.test(test)
+    result = pipeline.test(test, online_start=pd.to_datetime("2018-12-30"))
     print("Testing finished")
diff --git a/pywatts/core/base_step.py b/pywatts/core/base_step.py
index 8dbb77f..072bc99 100644
--- a/pywatts/core/base_step.py
+++ b/pywatts/core/base_step.py
@@ -121,19 +121,21 @@ class BaseStep(ABC):
     def _pack_data(self, start, end, buffer_element=None, return_all=False):
         # Provide requested data
         time_index = _get_time_indexes(self.buffer)
-        if end and start and end > start:
+        if start:
             index = list(self.buffer.values())[0].indexes[time_index[0]]
-            start = max(index[0], start.to_numpy())
+            start = start.to_numpy()
+            # If end is not set, all values should be considered. Thus we add a small timedelta to the last index entry.
+            end = end.to_numpy() if end is not None else (index[-1] + pd.Timedelta(nanoseconds=1)).to_numpy()
             # After sel copy is not needed, since it returns a new array.
             if buffer_element is not None:
                 return self.buffer[buffer_element].sel(
-                    **{time_index[0]: index[(index >= start) & (index < end.to_numpy())]})
+                    **{time_index[0]: index[(index >= start) & (index < end)]})
             elif return_all:
-                return {key: b.sel(**{time_index[0]: index[(index >= start) & (index < end.to_numpy())]}) for
+                return {key: b.sel(**{time_index[0]: index[(index >= start) & (index < end)]}) for
                         key, b in self.buffer.items()}
             else:
                 return list(self.buffer.values())[0].sel(
-                    **{time_index[0]: index[(index >= start) & (index < end.to_numpy())]})
+                    **{time_index[0]: index[(index >= start) & (index < end)]})
         else:
             self.finished = True
             if buffer_element is not None:
@@ -218,11 +220,13 @@ class BaseStep(ABC):
     def _input_stopped(input_data):
         return (input_data is not None and len(input_data) > 0 and any(map(lambda x: x is None, input_data.values())))
 
-    def reset(self):
+    def reset(self, keep_buffer=False):
         """
         Resets all information of the step concerning a specific run.
+        :param keep_buffer: Flag indicating if the buffer should be resetted too.
         """
-        self.buffer = {}
+        if not keep_buffer:
+            self.buffer = {}
         self.finished = False
         self.current_run_setting = self.default_run_setting.clone()
 
diff --git a/pywatts/core/pipeline.py b/pywatts/core/pipeline.py
index c7125b8..b1f5d62 100644
--- a/pywatts/core/pipeline.py
+++ b/pywatts/core/pipeline.py
@@ -72,16 +72,31 @@ class Pipeline(BaseTransformer):
         :return:The transformed data
         :rtype: xr.DataArray
         """
+        if self.current_run_setting.online_start is not None:
+            time_index_name = _get_time_indexes(x)[0]
+            time_index = list(x.values())[0][time_index_name]
+            if time_index[-1].values < self.current_run_setting.online_start:
+                # Complete data should not be executed online and no summaries should be calculated. Thus
+                # _transform is called directly.
+                return self._transform(x, None)
+            else:
+                return self._comp(x, self.current_run_setting.summary_formatter, self.batch)[0]
+        else:
+            return self._comp(x, self.current_run_setting.summary_formatter, self.batch)[0]
+
+    def _transform(self, x, batch=None):
         for key, (start_step, _) in self.start_steps.items():
-            start_step.buffer = {key: x[key].copy()}
+            if not start_step.buffer:
+                start_step.buffer = {key: x[key].copy()}
+            else:
+                dim = _get_time_indexes(x)[0]
+                last = start_step.buffer[key][dim].values[-1]
+                start_step.buffer[key] = xr.concat([start_step.buffer[key], x[key][x[key][dim] > last]], dim=dim)
             start_step.finished = True
-
         time_index = _get_time_indexes(x)
         self.counter = list(x.values())[0].indexes[time_index[0]][0]  # The start date of the input time series.
-
         last_steps = list(filter(lambda x: x.last, self.id_to_step.values()))
-
-        if not self.batch:
+        if not batch:
             return self._collect_results(last_steps)
         return self._collect_batches(last_steps, time_index)
 
@@ -153,7 +168,7 @@ class Pipeline(BaseTransformer):
         # TODO built the graph which should be drawn by starting with the last steps...
 
     def test(self, data: Union[pd.DataFrame, xr.Dataset], summary: bool = False,
-             summary_formatter: SummaryFormatter = SummaryMarkdown()):
+             summary_formatter: SummaryFormatter = SummaryMarkdown(), online_start=None):
         """
         Executes all modules in the pipeline in the correct order. This method call only transform on every module
         if the ComputationMode is Default. I.e. if no computationMode is specified during the addition of the module to
@@ -168,7 +183,7 @@ class Pipeline(BaseTransformer):
         :return: The result of all end points of the pipeline
         :rtype: Dict[xr.DataArray]
         """
-        return self._run(data, ComputationMode.Transform, summary, summary_formatter)
+        return self._run(data, ComputationMode.Transform, summary, summary_formatter, online_start)
 
     def train(self, data: Union[pd.DataFrame, xr.Dataset], summary: bool = False,
               summary_formatter: SummaryFormatter = SummaryMarkdown()):
@@ -190,19 +205,19 @@ class Pipeline(BaseTransformer):
         return self._run(data, ComputationMode.FitTransform, summary, summary_formatter)
 
     def _run(self, data: Union[pd.DataFrame, xr.Dataset], mode: ComputationMode, summary: bool,
-             summary_formatter: SummaryFormatter):
+             summary_formatter: SummaryFormatter, online_start=None):
+
+        self.current_run_setting = RunSetting(computation_mode=mode,
+                                              summary_formatter=summary_formatter,
+                                              online_start=online_start,
+                                              return_summary=summary)
 
         for step in self.id_to_step.values():
             step.reset()
-            step.set_run_setting(RunSetting(computation_mode=mode, summary_formatter=summary_formatter))
+            step.set_run_setting(self.current_run_setting)
 
         if isinstance(data, pd.DataFrame):
             data = data.to_xarray()
-
-        if isinstance(data, xr.Dataset):
-            result = self.transform(**{key: data[key] for key in data.data_vars})
-            sum = self._create_summary(summary_formatter)
-            return (result, sum) if summary else result
         elif isinstance(data, dict):
             for key in data:
                 if not isinstance(data[key], xr.DataArray):
@@ -211,9 +226,23 @@ class Pipeline(BaseTransformer):
                         "Make sure to pass Dict[str, xr.DataArray].",
                         self.name
                     )
-            result = self.transform(**data)
-            sum = self._create_summary(summary_formatter)
-            return (result, sum) if summary else result
+            data = xr.Dataset(data)
+
+        if isinstance(data, xr.Dataset):
+            if self.current_run_setting.online_start is not None:
+                # First only _transform should be called (no summary, no online) on the data before online_start.
+                # Afterwards, comp is called (_transform and summaries using online simulation)
+                index_name = _get_time_indexes(data)[0]
+                self._transform({key: data[key].sel(
+                    **{index_name: data[key][index_name] < self.current_run_setting.online_start }) for key in data.data_vars}, False)
+                for step in self.id_to_step.values():
+                    step.reset(keep_buffer=True)
+                    step.set_run_setting(self.current_run_setting.clone())
+                return self._comp({key: data[key].sel(
+                    **{index_name: data[key][index_name] >= self.current_run_setting.online_start}) for key in data.data_vars}
+                    , summary_formatter, self.batch, start=self.current_run_setting.online_start)
+            else:
+                return self._comp({key: data[key] for key in data.data_vars}, summary_formatter, self.batch)
 
         raise WrongParameterException(
             "Unkown data type to pass to pipeline steps.",
@@ -221,6 +250,12 @@ class Pipeline(BaseTransformer):
             self.name
         )
 
+    def _comp(self, data, summary_formatter, batch, start=None):
+        result = self._transform(data, batch)
+        summary = self._create_summary(summary_formatter, start)
+        return result, summary
+
+
     def add(self, *,
             module: Union[BaseStep],
             input_ids: List[int] = None,
@@ -414,11 +449,11 @@ class Pipeline(BaseTransformer):
             start_step.id = self.add(module=start_step, input_ids=[], target_ids=[])
         return self.start_steps[item][-1]
 
-    def _create_summary(self, summary_formatter):
+    def _create_summary(self, summary_formatter, start=None, end=None):
         summaries = []
         for step in self.id_to_step.values():
             if isinstance(step, SummaryStep):
-                summaries.append(step.get_summary())
+                summaries.append(step.get_summary(start, end))
             summaries.extend([step.transform_time, step.training_time])
         return summary_formatter.create_summary(summaries, self.file_manager)
 
diff --git a/pywatts/core/pipeline_step.py b/pywatts/core/pipeline_step.py
index 041c070..5f54fa4 100644
--- a/pywatts/core/pipeline_step.py
+++ b/pywatts/core/pipeline_step.py
@@ -39,16 +39,13 @@ class PipelineStep(Step):
         self.current_run_setting = self.default_run_setting.update(run_setting=run_setting)
         for step in self.module.id_to_step.values():
             step.set_run_setting(run_setting)
+        self.module.current_run_setting = self.current_run_setting
 
-    def _post_transform(self, result):
-        self.module._create_summary(self.current_run_setting.summary_formatter)
-        return super()._post_transform(result)
-
-    def reset(self):
+    def reset(self, keep_buffer=False):
         """
         Resets all information of the step concerning a specific run. Furthermore, it resets also all steps
         of the subpipeline.
         """
-        super().reset()
+        super().reset(keep_buffer=keep_buffer)
         for step in self.module.id_to_step.values():
-            step.reset()
+            step.reset(keep_buffer=keep_buffer)
diff --git a/pywatts/core/run_setting.py b/pywatts/core/run_setting.py
index 902cbb2..c01f3e9 100644
--- a/pywatts/core/run_setting.py
+++ b/pywatts/core/run_setting.py
@@ -14,9 +14,12 @@ class RunSetting:
     :type summary_formatter: SummaryFormatter
     """
 
-    def __init__(self, computation_mode: ComputationMode, summary_formatter: SummaryFormatter = SummaryMarkdown()):
+    def __init__(self, computation_mode: ComputationMode, summary_formatter: SummaryFormatter = SummaryMarkdown(),
+                 online_start=None, return_summary=False):
         self.computation_mode = computation_mode
         self.summary_formatter = summary_formatter
+        self.online_start = online_start
+        self.return_summary = return_summary
 
     def update(self, run_setting: 'RunSetting') -> 'RunSetting':
         """
@@ -31,6 +34,8 @@ class RunSetting:
         if setting.computation_mode == ComputationMode.Default:
             setting.computation_mode = run_setting.computation_mode
         setting.summary_formatter = run_setting.summary_formatter
+        setting.online_start = run_setting.online_start
+        setting.return_summary = run_setting.return_summary
         return setting
 
     def clone(self) -> 'RunSetting':
@@ -41,7 +46,9 @@ class RunSetting:
         """
         return RunSetting(
             computation_mode=self.computation_mode,
-            summary_formatter=self.summary_formatter
+            summary_formatter=self.summary_formatter,
+            online_start=self.online_start,
+            return_summary=self.return_summary
         )
 
     def save(self) -> Dict:
diff --git a/pywatts/core/step.py b/pywatts/core/step.py
index 825a090..368b732 100644
--- a/pywatts/core/step.py
+++ b/pywatts/core/step.py
@@ -1,5 +1,6 @@
 import logging
 import time
+import warnings
 from typing import Optional, Dict, Union, Callable, List
 
 import cloudpickle
@@ -64,6 +65,11 @@ class Step(BaseStep):
         self.retrain_batch = retrain_batch
         self.callbacks = callbacks
         self.batch_size = batch_size
+        if self.current_run_setting.computation_mode is not ComputationMode.Refit and train_if is not None:
+            message = "You added a refit_condition without setting the computation_mode to refit." \
+                      " The condition will be ignored."
+            warnings.warn(message)
+            logger.warning(message)
         self.lag = lag
         self.train_if = train_if
         self.result_steps: Dict[str, ResultStep] = {}
@@ -75,13 +81,20 @@ class Step(BaseStep):
     def _callbacks(self):
         # plots and writs the data if the step is finished.
         for callback in self.callbacks:
+            dim = _get_time_indexes(self.buffer)[0]
+
+            if self.current_run_setting.online_start is not None:
+                to_plot = {k: self.buffer[k][self.buffer[k][dim] >= self.current_run_setting.online_start] for k in
+                           self.buffer.keys()}
+            else:
+                to_plot = self.buffer
             if isinstance(callback, BaseCallback):
                 callback.set_filemanager(self.file_manager)
             if isinstance(self.buffer, xr.DataArray) or isinstance(self.buffer, xr.Dataset):
                 # DEPRECATED: direct DataArray or Dataset passing is depricated
                 callback({"deprecated": self.buffer})
             else:
-                callback(self.buffer)
+                callback(to_plot)
 
     def _transform(self, input_step):
         if isinstance(self.module, BaseEstimator) and not self.module.is_fitted:
diff --git a/pywatts/core/summary_step.py b/pywatts/core/summary_step.py
index fecab65..ca692e2 100644
--- a/pywatts/core/summary_step.py
+++ b/pywatts/core/summary_step.py
@@ -60,11 +60,11 @@ class SummaryStep(Step):
             "name": self.name,
         }
 
-    def get_summary(self) -> SummaryObject:
+    def get_summary(self, start, end) -> SummaryObject:
         """
         Calculates a summary for the input data.
         :return: The summary as markdown formatted string
         :rtype: Str
         """
-        input_data = self._get_input(None, None)
+        input_data = self._get_input(start, end)
         return self._transform(input_data)
diff --git a/pywatts/modules/wrappers/sklearn_wrapper.py b/pywatts/modules/wrappers/sklearn_wrapper.py
index 67a4be9..978552b 100644
--- a/pywatts/modules/wrappers/sklearn_wrapper.py
+++ b/pywatts/modules/wrappers/sklearn_wrapper.py
@@ -1,10 +1,11 @@
-import pickle
+import cloudpickle
 from typing import List, Tuple
 
 import numpy as np
 import sklearn
 import xarray as xr
 from sklearn.base import TransformerMixin
+from sklearn.feature_selection import SelectorMixin
 
 from pywatts.core.exceptions.kind_of_transform_does_not_exist_exception import KindOfTransformDoesNotExistException, \
     KindOfTransform
@@ -61,8 +62,9 @@ class SKLearnWrapper(BaseWrapper):
         self.targets = list(targets.keys())
         x = self._dataset_to_sklearn_input(inputs)
         target = self._dataset_to_sklearn_input(targets)
-        self.targets = list(
-            zip(targets.keys(), map(lambda x: x.shape[-1] if len(x.shape) > 1 else 1, list(targets.values()))))
+        self.targets = list(zip(targets.keys(),
+                                map(lambda t: t.values.reshape((t.shape[0], -1)).shape[-1] if len(t.shape) > 1 else 1,
+                                    list(targets.values()))))
         self.fit_result = self.module.fit(x, target)
         self.is_fitted = True
 
@@ -82,13 +84,13 @@ class SKLearnWrapper(BaseWrapper):
     def _sklearn_output_to_dataset(kwargs: xr.DataArray, prediction, targets: List[Tuple[str, int]]):
         reference = kwargs[list(kwargs)[0]]
         time_index = reference.indexes[_get_time_indexes(reference)[0]]
-        if len(targets) == 0:
+        if len(targets) == 0:  # sklearn module is a transformer
             coords = (
                 # first dimension is number of batches. We assume that this is the time.
                 ("time", time_index.values),
                 *[(f"dim_{j}", list(range(size))) for j, size in enumerate(prediction.shape[1:])])
             result = xr.DataArray(prediction, coords=coords)
-        else:
+        else:  # sklearn module is an estimator
             result = {}
             position = 0
             prediction = prediction.reshape(len(list(reference.coords.values())[0]), -1)
@@ -105,8 +107,12 @@ class SKLearnWrapper(BaseWrapper):
         :return: the transformed output
         """
         x_np = self._dataset_to_sklearn_input(kwargs)
+        target_names = self.targets
 
-        if isinstance(self.module, TransformerMixin):
+        if isinstance(self.module, SelectorMixin):
+            prediction = self.module.transform(x_np)
+            target_names = []  # output of selector must not match the shape of the target
+        elif isinstance(self.module, TransformerMixin):
             prediction = self.module.transform(x_np)
         elif "predict" in dir(self.module):
             prediction = self.module.predict(x_np)
@@ -115,7 +121,7 @@ class SKLearnWrapper(BaseWrapper):
                 f"The sklearn-module in {self.name} does not have a predict or transform method",
                 KindOfTransform.PREDICT_TRANSFORM)
 
-        return self._sklearn_output_to_dataset(kwargs, prediction, self.targets)
+        return self._sklearn_output_to_dataset(kwargs, prediction, target_names)
 
     def inverse_transform(self, **kwargs: xr.DataArray) -> xr.DataArray:
         """
@@ -124,14 +130,18 @@ class SKLearnWrapper(BaseWrapper):
         :return: the transformed output
         """
         x_np = self._dataset_to_sklearn_input(kwargs)
+        target_names = self.targets
+
         if self.has_inverse_transform:
             prediction = self.module.inverse_transform(x_np)
+            if isinstance(self.module, SelectorMixin):
+                target_names = []  # output of selector must not match the shape of the target
         else:
             raise KindOfTransformDoesNotExistException(
                 f"The sklearn-module in {self.name} does not have a inverse transform method",
                 KindOfTransform.INVERSE_TRANSFORM)
 
-        return self._sklearn_output_to_dataset(kwargs, prediction, self.targets)
+        return self._sklearn_output_to_dataset(kwargs, prediction, target_names)
 
     def predict_proba(self, **kwargs) -> xr.DataArray:
         """
@@ -150,12 +160,14 @@ class SKLearnWrapper(BaseWrapper):
         return self._sklearn_output_to_dataset(kwargs, prediction, self.targets)
 
     def save(self, fm: FileManager):
-        json = super().save(fm)
+        json_module = super().save(fm)
+        json_module["targets"] = self.targets
         file_path = fm.get_path(f'{self.name}.pickle')
         with open(file_path, 'wb') as outfile:
-            pickle.dump(obj=self.module, file=outfile)
-        json.update({"sklearn_module": file_path})
-        return json
+            cloudpickle.dump(obj=self.module, file=outfile)
+        json_module.update({"sklearn_module": file_path})
+        json_module.pop("params")
+        return json_module
 
     @classmethod
     def load(cls, load_information) -> 'SKLearnWrapper':
@@ -172,7 +184,8 @@ class SKLearnWrapper(BaseWrapper):
         """
         name = load_information["name"]
         with open(load_information["sklearn_module"], 'rb') as pickle_file:
-            module = pickle.load(pickle_file)
+            module = cloudpickle.load(pickle_file)
         module = cls(module=module, name=name)
         module.is_fitted = load_information["is_fitted"]
+        module.targets = load_information["targets"]
         return module


Test patch:
diff --git a/tests/unit/core/test_pipeline.py b/tests/unit/core/test_pipeline.py
index 96063dc..1186ffe 100644
--- a/tests/unit/core/test_pipeline.py
+++ b/tests/unit/core/test_pipeline.py
@@ -21,18 +21,14 @@ pipeline_json = {'id': 1,
                               'is_fitted': False,
                               'module': 'pywatts.modules.wrappers.sklearn_wrapper',
                               'name': 'StandardScaler',
-                              'params': {'copy': True, 'with_mean': True, 'with_std': True},
-                              'sklearn_module': os.path.join('test_pipeline', 'StandardScaler.pickle')},
+                              'sklearn_module': os.path.join('test_pipeline', 'StandardScaler.pickle'),
+                              'targets': []},
                              {'class': 'SKLearnWrapper',
                               'is_fitted': False,
                               'module': 'pywatts.modules.wrappers.sklearn_wrapper',
                               'name': 'LinearRegression',
-                              'params': {'copy_X': True,
-                                         'fit_intercept': True,
-                                         'n_jobs': None,
-                                         'positive': False,
-                                         'normalize': 'deprecated'},
-                              'sklearn_module': os.path.join('test_pipeline', 'LinearRegression.pickle')}],
+                              'sklearn_module': os.path.join('test_pipeline', 'LinearRegression.pickle'),
+                              'targets': []}],
                  'steps': [{'class': 'StartStep',
                             'default_run_setting': {'computation_mode': 4},
                             'id': 1,
@@ -147,7 +143,7 @@ class TestPipeline(unittest.TestCase):
         assert kwargs["obj"]["steps"] == pipeline_json["steps"]
 
     @patch('pywatts.core.pipeline.FileManager')
-    @patch('pywatts.modules.sklearn_wrapper.pickle')
+    @patch('pywatts.modules.sklearn_wrapper.cloudpickle')
     @patch('pywatts.core.pipeline.json')
     @patch("builtins.open", new_callable=mock_open)
     @patch('pywatts.core.pipeline.os.path.isdir')
@@ -223,8 +219,8 @@ class TestPipeline(unittest.TestCase):
 
         for step in self.pipeline.id_to_step.values():
             assert step.current_run_setting.computation_mode == ComputationMode.FitTransform
-
-        create_summary_mock.assert_has_calls([call(summary_formatter_mock), call(summary_formatter_mock)])
+        assert 2 == create_summary_mock.call_count
+        create_summary_mock.assert_has_calls([call(summary_formatter_mock, None), call(summary_formatter_mock, None)], any_order=True)
 
     @patch('pywatts.core.pipeline.FileManager')
     def test_add_pipeline_to_pipeline_and_test(self, fm_mock):
@@ -386,6 +382,7 @@ class TestPipeline(unittest.TestCase):
         pipeline.start_steps["foo"][0].last = False
         step_one.further_elements.side_effect = [True, True, True, True, False]
         pipeline.add(module=step_one, input_ids=[1])
+        pipeline.current_run_setting = RunSetting(computation_mode=ComputationMode.Transform)
 
         result = pipeline.transform(foo=da)
 
@@ -404,6 +401,7 @@ class TestPipeline(unittest.TestCase):
         step_two.name = "mock"
         step_two.get_result.return_value = {"mock": result_mock}
         self.pipeline.add(module=step_two, input_ids=[1])
+        self.pipeline.current_run_setting = RunSetting(computation_mode=ComputationMode.Transform)
 
         result = self.pipeline.transform(x=input_mock)
 
@@ -459,6 +457,7 @@ class TestPipeline(unittest.TestCase):
         pipeline.start_steps["foo"] = StartStep("foo"), None
         pipeline.start_steps["foo"][0].last = False
         step_one.further_elements.side_effect = [True, True, True, True, True, True, True, False]
+        pipeline.current_run_setting = RunSetting(computation_mode=ComputationMode.Transform)
         pipeline.add(module=step_one, input_ids=[1])
 
         result = pipeline.transform(foo=da)
@@ -581,3 +580,26 @@ class TestPipeline(unittest.TestCase):
         self.pipeline.refit(pd.Timestamp("2000.01.02"), pd.Timestamp("2022.01.02"))
 
         first_step.refit.assert_called_once_with(pd.Timestamp("2000.01.01"), pd.Timestamp("2022.01.01"))
+
+    def test_online_with_filled_buffer(self):
+
+        time = pd.date_range('2000-01-01', freq='1D', periods=7)
+        da = xr.DataArray([2, 3, 4, 3, 3, 1, 2], dims=["time"], coords={'time': time})
+        pipeline = Pipeline(batch=pd.Timedelta(days=1))
+        step_one = MagicMock()
+        step_one.get_result.return_value = {"step": da}
+
+        pipeline.start_steps["foo"] = StartStep("foo"), None
+        pipeline.start_steps["foo"][0].last = False
+        step_one.further_elements.side_effect = [True, True, True, True, False]
+        pipeline.add(module=step_one, input_ids=[1])
+
+        pipeline.test(pd.DataFrame({"foo": [1, 2, 2, 3, 4,5,6]},
+                      index=pd.DatetimeIndex(pd.date_range('2000-01-01', freq='24H', periods=7))),
+                      online_start=pd.to_datetime('2000-01-04'))
+
+        self.assertEqual(5, step_one.get_result.call_count)
+        self.assertEqual(2, step_one.reset.call_count)
+        reset_calls = [call(), call(keep_buffer=True)]
+        step_one.reset.assert_has_calls(reset_calls)
+        self.assertEqual(5, step_one.further_elements.call_count)
diff --git a/tests/unit/core/test_summary_step.py b/tests/unit/core/test_summary_step.py
index b862834..644324a 100644
--- a/tests/unit/core/test_summary_step.py
+++ b/tests/unit/core/test_summary_step.py
@@ -30,7 +30,7 @@ class TestSummaryStep(unittest.TestCase):
         self.fm_mock = None
 
     def test_get_summary(self):
-        result = self.summary.get_summary()
+        result = self.summary.get_summary(None, None)
 
         self.step_mock.get_result.assert_called_once_with(None, None)
         self.module_mock.transform.assert_called_once_with(file_manager=self.fm_mock, input=self.result_mock)
diff --git a/tests/unit/wrapper/test_sklearn_wrapper.py b/tests/unit/wrapper/test_sklearn_wrapper.py
index 7b62a42..14a32ed 100644
--- a/tests/unit/wrapper/test_sklearn_wrapper.py
+++ b/tests/unit/wrapper/test_sklearn_wrapper.py
@@ -7,6 +7,7 @@ from sklearn.cluster import KMeans
 from sklearn.linear_model import LinearRegression
 from sklearn.mixture import GaussianMixture
 from sklearn.multioutput import MultiOutputRegressor
+from sklearn.feature_selection import SelectKBest, f_regression
 
 from sklearn.preprocessing import StandardScaler
 from sklearn.svm import SVC, SVR
@@ -33,18 +34,31 @@ class TestSklearnWrapper(unittest.TestCase):
         wrapper = SKLearnWrapper(module=scaler)
         self.assertFalse("mean_" in scaler.__dir__())
 
-        wrapper.fit(test = xr.DataArray([1, 2, 3, 4, 5]))
+        wrapper.fit(test=xr.DataArray([1, 2, 3, 4, 5]))
 
         self.assertTrue("mean_" in scaler.__dir__())
         self.assertIsNotNone(scaler.mean_)
 
+    def test_transform_TransformerMixin(self):
+        scaler = StandardScaler()
+        wrapper = SKLearnWrapper(module=scaler)
+        self.assertFalse("mean_" in scaler.__dir__())
+        time = pd.date_range('2000-01-08', freq='24H', periods=5)
+        test = xr.DataArray([2, 2, 2, 2, 2], dims=["time"], coords={'time': time})
+
+        wrapper.fit(test=test)
+        result = wrapper.transform(test=test)
+
+        self.assertListEqual(list(result), [0, 0, 0, 0, 0])
+        self.assertEqual(result.shape, (5, 1))
+
     def test_fit_RegressorMixin(self):
         lin_reg = LinearRegression()
         wrapper = SKLearnWrapper(module=lin_reg)
         self.assertFalse("coef_" in lin_reg.__dir__())
 
-        wrapper.fit(test = xr.DataArray([1, 2, 3, 4, 5]),
-                    target= xr.DataArray([2, 2, 2, 2, 2]))
+        wrapper.fit(test=xr.DataArray([1, 2, 3, 4, 5]),
+                    target=xr.DataArray([2, 2, 2, 2, 2]))
 
         self.assertTrue("coef_" in lin_reg.__dir__())
         self.assertIsNotNone(lin_reg.coef_)
@@ -55,13 +69,42 @@ class TestSklearnWrapper(unittest.TestCase):
         time = pd.date_range('2000-01-08', freq='24H', periods=1)
         bar = xr.DataArray([1], dims=["time"], coords={'time': time})
 
-        wrapper.fit(test = xr.DataArray([1, 2, 3, 4, 5]),
-                    target= xr.DataArray([2, 2, 2, 2, 2]))
-
+        wrapper.fit(test=xr.DataArray([1, 2, 3, 4, 5]),
+                    target=xr.DataArray([2, 2, 2, 2, 2]))
 
         result = wrapper.transform(bar=bar)
         assert result["target"].values[0] == 2.0
-        self.assertEqual(result["target"].shape, (1,1))
+        self.assertEqual(result["target"].shape, (1, 1))
+
+    def test_fit_SelectorMixin(self):
+        kbest = SelectKBest(score_func=f_regression, k=1)
+        wrapper = SKLearnWrapper(module=kbest)
+
+        eps = 0.001
+        wrapper.fit(feature1=xr.DataArray([x + eps for x in [2, 2, 3, 4, 4]]),
+                    feature2=xr.DataArray([1, 2, 3, 4, 5]),
+                    target=xr.DataArray([2, 2, 3, 4, 4]))
+
+        self.assertTrue("scores_" in kbest.__dir__())
+        self.assertIsNotNone(kbest.scores_)
+
+    def test_transform_SelectorMixin(self):
+        kbest = SelectKBest(score_func=f_regression, k=1)
+        wrapper = SKLearnWrapper(module=kbest)
+        eps = 0.001
+        time = pd.date_range('2000-01-08', freq='24H', periods=5)
+        target = xr.DataArray([2, 2, 3, 4, 4], dims=["time"], coords={'time': time})
+        feature1 = target + eps
+        feature2 = xr.DataArray([4, 4, 3, 2, 2], dims=["time"], coords={'time': time}) + eps
+
+        wrapper.fit(feature1=feature1,
+                    feature2=feature2,
+                    target=target)
+
+        result = wrapper.transform(feature1=feature1, feature2=feature2)
+
+        self.assertListEqual(list(result), list(feature2))
+        self.assertEqual(result.shape, (5, 1))
 
     def test_DensityMixin(self):
         gauss_density = GaussianMixture(n_components=2)
@@ -159,10 +202,8 @@ class TestSklearnWrapper(unittest.TestCase):
 
         wrapper.fit(bar=bar, target1=target, target2=target2)
 
-
         result = wrapper.transform(bar=foo)
         self.assertAlmostEqual(result["target1"].values[0], 2.0)
         self.assertAlmostEqual(result["target2"].values[0], 3.0)
         self.assertEqual(result["target1"].shape, (1, 1))
         self.assertEqual(result["target2"].shape, (1, 1))
-