in kats/models/reconciliation/thm.py [0:0]
def _predict_origin(self, steps: int, method="struc") -> Dict[int, np.ndarray]:
"""
Generate original forecasts from each base model (without time index).
Args:
steps: Number of forecasts for level 1.
methd: Reconciliation method.
Returns:
Dictionary of forecasts of each level, whose key is level and value
is forecast array.
"""
m = self.m
levels = self.levels
freq = self.freq
h = int(np.ceil(steps / m))
hf = steps // m
orig_fcst = {}
models = self.models
# generate forecasts for each level
for k in levels:
num = int(freq[k] * h)
# assert models is not None
# pyre-fixme[16]: `Optional` has no attribute `__getitem__`.
if models[k] is not None:
orig_fcst[k] = models[k].predict(steps=num, freq="D")["fcst"].values
else:
fcst_num = len(self.info_fcsts[k])
if fcst_num < num:
if fcst_num >= hf * freq[k]:
# since the final output only needs hf*freq[k] forecasts
# for level k, we pad the forecast array to desirable
# length. (note that the padding values would be ignored
# in the final output.)
orig_fcst[k] = np.concatenate(
[
self.info_fcsts[k],
[self.info_fcsts[k][-1]] * (num - fcst_num),
]
)
elif method == "bu" and k != 1:
# for 'bu' only level 1 is needed.
orig_fcst[k] = self.info_fcsts[k]
else:
msg = (
f"{hf*freq[k]} steps of forecasts for level {k} are"
f" needed, but only receive {fcst_num} steps (and "
"forecast model is None)."
)
raise _log_error(msg)
else:
orig_fcst[k] = self.info_fcsts[k][:num]
return orig_fcst