src/forecaster/models/base.py

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)