def fit()

in kats/models/ensemble/kats_ensemble.py [0:0]

• 39 lines of code
• 9 McCabe index (conditional complexity)

    def fit(self, err_method: str = "mape") -> KatsEnsemble:
        """Fit individual forecasting models via calling fitExecutor

        This is the fit methdo to fit individual forecasting model
        """

        self.seasonality = KatsEnsemble.seasonality_detector(self.data)

        # check if self.params["seasonality_length"] is given
        if self.seasonality and self.params["seasonality_length"] is None:
            msg = "The given time series contains seasonality,\
            a `seasonality_length` must be given in params."
            raise _logged_error(msg)

        # set up auto backtesting flag
        auto_backtesting = False if self.params["aggregation"] == "median" else True

        # check fitExecutor
        fitExecutor = self.params.get("fitExecutor")
        if fitExecutor is None:
            fitExecutor = self.fitExecutor

        if self.seasonality:
            # STL decomposition
            sea_data, desea_data = KatsEnsemble.deseasonalize(
                self.data, self.decomposition_method
            )
            self.sea_data = sea_data
            self.desea_data = desea_data

            # we created extra models
            given_models = copy(self.params["models"].models)
            for m in self.params["models"].models:
                if m.model_name.lower() in SMODELS.keys():
                    tmp = copy(m)
                    tmp.model_name = m.model_name + "_smodel"
                    given_models.append(tmp)

            self.model_params = model_params = EnsembleParams(given_models)
            self.fitted, self.weights = fitExecutor(
                data=desea_data,
                models=model_params,
                should_auto_backtest=auto_backtesting,
            )
        else:
            # fit models on the original data
            self.model_params = model_params = EnsembleParams(
                self.params["models"].models
            )
            self.fitted, self.weights = fitExecutor(
                data=self.data,
                models=model_params,
                should_auto_backtest=auto_backtesting,
                err_method=err_method,
            )
        return self