diff --git a/pts/model/forecast.py b/pts/model/forecast.py index d98f179..de0b8bf 100644 --- a/pts/model/forecast.py +++ b/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: diff --git a/test/model/test_forecast.py b/test/model/test_forecast.py new file mode 100644 index 0000000..9689758 --- /dev/null +++ b/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)