initial forecast tests

pts/model/forecast.py

@@ -30,41 +30,41 @@ class Forecast(ABC):
     mean: np.ndarray
     _index = None
 
-    @abstractmethod
-    def quantile(self, q: Union[float, str]) -> np.ndarray:
-        """
-        Computes a quantile from the predicted distribution.
+    # @abstractmethod
+    # def quantile(self, q: Union[float, str]) -> np.ndarray:
+    #     """
+    #     Computes a quantile from the predicted distribution.
 
-        Parameters
-        ----------
-        q
-            Quantile to compute.
+    #     Parameters
+    #     ----------
+    #     q
+    #         Quantile to compute.
 
-        Returns
-        -------
-        numpy.ndarray
-            Value of the quantile across the prediction range.
-        """
-        pass
+    #     Returns
+    #     -------
+    #     numpy.ndarray
+    #         Value of the quantile across the prediction range.
+    #     """
+    #     pass
 
-    @abstractmethod
-    def dim(self) -> int:
-        """
-        Returns the dimensionality of the forecast object.
-        """
-        pass
+    # @abstractmethod
+    # def dim(self) -> int:
+    #     """
+    #     Returns the dimensionality of the forecast object.
+    #     """
+    #     pass
 
-    @abstractmethod
-    def copy_dim(self, dim: int):
-        """
-        Returns a new Forecast object with only the selected sub-dimension.
+    # @abstractmethod
+    # def copy_dim(self, dim: int):
+    #     """
+    #     Returns a new Forecast object with only the selected sub-dimension.
 
-        Parameters
-        ----------
-        dim
-            The returned forecast object will only represent this dimension.
-        """
-        pass
+    #     Parameters
+    #     ----------
+    #     dim
+    #         The returned forecast object will only represent this dimension.
+    #     """
+    #     pass
 
     def as_json_dict(self, config: "Config") -> dict:
         result = {}
@@ -379,7 +379,7 @@ class DistributionForecast(Forecast):
         if self._mean is not None:
             return self._mean
         else:
-            self._mean = self.distribution.mean.asnumpy()
+            self._mean = self.distribution.mean.numpy()
             return self._mean
 
     @property
@@ -391,7 +391,7 @@ class DistributionForecast(Forecast):
 
     def quantile(self, level):
         level = Quantile.parse(level).value
-        q = self.distribution.quantile(mx.nd.array([level])).asnumpy()[0]
+        q = self.distribution.icdf(torch.tensor([level])).numpy()[0]
         return q
 
     def to_sample_forecast(self, num_samples: int = 200) -> SampleForecast:

test/model/test_forecast.py

@@ -0,0 +1,96 @@
+# 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.
+
+# Third-party imports
+import numpy as np
+import pandas as pd
+import pytest
+import torch
+
+# First-party imports
+from pts.model import (
+    QuantileForecast,
+    SampleForecast,
+    DistributionForecast,
+)
+
+from torch.distributions import Uniform
+
+QUANTILES = np.arange(1, 100) / 100
+SAMPLES = np.arange(101).reshape(101, 1) / 100
+START_DATE = pd.Timestamp(2017, 1, 1, 12)
+FREQ = "1D"
+
+FORECASTS = {
+    "QuantileForecast": QuantileForecast(
+        forecast_arrays=QUANTILES.reshape(-1, 1),
+        start_date=START_DATE,
+        forecast_keys=np.array(QUANTILES, str),
+        freq=FREQ,
+    ),
+    "SampleForecast": SampleForecast(
+        samples=SAMPLES, start_date=START_DATE, freq=FREQ
+    ),
+    "DistributionForecast": DistributionForecast(
+        distribution=Uniform(low=torch.zeros(1), high=torch.ones(1)),
+        start_date=START_DATE,
+        freq=FREQ,
+    ),
+}
+
+
+@pytest.mark.parametrize("name", FORECASTS.keys())
+def test_Forecast(name):
+    forecast = FORECASTS[name]
+
+    def percentile(value):
+        return f"p{int(round(value * 100)):02d}"
+
+    num_samples, pred_length = SAMPLES.shape
+
+    for quantile in QUANTILES:
+        test_cases = [quantile, str(quantile), percentile(quantile)]
+        for quant_pred in map(forecast.quantile, test_cases):
+            assert np.isclose(
+                quant_pred[0], quantile
+            ), f"Expected {quantile} quantile {quantile}. Obtained {quant_pred}."
+
+    assert forecast.prediction_length == 1
+    assert len(forecast.index) == pred_length
+    assert forecast.index[0] == pd.Timestamp(START_DATE)
+
+
+def test_DistributionForecast():
+    forecast = DistributionForecast(
+        distribution=Uniform(
+            low=torch.tensor([0.0, 0.0]), high=torch.tensor([1.0, 2.0])
+        ),
+        start_date=START_DATE,
+        freq=FREQ,
+    )
+
+    def percentile(value):
+        return f"p{int(round(value * 100)):02d}"
+
+    for quantile in QUANTILES:
+        test_cases = [quantile, str(quantile), percentile(quantile)]
+        for quant_pred in map(forecast.quantile, test_cases):
+            expected = quantile * np.array([1.0, 2.0])
+            assert np.allclose(
+                quant_pred, expected
+            ), f"Expected {quantile} quantile {quantile}. Obtained {quant_pred}."
+
+    pred_length = 2
+    assert forecast.prediction_length == pred_length
+    assert len(forecast.index) == pred_length
+    assert forecast.index[0] == pd.Timestamp(START_DATE)