Spaces:
Runtime error
Runtime error
File size: 2,232 Bytes
8cf4695 |
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 |
import logging
from sktime.split import ExpandingWindowSplitter
from sktime.forecasting.compose import make_reduction
from sktime.forecasting.model_evaluation import evaluate
from sktime.forecasting.model_selection import ForecastingRandomizedSearchCV
from sktime.performance_metrics.forecasting import mean_absolute_percentage_error
import pandas as pd
class ModelBase():
def __init__(
self,
estimator,
param_grid
) -> None:
print('Init model ...')
self.estimator = estimator
self.param_grid = param_grid
def fit(
self,
y: pd.DataFrame,
cv,
window_length,
X: pd.DataFrame = None,
**kwargs
):
'''
datetime: pd.DateTimeIndex, required
this is the datetime value of y, or the historical value of forecasting horizon
'''
forecaster = make_reduction(
self.estimator,
strategy='recursive',
window_length=window_length)
self.gscv = ForecastingRandomizedSearchCV(
forecaster,
cv=cv,
param_distributions=self.param_grid,
n_iter=100,
random_state=42,
scoring=mean_absolute_percentage_error,
error_score='raise')
self.gscv.fit(y=y, X=X)
def forecast(self, fh, X):
y_pred = self.gscv.predict(fh=fh, X=X)
return {
"forecast": y_pred,
"best_score": self.gscv.best_score_,
"best_params": self.gscv.best_params_}
# def forecast(
# self,
# fh,
# exog=None
# ):
# predict_args = {
# 'fh': fh
# }
# if exog is not None:
# predict_args['X'] = exog
# return self.forecaster.predict(**predict_args)
# def evaluate(self):
# cv_args = {
# 'initial_window': self.n_predict * 2,
# 'step_length': self.n_predict,
# 'fh': list(range(1, self.n_predict+1))}
# cv = ExpandingWindowSplitter(**cv_args)
# return evaluate(
# forecaster=self.forecaster,
# y=self.y,
# cv=cv)
|