Pipeline
Pipeline 可用于创建包含多个 特征工程 方法与一个学习器的时序建模机器学习工作流。
1. 组装特征工程工作流
时序数据预处理通常包含多个步骤。我们可将多个特征工程方法组装为一个Pipeline。
1.1. 准备数据
>>> import pandas as pd
>>> import numpy as np
>>> from paddlets.datasets.tsdataset import TimeSeries, TSDataset
>>> target = TimeSeries.load_from_dataframe(
>>> pd.Series(
>>> np.random.randn(10).astype(np.float32),
>>> index=pd.date_range("2022-01-01", periods=10, freq="15T"),
>>> name="target"
>>> ))
>>> observed_cov = TimeSeries.load_from_dataframe(
>>> pd.DataFrame(
>>> np.random.randn(10, 2).astype(np.float32),
>>> index=pd.date_range("2022-01-01", periods=10, freq="15T"),
>>> columns=["observed_a", "observed_b"]
>>> ))
>>> known_cov = TimeSeries.load_from_dataframe(
>>> pd.DataFrame(
>>> np.random.randn(15, 2).astype(np.float32),
>>> index=pd.date_range("2022-01-01", periods=15, freq="15T"),
>>> columns=["known_c", "known_d"]
>>> ))
>>> tsdataset = TSDataset(target, observed_cov, known_cov)
>>> train_dataset, test_dataset = tsdataset.split(0.7)
>>> train_dataset
target observed_a observed_b known_c known_d
2022-01-01 00:00:00 1.044207 0.693854 -0.175082 -1.890724 1.364049
2022-01-01 00:15:00 -1.011910 -0.471103 0.001718 -0.307345 0.100475
2022-01-01 00:30:00 -0.953874 0.473357 -0.620498 0.426992 -1.976870
2022-01-01 00:45:00 0.811422 0.679846 0.401000 1.026558 -0.281426
2022-01-01 01:00:00 -0.343195 -0.657656 2.705243 0.997001 0.856995
2022-01-01 01:15:00 1.992397 -0.281009 1.666330 1.136454 0.154997
2022-01-01 01:30:00 0.070085 -0.291660 0.449088 0.034393 -1.101410
2022-01-01 01:45:00 NaN NaN NaN 0.387557 0.557443
2022-01-01 02:00:00 NaN NaN NaN -0.927851 1.425830
2022-01-01 02:15:00 NaN NaN NaN 0.056782 -0.112722
2022-01-01 02:30:00 NaN NaN NaN 0.545976 -2.725172
2022-01-01 02:45:00 NaN NaN NaN 0.436852 -2.653046
2022-01-01 03:00:00 NaN NaN NaN 0.579058 -1.125973
2022-01-01 03:15:00 NaN NaN NaN -0.480516 1.002109
2022-01-01 03:30:00 NaN NaN NaN 0.220903 0.325239
>>> test_dataset
target observed_a observed_b known_c known_d
2022-01-01 00:00:00 NaN NaN NaN -1.890724 1.364049
2022-01-01 00:15:00 NaN NaN NaN -0.307345 0.100475
2022-01-01 00:30:00 NaN NaN NaN 0.426992 -1.976870
2022-01-01 00:45:00 NaN NaN NaN 1.026558 -0.281426
2022-01-01 01:00:00 NaN NaN NaN 0.997001 0.856995
2022-01-01 01:15:00 NaN NaN NaN 1.136454 0.154997
2022-01-01 01:30:00 NaN NaN NaN 0.034393 -1.101410
2022-01-01 01:45:00 -0.833165 -1.301204 -1.073407 0.387557 0.557443
2022-01-01 02:00:00 0.767193 1.543675 0.985133 -0.927851 1.425830
2022-01-01 02:15:00 0.458591 0.453419 -2.523344 0.056782 -0.112722
2022-01-01 02:30:00 NaN NaN NaN 0.545976 -2.725172
2022-01-01 02:45:00 NaN NaN NaN 0.436852 -2.653046
2022-01-01 03:00:00 NaN NaN NaN 0.579058 -1.125973
2022-01-01 03:15:00 NaN NaN NaN -0.480516 1.002109
2022-01-01 03:30:00 NaN NaN NaN 0.220903 0.325239
对于上述数据的处理，我们可能希望利用 KSigma 对观察协变量与已知协变量做异常检测，然后利用 TimeFeatureGenerator 生成时间特征。
1.2. 构建
Pipeline 使用包含 (key, value) 对的列表进行初始化。 key 是特征工程方法的类名，value 是此方法的初始化参数。
该 Pipeline 由以下部分组成：
用于异常值检测的 Ksigma 特征工程方法。
用于生成时间特征的 TimeFeatureGenerator 特征工程方法。
>>> from paddlets.pipeline.pipeline import Pipeline
>>> from paddlets.transform import KSigma, TimeFeatureGenerator
>>> pipeline = Pipeline([(KSigma, {"cols":["observed_a", "observed_b", "known_c", "known_d"], "k": 1}), (TimeFeatureGenerator, {})])
1.3. 执行特征工程
拟合 Pipeline 并执行特征工程。
>>> pipeline.fit(train_dataset)
>>> tsdataset_preprocessed = pipeline.transform(test_dataset)
>>> tsdataset_preprocessed
target observed_a observed_b known_c known_d year month day weekday hour quarter dayofyear weekofyear is_holiday is_workday
2022-01-01 00:00:00 NaN NaN NaN 0.149473 -0.279299 2022 1 1 5 0 1 1 52 1.0 0.0
2022-01-01 00:15:00 NaN NaN NaN -0.307345 0.100475 2022 1 1 5 0 1 1 52 1.0 0.0
2022-01-01 00:30:00 NaN NaN NaN 0.426992 -0.279299 2022 1 1 5 0 1 1 52 1.0 0.0
2022-01-01 00:45:00 NaN NaN NaN 0.149473 -0.281426 2022 1 1 5 0 1 1 52 1.0 0.0
2022-01-01 01:00:00 NaN NaN NaN 0.149473 0.856995 2022 1 1 5 1 1 1 52 1.0 0.0
2022-01-01 01:15:00 NaN NaN NaN 0.149473 0.154997 2022 1 1 5 1 1 1 52 1.0 0.0
2022-01-01 01:30:00 NaN NaN NaN 0.034393 -1.101410 2022 1 1 5 1 1 1 52 1.0 0.0
2022-01-01 01:45:00 -0.833165 0.020804 0.632543 0.387557 0.557443 2022 1 1 5 1 1 1 52 1.0 0.0
2022-01-01 02:00:00 0.767193 0.020804 0.985133 0.149473 -0.279299 2022 1 1 5 2 1 1 52 1.0 0.0
2022-01-01 02:15:00 0.458591 0.453419 0.632543 0.056782 -0.112722 2022 1 1 5 2 1 1 52 1.0 0.0
2022-01-01 02:30:00 NaN NaN NaN 0.545976 -0.279299 2022 1 1 5 2 1 1 52 1.0 0.0
2022-01-01 02:45:00 NaN NaN NaN 0.436852 -0.279299 2022 1 1 5 2 1 1 52 1.0 0.0
2022-01-01 03:00:00 NaN NaN NaN 0.579058 -1.125973 2022 1 1 5 3 1 1 52 1.0 0.0
2022-01-01 03:15:00 NaN NaN NaN -0.480516 1.002109 2022 1 1 5 3 1 1 52 1.0 0.0
2022-01-01 03:30:00 NaN NaN NaN 0.220903 0.325239 2022 1 1 5 3 1 1 52 1.0 0.0
2. 组装含有学习器的工作流
Pipeline 的最后一个步骤可以是一个学习器，你仅需调用一次 fit 即可实现 Pipeline 中的各个特征工程方法与学习器对数据的拟合。
2.1. 构建
该 Pipeline 由以下部分组成：
用于异常值检测的 Ksigma 特征工程方法。
用于生成时间特征的 TimeFeatureGenerator 特征工程方法。
用于对时序数据建立时序模型的 MLPRegressor 学习器。
>>> from paddlets.models.dl.paddlepaddle import MLPRegressor
>>> mlp_params = {
>>> 'in_chunk_len': 3,
>>> 'out_chunk_len': 2,
>>> 'skip_chunk_len': 0,
>>> 'eval_metrics': ["mse", "mae"]
>>> }
>>> pipeline = Pipeline([(KSigma, {"cols":["observed_a", "observed_b", "known_c", "known_d"], "k": 1}), (TimeFeatureGenerator, {}), (MLPRegressor, mlp_params)])
2.2. 拟合 Pipeline 并执行预测
对于拟合后的 Pipeline，你可以使用 pipeline.predict 进行时序预测或使用 recursive_predict 进行多步时序预测。
>>> pipeline.fit(train_dataset)
>>> predicted_results = pipeline.predict(train_dataset)
>>> predicted_results
target
2022-01-01 01:45:00 2.543621
2022-01-01 02:00:00 -0.368826
2.3. 递归多步预测
递归多步预测的策略是：一步步利用 pipeline.predict 方法实现多步时序预测。当前时刻的预测结果会被添加至 TSDataset 时序数据集 的目标列中，并在模型预测时，被添加至历史目标值的滑动窗口中用于下一时刻预测。
注意：每次调用 pipeline.predict 的输出长度为 out_chunk_len， 所以 pipeline.predict 会被调用 ceiling(predict_length/out_chunk_len) 次，以满足多步时序的输出长度需求。例如，上述例子中的 Pipeline 的 out_chunk_length 是2，而 recursive_predict 允许你将 predict_length 设置为5或者更多：
>>> recursive_predicted_results = pipeline.recursive_predict(train_dataset, predict_length=5)
>>> recursive_predicted_results
target
2022-01-01 01:45:00 2.543621
2022-01-01 02:00:00 -0.368826
2022-01-01 02:15:00 3.192380
2022-01-01 02:30:00 -0.752583
2022-01-01 02:45:00 4.176333
注意： pipeline.recursive_predict 函数在 pipeline.skip_chunk != 0 时，无法使用。
注意： 在递归预测中，预测误差会被累计，因此预测性能会随着预测长度的增加而下降。
更多使用细节，请参考：API: pipeline.recursive_predict