source

​
mstl_decomposition

 mstl_decomposition
                     (df:Union[pandas.core.frame.DataFrame,polars.datafram
                     e.frame.DataFrame], model:statsforecast.models.MSTL,
                     freq:str, h:int)
Decompose the series into trend and seasonal using the MSTL model.
TypeDetails
dfUnionDataFrame with columns [unique_id, ds, y].
modelMSTLModel to use for the decomposition.
freqstrFrequency of the data (pandas alias)
hintForecast horizon.
ReturnsTupleOriginal dataframe with the ‘trend’ and ‘seasonal’ columns added.
import pandas as pd
from fastcore.test import test_fail
from utilsforecast.losses import smape

from statsforecast.models import Naive
from statsforecast.utils import generate_series

series = generate_series(10, freq='D')
series['unique_id'] = series['unique_id'].astype('int64')

test_fail(lambda: mstl_decomposition(series, Naive(), 'D', 14), contains='must be an MSTL instance')

horizon = 14
model = MSTL(season_length=7)
series = series.sample(frac=1.0)
train_df, X_df = mstl_decomposition(series, model, 'D', horizon)

series_pl = generate_series(10, freq='D', engine='polars')
series_pl = series_pl.with_columns(unique_id=pl.col('unique_id').cast(pl.Int64))
train_df_pl, X_df_pl = mstl_decomposition(series_pl, model, '1d', horizon)

pd.testing.assert_series_equal(
    train_df.groupby('unique_id')['ds'].max() + pd.offsets.Day(),
    X_df.groupby('unique_id')['ds'].min()
)
assert X_df.shape[0] == train_df['unique_id'].nunique() * horizon
pd.testing.assert_frame_equal(train_df, train_df_pl.to_pandas())
pd.testing.assert_frame_equal(X_df, X_df_pl.to_pandas())
with_estimate = train_df_pl.with_columns(estimate=pl.col('trend') + pl.col('seasonal'))
assert smape(with_estimate, models=['estimate'])['estimate'].mean() < 0.1

model = MSTL(season_length=[7, 28])
train_df, X_df = mstl_decomposition(series, model, 'D', horizon)
assert train_df.columns.intersection(X_df.columns).tolist() == ['unique_id', 'ds', 'trend', 'seasonal7', 'seasonal28']