/
_predictor.py
206 lines (174 loc) · 6.81 KB
/
_predictor.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
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 os
from typing import Dict, Iterator, Optional
# Third-party imports
import numpy as np
import pandas as pd
# First-party imports
from gluonts.core.component import validated
from gluonts.dataset.common import Dataset
from gluonts.evaluation import get_seasonality
from gluonts.model.forecast import SampleForecast
from gluonts.model.predictor import RepresentablePredictor
USAGE_MESSAGE = """
The RForecastPredictor is a thin wrapper for calling the R forecast package.
In order to use it you need to install R and run
pip install 'rpy2>=2.9.*,<3.*'
R -e 'install.packages(c("forecast", "nnfor"), repos="https://cloud.r-project.org")'
"""
class RForecastPredictor(RepresentablePredictor):
"""
Wrapper for calling the `R forecast package
<http://pkg.robjhyndman.com/forecast/>`_.
The `RForecastPredictor` is a thin wrapper for calling the R forecast
package. In order to use it you need to install R and run::
pip install 'rpy2>=2.9.*,<3.*'
R -e 'install.packages(c("forecast", "nnfor"), repos="https://cloud.r-project.org")'
Parameters
----------
method
The method from rforecast to be used one of
"ets", "arima", "tbats", "croston", "mlp".
prediction_length
Number of time points to be predicted.
freq
The granularity of the time series (e.g. '1H')
period
The period to be used (this is called `frequency` in the R forecast
package), result to a tentative reasonable default if not specified
(for instance 24 for hourly freq '1H')
trunc_length
Maximum history length to feed to the model (some models become slow
with very long series).
params
Parameters to be used when calling the forecast method default.
Note that currently only `output_type = 'samples'` is supported.
"""
@validated()
def __init__(
self,
freq: str,
prediction_length: int,
method_name: str = "ets",
period: int = None,
trunc_length: Optional[int] = None,
params: Optional[Dict] = None,
) -> None:
try:
from rpy2 import robjects, rinterface
import rpy2.robjects.packages as rpackages
from rpy2.rinterface import RRuntimeError
except ImportError as e:
raise ImportError(str(e) + USAGE_MESSAGE) from e
self._robjects = robjects
self._rinterface = rinterface
self._rinterface.initr()
self._rpackages = rpackages
this_dir = os.path.dirname(os.path.realpath(__file__))
r_files = [
n[:-2] for n in os.listdir(f"{this_dir}/R/") if n[-2:] == ".R"
]
for n in r_files:
try:
robjects.r(f'source("{this_dir}/R/{n}.R")')
except RRuntimeError as er:
raise RRuntimeError(str(er) + USAGE_MESSAGE) from er
supported_methods = ["ets", "arima", "tbats", "croston", "mlp"]
assert (
method_name in supported_methods
), f"method {method_name} is not supported please use one of {supported_methods}"
self.method_name = method_name
self._stats_pkg = rpackages.importr("stats")
self._r_method = robjects.r[method_name]
self.prediction_length = prediction_length
self.freq = freq
self.period = period if period is not None else get_seasonality(freq)
self.trunc_length = trunc_length
self.params = {
"prediction_length": self.prediction_length,
"output_types": ["samples"],
"frequency": self.period,
}
if params is not None:
self.params.update(params)
def _unlist(self, l):
if type(l).__name__.endswith("Vector"):
return [self._unlist(x) for x in l]
else:
return l
def _run_r_forecast(self, d, params, save_info):
buf = []
def save_to_buf(x):
buf.append(x)
def dont_save(x):
pass
f = save_to_buf if save_info else dont_save
# save output from the R console in buf
self._rinterface.set_writeconsole_regular(f)
self._rinterface.set_writeconsole_warnerror(f)
make_ts = self._stats_pkg.ts
r_params = self._robjects.vectors.ListVector(params)
vec = self._robjects.FloatVector(d["target"])
ts = make_ts(vec, frequency=self.period)
forecast = self._r_method(ts, r_params)
forecast_dict = dict(
zip(forecast.names, map(self._unlist, list(forecast)))
)
# FOR NOW ONLY SAMPLES...
# if "quantiles" in forecast_dict:
# forecast_dict["quantiles"] = dict(zip(params["quantiles"], forecast_dict["quantiles"]))
self._rinterface.set_writeconsole_regular(
self._rinterface.consolePrint
)
self._rinterface.set_writeconsole_warnerror(
self._rinterface.consolePrint
)
return forecast_dict, buf
def predict(
self,
dataset: Dataset,
num_samples: int = 100,
save_info: bool = False,
**kwargs,
) -> Iterator[SampleForecast]:
for entry in dataset:
if isinstance(entry, dict):
data = entry
else:
data = entry.data
if self.trunc_length:
data = data[-self.trunc_length :]
params = self.params.copy()
params["num_samples"] = num_samples
forecast_dict, console_output = self._run_r_forecast(
data, params, save_info=save_info
)
forecast_start = (
pd.Timestamp(data["start"], freq=self.freq)
+ data["target"].shape[0]
)
samples = np.array(forecast_dict["samples"])
expected_shape = (params["num_samples"], self.prediction_length)
assert (
samples.shape == expected_shape
), f"Expected shape {expected_shape} but found {samples.shape}"
info = (
{"console_output": "\n".join(console_output)}
if save_info
else None
)
yield SampleForecast(
samples, forecast_start, forecast_start.freqstr, info=info
)