使用 Python 和 Facebook Kats 进行时间序列预测

作者：社区 / 产品, 用例
2022 年 10 月 12 日

本文由 Vidhi Chugh 撰写。向下滚动查看作者简介和照片。

时间序列分析是对在恒定周期内收集的数据点和记录序列的研究。该分析指示了一个变量或一组变量是如何变化的，并有助于发现潜在的趋势和模式。

时间序列数据通常用于预测问题，通过根据历史信息预测未来数据的可能性。天气预报、股票价格预测和行业增长预测是时间序列分析一些最流行的应用。

机器学习算法的最新进展，如长短期记忆 (LSTM)和Prophet，已显著提高了预测的准确性。

在本文中，您将学习如何使用 InfluxDB 存储和访问时间序列数据，并使用 Python API 进行访问，并使用 Facebook Kats 库对其进行分析。

什么是 Facebook Kats 工具包

有许多用于分析数据的 Python 库，包括 sktime、Prophet、Facebook Kats 和 Darts。这篇文章将重点介绍 Kats 库，因为它是一个轻量级且易于使用的框架，经常用于分析时间序列数据。它提供了各种功能，包括以下内容：

• 预测：Kats 库提供了一系列工具，包括预测算法、集成、具有超参数调整的元学习算法、回溯测试和经验预测区间。

• 检测：它检测模式，如趋势、季节性、异常和变化点。

• 自动特征工程和嵌入：Kats tsfeatures 模块中的时间序列特征为监督学习算法自动生成特征。

• 实用工具：Kats 库为学习和实验提供时间序列模拟器。

什么是 InfluxDB

时间序列分析需要一个数据库，该数据库适合有效且高效地存储和检索数据。在这里，您将使用 InfluxDB，它是构建时间序列应用程序的领先平台之一。它是一个高性能引擎，开源、社区覆盖范围广且易于使用。此外，它可以本地托管或在云端托管。

使用 Python 和 Facebook Kats 实现时间序列预测

在以下教程中，您将学习如何创建一个简单的时间序列预测数据集，该数据集使用 InfluxDB 存储数据，然后使用 Facebook Kats 对其进行分析。

先决条件

在开始之前，您应该对 Python 语法和命令行/终端命令有基本的了解。

本教程的所有代码都可以在这个 GitHub 存储库中找到。

连接到 InfluxDB Cloud 实例

要开始使用 InfluxDB Cloud 实例，请访问 InfluxData 网站，然后点击右上角的获取 InfluxDB

对于仅限云端的帐户界面，选择免费使用

然后您将被导航到注册页面，您可以在输入必要的信息后注册

您需要选择一个云服务提供商（这里选择了 Amazon Web Services - AWS）来存储您的 InfluxDB 时间序列数据。您无需熟悉任何这些服务，因为 InfluxDB 抽象掉了所有底层复杂性。选择云提供商后，添加您的公司名称并同意条款

现在您已准备好开始设置数据库。选择您偏好的计划。在本例中，免费计划就足够了

选择计划后，您将进入一个入门屏幕，其中列出了许多编程语言。为此演示选择 Python

在下一页上，您需要观看视频，该视频向您展示如何设置 InfluxData。完成后点击下一步

设置您的本地机器

要通过 Python 访问 InfluxDB，您需要在您的机器上安装 influxdb-client 库。您可以通过在终端中运行以下命令来完成此操作

pip3 install influxdb-client

请注意：pip3 用于在 Python 3.X 中安装库，这里也使用它。

通过导航到 API 令牌 > 生成 API 令牌 > 所有访问 API 令牌，从 Web 界面生成您的 API 令牌。本教程将使用所有访问 API 令牌，但如果您想选择用户的身份验证级别，您也可以选择生成自定义 API 令牌。

在终端或命令行中运行以下命令，将您的令牌添加为环境变量

export INFLUXDB_TOKEN = "your token"

令牌未包含在代码中，以保持跨团队的可共享性。

从这里开始，您将使用 Python IDE 或 Jupyter Notebook 来写入和读取 InfluxDB 的数据。

将数据写入 InfluxDB

要将数据写入 InfluxDB，您需要访问一些数据。在这里，您将使用 航空旅客数据集，这是一个模型数据集，表示 1949 年至 1960 年的每月航空旅客数量。

首先，使用以下命令在您的操作系统上安装 pandas 库

pip3 install pandas

创建一个名为 writePassengerData.py 的文件，并粘贴以下代码。并确保将 AirPassengers.csv 放在同一目录下

writePassengerData.py

import pandas as pd

import influxdb_client, os, time

from influxdb_client import InfluxDBClient, Point, WritePrecision

from influxdb_client.client.write_api import SYNCHRONOUS

token = os.environ.get("your influx db token")

org = "your influx db org name"

url = "your influx db custom url"

bucket = "your influx bucket name"

client = influxdb_client.InfluxDBClient(url=url, token=token, org=org)

write_api = client.write_api(write_options=SYNCHRONOUS)

df = pd.read_csv('AirPassengers.csv')

for i in df.index:

point = (

Point("passengers")

.tag("month", df.iloc[i,0])

.field("passengers", df.iloc[i,1])

)

write_api.write(bucket=bucket, org=org, record=point)

time.sleep(1) # separate points by 1 second

在之前的代码块中，请确保您使用在云平台生成的令牌和之前输入的组织名称。在这里，您导入必要的库，如 pandas 用于读取 CSV 数据，influxdb_client 用于将数据写入 InfluxDB Cloud 实例。然后，您声明字符串变量来保存令牌、组织、URL 和存储桶等信息。接下来，您使用 InfluxDBClient() 实例化客户端，并使用 write_api() 方法激活写入 API。CSV 文件使用 pandas.read_csv 读取并存储在数据帧对象中。然后，您遍历数据帧中的每一行，创建一个临时点对象，并将该点对象作为记录写入 InfluxDB。

接下来，从您的终端运行该文件

writePassengerData.py

通过转到 InfluxDB Cloud 并点击存储桶，验证数据是否已写入/上传到 InfluxDB Cloud 存储桶。然后点击您的存储桶，并验证测量名称 passengers 是否可用

从 InfluxDB 读取数据

为了能够从 InfluxDB 读取数据，您需要创建一个名为 timeSeriesAnalysis.ipynb 的文件，并在其中包含以下代码

timeSeriesAnalysis.ipynb

import influxdb_client, os, time

from influxdb_client import InfluxDBClient, Point, WritePrecision

from influxdb_client.client.write_api import SYNCHRONOUS

token = os.environ.get("your influx db token")

org = "your influx db org name"

url = "your influx db custom url"

client = influxdb_client.InfluxDBClient(url=url, token=token, org=org)

query_api = client.query_api()

query = """from(bucket:"your influx db bucket name")

|> range(start: -1000m)

|> filter(fn: (r) => r._measurement == "passengers")

|> mean()"""

tables = query_api.query(query, org=org)

results = []

for table in tables:

for record in table.records:

results.append(({'month': record.values.get('month'), record.get_field(): record.get_value()}))

在此代码块中，您重新导入所有库（如之前演示的那样）。接下来，您声明保存令牌、组织和 URL 等信息的变量。然后，您使用 InfluxDBClient() 实例化客户端，并使用 query_api() 激活查询 API。

请注意：在数据库语言中，查询用作读取的同义词。

激活查询 API 后，您可以将存储桶和测量名称等详细信息添加到 Flux 查询中。使用查询 API，获取数据并将其存储在表对象中，然后遍历表，解析所需的信息，并将其附加到列表中

运行此代码后，您将获得一个字典列表，其中包含如下所示的数据

[{'month': '1949-03', 'passengers': 132.0},

{'month': '1954-04', 'passengers': 227.0},

{'month': '1952-06', 'passengers': 218.0},

{'month': '1956-07', 'passengers': 413.0},

{'month': '1958-06', 'passengers': 435.0},

{'month': '1955-09', 'passengers': 312.0},

{'month': '1956-02', 'passengers': 277.0},

{'month': '1958-01', 'passengers': 340.0},

{'month': '1954-11', 'passengers': 203.0},

{'month': '1959-07', 'passengers': 548.0}]

现在是时候使其可读并将其转换为 pandas 数据帧了。

导入用于时间序列预测的库

现在您已经读取了数据，是时候开始分析它了。

为了进行分析，您需要安装和导入以下库

Pip3 install numpy

Pip3 install kats

Pip3 install statsmodels

Pip3 install warnings

Pip3 install matplotlib

pandas、NumPy 和 Matplotlib 用于数据操作和 可视化。您还需要从 Kats 导入 SARIMA、Holt-Winters 和 Prophet 用于时间序列分析。models 和 TimeSeriesData 的安装是为了将标准数据帧转换为 Kats 库可使用的时间序列对象

timeSeriesAnalysis.ipynb

import pandas as pd

import numpy as np

import sys

import matplotlib.pyplot as plt

import warnings

import statsmodels.api as sm

from kats.models.sarima import SARIMAModel, SARIMAParams

from kats.models.holtwinters import HoltWintersParams, HoltWintersModel

from kats.models.prophet import ProphetModel, ProphetParams

from kats.consts import TimeSeriesData

将数据转换为时间序列格式

要将数据转换为时间序列对象，您需要首先将结果列表转换为数据帧。按月份升序对数据帧值进行排序，并将列从“month”和“passengers”重命名为“time”和“value”（这是 Kats 库中的标准步骤）。最后，将数据帧对象转换为时间序列数据对象

timeSeriesAnalysis.ipynb

air_passengers_df = pd.DataFrame(results)

air_passengers_df.sort_values('month', inplace=True)

air_passengers_df.columns = ["time", "value"]

air_passengers_ts = TimeSeriesData(air_passengers_df)

检查平稳性

现在，是时候可视化时间序列是否平稳，这是建模时间序列数据的先决条件之一

timeSeriesAnalysis.ipynb

plt.figure(figsize=(35,20))

fig = plt.plot(air_passengers_df['time'], air_passengers_df["value"])

plt.xticks(rotation=90)

plt.show()

时间序列是非平稳的，因为均值和方差不是恒定的

您可以使用 增广迪基-福勒检验来确认这一点

timeSeriesAnalysis.ipynb

from statsmodels.tsa.stattools import adfuller

X = air_passengers_df["value"]

result = adfuller(X)

print('ADF Statistic: %f' % result[0])

print('p-value: %f' % result[1])

高 p-value 表明我们无法为这个时间序列建立平稳性，正如您从 ADF 检验的输出中看到的那样

ADF Statistic: 0.815369

p-value: 0.991880

当您运行以下代码时，可以通过差分使序列平稳

timeSeriesAnalysis.ipynb

plt.figure(figsize=(35,20))

fig = plt.plot(air_passengers_df['time'], air_passengers_df["value"].diff())

plt.xticks(rotation=90)

plt.show()

如您所见，增广迪基-福勒检验给出了更好的结果

timeSeriesAnalysis.ipynb

result = adfuller(X.diff()[1:])

print('ADF Statistic: %f' % result[0])

print('p-value: %f' % result[1])

您的输出将如下所示

ADF Statistic: -2.829267

p-value: 0.054213

现在序列几乎是平稳的，是时候绘制自相关函数 (ACF) 和偏自相关函数 (PACF) 图表了。

ACF 和 PACF 图

ACF 和 PACF 图表确定移动平均 (MA) 滞后 (q) 和自相关 (AR) 滞后 (p)

timeSeriesAnalysis.ipynb

fig, ax = plt.subplots(2,1)

fig.set_figheight(15)

fig.set_figwidth(15)

fig = sm.graphics.tsa.plot_acf(air_passengers_df["value"].diff()[1:], lags=50, ax=ax[0])

fig = sm.graphics.tsa.plot_pacf(air_passengers_df["value"].diff()[1:], lags=50, ax=ax[1])

plt.show()

根据以下 PACF 和 ACF 图表，p = 2 和 q = 1 是一个好的起始值。可以使用网格搜索和袋外 (OOB) 样本进一步调整超参数

SARIMA 模型

要训练和预测 SARIMA 模型，首先声明 p、d、q 和 m 参数，其中 d = 1 用于线性趋势数据，m 是十二个月的季节性指数（航空旅行季节性）。使用训练数据和参数实例化 SARIMA 模型对象并拟合模型。然后使用训练好的模型进行预测，并绘制数据和预测，如此处所示

timeSeriesAnalysis.ipynb

# declare SARIMA parameters - use acf/pacf charts and grid search

params = SARIMAParams(p = 2, d=1, q=1, seasonal_order=(1,0,1,12), trend = 'ct')

# train sarima model

m = SARIMAModel(data=air_passengers_ts, params=params)

m.fit()

#forecast for next 30 months

fcst = m.predict(steps=30, freq="MS")

# visualize predictions

m.plot()

虽然该模型能够识别季节性和自相关性，但它无法识别数据中不断增加的范围，从而导致较高的误差方差

Holt-Winters 模型

Holt-Winters 模型通过捕获季节性中不断增加的范围来克服 SARIMA 模型的缺点，并生成置信度更高的预测

timeSeriesAnalysis.ipynb

# declare parameters for Holt Winters model

params = HoltWintersParams(trend="add", seasonal="mul", seasonal_periods=12)

# fit a Holt Winters model

hw_model = HoltWintersModel(data=air_passengers_ts, params=params)

hw_model.fit()

# forecast for next 30 months

fcst = hw_model.predict(steps=30, alpha = 0.1)

# plot predictions

hw_model.plot()

Prophet 时间序列模型

Prophet 时间序列模型建立在 Holt-Winters 模型的基础上，以进一步改进预测。遵循的步骤与之前讨论的两个示例类似

timeSeriesAnalysis.ipynb

# declare parameters for Prophet model - choose between additive or multiplicative

# multiplicative gives better results

params = ProphetParams(seasonality_mode='multiplicative')

# fit a prophet model instance

model = ProphetModel(air_passengers_ts, params)

model.fit()

# forecast for next 30 months

fcst = model.predict(steps=30, freq="MS")

# visualize predictions

model.plot()

Prophet 模型预测具有最小的误差方差，因此具有最高的置信度

本教程的所有代码都可以在这个 GitHub 存储库中找到。

结论

在本文中，您了解了时间序列数据的重要性及其在多个关键行业的应用。您还学习了如何设置 InfluxDB Cloud 实例并使用 Python API 写入 CSV。从那里，您从 InfluxDB 读取数据以使用 Facebook Kats 进行分析。本文最后逐步介绍了三种流行的时间序列算法，即 SARIMA、Holt-Winters 和 Prophet，以及它们的性能比较。

如果您正在寻找用于构建和运营时间序列应用程序的平台，请查看 InfluxDB。InfluxDB 是开源的，使开发人员能够更快地构建和部署变革性的监控、分析和物联网应用程序并进行扩展。该平台可以处理来自网络、物联网设备、应用程序和容器的大量时间序列数据。

关于作者

Vidhi Chugh 是一位屡获殊荣的 AI/ML 创新领导者，也是数据治理领域的领先专家，其愿景是构建值得信赖的 AI 解决方案。她在数据科学、产品和研究团队的交叉领域工作，以在沃尔玛全球技术印度公司交付业务价值和洞察力。在她的 LinkedIn 页面和 Medium 个人资料上了解更多信息。