From f050f80feb7a131a1ef888f3748fd58dba71a3d7 Mon Sep 17 00:00:00 2001 From: "Dr. Kashif Rasul" Date: Sat, 4 Jan 2020 12:22:50 +0100 Subject: [PATCH] added multivariate grouper --- pts/dataset/__init__.py | 10 +- pts/dataset/common.py | 9 ++ pts/dataset/multivariate_grouper.py | 206 ++++++++++++++++++++++++++++ pts/evaluation/__init__.py | 2 +- test/model/test_deepvar.py | 118 ++++++++++++++++ 5 files changed, 343 insertions(+), 2 deletions(-) create mode 100644 pts/dataset/multivariate_grouper.py create mode 100644 test/model/test_deepvar.py diff --git a/pts/dataset/__init__.py b/pts/dataset/__init__.py index abbc308..208479c 100644 --- a/pts/dataset/__init__.py +++ b/pts/dataset/__init__.py @@ -1,4 +1,11 @@ -from .common import DataEntry, FieldName, Dataset, MetaData, TrainDatasets +from .common import ( + DataEntry, + FieldName, + Dataset, + MetaData, + TrainDatasets, + DateConstants, +) from .list_dataset import ListDataset from .file_dataset import FileDataset from .loader import TrainDataLoader, InferenceDataLoader @@ -21,3 +28,4 @@ from .artificial import ( default_synthetic, generate_sf2, ) +from .multivariate_grouper import MultivariateGrouper diff --git a/pts/dataset/common.py b/pts/dataset/common.py index 45ab4e5..9c1f10d 100644 --- a/pts/dataset/common.py +++ b/pts/dataset/common.py @@ -1,6 +1,7 @@ from abc import ABC, abstractmethod from typing import Any, Dict, Iterable, NamedTuple, Sized, List, Optional, Iterator +import pandas as pd from pydantic import BaseModel DataEntry = Dict[str, Any] @@ -76,3 +77,11 @@ class TrainDatasets(NamedTuple): metadata: MetaData train: Dataset test: Optional[Dataset] = None + +class DateConstants: + """ + Default constants for specific dates. + """ + + OLDEST_SUPPORTED_TIMESTAMP = pd.Timestamp(1800, 1, 1, 12) + LATEST_SUPPORTED_TIMESTAMP = pd.Timestamp(2200, 1, 1, 12) diff --git a/pts/dataset/multivariate_grouper.py b/pts/dataset/multivariate_grouper.py new file mode 100644 index 0000000..25b88e4 --- /dev/null +++ b/pts/dataset/multivariate_grouper.py @@ -0,0 +1,206 @@ +# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). +# You may not use this file except in compliance with the License. +# A copy of the License is located at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# or in the "license" file accompanying this file. This file is distributed +# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +# express or implied. See the License for the specific language governing +# permissions and limitations under the License. + +# Standard library imports +import logging +import numpy as np +import pandas as pd +from typing import Callable, Optional + +# First-party imports +from .common import DataEntry, Dataset, FieldName, DateConstants +from .list_dataset import ListDataset + +class MultivariateGrouper: + """ + The MultivariateGrouper takes a univariate dataset and groups it into a + single multivariate time series. Therefore, this class allows the user + to convert a univariate dataset into a multivariate dataset without making + a separate copy of the dataset. + + The Multivariate Grouper has two different modes: + + Training: For training data, the univariate time series get aligned to the + earliest time stamp in the dataset. Time series will be left and right + padded to produce an array of shape (dim, num_time_steps) + + Test: The test dataset might have multiple start dates (usually because + the test dataset mimics a rolling evaluation scenario). In this case, + the univariate dataset will be split into n multivariate time series, + where n is the number of evaluation dates. Again, the + time series will be grouped but only left padded. Note that the + padded value will influence the prediction if the context length is + longer than the length of the time series. + + Rules for padding for training and test datasets can be specified by the + user. + + Parameters + ---------- + max_target_dim + Set maximum dimensionality (for faster testing or when hitting + constraints of multivariate model). Takes the last max_target_dim + time series and groups them to multivariate time series. + num_test_dates + Number of test dates in the test set. This can be more than one if + the test set contains more than one forecast start date (often the + case in a rolling evaluation scenario). Must be set to convert test + data. + train_fill_rule + Implements the rule that fills missing data after alignment of the + time series for the training dataset. + test_fill_rule + Implements the rule that fills missing data after alignment of the + time series for the test dataset. + + """ + + def __init__( + self, + max_target_dim: Optional[int] = None, + num_test_dates: Optional[int] = None, + train_fill_rule: Callable = np.mean, + test_fill_rule: Callable = lambda x: 0.0, + ) -> None: + self.num_test_dates = num_test_dates + self.max_target_dimension = max_target_dim + self.train_fill_function = train_fill_rule + self.test_fill_rule = test_fill_rule + + self.first_timestamp = DateConstants.LATEST_SUPPORTED_TIMESTAMP + self.last_timestamp = DateConstants.OLDEST_SUPPORTED_TIMESTAMP + self.frequency = "" + + def __call__(self, dataset: Dataset) -> Dataset: + self._preprocess(dataset) + return self._group_all(dataset) + + def _preprocess(self, dataset: Dataset) -> None: + """ + The preprocess function iterates over the dataset to gather data that + is necessary for alignment. + This includes + 1) Storing first/last timestamp in the dataset + 2) Storing the frequency of the dataset + """ + for data in dataset: + timestamp = data[FieldName.START] + self.first_timestamp = min(self.first_timestamp, timestamp) + self.last_timestamp = max( + self.last_timestamp, + timestamp + (len(data[FieldName.TARGET]) - 1) * timestamp.freq, + ) + self.frequency = timestamp.freq + logging.info( + f"first/last timestamp found: " + f"{self.first_timestamp}/{self.last_timestamp}" + ) + + def _group_all(self, dataset: Dataset) -> Dataset: + if self.num_test_dates is None: + grouped_dataset = self._prepare_train_data(dataset) + else: + grouped_dataset = self._prepare_test_data(dataset) + return grouped_dataset + + def _prepare_train_data(self, dataset: Dataset) -> ListDataset: + logging.info("group training time-series to datasets") + + grouped_data = self._transform_target(self._align_data_entry, dataset) + grouped_data = self._restrict_max_dimensionality(grouped_data) + grouped_data[FieldName.START] = self.first_timestamp + grouped_data[FieldName.FEAT_STATIC_CAT] = [0] + + return ListDataset([grouped_data], freq=self.frequency, one_dim_target=False) + + def _prepare_test_data(self, dataset: Dataset) -> ListDataset: + logging.info("group test time-series to datasets") + + grouped_data = self._transform_target(self._left_pad_data, dataset) + # splits test dataset with rolling date into N R^d time series where + # N is the number of rolling evaluation dates + split_dataset = np.split(grouped_data[FieldName.TARGET], self.num_test_dates) + + all_entries = list() + for dataset_at_test_date in split_dataset: + grouped_data = dict() + grouped_data[FieldName.TARGET] = np.array( + list(dataset_at_test_date), dtype=np.float32 + ) + grouped_data = self._restrict_max_dimensionality(grouped_data) + grouped_data[FieldName.START] = self.first_timestamp + grouped_data[FieldName.FEAT_STATIC_CAT] = [0] + all_entries.append(grouped_data) + + return ListDataset(all_entries, freq=self.frequency, one_dim_target=False) + + def _align_data_entry(self, data: DataEntry) -> np.array: + ts = self.to_ts(data) + return ts.reindex( + pd.date_range( + start=self.first_timestamp, + end=self.last_timestamp, + freq=data[FieldName.START].freq, + ), + fill_value=self.train_fill_function(ts), + ).values + + def _left_pad_data(self, data: DataEntry) -> np.array: + ts = self.to_ts(data) + return ts.reindex( + pd.date_range( + start=self.first_timestamp, + end=ts.index[-1], + freq=data[FieldName.START].freq, + ), + fill_value=self.test_fill_rule(ts), + ).values + + @staticmethod + def _transform_target(funcs, dataset: Dataset) -> DataEntry: + return {FieldName.TARGET: np.array([funcs(data) for data in dataset])} + + def _restrict_max_dimensionality(self, data: DataEntry) -> DataEntry: + """ + Takes the last max_target_dimension dimensions from a multivariate + dataentry. + + Parameters + ---------- + data + multivariate data entry with (dim, num_timesteps) target field + + Returns + ------- + DataEntry + data multivariate data entry with + (max_target_dimension, num_timesteps) target field + """ + + if self.max_target_dimension is not None: + # restrict maximum dimensionality (for faster testing) + data[FieldName.TARGET] = data[FieldName.TARGET][ + -self.max_target_dimension :, : + ] + return data + + @staticmethod + def to_ts(data: DataEntry) -> pd.Series: + return pd.Series( + data[FieldName.TARGET], + index=pd.date_range( + start=data[FieldName.START], + periods=len(data[FieldName.TARGET]), + freq=data[FieldName.START].freq, + ), + ) diff --git a/pts/evaluation/__init__.py b/pts/evaluation/__init__.py index 045debb..42273d1 100644 --- a/pts/evaluation/__init__.py +++ b/pts/evaluation/__init__.py @@ -1,2 +1,2 @@ from .evaluator import Evaluator, MultivariateEvaluator -from .backtest import make_evaluation_predictions +from .backtest import make_evaluation_predictions, backtest_metrics diff --git a/test/model/test_deepvar.py b/test/model/test_deepvar.py new file mode 100644 index 0000000..4d0e7a5 --- /dev/null +++ b/test/model/test_deepvar.py @@ -0,0 +1,118 @@ +# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). +# You may not use this file except in compliance with the License. +# A copy of the License is located at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# or in the "license" file accompanying this file. This file is distributed +# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either +# express or implied. See the License for the specific language governing +# permissions and limitations under the License. + +# First-party imports +import pytest + +from pts.dataset.artificial import constant_dataset +from pts.modules import ( + # MultivariateGaussianOutput, + LowRankMultivariateNormalOutput, +) +from pts.evaluation import backtest_metrics +from pts.model.deepvar import DeepVAREstimator +from pts.dataset import TrainDatasets, MultivariateGrouper +from pts import Trainer +from pts.evaluation import MultivariateEvaluator + + +def load_multivariate_constant_dataset(): + dataset_info, train_ds, test_ds = constant_dataset() + grouper_train = MultivariateGrouper(max_target_dim=10) + grouper_test = MultivariateGrouper(num_test_dates=1, max_target_dim=10) + metadata = dataset_info.metadata + metadata.prediction_length = dataset_info.prediction_length + return TrainDatasets( + metadata=dataset_info.metadata, + train=grouper_train(train_ds), + test=grouper_test(test_ds), + ) + + +dataset = load_multivariate_constant_dataset() +target_dim = int(dataset.metadata.feat_static_cat[0].cardinality) +metadata = dataset.metadata +estimator = DeepVAREstimator + + +@pytest.mark.timeout(10) +@pytest.mark.parametrize( + "distr_output, num_batches_per_epoch, Estimator, " "use_marginal_transformation", + [ + ( + LowRankMultivariateNormalOutput(dim=target_dim, rank=2), + 10, + estimator, + True, + ), + ( + LowRankMultivariateNormalOutput(dim=target_dim, rank=2), + 10, + estimator, + False, + ), + ( + LowRankMultivariateNormalOutput(dim=target_dim, rank=2), + 10, + estimator, + False, + ), + (None, 10, estimator, True), + # ( + # MultivariateGaussianOutput(dim=target_dim), + # 10, + # estimator, + # True, + # ), + # ( + # MultivariateGaussianOutput(dim=target_dim), + # 10, + # estimator, + # True, + # ), + ], +) +def test_deepvar( + distr_output, num_batches_per_epoch, Estimator, use_marginal_transformation, +): + + estimator = Estimator( + input_size=10, + num_cells=20, + num_layers=1, + pick_incomplete=True, + target_dim=target_dim, + prediction_length=metadata.prediction_length, + # target_dim=target_dim, + freq=metadata.freq, + distr_output=distr_output, + scaling=False, + use_marginal_transformation=use_marginal_transformation, + trainer=Trainer( + epochs=1, + batch_size=8, + learning_rate=1e-10, + num_batches_per_epoch=num_batches_per_epoch, + ), + ) + + agg_metrics, _ = backtest_metrics( + train_dataset=dataset.train, + test_dataset=dataset.test, + forecaster=estimator, + evaluator=MultivariateEvaluator( + quantiles=(0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9) + ), + ) + + assert agg_metrics["ND"] < 1.5