mirror of
https://github.com/wassname/pytorch-ts.git
synced 2026-07-16 11:21:03 +08:00
added multivariate grouper
This commit is contained in:
Dr. Kashif Rasul
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
),
|
||||
)
|
||||
@@ -1,2 +1,2 @@
|
||||
from .evaluator import Evaluator, MultivariateEvaluator
|
||||
from .backtest import make_evaluation_predictions
|
||||
from .backtest import make_evaluation_predictions, backtest_metrics
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user