Feat/strided dataset for torch and regression models

CHANGELOG.md

@@ -17,6 +17,7 @@ but cannot always guarantee backwards compatibility. Changes that may **break co
 - Added more resampling methods to `TimeSeries.resample()`. This allows to aggregate values when down-sampling and to fill or keep the holes when up-sampling. [#2654](https://github.com/unit8co/darts/pull/2654) by [Jonas Blanc](https://github.com/jonasblanc)
 - Added general function `darts.slice_intersect()` to intersect a sequence of `TimeSeries` along the time index. [#2592](https://github.com/unit8co/darts/pull/2592) by [Yoav Matzkevich](https://github.com/ymatzkevich).
 - Added new time aggregated metric `wmape()` (Weighted Mean Absolute Percentage Error). [#2544](https://github.com/unit8co/darts/pull/2648) by [He Weilin](https://github.com/cnhwl).
+- Added a `stride` argument to the `Dataset` classes and the `fit()/predict()` methods of the `RegressionModels` and torch-based models to reduce the size of the training set or apply elaborate training approaches. [#2624](https://github.com/unit8co/darts/pull/2529) by [Antoine Madrona](https://github.com/madtoinou)
 
 **Fixed**
 - Fixed a bug when performing optimized historical forecasts with `stride=1` using a `RegressionModel` with `output_chunk_shift>=1` and `output_chunk_length=1`, where the forecast time index was not properly shifted. [#2634](https://github.com/unit8co/darts/pull/2634) by [Mattias De Charleroy](https://github.com/MattiasDC).

darts/models/forecasting/global_baseline_models.py

@@ -256,6 +256,7 @@ def _build_train_dataset(
         future_covariates: Optional[Sequence[TimeSeries]],
         sample_weight: Optional[Sequence[TimeSeries]],
         max_samples_per_ts: Optional[int],
+        stride: int,
     ) -> MixedCovariatesTrainingDataset:
         return MixedCovariatesSequentialDataset(
             target_series=target,
@@ -264,6 +265,7 @@ def _build_train_dataset(
             input_chunk_length=self.input_chunk_length,
             output_chunk_length=0,
             output_chunk_shift=self.output_chunk_shift,
+            stride=stride,
             max_samples_per_ts=max_samples_per_ts,
             use_static_covariates=self.uses_static_covariates,
             sample_weight=sample_weight,

darts/models/forecasting/regression_model.py

@@ -600,6 +600,7 @@ def _create_lagged_data(
         future_covariates: Sequence[TimeSeries],
         max_samples_per_ts: int,
         sample_weight: Optional[Union[TimeSeries, str]] = None,
+        stride: int = 1,
         last_static_covariates_shape: Optional[tuple[int, int]] = None,
     ):
         (
@@ -624,6 +625,7 @@ def _create_lagged_data(
             check_inputs=False,
             concatenate=False,
             sample_weight=sample_weight,
+            stride=stride,
         )
 
         expected_nb_feat = (
@@ -675,6 +677,7 @@ def _fit_model(
         future_covariates: Sequence[TimeSeries],
         max_samples_per_ts: int,
         sample_weight: Optional[Union[Sequence[TimeSeries], str]],
+        stride: int,
         val_series: Optional[Sequence[TimeSeries]] = None,
         val_past_covariates: Optional[Sequence[TimeSeries]] = None,
         val_future_covariates: Optional[Sequence[TimeSeries]] = None,
@@ -692,6 +695,7 @@ def _fit_model(
             max_samples_per_ts=max_samples_per_ts,
             sample_weight=sample_weight,
             last_static_covariates_shape=None,
+            stride=stride,
         )
 
         if self.supports_val_set and val_series is not None:
@@ -741,6 +745,7 @@ def fit(
         max_samples_per_ts: Optional[int] = None,
         n_jobs_multioutput_wrapper: Optional[int] = None,
         sample_weight: Optional[Union[TimeSeries, Sequence[TimeSeries], str]] = None,
+        stride: int = 1,
         **kwargs,
     ):
         """
@@ -774,6 +779,10 @@ def fit(
             `"linear"` or `"exponential"` decay - the further in the past, the lower the weight. The weights are
             computed globally based on the length of the longest series in `series`. Then for each series, the weights
             are extracted from the end of the global weights. This gives a common time weighting across all series.
+        stride
+            The number of time steps between consecutive samples (windows of lagged values extracted from the target
+            series), applied starting from the end of the series. This should be used with caution as it might
+            introduce bias in the forecasts.
         **kwargs
             Additional keyword arguments passed to the `fit` method of the model.
         """
@@ -952,6 +961,7 @@ def fit(
             sample_weight=sample_weight,
             val_sample_weight=val_sample_weight,
             max_samples_per_ts=max_samples_per_ts,
+            stride=stride,
             **kwargs,
         )
         return self
@@ -993,11 +1003,11 @@ def predict(
             If set to `True`, the model predicts the parameters of its `likelihood` instead of the target. Only
             supported for probabilistic models with a likelihood, `num_samples = 1` and `n<=output_chunk_length`.
             Default: ``False``
+        show_warnings
+            Optionally, control whether warnings are shown. Not effective for all models.
         **kwargs : dict, optional
             Additional keyword arguments passed to the `predict` method of the model. Only works with
             univariate target series.
-        show_warnings
-            Optionally, control whether warnings are shown. Not effective for all models.
         """
         if series is None:
             # then there must be a single TS, and that was saved in super().fit as self.training_series

darts/models/forecasting/rnn_model.py

@@ -566,12 +566,14 @@ def _build_train_dataset(
         future_covariates: Optional[Sequence[TimeSeries]],
         sample_weight: Optional[Sequence[TimeSeries]],
         max_samples_per_ts: Optional[int],
+        stride: int,
     ) -> DualCovariatesShiftedDataset:
         return DualCovariatesShiftedDataset(
             target_series=target,
             covariates=future_covariates,
             length=self.training_length,
             shift=1,
+            stride=stride,
             max_samples_per_ts=max_samples_per_ts,
             use_static_covariates=self.uses_static_covariates,
             sample_weight=sample_weight,

darts/models/forecasting/tcn_model.py

@@ -538,12 +538,14 @@ def _build_train_dataset(
         future_covariates: Optional[Sequence[TimeSeries]],
         sample_weight: Optional[Sequence[TimeSeries]],
         max_samples_per_ts: Optional[int],
+        stride: int,
     ) -> PastCovariatesShiftedDataset:
         return PastCovariatesShiftedDataset(
             target_series=target,
             covariates=past_covariates,
             length=self.input_chunk_length,
             shift=self.output_chunk_length + self.output_chunk_shift,
+            stride=stride,
             max_samples_per_ts=max_samples_per_ts,
             use_static_covariates=self.uses_static_covariates,
             sample_weight=sample_weight,

darts/models/forecasting/tft_model.py

@@ -1162,6 +1162,7 @@ def _build_train_dataset(
         future_covariates: Optional[Sequence[TimeSeries]],
         sample_weight: Optional[Sequence[TimeSeries]],
         max_samples_per_ts: Optional[int],
+        stride: int,
     ) -> MixedCovariatesSequentialDataset:
         raise_if(
             future_covariates is None and not self.add_relative_index,
@@ -1179,6 +1180,7 @@ def _build_train_dataset(
             input_chunk_length=self.input_chunk_length,
             output_chunk_length=self.output_chunk_length,
             output_chunk_shift=self.output_chunk_shift,
+            stride=stride,
             max_samples_per_ts=max_samples_per_ts,
             use_static_covariates=self.uses_static_covariates,
             sample_weight=sample_weight,

darts/models/forecasting/torch_forecasting_model.py

@@ -567,6 +567,7 @@ def _build_train_dataset(
         future_covariates: Optional[Sequence[TimeSeries]],
         sample_weight: Optional[Union[Sequence[TimeSeries], str]],
         max_samples_per_ts: Optional[int],
+        stride: int,
     ) -> TrainingDataset:
         """
         Each model must specify the default training dataset to use.
@@ -659,6 +660,8 @@ def fit(
         val_sample_weight: Optional[
             Union[TimeSeries, Sequence[TimeSeries], str]
         ] = None,
+        stride: int = 1,
+        eval_stride: int = 1,
     ) -> "TorchForecastingModel":
         """Fit/train the model on one or multiple series.
 
@@ -732,7 +735,12 @@ def fit(
             are extracted from the end of the global weights. This gives a common time weighting across all series.
         val_sample_weight
             Same as for `sample_weight` but for the evaluation dataset.
-
+        stride
+            The number of time steps between consecutive samples (windows of lagged values extracted from the target
+            series), applied starting from the end of the series. This should be used with caution as it might
+            introduce bias in the forecasts.
+        eval_stride
+            Same as for `stride` but for the evaluation dataset.
         Returns
         -------
         self
@@ -750,10 +758,12 @@ def fit(
             past_covariates=past_covariates,
             future_covariates=future_covariates,
             sample_weight=sample_weight,
+            stride=stride,
             val_series=val_series,
             val_past_covariates=val_past_covariates,
             val_future_covariates=val_future_covariates,
             val_sample_weight=val_sample_weight,
+            eval_stride=eval_stride,
             trainer=trainer,
             verbose=verbose,
             epochs=epochs,
@@ -774,12 +784,14 @@ def _setup_for_fit_from_dataset(
         past_covariates: Optional[Union[TimeSeries, Sequence[TimeSeries]]] = None,
         future_covariates: Optional[Union[TimeSeries, Sequence[TimeSeries]]] = None,
         sample_weight: Optional[Union[TimeSeries, Sequence[TimeSeries], str]] = None,
+        stride: int = 1,
         val_series: Optional[Union[TimeSeries, Sequence[TimeSeries]]] = None,
         val_past_covariates: Optional[Union[TimeSeries, Sequence[TimeSeries]]] = None,
         val_future_covariates: Optional[Union[TimeSeries, Sequence[TimeSeries]]] = None,
         val_sample_weight: Optional[
             Union[TimeSeries, Sequence[TimeSeries], str]
         ] = None,
+        eval_stride: int = 1,
         trainer: Optional[pl.Trainer] = None,
         verbose: Optional[bool] = None,
         epochs: int = 0,
@@ -855,6 +867,7 @@ def _setup_for_fit_from_dataset(
             future_covariates=future_covariates,
             sample_weight=sample_weight,
             max_samples_per_ts=max_samples_per_ts,
+            stride=stride,
         )
 
         if val_series is not None:
@@ -864,6 +877,7 @@ def _setup_for_fit_from_dataset(
                 future_covariates=val_future_covariates,
                 sample_weight=val_sample_weight,
                 max_samples_per_ts=max_samples_per_ts,
+                stride=eval_stride,
             )
         else:
             val_dataset = None