家庭用電機器學習的多步時間序列預測 · Machine Learning Mastery 博客文章翻譯

# 家庭用電機器學習的多步時間序列預測 > 原文： [https://machinelearningmastery.com/multi-step-time-series-forecasting-with-machine-learning-models-for-household-electricity-consumption/](https://machinelearningmastery.com/multi-step-time-series-forecasting-with-machine-learning-models-for-household-electricity-consumption/) 鑒于智能電表的興起以及太陽能電池板等發電技術的廣泛采用，可提供大量的用電數據。該數據代表了多變量時間序列的功率相關變量，而這些變量又可用于建模甚至預測未來的電力消耗。機器學習算法預測單個值，不能直接用于多步預測。可以使用機器學習算法進行多步預測的兩種策略是遞歸和直接方法。在本教程中，您將了解如何使用機器學習算法開發遞歸和直接多步預測模型。完成本教程后，您將了解： * 如何開發一個評估線性，非線性和集成機器學習算法的框架，用于多步時間序列預測。 * 如何使用遞歸多步時間序列預測策略評估機器學習算法。 * 如何使用直接的每日和每個引導時間多步時間序列預測策略來評估機器學習算法。讓我們開始吧。 ![Multi-step Time Series Forecasting with Machine Learning Models for Household Electricity Consumption](https://img.kancloud.cn/e6/d4/e6d4baed9d9554962fbc46c10a941f29_640x398.jpg) 用于家庭用電的機器學習模型的多步驟時間序列預測照片由 [Sean McMenemy](https://www.flickr.com/photos/seanfx/9827244314/) ，保留一些權利。 ## 教程概述本教程分為五個部分;他們是： 1. 問題描述 2. 加載并準備數據集 3. 模型評估 4. 遞歸多步預測 5. 直接多步預測 ## 問題描述 '[家庭用電量](https://archive.ics.uci.edu/ml/datasets/individual+household+electric+power+consumption)'數據集是一個多變量時間序列數據集，描述了四年內單個家庭的用電量。該數據是在 2006 年 12 月至 2010 年 11 月之間收集的，并且每分鐘收集家庭內的能耗觀察結果。它是一個多變量系列，由七個變量組成（除日期和時間外）;他們是： * **global_active_power** ：家庭消耗的總有功功率（千瓦）。 * **global_reactive_power** ：家庭消耗的總無功功率（千瓦）。 * **電壓**：平均電壓（伏特）。 * **global_intensity** ：平均電流強度（安培）。 * **sub_metering_1** ：廚房的有功電能（瓦特小時的有功電能）。 * **sub_metering_2** ：用于洗衣的有功能量（瓦特小時的有功電能）。 * **sub_metering_3** ：氣候控制系統的有功電能（瓦特小時的有功電能）。有功和無功電能參考[交流電](https://en.wikipedia.org/wiki/AC_power)的技術細節。可以通過從總活動能量中減去三個定義的子計量變量的總和來創建第四個子計量變量，如下所示： ```py sub_metering_remainder = (global_active_power * 1000 / 60) - (sub_metering_1 + sub_metering_2 + sub_metering_3) ``` ## 加載并準備數據集數據集可以從 UCI 機器學習庫下載為單個 20 兆字節的.zip 文件： * [household_power_consumption.zip](https://archive.ics.uci.edu/ml/machine-learning-databases/00235/household_power_consumption.zip) 下載數據集并將其解壓縮到當前工作目錄中。您現在將擁有大約 127 兆字節的文件“ _household_power_consumption.txt_ ”并包含所有觀察結果。我們可以使用 _read_csv（）_ 函數來加載數據，并將前兩列合并到一個日期時間列中，我們可以將其用作索引。 ```py # load all data dataset = read_csv('household_power_consumption.txt', sep=';', header=0, low_memory=False, infer_datetime_format=True, parse_dates={'datetime':[0,1]}, index_col=['datetime']) ``` 接下來，我們可以用'_ 標記所有[缺失值](https://machinelearningmastery.com/handle-missing-timesteps-sequence-prediction-problems-python/)？_ '具有 _NaN_ 值的字符，這是一個浮點數。這將允許我們將數據作為一個浮點值數組而不是混合類型（效率較低）。 ```py # mark all missing values dataset.replace('?', nan, inplace=True) # make dataset numeric dataset = dataset.astype('float32') ``` 我們還需要填寫缺失值，因為它們已被標記。一種非常簡單的方法是從前一天的同一時間復制觀察。我們可以在一個名為 _fill_missing（）_ 的函數中實現它，該函數將從 24 小時前獲取數據的 NumPy 數組并復制值。 ```py # fill missing values with a value at the same time one day ago def fill_missing(values): one_day = 60 * 24 for row in range(values.shape[0]): for col in range(values.shape[1]): if isnan(values[row, col]): values[row, col] = values[row - one_day, col] ``` 我們可以將此函數直接應用于 DataFrame 中的數據。 ```py # fill missing fill_missing(dataset.values) ``` 現在，我們可以使用上一節中的計算創建一個包含剩余子計量的新列。 ```py # add a column for for the remainder of sub metering values = dataset.values dataset['sub_metering_4'] = (values[:,0] * 1000 / 60) - (values[:,4] + values[:,5] + values[:,6]) ``` 我們現在可以將清理后的數據集版本保存到新文件中;在這種情況下，我們只需將文件擴展名更改為.csv，并將數據集保存為“ _household_power_consumption.csv_ ”。 ```py # save updated dataset dataset.to_csv('household_power_consumption.csv') ``` 將所有這些結合在一起，下面列出了加載，清理和保存數據集的完整示例。 ```py # load and clean-up data from numpy import nan from numpy import isnan from pandas import read_csv from pandas import to_numeric # fill missing values with a value at the same time one day ago def fill_missing(values): one_day = 60 * 24 for row in range(values.shape[0]): for col in range(values.shape[1]): if isnan(values[row, col]): values[row, col] = values[row - one_day, col] # load all data dataset = read_csv('household_power_consumption.txt', sep=';', header=0, low_memory=False, infer_datetime_format=True, parse_dates={'datetime':[0,1]}, index_col=['datetime']) # mark all missing values dataset.replace('?', nan, inplace=True) # make dataset numeric dataset = dataset.astype('float32') # fill missing fill_missing(dataset.values) # add a column for for the remainder of sub metering values = dataset.values dataset['sub_metering_4'] = (values[:,0] * 1000 / 60) - (values[:,4] + values[:,5] + values[:,6]) # save updated dataset dataset.to_csv('household_power_consumption.csv') ``` 運行該示例將創建新文件' _household_power_consumption.csv_ '，我們可以將其用作建模項目的起點。 ## 模型評估在本節中，我們將考慮如何開發和評估家庭電力數據集的預測模型。本節分為四個部分;他們是： 1. 問題框架 2. 評估指標 3. 訓練和測試集 4. 前瞻性驗證 ### 問題框架有許多方法可以利用和探索家庭用電量數據集。在本教程中，我們將使用這些數據來探索一個非常具體的問題;那是： > 鑒于最近的耗電量，未來一周的預期耗電量是多少？這要求預測模型預測未來七天每天的總有功功率。從技術上講，考慮到多個預測步驟，這個問題的框架被稱為多步驟時間序列預測問題。利用多個輸入變量的模型可以稱為多變量多步時間序列預測模型。這種類型的模型在規劃支出方面可能有助于家庭。在供應方面，它也可能有助于規劃特定家庭的電力需求。數據集的這種框架還表明，將每分鐘功耗的觀察結果下采樣到每日總數是有用的。這不是必需的，但考慮到我們對每天的總功率感興趣，這是有道理的。我們可以使用 pandas DataFrame 上的 [resample（）函數](https://pandas.pydata.org/pandas-docs/stable/generated/pandas.DataFrame.resample.html)輕松實現這一點。使用參數' _D_ '調用此函數允許按日期時間索引的加載數據按天分組（[查看所有偏移別名](http://pandas.pydata.org/pandas-docs/stable/timeseries.html#offset-aliases)）。然后，我們可以計算每天所有觀測值的總和，并為八個變量中的每一個創建每日耗電量數據的新數據集。下面列出了完整的示例。 ```py # resample minute data to total for each day from pandas import read_csv # load the new file dataset = read_csv('household_power_consumption.csv', header=0, infer_datetime_format=True, parse_dates=['datetime'], index_col=['datetime']) # resample data to daily daily_groups = dataset.resample('D') daily_data = daily_groups.sum() # summarize print(daily_data.shape) print(daily_data.head()) # save daily_data.to_csv('household_power_consumption_days.csv') ``` 運行該示例將創建一個新的每日總功耗數據集，并將結果保存到名為“ _household_power_consumption_days.csv_ ”的單獨文件中。我們可以將其用作數據集，用于擬合和評估所選問題框架的預測模型。 ### 評估指標預測將包含七個值，一個用于一周中的每一天。多步預測問題通常分別評估每個預測時間步長。這有助于以下幾個原因： * 在特定提前期評論技能（例如+1 天 vs +3 天）。 * 在不同的交付時間基于他們的技能對比模型（例如，在+1 天的模型和在日期+5 的模型良好的模型）。總功率的單位是千瓦，并且具有也在相同單位的誤差度量將是有用的。均方根誤差（RMSE）和平均絕對誤差（MAE）都符合這個要求，盡管 RMSE 更常用，將在本教程中采用。與 MAE 不同，RMSE 更能預測預測誤差。此問題的表現指標是從第 1 天到第 7 天的每個提前期的 RMSE。作為捷徑，使用單個分數總結模型的表現以幫助模型選擇可能是有用的。可以使用的一個可能的分數是所有預測天數的 RMSE。下面的函數 _evaluate_forecasts（）_ 將實現此行為并基于多個七天預測返回模型的表現。 ```py # evaluate one or more weekly forecasts against expected values def evaluate_forecasts(actual, predicted): scores = list() # calculate an RMSE score for each day for i in range(actual.shape[1]): # calculate mse mse = mean_squared_error(actual[:, i], predicted[:, i]) # calculate rmse rmse = sqrt(mse) # store scores.append(rmse) # calculate overall RMSE s = 0 for row in range(actual.shape[0]): for col in range(actual.shape[1]): s += (actual[row, col] - predicted[row, col])**2 score = sqrt(s / (actual.shape[0] * actual.shape[1])) return score, scores ``` 運行該函數將首先返回整個 RMSE，無論白天，然后每天返回一系列 RMSE 分數。 ### 訓練和測試集我們將使用前三年的數據來訓練預測模型和評估模型的最后一年。給定數據集中的數據將分為標準周。這些是從周日開始到周六結束的周。這是使用所選模型框架的現實且有用的方法，其中可以預測未來一周的功耗。它也有助于建模，其中模型可用于預測特定日期（例如星期三）或整個序列。我們將數據拆分為標準周，從測試數據集向后工作。數據的最后一年是 2010 年，2010 年的第一個星期日是 1 月 3 日。數據于 2010 年 11 月中旬結束，數據中最接近的最后一個星期六是 11 月 20 日。這給出了 46 周的測試數據。下面提供了測試數據集的每日數據的第一行和最后一行以供確認。 ```py 2010-01-03,2083.4539999999984,191.61000000000055,350992.12000000034,8703.600000000033,3842.0,4920.0,10074.0,15888.233355799992 ... 2010-11-20,2197.006000000004,153.76800000000028,346475.9999999998,9320.20000000002,4367.0,2947.0,11433.0,17869.76663959999 ``` 每日數據從 2006 年底開始。數據集中的第一個星期日是 12 月 17 日，這是第二行數據。將數據組織到標準周內為訓練預測模型提供了 159 個完整的標準周。 ```py 2006-12-17,3390.46,226.0059999999994,345725.32000000024,14398.59999999998,2033.0,4187.0,13341.0,36946.66673200004 ... 2010-01-02,1309.2679999999998,199.54600000000016,352332.8399999997,5489.7999999999865,801.0,298.0,6425.0,14297.133406600002 ``` 下面的函數 _split_dataset（）_ 將每日數據拆分為訓練集和測試集，并將每個數據組織成標準周。使用特定行偏移來使用數據集的知識來分割數據。然后使用 NumPy [split（）函數](https://docs.scipy.org/doc/numpy/reference/generated/numpy.split.html)將分割數據集組織成每周數據。 ```py # split a univariate dataset into train/test sets def split_dataset(data): # split into standard weeks train, test = data[1:-328], data[-328:-6] # restructure into windows of weekly data train = array(split(train, len(train)/7)) test = array(split(test, len(test)/7)) return train, test ``` 我們可以通過加載每日數據集并打印訓練和測試集的第一行和最后一行數據來測試此功能，以確認它們符合上述預期。完整的代碼示例如下所示。 ```py # split into standard weeks from numpy import split from numpy import array from pandas import read_csv # split a univariate dataset into train/test sets def split_dataset(data): # split into standard weeks train, test = data[1:-328], data[-328:-6] # restructure into windows of weekly data train = array(split(train, len(train)/7)) test = array(split(test, len(test)/7)) return train, test # load the new file dataset = read_csv('household_power_consumption_days.csv', header=0, infer_datetime_format=True, parse_dates=['datetime'], index_col=['datetime']) train, test = split_dataset(dataset.values) # validate train data print(train.shape) print(train[0, 0, 0], train[-1, -1, 0]) # validate test print(test.shape) print(test[0, 0, 0], test[-1, -1, 0]) ``` 運行該示例表明，訓練數據集確實有 159 周的數據，而測試數據集有 46 周。我們可以看到，第一行和最后一行的訓練和測試數據集的總有效功率與我們定義為每組標準周界限的特定日期的數據相匹配。 ```py (159, 7, 8) 3390.46 1309.2679999999998 (46, 7, 8) 2083.4539999999984 2197.006000000004 ``` ### 前瞻性驗證將使用稱為前向驗證的方案來評估模型。這是需要模型進行一周預測的地方，然后該模型的實際數據可用于模型，以便它可以用作在隨后一周進行預測的基礎。這對于如何在實踐中使用模型以及對模型有益，使其能夠利用最佳可用數據都是現實的。我們可以通過分離輸入數據和輸出/預測數據來證明這一點。 ```py Input, Predict [Week1] Week2 [Week1 + Week2] Week3 [Week1 + Week2 + Week3] Week4 ... ``` 評估此數據集上的預測模型的前瞻性驗證方法在下面的函數中提供，名為 _evaluate_model（）_。提供 scikit-learn 模型對象作為函數的參數，以及訓練和測試數據集。提供了另一個參數 _n_input_ ，用于定義模型將用作輸入以進行預測的先前觀察的數量。關于 scikit-learn 模型如何適合并進行預測的細節將在后面的章節中介紹。然后使用先前定義的 _evaluate_forecasts（）_ 函數，針對測試數據集評估模型所做的預測。 ```py # evaluate a single model def evaluate_model(model, train, test, n_input): # history is a list of weekly data history = [x for x in train] # walk-forward validation over each week predictions = list() for i in range(len(test)): # predict the week yhat_sequence = ... # store the predictions predictions.append(yhat_sequence) # get real observation and add to history for predicting the next week history.append(test[i, :]) predictions = array(predictions) # evaluate predictions days for each week score, scores = evaluate_forecasts(test[:, :, 0], predictions) return score, scores ``` 一旦我們對模型進行評估，我們就可以總結表現。下面的函數名為 _summarize_scores（）_，將模型的表現顯示為單行，以便與其他模型進行比較。 ```py # summarize scores def summarize_scores(name, score, scores): s_scores = ', '.join(['%.1f' % s for s in scores]) print('%s: [%.3f] %s' % (name, score, s_scores)) ``` 我們現在已經開始評估數據集上的預測模型的所有元素。 ## 遞歸多步預測大多數預測建模算法將采用一些觀測值作為輸入并預測單個輸出值。因此，它們不能直接用于進行多步驟時間序列預測。這適用于大多數線性，非線性和整體機器學習算法。可以使用機器學習算法進行多步時間序列預測的一種方法是遞歸地使用它們。這涉及對一個時間步進行預測，進行預測，并將其作為輸入饋送到模型中，以便預測隨后的時間步。重復該過程直到預測到所需數量的步驟。例如： ```py X = [x1, x2, x3] y1 = model.predict(X) X = [x2, x3, y1] y2 = model.predict(X) X = [x3, y1, y2] y3 = model.predict(X) ... ``` 在本節中，我們將開發一個測試工具，用于擬合和評估 scikit-learn 中提供的機器學習算法，使用遞歸模型進行多步預測。第一步是將窗口格式的準備好的訓練數據轉換為單個單變量系列。下面的 _to_series（）_ 功能會將每周多變量數據列表轉換為每日消耗的單變量單變量系列。 ```py # convert windows of weekly multivariate data into a series of total power def to_series(data): # extract just the total power from each week series = [week[:, 0] for week in data] # flatten into a single series series = array(series).flatten() return series ``` 接下來，需要將每日電力的順序轉換成適合于監督學習問題的輸入和輸出。預測將是前幾天消耗的總功率的一些函數。我們可以選擇用作輸入的前幾天數，例如一周或兩周。總會有一個輸出：第二天消耗的總功率。該模型將適合先前時間步驟的真實觀察結果。我們需要迭代消耗的每日功率序列并將其分成輸入和輸出。這稱為滑動窗口數據表示。下面的 _to_supervised（）_ 函數實現了這種行為。它將每周數據列表作為輸入，以及用作創建的每個樣本的輸入的前幾天的數量。第一步是將歷史記錄轉換為單個數據系列。然后枚舉該系列，每個時間步創建一個輸入和輸出對。這個問題的框架將允許模型學習根據前幾天的觀察結果預測一周中的任何一天。該函數返回輸入（X）和輸出（y），以便訓練模型。 ```py # convert history into inputs and outputs def to_supervised(history, n_input): # convert history to a univariate series data = to_series(history) X, y = list(), list() ix_start = 0 # step over the entire history one time step at a time for i in range(len(data)): # define the end of the input sequence ix_end = ix_start + n_input # ensure we have enough data for this instance if ix_end < len(data): X.append(data[ix_start:ix_end]) y.append(data[ix_end]) # move along one time step ix_start += 1 return array(X), array(y) ``` scikit-learn 庫允許將模型用作管道的一部分。這允許在擬合模型之前自動應用數據變換。更重要的是，變換以正確的方式準備，在那里準備或適合訓練數據并應用于測試數據。這可以在評估模型時防止數據泄漏。在通過在訓練數據集上擬合每個模型之前創建管道來評估模型時，我們可以使用此功能。在使用模型之前，我們將標準化和標準化數據。下面的 _make_pipeline（）_ 函數實現了這種行為，返回一個可以像模型一樣使用的 Pipeline，例如它可以適合它，它可以做出預測。每列執行標準化和規范化操作。在 _to_supervised（）_ 功能中，我們基本上將一列數據（總功率）分成多個列，例如，七天七天的輸入觀察。這意味著輸入數據中的七列中的每一列將具有用于標準化的不同均值和標準偏差以及用于歸一化的不同最小值和最大值。鑒于我們使用了滑動窗口，幾乎所有值都會出現在每列中，因此，這可能不是問題。但重要的是要注意，在將數據拆分為輸入和輸出之前將數據縮放為單個列會更加嚴格。 ```py # create a feature preparation pipeline for a model def make_pipeline(model): steps = list() # standardization steps.append(('standardize', StandardScaler())) # normalization steps.append(('normalize', MinMaxScaler())) # the model steps.append(('model', model)) # create pipeline pipeline = Pipeline(steps=steps) return pipeline ``` 我們可以將這些元素組合成一個名為 _sklearn_predict（）_ 的函數，如下所示。該函數采用 scikit-learn 模型對象，訓練數據，稱為歷史記錄以及用作輸入的指定數量的前幾天。它將訓練數據轉換為輸入和輸出，將模型包裝在管道中，使其適合，并使用它進行預測。 ```py # fit a model and make a forecast def sklearn_predict(model, history, n_input): # prepare data train_x, train_y = to_supervised(history, n_input) # make pipeline pipeline = make_pipeline(model) # fit the model pipeline.fit(train_x, train_y) # predict the week, recursively yhat_sequence = forecast(pipeline, train_x[-1, :], n_input) return yhat_sequence ``` 模型將使用訓練數據集中的最后一行作為輸入以進行預測。 _forecast（）_ 函數將使用該模型進行遞歸多步預測。遞歸預測涉及迭代多步預測所需的七天中的每一天。將模型的輸入數據作為 _input_data_ 列表的最后幾個觀察值。此列表附有訓練數據最后一行的所有觀察結果，當我們使用模型進行預測時，它們會添加到此列表的末尾。因此，我們可以從該列表中獲取最后的 _n_input_ 觀測值，以實現提供先前輸出作為輸入的效果。該模型用于對準備好的輸入數據進行預測，并將輸出添加到我們將返回的實際輸出序列的列表和輸入數據列表中，我們將從中輸出觀察值作為模型的輸入。下一次迭代。 ```py # make a recursive multi-step forecast def forecast(model, input_x, n_input): yhat_sequence = list() input_data = [x for x in input_x] for j in range(7): # prepare the input data X = array(input_data[-n_input:]).reshape(1, n_input) # make a one-step forecast yhat = model.predict(X)[0] # add to the result yhat_sequence.append(yhat) # add the prediction to the input input_data.append(yhat) return yhat_sequence ``` 我們現在擁有使用遞歸多步預測策略來擬合和評估 scikit-learn 模型的所有元素。我們可以更新上一節中定義的 _evaluate_model（）_ 函數來調用 _sklearn_predict（）_ 函數。更新的功能如下所示。 ```py # evaluate a single model def evaluate_model(model, train, test, n_input): # history is a list of weekly data history = [x for x in train] # walk-forward validation over each week predictions = list() for i in range(len(test)): # predict the week yhat_sequence = sklearn_predict(model, history, n_input) # store the predictions predictions.append(yhat_sequence) # get real observation and add to history for predicting the next week history.append(test[i, :]) predictions = array(predictions) # evaluate predictions days for each week score, scores = evaluate_forecasts(test[:, :, 0], predictions) return score, scores ``` 一個重要的最終函數是 _get_models（）_，它定義了一個 scikit-learn 模型對象的字典，映射到我們可以用于報告的簡寫名稱。我們將從評估一套線性算法開始。我們期望這些表現類似于自回歸模型（例如，如果使用七天的輸入，則為 AR（7））。具有十個線性模型的 _get_models（）_ 函數定義如下。這是一個抽查，我們對各種算法的一般表現感興趣，而不是優化任何給定的算法。 ```py # prepare a list of ml models def get_models(models=dict()): # linear models models['lr'] = LinearRegression() models['lasso'] = Lasso() models['ridge'] = Ridge() models['en'] = ElasticNet() models['huber'] = HuberRegressor() models['lars'] = Lars() models['llars'] = LassoLars() models['pa'] = PassiveAggressiveRegressor(max_iter=1000, tol=1e-3) models['ranscac'] = RANSACRegressor() models['sgd'] = SGDRegressor(max_iter=1000, tol=1e-3) print('Defined %d models' % len(models)) return models ``` 最后，我們可以將所有這些結合在一起。首先，加載數據集并將其拆分為訓練集和測試集。 ```py # load the new file dataset = read_csv('household_power_consumption_days.csv', header=0, infer_datetime_format=True, parse_dates=['datetime'], index_col=['datetime']) # split into train and test train, test = split_dataset(dataset.values) ``` 然后，我們可以準備模型字典并定義觀察的前幾天的數量，以用作模型的輸入。 ```py # prepare the models to evaluate models = get_models() n_input = 7 ``` 然后枚舉字典中的模型，評估每個模型，總結其分數，并將結果添加到線圖中。下面列出了完整的示例。 ```py # recursive multi-step forecast with linear algorithms from math import sqrt from numpy import split from numpy import array from pandas import read_csv from sklearn.metrics import mean_squared_error from matplotlib import pyplot from sklearn.preprocessing import StandardScaler from sklearn.preprocessing import MinMaxScaler from sklearn.pipeline import Pipeline from sklearn.linear_model import LinearRegression from sklearn.linear_model import Lasso from sklearn.linear_model import Ridge from sklearn.linear_model import ElasticNet from sklearn.linear_model import HuberRegressor from sklearn.linear_model import Lars from sklearn.linear_model import LassoLars from sklearn.linear_model import PassiveAggressiveRegressor from sklearn.linear_model import RANSACRegressor from sklearn.linear_model import SGDRegressor # split a univariate dataset into train/test sets def split_dataset(data): # split into standard weeks train, test = data[1:-328], data[-328:-6] # restructure into windows of weekly data train = array(split(train, len(train)/7)) test = array(split(test, len(test)/7)) return train, test # evaluate one or more weekly forecasts against expected values def evaluate_forecasts(actual, predicted): scores = list() # calculate an RMSE score for each day for i in range(actual.shape[1]): # calculate mse mse = mean_squared_error(actual[:, i], predicted[:, i]) # calculate rmse rmse = sqrt(mse) # store scores.append(rmse) # calculate overall RMSE s = 0 for row in range(actual.shape[0]): for col in range(actual.shape[1]): s += (actual[row, col] - predicted[row, col])**2 score = sqrt(s / (actual.shape[0] * actual.shape[1])) return score, scores # summarize scores def summarize_scores(name, score, scores): s_scores = ', '.join(['%.1f' % s for s in scores]) print('%s: [%.3f] %s' % (name, score, s_scores)) # prepare a list of ml models def get_models(models=dict()): # linear models models['lr'] = LinearRegression() models['lasso'] = Lasso() models['ridge'] = Ridge() models['en'] = ElasticNet() models['huber'] = HuberRegressor() models['lars'] = Lars() models['llars'] = LassoLars() models['pa'] = PassiveAggressiveRegressor(max_iter=1000, tol=1e-3) models['ranscac'] = RANSACRegressor() models['sgd'] = SGDRegressor(max_iter=1000, tol=1e-3) print('Defined %d models' % len(models)) return models # create a feature preparation pipeline for a model def make_pipeline(model): steps = list() # standardization steps.append(('standardize', StandardScaler())) # normalization steps.append(('normalize', MinMaxScaler())) # the model steps.append(('model', model)) # create pipeline pipeline = Pipeline(steps=steps) return pipeline # make a recursive multi-step forecast def forecast(model, input_x, n_input): yhat_sequence = list() input_data = [x for x in input_x] for j in range(7): # prepare the input data X = array(input_data[-n_input:]).reshape(1, n_input) # make a one-step forecast yhat = model.predict(X)[0] # add to the result yhat_sequence.append(yhat) # add the prediction to the input input_data.append(yhat) return yhat_sequence # convert windows of weekly multivariate data into a series of total power def to_series(data): # extract just the total power from each week series = [week[:, 0] for week in data] # flatten into a single series series = array(series).flatten() return series # convert history into inputs and outputs def to_supervised(history, n_input): # convert history to a univariate series data = to_series(history) X, y = list(), list() ix_start = 0 # step over the entire history one time step at a time for i in range(len(data)): # define the end of the input sequence ix_end = ix_start + n_input # ensure we have enough data for this instance if ix_end < len(data): X.append(data[ix_start:ix_end]) y.append(data[ix_end]) # move along one time step ix_start += 1 return array(X), array(y) # fit a model and make a forecast def sklearn_predict(model, history, n_input): # prepare data train_x, train_y = to_supervised(history, n_input) # make pipeline pipeline = make_pipeline(model) # fit the model pipeline.fit(train_x, train_y) # predict the week, recursively yhat_sequence = forecast(pipeline, train_x[-1, :], n_input) return yhat_sequence # evaluate a single model def evaluate_model(model, train, test, n_input): # history is a list of weekly data history = [x for x in train] # walk-forward validation over each week predictions = list() for i in range(len(test)): # predict the week yhat_sequence = sklearn_predict(model, history, n_input) # store the predictions predictions.append(yhat_sequence) # get real observation and add to history for predicting the next week history.append(test[i, :]) predictions = array(predictions) # evaluate predictions days for each week score, scores = evaluate_forecasts(test[:, :, 0], predictions) return score, scores # load the new file dataset = read_csv('household_power_consumption_days.csv', header=0, infer_datetime_format=True, parse_dates=['datetime'], index_col=['datetime']) # split into train and test train, test = split_dataset(dataset.values) # prepare the models to evaluate models = get_models() n_input = 7 # evaluate each model days = ['sun', 'mon', 'tue', 'wed', 'thr', 'fri', 'sat'] for name, model in models.items(): # evaluate and get scores score, scores = evaluate_model(model, train, test, n_input) # summarize scores summarize_scores(name, score, scores) # plot scores pyplot.plot(days, scores, marker='o', label=name) # show plot pyplot.legend() pyplot.show() ``` 運行該示例將評估十個線性算法并總結結果。評估每個算法并使用單行摘要報告表現，包括總體 RMSE 以及每次步驟 RMSE。我們可以看到，大多數評估模型表現良好，整周誤差低于 400 千瓦，隨機隨機梯度下降（SGD）回歸量表現最佳，總體 RMSE 約為 383。 ```py Defined 10 models lr: [388.388] 411.0, 389.1, 338.0, 370.8, 408.5, 308.3, 471.1 lasso: [386.838] 403.6, 388.9, 337.3, 371.1, 406.1, 307.6, 471.6 ridge: [387.659] 407.9, 388.6, 337.5, 371.2, 407.0, 307.7, 471.7 en: [469.337] 452.2, 451.9, 435.8, 485.7, 460.4, 405.8, 575.1 huber: [392.465] 412.1, 388.0, 337.9, 377.3, 405.6, 306.9, 492.5 lars: [388.388] 411.0, 389.1, 338.0, 370.8, 408.5, 308.3, 471.1 llars: [388.406] 396.1, 387.8, 339.3, 377.8, 402.9, 310.3, 481.9 pa: [399.402] 410.0, 391.7, 342.2, 389.7, 409.8, 315.9, 508.4 ranscac: [439.945] 454.0, 424.0, 369.5, 421.5, 457.5, 409.7, 526.9 sgd: [383.177] 400.3, 386.0, 333.0, 368.9, 401.5, 303.9, 466.9 ``` 還創建了 10 個分類器中每個分類器的每日 RMSE 的線圖。我們可以看到，除了兩種方法之外，其他所有方法都會聚集在一起，在七天預測中表現同樣出色。 ![Line Plot of Recursive Multi-step Forecasts With Linear Algorithms](https://img.kancloud.cn/e4/90/e4902118850c61ec91c68f067fb49662_1280x960.jpg) 線性算法的遞歸多步預測線圖通過調整一些表現更好的算法的超參數可以獲得更好的結果。此外，更新示例以測試一套非線性和集合算法可能會很有趣。一個有趣的實驗可能是評估一個或幾個表現更好的算法的表現，前一天或多或少作為輸入。 ## 直接多步預測多步預測的遞歸策略的替代方案是對每個要預測的日子使用不同的模型。這稱為直接多步預測策略。因為我們有興趣預測七天，所以需要準備七個不同的模型，每個模型專門用于預測不同的一天。訓練這種模型有兩種方法： * **預測日**。可以準備模型來預測標準周的特定日期，例如星期一。 * **預測提前期**。可以準備模型來預測特定的提前期，例如第 1 天預測一天將更具體，但意味著每個模型可以使用更少的訓練數據。預測提前期會使用更多的訓練數據，但需要模型在一周的不同日期進行推廣。我們將在本節中探討這兩種方法。 ### 直接日方法首先，我們必須更新 _to_supervised（）_ 函數以準備數據，例如用作輸入的前一周觀察數據以及用作輸出的下周特定日期的觀察結果。下面列出了實現此行為的更新的 _to_supervised（）_ 函數。它需要一個參數 _output_ix_ 來定義下一周的日[0,6]以用作輸出。 ```py # convert history into inputs and outputs def to_supervised(history, output_ix): X, y = list(), list() # step over the entire history one time step at a time for i in range(len(history)-1): X.append(history[i][:,0]) y.append(history[i + 1][output_ix,0]) return array(X), array(y) ``` 此功能可以調用七次，對于所需的七種型號中的每一種都可以調用一次。接下來，我們可以更新 _sklearn_predict（）_ 函數，為一周預測中的每一天創建一個新數據集和一個新模型。函數的主體大部分沒有變化，只是在輸出序列中每天在循環中使用它，其中“ _i_ ”的索引被傳遞給 _to_supervised 的調用（）_ 為了準備一個特定的數據集來訓練模型來預測那一天。該函數不再采用 _n_input_ 參數，因為我們已將輸入修復為前一周的七天。 ```py # fit a model and make a forecast def sklearn_predict(model, history): yhat_sequence = list() # fit a model for each forecast day for i in range(7): # prepare data train_x, train_y = to_supervised(history, i) # make pipeline pipeline = make_pipeline(model) # fit the model pipeline.fit(train_x, train_y) # forecast x_input = array(train_x[-1, :]).reshape(1,7) yhat = pipeline.predict(x_input)[0] # store yhat_sequence.append(yhat) return yhat_sequence ``` 下面列出了完整的示例。 ```py # direct multi-step forecast by day from math import sqrt from numpy import split from numpy import array from pandas import read_csv from sklearn.metrics import mean_squared_error from matplotlib import pyplot from sklearn.preprocessing import StandardScaler from sklearn.preprocessing import MinMaxScaler from sklearn.pipeline import Pipeline from sklearn.linear_model import LinearRegression from sklearn.linear_model import Lasso from sklearn.linear_model import Ridge from sklearn.linear_model import ElasticNet from sklearn.linear_model import HuberRegressor from sklearn.linear_model import Lars from sklearn.linear_model import LassoLars from sklearn.linear_model import PassiveAggressiveRegressor from sklearn.linear_model import RANSACRegressor from sklearn.linear_model import SGDRegressor # split a univariate dataset into train/test sets def split_dataset(data): # split into standard weeks train, test = data[1:-328], data[-328:-6] # restructure into windows of weekly data train = array(split(train, len(train)/7)) test = array(split(test, len(test)/7)) return train, test # evaluate one or more weekly forecasts against expected values def evaluate_forecasts(actual, predicted): scores = list() # calculate an RMSE score for each day for i in range(actual.shape[1]): # calculate mse mse = mean_squared_error(actual[:, i], predicted[:, i]) # calculate rmse rmse = sqrt(mse) # store scores.append(rmse) # calculate overall RMSE s = 0 for row in range(actual.shape[0]): for col in range(actual.shape[1]): s += (actual[row, col] - predicted[row, col])**2 score = sqrt(s / (actual.shape[0] * actual.shape[1])) return score, scores # summarize scores def summarize_scores(name, score, scores): s_scores = ', '.join(['%.1f' % s for s in scores]) print('%s: [%.3f] %s' % (name, score, s_scores)) # prepare a list of ml models def get_models(models=dict()): # linear models models['lr'] = LinearRegression() models['lasso'] = Lasso() models['ridge'] = Ridge() models['en'] = ElasticNet() models['huber'] = HuberRegressor() models['lars'] = Lars() models['llars'] = LassoLars() models['pa'] = PassiveAggressiveRegressor(max_iter=1000, tol=1e-3) models['ranscac'] = RANSACRegressor() models['sgd'] = SGDRegressor(max_iter=1000, tol=1e-3) print('Defined %d models' % len(models)) return models # create a feature preparation pipeline for a model def make_pipeline(model): steps = list() # standardization steps.append(('standardize', StandardScaler())) # normalization steps.append(('normalize', MinMaxScaler())) # the model steps.append(('model', model)) # create pipeline pipeline = Pipeline(steps=steps) return pipeline # convert history into inputs and outputs def to_supervised(history, output_ix): X, y = list(), list() # step over the entire history one time step at a time for i in range(len(history)-1): X.append(history[i][:,0]) y.append(history[i + 1][output_ix,0]) return array(X), array(y) # fit a model and make a forecast def sklearn_predict(model, history): yhat_sequence = list() # fit a model for each forecast day for i in range(7): # prepare data train_x, train_y = to_supervised(history, i) # make pipeline pipeline = make_pipeline(model) # fit the model pipeline.fit(train_x, train_y) # forecast x_input = array(train_x[-1, :]).reshape(1,7) yhat = pipeline.predict(x_input)[0] # store yhat_sequence.append(yhat) return yhat_sequence # evaluate a single model def evaluate_model(model, train, test): # history is a list of weekly data history = [x for x in train] # walk-forward validation over each week predictions = list() for i in range(len(test)): # predict the week yhat_sequence = sklearn_predict(model, history) # store the predictions predictions.append(yhat_sequence) # get real observation and add to history for predicting the next week history.append(test[i, :]) predictions = array(predictions) # evaluate predictions days for each week score, scores = evaluate_forecasts(test[:, :, 0], predictions) return score, scores # load the new file dataset = read_csv('household_power_consumption_days.csv', header=0, infer_datetime_format=True, parse_dates=['datetime'], index_col=['datetime']) # split into train and test train, test = split_dataset(dataset.values) # prepare the models to evaluate models = get_models() # evaluate each model days = ['sun', 'mon', 'tue', 'wed', 'thr', 'fri', 'sat'] for name, model in models.items(): # evaluate and get scores score, scores = evaluate_model(model, train, test) # summarize scores summarize_scores(name, score, scores) # plot scores pyplot.plot(days, scores, marker='o', label=name) # show plot pyplot.legend() pyplot.show() ``` 首先運行該示例總結了每個模型的表現。我們可以看到表現比這個問題的遞歸模型略差。 ```py Defined 10 models lr: [410.927] 463.8, 381.4, 351.9, 430.7, 387.8, 350.4, 488.8 lasso: [408.440] 458.4, 378.5, 352.9, 429.5, 388.0, 348.0, 483.5 ridge: [403.875] 447.1, 377.9, 347.5, 427.4, 384.1, 343.4, 479.7 en: [454.263] 471.8, 433.8, 415.8, 477.4, 434.4, 373.8, 551.8 huber: [409.500] 466.8, 380.2, 359.8, 432.4, 387.0, 351.3, 470.9 lars: [410.927] 463.8, 381.4, 351.9, 430.7, 387.8, 350.4, 488.8 llars: [406.490] 453.0, 378.8, 357.3, 428.1, 388.0, 345.0, 476.9 pa: [402.476] 428.4, 380.9, 356.5, 426.7, 390.4, 348.6, 471.4 ranscac: [497.225] 456.1, 423.0, 445.9, 547.6, 521.9, 451.5, 607.2 sgd: [403.526] 441.4, 378.2, 354.5, 423.9, 382.4, 345.8, 480.3 ``` 還創建了每個模型的每日 RMSE 得分的線圖，顯示了與遞歸模型所見的類似的模型分組。 ![Line Plot of Direct Per-Day Multi-step Forecasts With Linear Algorithms](https://img.kancloud.cn/aa/ef/aaefcbce6795e518eb0229db28c5c0eb_1280x960.jpg) 線性算法的直接每日多步預測線圖 ### 直接提前期方法直接提前期方法是相同的，除了 _to_supervised（）_ 使用更多的訓練數據集。該函數與遞歸模型示例中定義的函數相同，只是它需要額外的 _output_ix_ 參數來定義下一周中用作輸出的日期。下面列出了直接每個引導時間策略的更新 _to_supervised（）_ 函數。與每日策略不同，此版本的功能支持可變大小的輸入（不僅僅是七天），如果您愿意，可以進行實驗。 ```py # convert history into inputs and outputs def to_supervised(history, n_input, output_ix): # convert history to a univariate series data = to_series(history) X, y = list(), list() ix_start = 0 # step over the entire history one time step at a time for i in range(len(data)): # define the end of the input sequence ix_end = ix_start + n_input ix_output = ix_end + output_ix # ensure we have enough data for this instance if ix_output < len(data): X.append(data[ix_start:ix_end]) y.append(data[ix_output]) # move along one time step ix_start += 1 return array(X), array(y) ``` 下面列出了完整的示例。 ```py # direct multi-step forecast by lead time from math import sqrt from numpy import split from numpy import array from pandas import read_csv from sklearn.metrics import mean_squared_error from matplotlib import pyplot from sklearn.preprocessing import StandardScaler from sklearn.preprocessing import MinMaxScaler from sklearn.pipeline import Pipeline from sklearn.linear_model import LinearRegression from sklearn.linear_model import Lasso from sklearn.linear_model import Ridge from sklearn.linear_model import ElasticNet from sklearn.linear_model import HuberRegressor from sklearn.linear_model import Lars from sklearn.linear_model import LassoLars from sklearn.linear_model import PassiveAggressiveRegressor from sklearn.linear_model import RANSACRegressor from sklearn.linear_model import SGDRegressor # split a univariate dataset into train/test sets def split_dataset(data): # split into standard weeks train, test = data[1:-328], data[-328:-6] # restructure into windows of weekly data train = array(split(train, len(train)/7)) test = array(split(test, len(test)/7)) return train, test # evaluate one or more weekly forecasts against expected values def evaluate_forecasts(actual, predicted): scores = list() # calculate an RMSE score for each day for i in range(actual.shape[1]): # calculate mse mse = mean_squared_error(actual[:, i], predicted[:, i]) # calculate rmse rmse = sqrt(mse) # store scores.append(rmse) # calculate overall RMSE s = 0 for row in range(actual.shape[0]): for col in range(actual.shape[1]): s += (actual[row, col] - predicted[row, col])**2 score = sqrt(s / (actual.shape[0] * actual.shape[1])) return score, scores # summarize scores def summarize_scores(name, score, scores): s_scores = ', '.join(['%.1f' % s for s in scores]) print('%s: [%.3f] %s' % (name, score, s_scores)) # prepare a list of ml models def get_models(models=dict()): # linear models models['lr'] = LinearRegression() models['lasso'] = Lasso() models['ridge'] = Ridge() models['en'] = ElasticNet() models['huber'] = HuberRegressor() models['lars'] = Lars() models['llars'] = LassoLars() models['pa'] = PassiveAggressiveRegressor(max_iter=1000, tol=1e-3) models['ranscac'] = RANSACRegressor() models['sgd'] = SGDRegressor(max_iter=1000, tol=1e-3) print('Defined %d models' % len(models)) return models # create a feature preparation pipeline for a model def make_pipeline(model): steps = list() # standardization steps.append(('standardize', StandardScaler())) # normalization steps.append(('normalize', MinMaxScaler())) # the model steps.append(('model', model)) # create pipeline pipeline = Pipeline(steps=steps) return pipeline # # convert windows of weekly multivariate data into a series of total power def to_series(data): # extract just the total power from each week series = [week[:, 0] for week in data] # flatten into a single series series = array(series).flatten() return series # convert history into inputs and outputs def to_supervised(history, n_input, output_ix): # convert history to a univariate series data = to_series(history) X, y = list(), list() ix_start = 0 # step over the entire history one time step at a time for i in range(len(data)): # define the end of the input sequence ix_end = ix_start + n_input ix_output = ix_end + output_ix # ensure we have enough data for this instance if ix_output < len(data): X.append(data[ix_start:ix_end]) y.append(data[ix_output]) # move along one time step ix_start += 1 return array(X), array(y) # fit a model and make a forecast def sklearn_predict(model, history, n_input): yhat_sequence = list() # fit a model for each forecast day for i in range(7): # prepare data train_x, train_y = to_supervised(history, n_input, i) # make pipeline pipeline = make_pipeline(model) # fit the model pipeline.fit(train_x, train_y) # forecast x_input = array(train_x[-1, :]).reshape(1,n_input) yhat = pipeline.predict(x_input)[0] # store yhat_sequence.append(yhat) return yhat_sequence # evaluate a single model def evaluate_model(model, train, test, n_input): # history is a list of weekly data history = [x for x in train] # walk-forward validation over each week predictions = list() for i in range(len(test)): # predict the week yhat_sequence = sklearn_predict(model, history, n_input) # store the predictions predictions.append(yhat_sequence) # get real observation and add to history for predicting the next week history.append(test[i, :]) predictions = array(predictions) # evaluate predictions days for each week score, scores = evaluate_forecasts(test[:, :, 0], predictions) return score, scores # load the new file dataset = read_csv('household_power_consumption_days.csv', header=0, infer_datetime_format=True, parse_dates=['datetime'], index_col=['datetime']) # split into train and test train, test = split_dataset(dataset.values) # prepare the models to evaluate models = get_models() n_input = 7 # evaluate each model days = ['sun', 'mon', 'tue', 'wed', 'thr', 'fri', 'sat'] for name, model in models.items(): # evaluate and get scores score, scores = evaluate_model(model, train, test, n_input) # summarize scores summarize_scores(name, score, scores) # plot scores pyplot.plot(days, scores, marker='o', label=name) # show plot pyplot.legend() pyplot.show() ``` 運行該示例總結了每個評估的線性模型的總體和每日 RMSE。我們可以看到，通常每個引導時間方法產生的表現優于每日版本。這可能是因為該方法使更多的訓練數據可用于模型。 ```py Defined 10 models lr: [394.983] 411.0, 400.7, 340.2, 382.9, 385.1, 362.8, 469.4 lasso: [391.767] 403.6, 394.4, 336.1, 382.7, 384.2, 360.4, 468.1 ridge: [393.444] 407.9, 397.8, 338.9, 383.2, 383.2, 360.4, 469.6 en: [461.986] 452.2, 448.3, 430.3, 480.4, 448.9, 396.0, 560.6 huber: [394.287] 412.1, 394.0, 333.4, 384.1, 383.1, 364.3, 474.4 lars: [394.983] 411.0, 400.7, 340.2, 382.9, 385.1, 362.8, 469.4 llars: [390.075] 396.1, 390.1, 334.3, 384.4, 385.2, 355.6, 470.9 pa: [389.340] 409.7, 380.6, 328.3, 388.6, 370.1, 351.8, 478.4 ranscac: [439.298] 387.2, 462.4, 394.4, 427.7, 412.9, 447.9, 526.8 sgd: [390.184] 396.7, 386.7, 337.6, 391.4, 374.0, 357.1, 473.5 ``` 再次創建了每日 RMSE 分數的線圖。 ![Line Plot of Direct Per-Lead Time Multi-step Forecasts With Linear Algorithms](https://img.kancloud.cn/02/7f/027feae3e842e3ab7793d030fcd59399_1280x960.jpg) 線性算法的直接每個引導時間多步預測的線圖探索混合使用每日和每時間步驟來模擬問題可能會很有趣。如果增加用作每個引導時間的輸入的前幾天的數量是否改善了表現，例如，也可能是有趣的。使用兩周的數據而不是一周。 ## 擴展本節列出了一些擴展您可能希望探索的教程的想法。 * **調諧模型**。選擇一個表現良好的模型并調整模型超參數以進一步提高表現。 * **調整數據準備**。在擬合每個模型之前，所有數據都被標準化并標準化;探索這些方法是否必要以及數據縮放方法的更多或不同組合是否可以帶來更好的表現。 * **探索輸入大小**。輸入大小限于先前觀察的七天;探索越來越少的觀察日作為輸入及其對模型表現的影響。 * **非線性算法**。探索一套非線性和集成機器學習算法，看看它們是否能提升表現，例如 SVM 和隨機森林。 * **多變量直接模型**。開發直接模型，利用前一周的所有輸入變量，而不僅僅是每日消耗的總功率。這將需要將八個變量的七天的二維陣列展平為一維向量。如果你探索任何這些擴展，我很想知道。 ## 進一步閱讀如果您希望深入了解，本節將提供有關該主題的更多資源。 ### API * [pandas.read_csv API](https://pandas.pydata.org/pandas-docs/stable/generated/pandas.read_csv.html) * [pandas.DataFrame.resample API](https://pandas.pydata.org/pandas-docs/stable/generated/pandas.DataFrame.resample.html) * [重采樣偏移別名](http://pandas.pydata.org/pandas-docs/stable/timeseries.html#offset-aliases) * [sklearn.metrics.mean_squared_error API](http://scikit-learn.org/stable/modules/generated/sklearn.metrics.mean_squared_error.html) * [numpy.split API](https://docs.scipy.org/doc/numpy/reference/generated/numpy.split.html) ### 用品 * [個人家庭用電數據集，UCI 機器學習庫。](https://archive.ics.uci.edu/ml/datasets/individual+household+electric+power+consumption) * [交流電源，維基百科。](https://en.wikipedia.org/wiki/AC_power) * [多步時間序列預測的 4 種策略](https://machinelearningmastery.com/multi-step-time-series-forecasting/) ## 摘要在本教程中，您了解了如何使用機器學習算法開發遞歸和直接多步預測模型。具體來說，你學到了： * 如何開發一個評估線性，非線性和集成機器學習算法的框架，用于多步時間序列預測。 * 如何使用遞歸多步時間序列預測策略評估機器學習算法。 * 如何使用直接的每日和每個引導時間多步時間序列預測策略來評估機器學習算法。你有任何問題嗎？在下面的評論中提出您的問題，我會盡力回答。