feat(Pytorch): added custom model to pytorch-forecasting (#8)

* feat: Refractored and created new model. Pipeline not ready yet.

* feat: Implemented and refactored a data pipeline.

* ref: Refractored to make more sense.

* feat: Training works now with models that you can change.

* feat: Added predict function but without working instructions.

* feat: gitignore.

lightning/models/built_in_models.py

@@ -0,0 +1,19 @@
+from pytorch_forecasting import TemporalFusionTransformer
+from pytorch_forecasting.metrics import QuantileLoss
+
+def create_TemporalFusionTransformer(training_dataset, model_options):
+    # create the model
+    tft = TemporalFusionTransformer.from_dataset(
+        training_dataset,
+        learning_rate=0.03,
+        hidden_size=32,
+        attention_head_size=1,
+        dropout=0.1,
+        hidden_continuous_size=16,
+        output_size=7,
+        loss=QuantileLoss(),
+        log_interval=2,
+        reduce_on_plateau_patience=4
+    )
+    print(f"Number of parameters in network: {tft.size()/1e3:.1f}k")
+    return tft

lightning/models/custom_model.py

@@ -0,0 +1,98 @@
+#%%
+import warnings
+from typing import Dict
+
+warnings.filterwarnings("ignore")
+
+import torch
+from torch import nn
+
+from pytorch_forecasting.models import BaseModel
+from pytorch_forecasting import TimeSeriesDataSet
+
+
+#%%
+class FullyConnectedModule(nn.Module):
+    def __init__(self, input_size: int, output_size: int, hidden_size: int, n_hidden_layers: int):
+        super().__init__()
+
+        # input layer
+        module_list = [nn.Linear(input_size, hidden_size), nn.ReLU()]
+        # hidden layers
+        for _ in range(n_hidden_layers):
+            module_list.extend([nn.Linear(hidden_size, hidden_size), nn.ReLU()])
+        # output layer
+        module_list.append(nn.Linear(hidden_size, output_size))
+
+        self.sequential = nn.Sequential(*module_list)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        # x of shape: batch_size x n_timesteps_in
+        # output of shape batch_size x n_timesteps_out
+        return self.sequential(x)
+
+
+
+#%%
+class FullyConnectedModel(BaseModel):
+    def __init__(self, input_size: int, output_size: int, hidden_size: int, n_hidden_layers: int, **kwargs):
+        # saves arguments in signature to `.hparams` attribute, mandatory call - do not skip this
+        self.save_hyperparameters()
+        # pass additional arguments to BaseModel.__init__, mandatory call - do not skip this
+        super().__init__(**kwargs)
+        self.network = FullyConnectedModule(
+            input_size=self.hparams.input_size,
+            output_size=self.hparams.output_size,
+            hidden_size=self.hparams.hidden_size,
+            n_hidden_layers=self.hparams.n_hidden_layers,
+        )
+
+    def forward(self, x: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
+        # x is a batch generated based on the TimeSeriesDataset
+        network_input = x["encoder_cont"].squeeze(-1)
+        prediction = self.network(network_input)
+
+        # rescale predictions into target space
+        prediction = self.transform_output(prediction, target_scale=x["target_scale"])
+
+        # We need to return a dictionary that at least contains the prediction
+        # The parameter can be directly forwarded from the input.
+        # The conversion to a named tuple can be directly achieved with the `to_network_output` function.
+        return self.to_network_output(prediction=prediction)
+    
+    @classmethod
+    def from_dataset(cls, dataset: TimeSeriesDataSet, **kwargs):
+        new_kwargs = {
+            "output_size": dataset.max_prediction_length,
+            "input_size": dataset.max_encoder_length,
+        }
+        new_kwargs.update(kwargs)  # use to pass real hyperparameters and override defaults set by dataset
+        # example for dataset validation
+        assert dataset.max_prediction_length == dataset.min_prediction_length, "Decoder only supports a fixed length"
+        assert dataset.min_encoder_length == dataset.max_encoder_length, "Encoder only supports a fixed length"
+        assert (
+            len(dataset.time_varying_known_categoricals) == 0
+            and len(dataset.time_varying_known_reals) == 0
+            and len(dataset.time_varying_unknown_categoricals) == 0
+            and len(dataset.static_categoricals) == 0
+            and len(dataset.static_reals) == 0
+            and len(dataset.time_varying_unknown_reals) == 1
+            and dataset.time_varying_unknown_reals[0] == dataset.target
+        ), "Only covariate should be the target in 'time_varying_unknown_reals'"
+
+        return super().from_dataset(dataset, **new_kwargs)
+    
+    def calculate_prediction_actual_by_variable(x, train):
+        pass
+    
+
+
+
+
+# %%
+def create_FullyConnectedModel(training_dataset, kwargs):
+    model = FullyConnectedModel.from_dataset(training_dataset, **kwargs)
+    model.summarize("full")  # print model summary
+    model.hparams
+    
+    return model