--- title: Core keywords: fastai sidebar: home_sidebar nb_path: "nbs/core.ipynb" ---
{% raw %}
{% endraw %} {% raw %}
{% endraw %} {% raw %}
from fastcore.test import test_eq

from statsforecast.models import (
    adida,
    auto_arima,
    croston_classic,
    croston_optimized,
    croston_sba,
    historic_average,
    imapa,
    naive,
    random_walk_with_drift,
    seasonal_exponential_smoothing,
    seasonal_naive,
    seasonal_window_average,
    ses,
    tsb,
    window_average,
)
from statsforecast.utils import generate_series
{% endraw %} {% raw %}

class StatsForecast[source]

StatsForecast(df, models, freq, n_jobs=1, ray_address=None)

{% endraw %} {% raw %}
{% endraw %}

Daily data

{% raw %}
fcst = StatsForecast(
    series,
    [adida, croston_classic, croston_optimized,
     croston_sba, historic_average, imapa, naive, 
     random_walk_with_drift, (seasonal_exponential_smoothing, 7, 0.1),
     (seasonal_naive, 7), (seasonal_window_average, 7, 4),
     (ses, 0.1), (tsb, 0.1, 0.3), (window_average, 4)],
    freq='D',
)
res = fcst.forecast(14)
res
{% endraw %}

Parallel

{% raw %}
try: from nbdev.imports import IN_NOTEBOOK
except: IN_NOTEBOOK=False
if __name__=="__main__" and not IN_NOTEBOOK:
    fcst = StatsForecast(
        series,
        [adida, (ses, 0.1), historic_average, croston_classic],
        freq='D',
        n_jobs=2
    )
    res = fcst.forecast(14)
    print(res)
{% endraw %}

Monthly data

{% raw %}
monthly_series = generate_series(10_000, freq='M', min_length=10, max_length=20, equal_ends=True)
monthly_series
{% endraw %} {% raw %}
fcst = StatsForecast(
    monthly_series,
    [adida, (ses, 0.1), historic_average, croston_classic],
    freq='M'
)
%time monthly_res = fcst.forecast(4)
monthly_res
{% endraw %} {% raw %}
fcst.ga.split(2)[1].data
{% endraw %}

Integer datestamp

{% raw %}
from statsforecast.utils import AirPassengers as ap
{% endraw %} {% raw %}
int_ds_df = pd.DataFrame({'ds': np.arange(1, len(ap) + 1), 'y': ap})
int_ds_df.insert(0, 'unique_id', 'AirPassengers')
int_ds_df.set_index('unique_id', inplace=True)
int_ds_df.head()
{% endraw %} {% raw %}
int_ds_df.tail()
{% endraw %} {% raw %}
fcst = StatsForecast(int_ds_df, models=[historic_average], freq='D')
horizon = 7
forecast = fcst.forecast(horizon)
forecast.head()
{% endraw %} {% raw %}
last_date = int_ds_df['ds'].max()
test_eq(forecast['ds'].values, np.arange(last_date + 1, last_date + 1 + horizon))
{% endraw %}

External regressors

Every column after y is considered an external regressor and will be passed to the models that allow them. If you use them you must supply the future values to the forecast method.

{% raw %}
def linear_regression(X, h, future_xreg):
    y = X[:, 0]
    xreg = X[:, 1:]
    coefs, *_ = np.linalg.lstsq(xreg, y, rcond=None)
    return future_xreg @ coefs
{% endraw %} {% raw %}
series_xreg = series = generate_series(10_000, equal_ends=True)
series_xreg['intercept'] = 1
series_xreg['dayofweek'] = series_xreg['ds'].dt.dayofweek
series_xreg = pd.get_dummies(series_xreg, columns=['dayofweek'], drop_first=True)
series_xreg
{% endraw %} {% raw %}
dates = sorted(series_xreg['ds'].unique())
valid_start = dates[-14]
train_mask = series_xreg['ds'] < valid_start
series_train = series_xreg[train_mask]
series_valid = series_xreg[~train_mask]
X_valid = series_valid.drop(columns=['y'])
fcst = StatsForecast(
    series_train,
    [linear_regression],
    freq='D',
)
%time xreg_res = fcst.forecast(14, xreg=X_valid)
xreg_res['y'] = series_valid['y'].values
{% endraw %} {% raw %}
xreg_res.groupby('ds').mean().plot();
{% endraw %}

Confidence intervals

{% raw %}
ap_df = pd.DataFrame({'ds': np.arange(ap.size), 'y': ap}, index=pd.Index([0] * ap.size, name='unique_id'))
fcst = StatsForecast(
    ap_df,
    [(seasonal_naive, 12), (auto_arima, 12)],
    freq='M',
)
ap_ci = fcst.forecast(12, level=(80, 95))
ap_ci.set_index('ds').plot(marker='.', figsize=(10, 6));
{% endraw %}

n jobs

{% raw %}
ap_df_2 = pd.DataFrame(
    {'ds': np.hstack([np.arange(ap.size), np.arange(ap.size)]), 
     'y': np.hstack([ap, ap])}, 
    index=pd.Index([0] * ap.size + [1] * ap.size, name='unique_id')
)
{% endraw %} {% raw %}
if __name__=="__main__" and not IN_NOTEBOOK:
    ap_df = pd.DataFrame({'ds': np.arange(ap.size), 'y': ap}, index=pd.Index([0] * ap.size, name='unique_id'))
    fcst = StatsForecast(
        ap_df,
        [(seasonal_naive, 12), (auto_arima, 12)],
        freq='M',
        n_jobs=101
    )
    ap_ci = fcst.forecast(12, level=(80, 95))
    ap_ci.set_index('ds').plot(marker='.', figsize=(10, 6))
{% endraw %}