如何使用XGBoost和InluxDB進行時間序列預測

作者：李睿 2023-03-27 07:34:28

XGBoost是一個流行的開源機器學習庫，可用于解決各種預測問題。人們需要了解如何使用它與InfluxDB進行時間序列預測。

譯者 | 李睿

審校 | 孫淑娟

XGBoost是一個開源的機器學習庫，它實現了優(yōu)化的分布式梯度增強算法。XGBoost使用并行處理實現快速性能，很好地處理缺失值，在小型數據集上執(zhí)行良好，并防止過擬合。所有這些優(yōu)點使XGBoost成為回歸問題(例如預測)的一種流行解決方案。

預測是各種業(yè)務目標的關鍵任務，例如預測分析、預測維護、產品規(guī)劃、預算等。許多預測或預測問題都涉及到時間序列數據。這使得XGBoost成為開源時間序列數據庫InfluxDB的出色伙伴。

本教程將學習如何使用XGBoost的Python包預測來自InfluxDB時間序列數據庫的數據。還將使用InfluxDB Python客戶端庫從InfluxDB查詢數據，并將數據轉換為Pandas DataFrame，以便更容易地使用時間序列數據，然后再做預測。此外，還將更詳細地討論XGBoost的優(yōu)點。

一、要求

本教程是在通過Homebrew安裝Python 3的macOS系統(tǒng)上執(zhí)行的。建議設置額外的工具，如virtualenv、pyenv或conda-env，以簡化Python和客戶端安裝。否則，其全部要求如下：

influxdb-client=1.30.0
pandas = 1.4.3
xgboost>=1.7.3
influxdb-client>=1.30.0
pandas>=1.4.3
matplotlib>=3.5.2
sklearn>=1.1.1

本教程還假設有一個免費的層InfluxDB云帳戶，并且已經創(chuàng)建了一個存儲桶和一個令牌，可以將存儲桶視為數據庫或InfluxDB中數據組織的最高層次結構。在本教程中，將創(chuàng)建一個名為NOAA的存儲桶。

二、決策樹、隨機森林和梯度增強

為了理解XGBoost是什么，必須理解決策樹、隨機森林和梯度增強。決策樹是一種有監(jiān)督的學習方法，由一系列特征測試組成。每個節(jié)點都是一個測試，所有節(jié)點都被組織在一個流程圖結構中。分支表示最終決定將哪個葉標簽或類標簽分配給輸入數據的條件。

機器學習中的決策樹用于確定明天是否會下雨。經過編輯以顯示決策樹的組件：葉、分支和節(jié)點。

決策樹、隨機森林和梯度增強背后的指導原則是，多個“弱學習者”或分類器共同做出強大的預測。

隨機森林包含多個決策樹。決策樹中的每個節(jié)點都被認為是弱學習者，隨機森林中的每個決策樹被認為是隨機森林模型中許多弱學習者中的一個。通常情況下，所有的數據都被隨機劃分為子集，并通過不同的決策樹進行傳遞。

使用決策樹和隨機森林的梯度增強是相似的，但它們的結構方式不同。梯度增強樹也包含決策樹森林，但這些決策樹是額外構建的，所有數據都經過決策樹集合。梯度增強樹可能包含一組分類樹或回歸樹，分類樹用于離散值(例如貓或狗)。回歸樹用于連續(xù)值(例如0到100)。

三、什么是XGBoost?

梯度增強是一種用于分類和預測的機器學習算法。XGBoost只是一個極端類型的梯度增強。它的極端之處在于，可以通過并行處理的能力更有效地執(zhí)行梯度增強。XGBoost文檔中的下圖說明了如何使用梯度增強來預測某人是否會喜歡一款電子游戲。

采用兩棵決策樹被用來決定某人是否可能喜歡一款電子游戲。將兩棵樹的葉子得分相加，以確定哪一個人最有可能喜歡這款電子游戲。

XGBoost的一些優(yōu)點：

相對容易理解。
適用于具有很少特征的小型、結構化和規(guī)則數據。

XGBoost的一些缺點：

易于過擬合，對異常值敏感。在XGBoost中使用時間序列數據的物化視圖進行預測可能是一個好主意。
在稀疏或無監(jiān)督數據上表現不佳。

四、使用XGBoost進行時間序列預測

在這里使用的是空氣傳感器樣本數據集，它是由InfluxDB提供的。該數據集包含來自多個傳感器的溫度數據。正在為單個傳感器創(chuàng)建溫度預測，其數據是這樣的：

使用以下Flux代碼導入單個時間序列的數據集和過濾器。(Flux是InfluxDB的查詢語言)

import "join"

import "influxdata/influxdb/sample"

//dataset is regular time series at 10 second intervals

data = sample.data(set: "airSensor")

  |> filter(fn: (r) => r._field == "temperature" and r.sensor_id == "TLM0100")

隨機森林和梯度增強可以用于時間序列預測，但它們需要將數據轉換為監(jiān)督學習。這意味著必須以滑動窗口方法或緩慢移動方法將數據向前移動，以將時間序列數據轉換為監(jiān)督學習集，也可以用Flux準備數據。在理想情況下，應該首先執(zhí)行一些自相關分析，以確定要使用的最佳方法。為簡潔起見，將使用以下Flux代碼按一個常規(guī)時間間隔移動數據。

import "join"

import "influxdata/influxdb/sample"

data = sample.data(set: "airSensor")

  |> filter(fn: (r) => r._field == "temperature" and r.sensor_id == "TLM0100")

shiftedData = data

  |> timeShift(duration: 10s , columns: ["_time"] )

join.time(left: data, right: shiftedData, as: (l, r) => ({l with data: l._value, shiftedData: r._value}))

  |> drop(columns: ["_measurement", "_time", "_value", "sensor_id", "_field"])

左右滑動查看完整代碼

如果想向模型輸入中添加額外的滯后數據，可以改為遵循以下Flux邏輯。

import "experimental"

import "influxdata/influxdb/sample"

data = sample.data(set: "airSensor")

|> filter(fn: (r) => r._field == "temperature" and r.sensor_id == "TLM0100")



shiftedData1 = data

|> timeShift(duration: 10s , columns: ["_time"] )

|> set(key: "shift" , value: "1" )



shiftedData2 = data

|> timeShift(duration: 20s , columns: ["_time"] )

|> set(key: "shift" , value: "2" )



shiftedData3 = data

|> timeShift(duration: 30s , columns: ["_time"] )

|> set(key: "shift" , value: "3")



shiftedData4 = data

|> timeShift(duration: 40s , columns: ["_time"] )

|> set(key: "shift" , value: "4")



union(tables: [shiftedData1, shiftedData2, shiftedData3, shiftedData4])

|> pivot(rowKey:["_time"], columnKey: ["shift"], valueColumn: "_value")

|> drop(columns: ["_measurement", "_time", "_value", "sensor_id", "_field"])

// remove the NaN values

|> limit(n:360)

|> tail(n: 356)

此外，我們必須使用向前驗證來訓練算法。這涉及到將數據集分為測試集和訓練集。然后利用XGB Regressor對XGBoost模型進行訓練，并用擬合方法進行預測。最后，我們使用平均絕對誤差 (MAE)來確定預測的準確性。對于10秒的延遲，計算出的平均絕對誤差 (MAE)為0.035。我們可以把這理解為96.5%的預測是非常正確的。下圖展示了我們從XGBoost得到的預測結果與從訓練/測試分割得到的期望值之間的對比。

以下是完整的腳本。這段代碼主要是從這里的教程中引入的。

import pandas as pd

from numpy import asarray

from sklearn.metrics import mean_absolute_error

from xgboost import XGBRegressor

from matplotlib import pyplot

from influxdb_client import InfluxDBClient

from influxdb_client.client.write_api import SYNCHRONOUS



# query data with the Python InfluxDB Client Library and transform data into a supervised learning problem with Flux

client = InfluxDBClient(url="https://us-west-2-1.aws.cloud2.influxdata.com", token="NyP-HzFGkObUBI4Wwg6Rbd-_SdrTMtZzbFK921VkMQWp3bv_e9BhpBi6fCBr_0-6i0ev32_XWZcmkDPsearTWA==", org="0437f6d51b579000")



# write_api = client.write_api(write_optinotallow=SYNCHRONOUS)

query_api = client.query_api()

df = query_api.query_data_frame('import "join"'

'import "influxdata/influxdb/sample"'

'data = sample.data(set: "airSensor")'

  '|> filter(fn: (r) => r._field == "temperature" and r.sensor_id == "TLM0100")'

'shiftedData = data'

  '|> timeShift(duration: 10s , columns: ["_time"] )'

'join.time(left: data, right: shiftedData, as: (l, r) => ({l with data: l._value, shiftedData: r._value}))'

  '|> drop(columns: ["_measurement", "_time", "_value", "sensor_id", "_field"])'

  '|> yield(name: "converted to supervised learning dataset")'

)

df = df.drop(columns=['table', 'result'])

data = df.to_numpy()



# split a univariate dataset into train/test sets

def train_test_split(data, n_test):

     return data[:-n_test:], data[-n_test:]



# fit an xgboost model and make a one step prediction

def xgboost_forecast(train, testX):

     # transform list into array

     train = asarray(train)

     # split into input and output columns

     trainX, trainy = train[:, :-1], train[:, -1]

     # fit model

     model = XGBRegressor(objective='reg:squarederror', n_estimators=1000)

     model.fit(trainX, trainy)

     # make a one-step prediction

     yhat = model.predict(asarray([testX]))

     return yhat[0]



# walk-forward validation for univariate data

def walk_forward_validation(data, n_test):

     predictions = list()

     # split dataset

     train, test = train_test_split(data, n_test)

     history = [x for x in train]

     # step over each time-step in the test set

     for i in range(len(test)):

          # split test row into input and output columns

          testX, testy = test[i, :-1], test[i, -1]

          # fit model on history and make a prediction

          yhat = xgboost_forecast(history, testX)

          # store forecast in list of predictions

          predictions.append(yhat)

          # add actual observation to history for the next loop

          history.append(test[i])

          # summarize progress

          print('>expected=%.1f, predicted=%.1f' % (testy, yhat))

     # estimate prediction error

     error = mean_absolute_error(test[:, -1], predictions)

     return error, test[:, -1], predictions



# evaluate

mae, y, yhat = walk_forward_validation(data, 100)

print('MAE: %.3f' % mae)



# plot expected vs predicted

pyplot.plot(y, label='Expected')

pyplot.plot(yhat, label='Predicted')

pyplot.legend()

pyplot.show()

五、結論

希望這篇博文能夠激勵人們利用XGBoost和InfluxDB進行預測。為此建議查看相關的報告，其中包括如何使用本文描述的許多算法和InfluxDB來進行預測和執(zhí)行異常檢測的示例。

原文鏈接：https://www.infoworld.com/article/3682070/time-series-forecasting-with-xgboost-and-influxdb.html

責任編輯：武曉燕來源： 51CTO技術棧