diff --git a/pts/dataset/repository/_m4.py b/pts/dataset/repository/_m4.py index d24dd94..bda0250 100644 --- a/pts/dataset/repository/_m4.py +++ b/pts/dataset/repository/_m4.py @@ -78,3 +78,4 @@ def generate_m4_dataset( for cat, target in enumerate(test_target_values) ], ) + \ No newline at end of file diff --git a/pts/model/simple_feedforward/__init__.py b/pts/model/simple_feedforward/__init__.py new file mode 100644 index 0000000..27dc9a9 --- /dev/null +++ b/pts/model/simple_feedforward/__init__.py @@ -0,0 +1,6 @@ +from .simple_feedforward_estimator import SimpleFeedForwardEstimator +from .simple_feedforward_network import ( + SimpleFeedForwardTrainingNetwork, + SimpleFeedForwardPredictionNetwork, +) + diff --git a/pts/model/simple_feedforward/simple_feedforward_estimator.py b/pts/model/simple_feedforward/simple_feedforward_estimator.py new file mode 100644 index 0000000..06202b4 --- /dev/null +++ b/pts/model/simple_feedforward/simple_feedforward_estimator.py @@ -0,0 +1,176 @@ +from typing import List, Optional + +import torch +import torch.nn as nn + +from pts import Trainer +from pts.model import PTSEstimator, PTSPredictor, copy_parameters +from pts.modules import DistributionOutput, StudentTOutput +from pts.dataset import FieldName +from pts.transform import ( + Transformation, + Chain, + InstanceSplitter, + ExpectedNumInstanceSampler, +) + +from .simple_feedforward_network import ( + SimpleFeedForwardTrainingNetwork, + SimpleFeedForwardPredictionNetwork, +) + + +class SimpleFeedForwardEstimator(PTSEstimator): + """ + SimpleFeedForwardEstimator shows how to build a simple MLP model predicting + the next target time-steps given the previous ones. + + Given that we want to define a pytorch model trainable by SGD, we inherit the + parent class `PTSEstimator` that handles most of the logic for fitting a + neural-network. + + We thus only have to define: + + 1. How the data is transformed before being fed to our model:: + + def create_transformation(self) -> Transformation + + 2. How the training happens:: + + def create_training_network(self) -> nn.Module + + 3. how the predictions can be made for a batch given a trained network:: + + def create_predictor( + self, + transformation: Transformation, + trained_net: nn.Module, + ) -> Predictor + + + Parameters + ---------- + freq + Time time granularity of the data + prediction_length + Length of the prediction horizon + trainer + Trainer object to be used (default: Trainer()) + num_hidden_dimensions + Number of hidden nodes in each layer (default: [40, 40]) + context_length + Number of time units that condition the predictions + (default: None, in which case context_length = prediction_length) + distr_output + Distribution to fit (default: StudentTOutput()) + batch_normalization + Whether to use batch normalization (default: False) + mean_scaling + Scale the network input by the data mean and the network output by + its inverse (default: True) + num_parallel_samples + Number of evaluation samples per time series to increase parallelism during inference. + This is a model optimization that does not affect the accuracy (default: 100) + """ + + def __init__( + self, + freq: str, + prediction_length: int, + input_size: int, + trainer: Trainer = Trainer(), + num_hidden_dimensions: Optional[List[int]] = None, + context_length: Optional[int] = None, + distr_output: DistributionOutput = StudentTOutput(), + batch_normalization: bool = False, + mean_scaling: bool = True, + num_parallel_samples: int = 100, + ) -> None: + """ + Defines an estimator. All parameters should be serializable. + """ + super().__init__(trainer=trainer) + + self.num_hidden_dimensions = ( + num_hidden_dimensions + if num_hidden_dimensions is not None + else list([40, 40]) + ) + self.input_size = input_size + self.prediction_length = prediction_length + self.context_length = ( + context_length if context_length is not None else prediction_length + ) + self.freq = freq + self.distr_output = distr_output + self.batch_normalization = batch_normalization + self.mean_scaling = mean_scaling + self.num_parallel_samples = num_parallel_samples + + # here we do only a simple operation to convert the input data to a form + # that can be digested by our model by only splitting the target in two, a + # conditioning part and a to-predict part, for each training example. + # For a more complex transformation example, see the `pts.model.deepar` + # transformation that includes time features, age feature, observed values + # indicator, etc. + def create_transformation(self) -> Transformation: + return Chain( + [ + InstanceSplitter( + target_field=FieldName.TARGET, + is_pad_field=FieldName.IS_PAD, + start_field=FieldName.START, + forecast_start_field=FieldName.FORECAST_START, + train_sampler=ExpectedNumInstanceSampler(num_instances=1), + past_length=self.context_length, + future_length=self.prediction_length, + time_series_fields=[], # [FieldName.FEAT_DYNAMIC_REAL] + ) + ] + ) + + # defines the network, we get to see one batch to initialize it. + # the network should return at least one tensor that is used as a loss to minimize in the training loop. + # several tensors can be returned for instance for analysis, see DeepARTrainingNetwork for an example. + def create_training_network( + self, device: torch.device + ) -> SimpleFeedForwardTrainingNetwork: + return SimpleFeedForwardTrainingNetwork( + input_size=self.input_size, + num_hidden_dimensions=self.num_hidden_dimensions, + prediction_length=self.prediction_length, + context_length=self.context_length, + distr_output=self.distr_output, + batch_normalization=self.batch_normalization, + mean_scaling=self.mean_scaling, + ).to(device) + + # we now define how the prediction happens given that we are provided a + # training network. + def create_predictor( + self, + transformation: Transformation, + trained_network: nn.Module, + device: torch.device, + ) -> PTSPredictor: + prediction_network = SimpleFeedForwardPredictionNetwork( + input_size=self.input_size, + num_hidden_dimensions=self.num_hidden_dimensions, + prediction_length=self.prediction_length, + context_length=self.context_length, + distr_output=self.distr_output, + batch_normalization=self.batch_normalization, + mean_scaling=self.mean_scaling, + num_parallel_samples=self.num_parallel_samples, + ) + + copy_parameters(trained_network, prediction_network) + + return PTSPredictor( + input_transform=transformation, + prediction_net=prediction_network, + batch_size=self.trainer.batch_size, + freq=self.freq, + prediction_length=self.prediction_length, + device=device, + ) diff --git a/pts/model/simple_feedforward/simple_feedforward_network.py b/pts/model/simple_feedforward/simple_feedforward_network.py new file mode 100644 index 0000000..71b365f --- /dev/null +++ b/pts/model/simple_feedforward/simple_feedforward_network.py @@ -0,0 +1,113 @@ +from typing import List + +import torch +import torch.nn as nn +import torch.nn.functional as F +from torch.distributions import Distribution + +from pts.modules import MeanScaler, NOPScaler, DistributionOutput, LambdaLayer + + +class SimpleFeedForwardNetworkBase(nn.Module): + """ + Abstract base class to implement feed-forward networks for probabilistic + time series prediction. + + This class does not implement hybrid_forward: this is delegated + to the two subclasses SimpleFeedForwardTrainingNetwork and + SimpleFeedForwardPredictionNetwork, that define respectively how to + compute the loss and how to generate predictions. + + Parameters + ---------- + num_hidden_dimensions + Number of hidden nodes in each layer. + prediction_length + Number of time units to predict. + context_length + Number of time units that condition the predictions. + batch_normalization + Whether to use batch normalization. + mean_scaling + Scale the network input by the data mean and the network output by + its inverse. + distr_output + Distribution to fit. + kwargs + """ + + def __init__( + self, + input_size: int, + num_hidden_dimensions: List[int], + prediction_length: int, + context_length: int, + batch_normalization: bool, + mean_scaling: bool, + distr_output: DistributionOutput, + ) -> None: + super().__init__() + + self.input_size = input_size + self.num_hidden_dimensions = num_hidden_dimensions + self.prediction_length = prediction_length + self.context_length = context_length + self.batch_normalization = batch_normalization + self.mean_scaling = mean_scaling + self.distr_output = distr_output + + modules = [] + dims = self.num_hidden_dimensions + # for i, dim in enumerate(dims[:-1]): + # modules.append(nn.Linear(dims[i], dim)) + # modules.append(nn.ReLU()) + # if self.batch_normalization: + # modules.append(nn.BatchNorm1d(dim)) + modules.append(nn.Linear(100, dims[-1] * prediction_length)) + modules.append( + LambdaLayer(lambda o: torch.reshape(o, (-1, prediction_length, dims[-1]))) + ) + self.mlp = nn.Sequential(*modules) + + self.distr_args_proj = self.distr_output.get_args_proj(dims[-1]) + + self.scaler = MeanScaler() if mean_scaling else NOPScaler() + + def get_distr(self, past_target: torch.Tensor) -> Distribution: + # (batch_size, seq_len, target_dim) and (batch_size, seq_len, target_dim) + scaled_target, target_scale = self.scaler( + past_target, + torch.ones_like(past_target), # TODO: pass the actual observed here + ) + + mlp_outputs = self.mlp(scaled_target) + distr_args = self.distr_args_proj(mlp_outputs) + return self.distr_output.distribution( + distr_args, scale=target_scale.unsqueeze(1) + ) + + +class SimpleFeedForwardTrainingNetwork(SimpleFeedForwardNetworkBase): + def forward( + self, past_target: torch.Tensor, future_target: torch.Tensor + ) -> torch.Tensor: + distr = self.get_distr(past_target) + + # (batch_size, prediction_length, target_dim) + loss = -distr.log_prob(future_target) + + return loss.mean() + + +class SimpleFeedForwardPredictionNetwork(SimpleFeedForwardNetworkBase): + def __init__(self, num_parallel_samples: int = 100, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.num_parallel_samples = num_parallel_samples + + def forward(self, past_target: torch.Tensor) -> torch.Tensor: + distr = self.get_distr(past_target) + + # (num_samples, batch_size, prediction_length) + samples = distr.sample((self.num_parallel_samples,)) + + return samples.permute(1, 0, 2)