Repository: KIT-IAI/pyWATTS

Issue:
[Online Learning] Modules has to specify how much past data they need.
Some modules need multiple value to calculate the current value. E.g. rolling_mean and sample_module. However, the current online/batch learning approach only provides the past batch as input, which might not be enough. 

## Solution:
* [ ] The modules need a method which returns the number of needed data.
* [ ] The step has to pay attention that enough data is fetched from the previous step.


see #116 

Discussion:

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3c9218b..fb7cd5b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,7 @@
 * Enable Multiple RefitConditions ([#192](https://github.com/KIT-IAI/pyWATTS/issues/192))
 * Periodic Refit Condition ([#192](https://github.com/KIT-IAI/pyWATTS/issues/192))
 * Add AnomalyGeneration modules ([#195](https://github.com/KIT-IAI/pyWATTS/issues/195))
+* Add get_min_data method in modules to enable them to specify how much data they need. ([#147](https://github.com/KIT-IAI/pyWATTS/issues/147))
 
 ### Changed
 * Retraining is triggered after all steps are transformed ([#117](https://github.com/KIT-IAI/pyWATTS/issues/117))
diff --git a/pywatts/core/base.py b/pywatts/core/base.py
index f4c8757..4a7da00 100644
--- a/pywatts/core/base.py
+++ b/pywatts/core/base.py
@@ -137,6 +137,9 @@ class Base(ABC):
         """
         return self.fit(**kwargs)
 
+    def get_min_data(self):
+        return 0
+
     def __call__(self,
                  use_inverse_transform: bool = False,
                  use_prob_transform: bool = False,
diff --git a/pywatts/core/base_step.py b/pywatts/core/base_step.py
index 072bc99..cc064d4 100644
--- a/pywatts/core/base_step.py
+++ b/pywatts/core/base_step.py
@@ -36,7 +36,6 @@ class BaseStep(ABC):
         self.input_steps: Dict[str, "BaseStep"] = dict() if input_steps is None else input_steps
         self.targets: Dict[str, "BaseStep"] = dict() if targets is None else targets
         self.condition = condition
-        self.cached_result = {"cached": None, "start": None, "end": None}
 
         self.name = name
 
@@ -49,7 +48,7 @@ class BaseStep(ABC):
         self.transform_time = SummaryObjectList(self.name + " Transform Time", category=SummaryCategory.TransformTime)
 
     def get_result(self, start: pd.Timestamp, end: Optional[pd.Timestamp], buffer_element: str = None,
-                   return_all=False):
+                   return_all=False, minimum_data=(0, pd.Timedelta(0))):
         """
         This method is responsible for providing the result of this step.
         Therefore,
@@ -74,9 +73,7 @@ class BaseStep(ABC):
         # Only execute the module if the step is not finished and the results are not yet calculated
         if not self.finished and not (end is not None and self._current_end is not None and end <= self._current_end):
             if not self.buffer or not self._current_end or end > self._current_end:
-                self.cached_result["cached"] = self._compute(start, end)
-                self.cached_result["start"] = start
-                self.cached_result["end"] = end
+                self._compute(start, end, minimum_data)
                 self._current_end = end
             if not end:
                 self.finished = True
@@ -87,15 +84,9 @@ class BaseStep(ABC):
             if self.finished:
                 self._callbacks()
 
-        # Check if the cached results fits to the request, if yes return it.
-        if self.cached_result["cached"] is not None and self.cached_result["start"] == start and self.cached_result[
-            "end"] == end:
-            return copy.deepcopy(self.cached_result["cached"]) if return_all else copy.deepcopy(self.cached_result["cached"][
-                    buffer_element]) if buffer_element is not None else copy.deepcopy(list(self.cached_result["cached"].values())[
-                    0])
-        return self._pack_data(start, end, buffer_element, return_all=return_all)
+        return self._pack_data(start, end, buffer_element, return_all=return_all, minimum_data=minimum_data)
 
-    def _compute(self, start, end) -> Dict[str, xr.DataArray]:
+    def _compute(self, start, end, minimum_data) -> Dict[str, xr.DataArray]:
         pass
 
     def further_elements(self, counter: pd.Timestamp) -> bool:
@@ -118,12 +109,16 @@ class BaseStep(ABC):
                 return False
         return True
 
-    def _pack_data(self, start, end, buffer_element=None, return_all=False):
+    def _pack_data(self, start, end, buffer_element=None, return_all=False, minimum_data=(0, pd.Timedelta(0))):
         # Provide requested data
         time_index = _get_time_indexes(self.buffer)
         if start:
             index = list(self.buffer.values())[0].indexes[time_index[0]]
-            start = start.to_numpy()
+            if len(index) > 1:
+                freq = index[1] - index[0]
+            else:
+                freq = 0
+            start = start.to_numpy() - pd.Timedelta(minimum_data[0] * freq) - minimum_data[1]
             # 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.
@@ -164,7 +159,8 @@ class BaseStep(ABC):
             # Time dimension is mandatory, consequently there dim has to exist
             dim = _get_time_indexes(result)[0]
             for key in self.buffer.keys():
-                self.buffer[key] = xr.concat([self.buffer[key], result[key]], dim=dim)
+                last = self.buffer[key][dim].values[-1]
+                self.buffer[key] = xr.concat([self.buffer[key], result[key][result[key][dim] > last]], dim=dim)
         return result
 
     def get_json(self, fm: FileManager) -> Dict:
@@ -201,10 +197,10 @@ class BaseStep(ABC):
         :return: The restored step.
         """
 
-    def _get_input(self, start, batch):
+    def _get_input(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
         return None
 
-    def _get_target(self, start, batch):
+    def _get_target(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
         return None
 
     def _should_stop(self, start, end) -> bool:
diff --git a/pywatts/core/either_or_step.py b/pywatts/core/either_or_step.py
index c7e0c6e..8708dec 100644
--- a/pywatts/core/either_or_step.py
+++ b/pywatts/core/either_or_step.py
@@ -1,3 +1,4 @@
+import pandas as pd
 from pywatts.core.base_step import BaseStep
 
 
@@ -14,14 +15,14 @@ class EitherOrStep(BaseStep):
         super().__init__(input_steps)
         self.name = "EitherOr"
 
-    def _compute(self, start, end):
-        input_data = self._get_input(start, end)
+    def _compute(self, start, end, minimum_data):
+        input_data = self._get_input(start, end, minimum_data)
         return self._transform(input_data)
 
-    def _get_input(self, start, batch):
+    def _get_input(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
         inputs = []
         for step in self.input_steps.values():
-            inp = step.get_result(start, batch)
+            inp = step.get_result(start, batch, minimum_data=minimum_data)
             inputs.append(inp)
         return inputs
 
diff --git a/pywatts/core/result_step.py b/pywatts/core/result_step.py
index 84c5a18..58576ca 100644
--- a/pywatts/core/result_step.py
+++ b/pywatts/core/result_step.py
@@ -15,14 +15,16 @@ class ResultStep(BaseStep):
         super().__init__(input_steps=input_steps)
         self.buffer_element = buffer_element
 
-    def get_result(self, start: pd.Timestamp, end: Optional[pd.Timestamp], *args, return_all=False, **kwargs):
+    def get_result(self, start: pd.Timestamp, end: Optional[pd.Timestamp], buffer_element: str = None,
+                   return_all=False, minimum_data=(0, pd.Timedelta(0))):
         """
         Returns the specified result of the previous step.
         """
+
         if not return_all:
-            return list(self.input_steps.values())[0].get_result(start, end, self.buffer_element)
+            return list(self.input_steps.values())[0].get_result(start, end, self.buffer_element, minimum_data=minimum_data)
         else:
-            return {self.buffer_element: list(self.input_steps.values())[0].get_result(start, end, self.buffer_element)}
+            return {self.buffer_element: list(self.input_steps.values())[0].get_result(start, end, self.buffer_element, minimum_data=minimum_data)}
 
     def get_json(self, fm: FileManager) -> Dict:
         """
diff --git a/pywatts/core/step.py b/pywatts/core/step.py
index cd89a8a..7d32b19 100644
--- a/pywatts/core/step.py
+++ b/pywatts/core/step.py
@@ -4,6 +4,7 @@ import warnings
 from typing import Optional, Dict, Union, Callable, List
 
 import cloudpickle
+import numpy as np
 import pandas as pd
 import xarray as xr
 
@@ -144,15 +145,15 @@ class Step(BaseStep):
 
         return step
 
-    def _compute(self, start, end):
-        input_data = self._get_input(start, end)
-        target = self._get_target(start, end)
+    def _compute(self, start, end, minimum_data):
+        input_data = self._get_input(start, end, minimum_data)
+        target = self._get_target(start, end, minimum_data)
         if self.current_run_setting.computation_mode in [ComputationMode.Default, ComputationMode.FitTransform,
                                                          ComputationMode.Train]:
             # Fetch input_data and target data
             if self.batch_size:
-                input_batch = self._get_input(end - self.batch_size, end)
-                target_batch = self._get_target(end - self.batch_size, end)
+                input_batch = self._get_input(end - self.batch_size, end, minimum_data)
+                target_batch = self._get_target(end - self.batch_size, end, minimum_data)
                 start_time = time.time()
                 self._fit(input_batch, target_batch)
                 self.training_time.set_kv("", time.time() - start_time)
@@ -173,14 +174,26 @@ class Step(BaseStep):
             result_dict[key] = res.sel(**{_get_time_indexes(res)[0]: index[(index >= start)]})
         return result_dict
 
-    def _get_target(self, start, batch):
+    def _get_target(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
+        min_data_module = self.module.get_min_data()
+        if isinstance(min_data_module, (int, np.integer)):
+            minimum_data = minimum_data[0] + min_data_module, minimum_data[1]
+        else:
+            minimum_data = minimum_data[0], minimum_data[1] + min_data_module
         return {
-            key: target.get_result(start, batch) for key, target in self.targets.items()
+            key: target.get_result(start, batch, minimum_data=minimum_data)
+            for key, target in self.targets.items()
         }
 
-    def _get_input(self, start, batch):
+    def _get_input(self, start, batch, minimum_data=(0, pd.Timedelta(0))):
+        min_data_module = self.module.get_min_data()
+        if isinstance(min_data_module, (int, np.integer)):
+            minimum_data = minimum_data[0] + min_data_module, minimum_data[1]
+        else:
+            minimum_data = minimum_data[0], minimum_data[1] + min_data_module
         return {
-            key: input_step.get_result(start, batch) for key, input_step in self.input_steps.items()
+            key: input_step.get_result(start, batch, minimum_data=minimum_data) for
+            key, input_step in self.input_steps.items()
         }
 
     def get_json(self, fm: FileManager):
diff --git a/pywatts/modules/feature_extraction/rolling_base.py b/pywatts/modules/feature_extraction/rolling_base.py
index 56e841f..54573fb 100644
--- a/pywatts/modules/feature_extraction/rolling_base.py
+++ b/pywatts/modules/feature_extraction/rolling_base.py
@@ -2,6 +2,7 @@ from abc import ABC, abstractmethod
 from enum import IntEnum
 from typing import Dict, Optional
 
+import pandas as pd
 import xarray as xr
 
 from pywatts.core.base import BaseTransformer
@@ -129,3 +130,6 @@ class RollingBase(BaseTransformer, ABC):
     @abstractmethod
     def _get_rolling(self, df):
         pass
+
+    def get_min_data(self):
+        return pd.Timedelta(f"{self.window_size}{self.window_size_unit}")
diff --git a/pywatts/modules/feature_extraction/trend_extraction.py b/pywatts/modules/feature_extraction/trend_extraction.py
index 78b9f95..d4f9d3b 100644
--- a/pywatts/modules/feature_extraction/trend_extraction.py
+++ b/pywatts/modules/feature_extraction/trend_extraction.py
@@ -83,3 +83,6 @@ class TrendExtraction(BaseTransformer):
                   range(1, self.length + 1)]
         trend = xr.DataArray(np.stack(trends, axis=-1), dims=(*x.dims, "length"), coords=x.coords)
         return trend.transpose(_get_time_indexes(x)[0], "length", ...)
+
+    def get_min_data(self):
+        return self.period * (self.length + 1)
diff --git a/pywatts/modules/preprocessing/clock_shift.py b/pywatts/modules/preprocessing/clock_shift.py
index 4ed6f7b..667059b 100644
--- a/pywatts/modules/preprocessing/clock_shift.py
+++ b/pywatts/modules/preprocessing/clock_shift.py
@@ -1,5 +1,7 @@
 from typing import Dict, List
 
+import numpy as np
+import pandas as pd
 import xarray as xr
 
 from pywatts.core.base import BaseTransformer
@@ -75,3 +77,6 @@ class ClockShift(BaseTransformer):
                 f"Not all indexes ({indexes}) are in the indexes of x ({list(x.indexes.keys())}).",
                 "Perhaps you set the wrong indexes with set_params or during the initialization of the ClockShift.",
                 module=self.name) from exc
+
+    def get_min_data(self):
+        return np.abs(self.lag)
diff --git a/pywatts/modules/preprocessing/resampler.py b/pywatts/modules/preprocessing/resampler.py
index 780496c..bceeea1 100644
--- a/pywatts/modules/preprocessing/resampler.py
+++ b/pywatts/modules/preprocessing/resampler.py
@@ -1,5 +1,6 @@
 from typing import Dict, Optional, Any
 
+import pandas as pd
 import xarray as xr
 
 from pywatts.core.base import BaseTransformer
@@ -108,3 +109,6 @@ class Resampler(BaseTransformer):
         return getattr(
             x.resample(**{self.time_index: self.target_time}), self.method
         )(**args)
+
+    def get_min_data(self):
+        return pd.Timedelta(self.target_time)
diff --git a/pywatts/modules/preprocessing/sampler.py b/pywatts/modules/preprocessing/sampler.py
index 8bad874..572f72e 100644
--- a/pywatts/modules/preprocessing/sampler.py
+++ b/pywatts/modules/preprocessing/sampler.py
@@ -1,5 +1,6 @@
 from typing import Dict, List
 
+import pandas as pd
 import xarray as xr
 
 from pywatts.core.base import BaseTransformer
@@ -33,6 +34,9 @@ class Sampler(BaseTransformer):
         self.sample_size = sample_size
         self.indexes = indexes
 
+    def get_min_data(self):
+        return self.sample_size
+
     def get_params(self) -> Dict[str, object]:
         """
         Returns a list of all hyperparameters/ user defined parameters


Test patch:
diff --git a/tests/unit/core/test_step.py b/tests/unit/core/test_step.py
index 8453a97..d9de8f1 100644
--- a/tests/unit/core/test_step.py
+++ b/tests/unit/core/test_step.py
@@ -112,6 +112,7 @@ class TestStep(unittest.TestCase):
 
         self.module_mock.transform.return_value = xr.DataArray([2, 3, 4, 3, 3, 1, 2], dims=["time"],
                                                                coords={'time': time2})
+        self.module_mock.get_min_data.return_value = 2
         step = Step(self.module_mock, {"x": input_step}, file_manager=MagicMock())
         da = xr.DataArray([2, 3, 4, 3, 3, 1, 2], dims=["time"], coords={'time': time})
         step.buffer = {"test": da}
@@ -122,8 +123,10 @@ class TestStep(unittest.TestCase):
 
         # Two calls, once in should_stop and once in _transform
         input_step.get_result.assert_has_calls(
-            [call(pd.Timestamp('2000-01-07 00:00:00'), pd.Timestamp('2020-01-14 00:00:00')),
-             call(pd.Timestamp('2000-01-07 00:00:00'), pd.Timestamp('2020-01-14 00:00:00'))])
+            [call(pd.Timestamp('2000-01-07 00:00:00'), pd.Timestamp('2020-01-14 00:00:00'),
+                  minimum_data=(2, pd.Timedelta(0))),
+             call(pd.Timestamp('2000-01-07 00:00:00'), pd.Timestamp('2020-01-14 00:00:00'),
+                  minimum_data=(2, pd.Timedelta(0)))])
         xr_mock.concat.assert_called_once()
 
         xr.testing.assert_equal(da, list(xr_mock.concat.call_args_list[0])[0][0][0])
@@ -147,8 +150,10 @@ class TestStep(unittest.TestCase):
 
         # Two calls, once in should_stop and once in _transform
         input_step.get_result.assert_has_calls(
-            [call(pd.Timestamp('2000-01-01 '), pd.Timestamp('2020-12-12 ')),
-             call(pd.Timestamp('2000-01-01 '), pd.Timestamp('2020-12-12 '))])
+            [call(pd.Timestamp("2000-01-01"), pd.Timestamp('2020-12-12 '),
+                  minimum_data=(0, self.module_mock.get_min_data().__radd__())),
+             call(pd.Timestamp('2000-01-01 '), pd.Timestamp('2020-12-12 '),
+                  minimum_data=(0, self.module_mock.get_min_data().__radd__()))])
 
         self.module_mock.transform.assert_called_once_with(x=input_step_result_mock)
 
@@ -271,27 +276,33 @@ class TestStep(unittest.TestCase):
         assert step._should_stop(None, None) == False
         assert step.finished == False
 
+
+    @patch('pywatts.core.step.Step._get_target', return_value={"target": 1})
+    @patch('pywatts.core.step.Step._get_input', return_value={"x": 2})
     @patch('pywatts.core.step.isinstance', side_effect=[True, False, True])
-    def test_refit_refit_conditions_false(self, isinstance_mock):
+    def test_refit_refit_conditions_false(self, isinstance_mock, get_input_mock, get_target_mock):
         step = Step(self.module_mock, {"x": self.step_mock}, file_manager=None, refit_conditions=[lambda x, y: False],
                     computation_mode=ComputationMode.Refit)
         step.refit(pd.Timestamp("2000.01.01"), pd.Timestamp("2020.01.01"))
         self.module_mock.refit.assert_not_called()
 
+    @patch('pywatts.core.step.Step._get_target', return_value={"target": 1})
+    @patch('pywatts.core.step.Step._get_input', return_value={"x": 2})
     @patch('pywatts.core.step.isinstance', side_effect=[True, False, True])
-    def test_refit_refit_conditions_true(self, isinstance_mock):
+    def test_refit_refit_conditions_true(self, isinstance_mock, get_input_mock, get_target_mock):
         step = Step(self.module_mock, {"x": self.step_mock},
                     targets={"target": self.step_mock}, file_manager=None, refit_conditions=[lambda x, y: True],
                     computation_mode=ComputationMode.Refit)
         step.refit(pd.Timestamp("2000.01.01"), pd.Timestamp("2020.01.01"))
-        self.module_mock.refit.assert_called_once_with(x=self.step_mock.get_result(),
-                                                       target=self.step_mock.get_result())
+        self.module_mock.refit.assert_called_once_with(x=2, target=1)
 
+    @patch('pywatts.core.step.Step._get_target', return_value={"target": 1})
+    @patch('pywatts.core.step.Step._get_input', return_value={"x": 2})
     @patch('pywatts.core.step.isinstance', side_effect=[True, False, True, False, True])
-    def test_multiple_refit_conditions(self, isinstance_mock):
+    def test_multiple_refit_conditions(self, isinstance_mock, get_input_mock, get_target_mock):
         step = Step(self.module_mock, {"x": self.step_mock},
                     targets={"target": self.step_mock}, file_manager=None, refit_conditions=[lambda x, y: False, lambda x, y: True],
                     computation_mode=ComputationMode.Refit)
         step.refit(pd.Timestamp("2000.01.01"), pd.Timestamp("2020.01.01"))
-        self.module_mock.refit.assert_called_once_with(x=self.step_mock.get_result(),
-                                                       target=self.step_mock.get_result())
+        self.module_mock.refit.assert_called_once_with(x=2, target=1)
+
diff --git a/tests/unit/core/test_summary_step.py b/tests/unit/core/test_summary_step.py
index 644324a..0bd495c 100644
--- a/tests/unit/core/test_summary_step.py
+++ b/tests/unit/core/test_summary_step.py
@@ -30,9 +30,10 @@ class TestSummaryStep(unittest.TestCase):
         self.fm_mock = None
 
     def test_get_summary(self):
+        self.module_mock.get_min_data.return_value = pd.Timedelta(hours=1)
         result = self.summary.get_summary(None, None)
 
-        self.step_mock.get_result.assert_called_once_with(None, None)
+        self.step_mock.get_result.assert_called_once_with(None, None, minimum_data=(0, pd.Timedelta(hours=1)))
         self.module_mock.transform.assert_called_once_with(file_manager=self.fm_mock, input=self.result_mock)
         self.assertEqual(result, "#I AM MARKDOWN")
 
diff --git a/tests/unit/modules/test_clock_shift.py b/tests/unit/modules/test_clock_shift.py
index b256bfa..bb815a8 100644
--- a/tests/unit/modules/test_clock_shift.py
+++ b/tests/unit/modules/test_clock_shift.py
@@ -60,3 +60,7 @@ class TestClockShift(unittest.TestCase):
         self.assertEqual(context.exception.message,
                          "Not all indexes (['FOO']) are in the indexes of x (['time']). "
                          "Perhaps you set the wrong indexes with set_params or during the initialization of the ClockShift.")
+
+    def test_get_min_data(self):
+        clock_shift = ClockShift(lag=24)
+        self.assertEqual(clock_shift.get_min_data(), 24)
diff --git a/tests/unit/modules/test_resampler.py b/tests/unit/modules/test_resampler.py
index fd8a935..f88fb75 100644
--- a/tests/unit/modules/test_resampler.py
+++ b/tests/unit/modules/test_resampler.py
@@ -88,3 +88,7 @@ class TestResampler(unittest.TestCase):
         expected_data = np.arange(0, self.times - 0.5, step=0.5)
         expected_time = pd.date_range(self.start_time, freq="1800s", periods=len(expected_data))
         self._test_base("linear", "1800s", "interpolate", expected_time, expected_data, method_args={"kind": "linear"})
+
+    def test_get_min_data(self):
+        resampler = Resampler(target_time="12h")
+        self.assertEqual(resampler.get_min_data(), pd.Timedelta(hours=12))
diff --git a/tests/unit/modules/test_rolling_mean.py b/tests/unit/modules/test_rolling_mean.py
index 05cd5a6..36ebc1c 100644
--- a/tests/unit/modules/test_rolling_mean.py
+++ b/tests/unit/modules/test_rolling_mean.py
@@ -83,3 +83,7 @@ class TestRollingMean(unittest.TestCase):
         expected_result = xr.DataArray([0., 0., 0., 0., 3, 3, 3.5, 3.5, 2, 2., 6.,6,4.5, 4.5], dims=["time"], coords={'time': time})
 
         xr.testing.assert_equal(result, expected_result)
+
+    def test_get_min_data(self):
+        r_mean = RollingMean(window_size=12, window_size_unit="h")
+        self.assertEqual(r_mean.get_min_data(), pd.Timedelta(hours=12))
diff --git a/tests/unit/modules/test_sampler.py b/tests/unit/modules/test_sampler.py
index fdac408..65f11ff 100644
--- a/tests/unit/modules/test_sampler.py
+++ b/tests/unit/modules/test_sampler.py
@@ -67,3 +67,7 @@ class TestSampler(unittest.TestCase):
         self.assertEqual(context.exception.message,
                          "Sample size cannot be less than or equal to zero. "
                          "Please define a sample size greater than zero.")
+
+    def test_get_min_data(self):
+        sampler = Sampler(sample_size=24)
+        self.assertEqual(sampler.get_min_data(), 24)
diff --git a/tests/unit/modules/test_trend_extraction.py b/tests/unit/modules/test_trend_extraction.py
index 4bcb85c..113f74b 100644
--- a/tests/unit/modules/test_trend_extraction.py
+++ b/tests/unit/modules/test_trend_extraction.py
@@ -49,3 +49,7 @@ class TestTrendExtraction(unittest.TestCase):
              [9, 7, 5]], dims=["time", "length"], coords={'time': time})
 
         xr.testing.assert_equal(result, expected_result)
+
+    def test_get_min_data(self):
+        trend = TrendExtraction(period=12, length=2)
+        self.assertEqual(trend.get_min_data(), 36)