in src/gluonts/model/rotbaum/_preprocess.py [0:0]
def preprocess_from_single_ts(self, time_series: Dict) -> Tuple:
"""
Takes a single time series, ts_list, and returns preprocessed data.
Note that the number of features is determined by the implementation
of make_features. The number of context windows is determined by
num_samples, see documentation under Parameters.
If stratify_targets is False, then the length of feature_data is:
(number of context windows) x (number of features)
And the length of target_data is:
(number of context windows) x (forecast_horizon)
If stratify_targets is False, then the length of feature_data is:
(number of context windows) * forecast_horizon x (number of features+1)
And the length of target_data is:
(number of context windows) * forecast_horizon x 1
Parameters
----------
time_series: dict
has 'target' and 'start' keys
Returns
-------
tuple
list of feature datapoints, list of target datapoints
"""
altered_time_series = time_series.copy()
if self.n_ignore_last > 0:
altered_time_series["target"] = altered_time_series["target"][
: -self.n_ignore_last
]
feature_data = []
target_data = []
max_num_context_windows = (
len(altered_time_series["target"])
- self.context_window_size
- self.forecast_horizon
+ 1
)
if max_num_context_windows < 1:
if not self.use_feat_static_real and not self.cardinality:
return [], []
else:
# will return featurized data containing no target
return (
self.make_features(
altered_time_series,
len(altered_time_series["target"]),
),
[],
)
if self.num_samples > 0:
locations = [
np.random.randint(max_num_context_windows)
for _ in range(self.num_samples)
]
else:
locations = range(max_num_context_windows)
for starting_index in locations:
if self.stratify_targets:
featurized_data = self.make_features(
altered_time_series, starting_index
)
for forecast_horizon_index in range(self.forecast_horizon):
feature_data.append(
list(featurized_data) + [forecast_horizon_index]
)
target_data.append(
[
time_series["target"][
starting_index
+ self.context_window_size
+ forecast_horizon_index
]
]
)
else:
featurized_data = self.make_features(
altered_time_series, starting_index
)
feature_data.append(featurized_data)
target_data.append(
time_series["target"][
starting_index
+ self.context_window_size : starting_index
+ self.context_window_size
+ self.forecast_horizon
]
)
return feature_data, target_data