def forecaster()

in afa/core.py [0:0]

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

def forecaster(func):
    """Forecast function decorator. Apply this to any forecasting function and
    it will handle any compulsory pre- and post-processing of arguments and
    return values.

    """

    def do_forecast(y, horiz, freq, **kwargs):
        assert horiz > 0
        
        local_model = kwargs.get("local_model", False)
        seasonal = kwargs.get("seasonal", False)
        trim_zeros = kwargs.get("trim_zeros", False)
        use_log = kwargs.get("use_log", False)
        dc = kwargs.get("dc", None)

        if trim_zeros:
            y = np.trim_zeros(y, trim="f")

        if use_log:
            y = np.log1p(y)
            
        if local_model:
            y = y[-TAIL_LEN[freq]:]
            
        if len(y) == 1:
            y = np.pad(y, [1,0], constant_values=1)

        y = np.nan_to_num(y)

        # forecast via seasonal decomposition
        if seasonal:
            if dc is None:
                yp = np.zeros(horiz)
            else:
                period = DC_PERIODS[freq]

                if len(y) < 2 * period:
                    period = int(len(y) / 2)

                kwargs.pop("seasonal")

                resid, trend, yp_seasonal = dc
                #dc = sm.tsa.seasonal.seasonal_decompose(y, period=period, two_sided=False)
                #yp_seasonal = fourier(seas, horiz, freq, seasonal=False)
                yp_trend = func(np.nan_to_num(trend), horiz, **kwargs)
                yp_resid = func(np.nan_to_num(resid), horiz, **kwargs)
                yp = yp_seasonal + yp_trend + yp_resid
        else:
            # ensure the input values are not null
            yp = func(y, horiz, **kwargs)

        # de-normalize from log-scale
        if use_log:
            yp = np.exp(yp)

        yp = np.nan_to_num(yp).clip(0).round(0)
        
        assert len(yp) == horiz

        return yp

    return do_forecast