MLForecast

​

数据

这展示了仅包含 M4 数据集中 4 个序列的示例。如果你想在所有序列上自己运行，可以参考这本笔记本。

import random
import tempfile

import lightgbm as lgb
import matplotlib.pyplot as plt
import numpy as np
import xgboost as xgb
from sklearn.linear_model import LinearRegression
from utilsforecast.feature_engineering import time_features
from utilsforecast.plotting import plot_series

from mlforecast.lag_transforms import ExpandingMean, ExponentiallyWeightedMean, RollingMean
from mlforecast.lgb_cv import LightGBMCV
from mlforecast.target_transforms import Differences, LocalStandardScaler
from mlforecast.utils import generate_daily_series

df = pd.read_parquet('https://datasets-nixtla.s3.amazonaws.com/m4-hourly.parquet')
ids = df['unique_id'].unique()
random.seed(0)
sample_ids = random.choices(ids, k=4)
sample_df = df[df['unique_id'].isin(sample_ids)]
sample_df

我们现在将这些数据分为训练集和验证集。

horizon = 48
valid = sample_df.groupby('unique_id').tail(horizon)
train = sample_df.drop(valid.index)
train.shape, valid.shape

((3840, 3), (192, 3))

​

MLForecast

 MLForecast (models:Union[sklearn.base.BaseEstimator,List[sklearn.base.Bas
             eEstimator],Dict[str,sklearn.base.BaseEstimator]],
             freq:Union[int,str], lags:Optional[Iterable[int]]=None, lag_t
             ransforms:Optional[Dict[int,List[Union[Callable,Tuple[Callabl
             e,Any]]]]]=None,
             date_features:Optional[Iterable[Union[str,Callable]]]=None,
             num_threads:int=1, target_transforms:Optional[List[Union[mlfo
             recast.target_transforms.BaseTargetTransform,mlforecast.targe
             t_transforms._BaseGroupedArrayTargetTransform]]]=None,
             lag_transforms_namer:Optional[Callable]=None)

预测管道

MLForecast 对象封装了特征工程 + 模型训练 + 预测的功能。

fcst = MLForecast(
    models=lgb.LGBMRegressor(random_state=0, verbosity=-1),
    freq=1,
    lags=[24 * (i+1) for i in range(7)],
    lag_transforms={
        48: [ExponentiallyWeightedMean(alpha=0.3)],
    },
    num_threads=1,
    target_transforms=[Differences([24])],
)
fcst

MLForecast(models=[LGBMRegressor], freq=1, lag_features=['lag24', 'lag48', 'lag72', 'lag96', 'lag120', 'lag144', 'lag168', 'exponentially_weighted_mean_lag48_alpha0.3'], date_features=[], num_threads=1)

完成此设置后，我们可以计算特征并拟合模型。

​

MLForecast.fit

 MLForecast.fit
                 (df:Union[pandas.core.frame.DataFrame,polars.dataframe.fr
                 ame.DataFrame], id_col:str='unique_id',
                 time_col:str='ds', target_col:str='y',
                 static_features:Optional[List[str]]=None,
                 dropna:bool=True, keep_last_n:Optional[int]=None,
                 max_horizon:Optional[int]=None, prediction_intervals:Opti
                 onal[mlforecast.utils.PredictionIntervals]=None,
                 fitted:bool=False, as_numpy:bool=False,
                 weight_col:Optional[str]=None)

应用特征工程并训练模型。

fcst = MLForecast(
    models=lgb.LGBMRegressor(random_state=0, verbosity=-1),
    freq=1,
    lags=[24 * (i+1) for i in range(7)],
    lag_transforms={
        48: [ExponentiallyWeightedMean(alpha=0.3)],
    },
    num_threads=1,
    target_transforms=[Differences([24])],
)

train2 = train.copy()
train2['weight'] = np.random.default_rng(seed=0).random(train2.shape[0])
fcst.fit(train2, weight_col='weight', as_numpy=True).predict(5)

fcst.cross_validation(train2, n_windows=2, h=5, weight_col='weight', as_numpy=True)

fcst.fit(train, fitted=True);

​

MLForecast.save

 MLForecast.save (path:Union[str,pathlib.Path])

保存预测对象

​

MLForecast.load

 MLForecast.load (path:Union[str,pathlib.Path])

加载预测对象

​

MLForecast.update

 MLForecast.update
                    (df:Union[pandas.core.frame.DataFrame,polars.dataframe
                    .frame.DataFrame])

更新存储序列的值。

​

MLForecast.make_future_dataframe

 MLForecast.make_future_dataframe (h:int)

创建一个包含所有 id 和预测范围内的未来时间的 dataframe。

expected_future = fcst.make_future_dataframe(h=1)
expected_future

​

MLForecast.get_missing_future

 MLForecast.get_missing_future (h:int, X_df:~DFType)

获取 X_df 中缺失的 id 和时间组合。

missing_future = fcst.get_missing_future(h=1, X_df=expected_future.head(2))
pd.testing.assert_frame_equal(
    missing_future,
    expected_future.tail(2).reset_index(drop=True)
)

​

MLForecast.forecast_fitted_values

 MLForecast.forecast_fitted_values
                                    (level:Optional[List[Union[int,float]]
                                    ]=None)

访问样本内预测。

fcst.forecast_fitted_values()

fcst.forecast_fitted_values(level=[90])

一旦运行完成，我们就可以计算预测结果了。

​

MLForecast.predict

 MLForecast.predict (h:int,
                     before_predict_callback:Optional[Callable]=None,
                     after_predict_callback:Optional[Callable]=None,
                     new_df:Optional[~DFType]=None,
                     level:Optional[List[Union[int,float]]]=None,
                     X_df:Optional[~DFType]=None,
                     ids:Optional[List[str]]=None)

计算未来 h 步的预测。

predictions = fcst.predict(horizon)

我们可以查看一些结果。

results = valid.merge(predictions, on=['unique_id', 'ds'])
fig = plot_series(forecasts_df=results)

​

预测区间

使用 MLForecast，你可以使用保形预测生成预测区间。要配置保形预测，你需要将 PredictionIntervals 类的一个实例传递给 fit 方法的 prediction_intervals 参数。该类接受三个参数：n_windows、h 和 method。

• n_windows 表示用于校准区间的交叉验证窗口数
• h 是预测范围
• method 可以是 conformal_distribution 或 conformal_error；conformal_distribution（默认）根据交叉验证误差创建预测路径，并使用这些路径计算分位数，而 conformal_error 计算误差分位数以生成预测区间。该策略将针对每个预测步长调整区间，导致每个步长的宽度不同。请注意，必须至少使用 2 个交叉验证窗口。

fcst.fit(
    train,
    prediction_intervals=PredictionIntervals(n_windows=3, h=48)
);

之后，你只需使用 level 参数将所需的置信水平包含到 predict 方法中即可。水平必须介于 0 到 100 之间。

predictions_w_intervals = fcst.predict(48, level=[50, 80, 95])
predictions_w_intervals.head()

让我们探索生成的区间。

results = valid.merge(predictions_w_intervals, on=['unique_id', 'ds'])
fig = plot_series(forecasts_df=results, level=[50, 80, 95])

如果你想减少计算时间并为整个预测范围生成相同宽度的区间，只需将 h=1 传递给 PredictionIntervals 类即可。这种策略的缺点是，在某些情况下，绝对残差的方差可能很小（甚至为零），因此区间可能会太窄。

fcst.fit(
    train,  
    prediction_intervals=PredictionIntervals(n_windows=3, h=1)
);

predictions_w_intervals_ws_1 = fcst.predict(48, level=[80, 90, 95])

让我们探索生成的区间。

results = valid.merge(predictions_w_intervals_ws_1, on=['unique_id', 'ds'])
fig = plot_series(forecasts_df=results, level=[90])

​

使用预训练模型进行预测

MLForecast 允许你使用预训练模型为新数据集生成预测。只需在调用 predict 方法时，将包含新观测值的 pandas dataframe 作为 new_df 参数的值即可。该 dataframe 应与用于拟合模型的 dataframe 具有相同的结构，包括任何特征和时间序列数据。然后，该函数将使用预训练模型为新观测值生成预测。这使得你可以轻松地将预训练模型应用于新数据集并生成预测，而无需重新训练模型。

ercot_df = pd.read_csv('https://datasets-nixtla.s3.amazonaws.com/ERCOT-clean.csv')
# we have to convert the ds column to integers
# since MLForecast was trained with that structure
ercot_df['ds'] = np.arange(1, len(ercot_df) + 1)
# use the `new_df` argument to pass the ercot dataset 
ercot_fcsts = fcst.predict(horizon, new_df=ercot_df)
fig = plot_series(ercot_df, ercot_fcsts, max_insample_length=48 * 2)

如果你想查看将用于训练模型的数据，可以调用 Forecast.preprocess。

​

MLForecast.preprocess

 MLForecast.preprocess (df:~DFType, id_col:str='unique_id',
                        time_col:str='ds', target_col:str='y',
                        static_features:Optional[List[str]]=None,
                        dropna:bool=True, keep_last_n:Optional[int]=None,
                        max_horizon:Optional[int]=None,
                        return_X_y:bool=False, as_numpy:bool=False,
                        weight_col:Optional[str]=None)

将特征添加到 data。

prep_df = fcst.preprocess(train)
prep_df

如果我们这样做，那么必须调用 Forecast.fit_models，因为这仅存储了序列信息。

​

MLForecast.fit_models

 MLForecast.fit_models (X:Union[pandas.core.frame.DataFrame,polars.datafra
                        me.frame.DataFrame,numpy.ndarray],
                        y:numpy.ndarray)

手动训练模型。如果你事先调用了 MLForecast.preprocess，请使用此方法。

X, y = prep_df.drop(columns=['unique_id', 'ds', 'y']), prep_df['y']
fcst.fit_models(X, y)

MLForecast(models=[LGBMRegressor], freq=1, lag_features=['lag24', 'lag48', 'lag72', 'lag96', 'lag120', 'lag144', 'lag168', 'exponentially_weighted_mean_lag48_alpha0.3'], date_features=[], num_threads=1)

predictions2 = fcst.predict(horizon)
pd.testing.assert_frame_equal(predictions, predictions2)

​

MLForecast.cross_validation

 MLForecast.cross_validation (df:~DFType, n_windows:int, h:int,
                              id_col:str='unique_id', time_col:str='ds',
                              target_col:str='y',
                              step_size:Optional[int]=None,
                              static_features:Optional[List[str]]=None,
                              dropna:bool=True,
                              keep_last_n:Optional[int]=None,
                              refit:Union[bool,int]=True,
                              max_horizon:Optional[int]=None, before_predi
                              ct_callback:Optional[Callable]=None, after_p
                              redict_callback:Optional[Callable]=None, pre
                              diction_intervals:Optional[mlforecast.utils.
                              PredictionIntervals]=None,
                              level:Optional[List[Union[int,float]]]=None,
                              input_size:Optional[int]=None,
                              fitted:bool=False, as_numpy:bool=False,
                              weight_col:Optional[str]=None)

执行时间序列交叉验证。创建 n_windows 个分割，每个窗口包含 h 个测试周期，然后训练模型，计算预测并合并实际值。

如果我们想知道特定模型和特征集下的预测效果如何，可以进行交叉验证。交叉验证的作用是将我们的数据分成两部分，第一部分用于训练，第二部分用于验证。由于数据具有时间依赖性，我们通常将数据的最后 x 个观测值作为验证集。

此过程在 MLForecast.cross_validation 中实现，它接收我们的数据并执行上述过程 n_windows 次，其中每个窗口包含 h 个验证样本。例如，如果我们有 100 个样本并想进行 2 次回测，每次大小为 14，则分割如下：

1. 训练集：1 到 72。验证集：73 到 86。
2. 训练集：1 到 86。验证集：87 到 100。

你可以使用 step_size 参数控制每个交叉验证窗口之间的距离。例如，如果我们有 100 个样本并想进行 2 次大小为 14 的回测，并在每个折叠中向前移动一步（step_size=1），则分割如下：

1. 训练集：1 到 85。验证集：86 到 99。
2. 训练集：1 到 86。验证集：87 到 100。

你还可以通过设置 refit=False 来执行交叉验证，而无需对每个窗口重新训练模型。这使得你可以使用多个窗口大小评估模型的性能，而无需每次都重新训练它们。

fcst = MLForecast(
    models=lgb.LGBMRegressor(random_state=0, verbosity=-1),
    freq=1,
    lags=[24 * (i+1) for i in range(7)],
    lag_transforms={
        1: [RollingMean(window_size=24)],
        24: [RollingMean(window_size=24)],
        48: [ExponentiallyWeightedMean(alpha=0.3)],
    },
    num_threads=1,
    target_transforms=[Differences([24])],
)
cv_results = fcst.cross_validation(
    train,
    n_windows=2,
    h=horizon,
    step_size=horizon,
    fitted=True,
)
cv_results

由于我们将 fitted 设置为 True，我们也可以通过 cross_validation_fitted_values 方法访问训练集的预测结果。

fcst.cross_validation_fitted_values()

我们还可以通过将配置传递给 prediction_intervals 以及通过 levels 传递宽度值来计算预测区间。

cv_results_intervals = fcst.cross_validation(
    train,
    n_windows=2,
    h=horizon,
    step_size=horizon,
    prediction_intervals=PredictionIntervals(h=horizon),
    level=[80, 90]
)
cv_results_intervals

refit 参数允许我们控制是否要在每个窗口中重新训练模型。它可以是

• 一个布尔值：True 将在每个窗口中重新训练，False 仅在第一个窗口中训练。
• 一个正整数：模型将在第一个窗口中训练，然后每隔 refit 个窗口训练一次。

fcst = MLForecast(
    models=LinearRegression(),
    freq=1,
    lags=[1, 24],
)
for refit, expected_models in zip([True, False, 2], [4, 1, 2]):
    fcst.cross_validation(
        train,
        n_windows=4,
        h=horizon,
        refit=refit,
    )
    test_eq(len(fcst.cv_models_), expected_models)

fig = plot_series(forecasts_df=cv_results.drop(columns='cutoff'))

fig = plot_series(forecasts_df=cv_results_intervals.drop(columns='cutoff'), level=[90])

​

MLForecast.from_cv

 MLForecast.from_cv (cv:mlforecast.lgb_cv.LightGBMCV)

一旦找到一组适用于你的问题的特征和参数，就可以使用 MLForecast.from_cv 从中构建一个预测对象，该方法接受训练好的 LightGBMCV 对象，并构建一个将使用相同特征和参数的 MLForecast 对象。然后，你就可以像往常一样调用 fit 和 predict。

cv = LightGBMCV(
    freq=1,
    lags=[24 * (i+1) for i in range(7)],
    lag_transforms={
        48: [ExponentiallyWeightedMean(alpha=0.3)],
    },
    num_threads=1,
    target_transforms=[Differences([24])]
)
hist = cv.fit(
    train,
    n_windows=2,
    h=horizon,
    params={'verbosity': -1},
)

[10] mape: 0.118569
[20] mape: 0.111506
[30] mape: 0.107314
[40] mape: 0.106089
[50] mape: 0.106630
Early stopping at round 50
Using best iteration: 40

fcst = MLForecast.from_cv(cv)
assert cv.best_iteration_ == fcst.models['LGBMRegressor'].n_estimators