Python 时间序列预测教程

作者：社区 / 用例
2023 年 1 月 18 日

本文最初发表于 The New Stack，经许可在此转载。

生活在一个快速变化的社会中的一个后果是，所有系统的状态都同样快速地变化，随之而来的是运营中的不一致性。但是，如果您可以预见这些不一致性呢？如果您可以窥视未来呢？这就是 时间序列数据 可以提供帮助的地方。

时间序列数据 是指描述特定系统在不同时间点的一系列数据点。时间间隔取决于特定系统，但通常，数据根据每个记录的日期和/或时间排列。这意味着系统的时间序列数据是其随时间推移的各种状态的详细记录。每个数据点都可以通过其时间戳唯一标识。有些人可能会争辩说，所有数据都是时间序列数据；但是，并非所有记录都以维护系统随时间变化的细节的方式记录。

为了维护系统的变化，您需要 时间序列预测，即使用可用的时间记录来预测系统在稍后时间的状态。因此，给定从远古时代到昨天为止的系统记录，预测系统今天和明天的状态的过程就是时间序列预测。您可以使用时间序列预测来预测天气或股票价格。它还有助于在更 工业环境 中进行预测性维护，以及预测能源资源的使用情况以进行适当的管理。

在本教程中，您将了解更多关于使用 InfluxDB 进行时间序列预测的信息，以及如何构建时间序列预测器以窥视未来。

理解时间序列预测

如前所述，时间序列预测是使用存储的特定系统过去的时间戳记录来预测其未来会发生什么的过程。

请注意，在开发过程中，过去和未来使用得更宽松。给定一个特定的参考点，所有在其之前记录的数据点都在过去，所有在其之后记录的数据都称为未来。

使用过去来预测未来可能听起来很棘手，因为有人可能会问：“如何在没有目标列的情况下将数据分割成特征和目标？” 首先，定义一个窗口。此窗口本质上是指您需要回溯多长时间才能做出预测。这有助于设置一个截止点，超过该截止点，数据点对预测的影响可以忽略不计。使用此窗口，您可以在数据集上滑动以生成训练数据。

假设您有关于一家公司全年销售额的数据，窗口期为 60 天：1 月 1 日至 2 月 29 日。这 60 天将用于预测 3 月 1 日的销售额（假设是闰年）。接下来，您将使用从 1 月 2 日到 3 月 1 日的数据来预测 3 月 2 日的销售额。这将逐步完成，直到您考虑 11 月 1 日到 12 月 30 日来预测 12 月 31 日的销售额。

除了销售额，时间序列预测还可以用于预测天气。根据有关温度、相对湿度和其他与天气相关的参数的历史信息，可以预测稍后日期的天气。

时间序列预测也已用于预测各种组织的股票价格，前提是它们有历史数据。同样，从标准货币到加密货币的货币价格也可以使用时间序列预测来预测。

在更工业化的方面，根据机械厂中各种设备的运行状况数据，可以预测未来的状态，这有助于尽早识别故障，以便在零件损坏和停止运行之前进行维护。这被称为预测性维护。

在能源效率领域，根据家庭或特定家庭的用电量信息，配电公司可以有效地供电，以便在特定时间需要更多电力的人获得足够的电力。组织和家庭也可以使用这些信息来了解他们的用电趋势，并实施法规以节省成本。

时间序列预测和其他形式的 时间序列数据分析 的用例是无穷无尽的，这就是为什么理解如何有效地使用时间序列预测非常重要。

使用 InfluxDB 实施时间序列预测

为了演示如何有效地进行时间序列预测，本教程包括使用 InfluxDB 进行数据准备和建模过程的演练。为了保留上下文，将要处理的示例问题与预测家庭能源消耗有关，给定每隔 15 分钟测量的能源消耗数据。

设置 InfluxDB

要设置 InfluxDB，请导航到 InfluxDB OSS 文档 并单击开始使用按钮。这是 InfluxDB 的开源版本，可以在本地服务器上设置。按照其 安装 InfluxDB 页面上的安装说明，针对您的特定操作系统（在本例中为 Windows）安装并启动 InfluxDB OSS

此时，您应该在本地服务器上运行 InfluxDB。在浏览器中输入此本地服务器的 URL 以访问 InfluxDB 界面。然后输入您的姓名、密码、组织和存储桶名称以完成 InfluxDB 设置。完成后，您将被带到 InfluxDB OSS 主页

开始使用 InfluxDB 的另一种选择是使用免费的 InfluxDB Cloud 实例，而无需在您的计算机上进行任何设置。

将数据加载到 InfluxDB 中

设置好 InfluxDB 后，现在您需要将数据加载到数据库中。首先从 此 Kaggle 页面 下载 CSV 数据。

在将 CSV 文件直接上传到 InfluxDB 之前，必须对其进行 注释。由于此 CSV 文件未注释，因此将使用 InfluxDB Python 客户端 将数据写入数据库。这将使您很好地了解如何将数据流式传输到 InfluxDB 中。

接下来，导航到 API 令牌 选项卡，如果尚不存在 API 令牌，请单击 生成 API 令牌 按钮。然后单击新创建的令牌的名称以查看和复制所有访问权限 API 令牌，如下所示。此令牌将用于验证您从客户端到 InfluxDB 的连接

有了 API 令牌、存储桶和组织后，导航到您首选的代码编辑器并为此项目创建一个文件夹。然后将先前下载的 CSV 文件传输到此父文件夹中的新 data/ 目录中。

使用以下 Windows 命令创建并激活名为 .venv 的 Python 虚拟环境

python -m venv .venv
.venv\Scripts\activate.bat

接下来，安装本教程所需的所有库，并将 API 令牌设置为环境变量

pip install pandas influxdb-client matplotlib
pip install fbprophet

完成后，创建一个脚本并导入必要的库

import os

from datetime import datetime

import pandas as pd

from influxdb_client import InfluxDBClient, Point, WritePrecision

from influxdb_client.client.write_api import SYNCHRONOUS

token = os.getenv("INFLUX_TOKEN")

organization = "forecasting"

bucket = "energy_consumption"

在这里，您安装 os 模块以加载环境变量，安装 pandas 库以加载 CSV 文件，并安装 InfluxDB 方法以方便写入过程。接下来，将 API 令牌、组织和存储桶名称加载到正确命名的变量中。

现在您需要创建 InfluxDB 客户端并实例化 write_API

PORT = 8086

client = InfluxDBClient(url=f"http://127.0.0.1:{PORT}", token=token, org=organization)

write_api = client.write_api(write_options=SYNCHRONOUS)

df = pd.read_csv('data/D202.csv')

在前面的代码中，您定义了 InfluxDB 服务器正在运行的 PORT 号。然后，您通过传入正在运行的服务器的 URL、API 令牌和组织名称作为参数来实例化 InfluxDB 客户端。

接下来，使用先前定义的 client 实例调用 write_API 方法。最后，在此代码片段中，您使用 read_csv 方法将 CSV 文件加载为 Pandas DataFrame。以下是数据集的外观

for index, row in df.iterrows():

    print(index, end=' ')

    stamp = datetime.strptime(f"{row['DATE']}, {row['START TIME']}", 

                                "%m/%d/%Y, %H:%M")

    p = Point(row["TYPE"])\
        .time(stamp, WritePrecision.NS)\
        .field("usage(KWh)", row["USAGE"])\
        .tag("cost", row["COST"])

    write_api.write(bucket=bucket, org=organization, record=p)

此代码片段包含实际的写入步骤。在这里，您循环遍历 Pandas DataFrame 并打印索引以跟踪进度。接下来，将开始时间转换为日期时间对象。然后，此开始时间用于定义该数据点的时间字段。

先前导入的 InfluxDB 点对象用于配置要上传的数据行。每个点实例都应具有时间、字段和一个或多个标签。时间 是指记录该数据点的时间。字段是以千瓦时 (kWh) 为单位的用量数据，成本被视为标签。

最后，在循环的每次迭代中，都会调用 write API 以将数据点写入数据库中的存储桶。

从 InfluxDB 读取数据

接下来，您必须将存储在 InfluxDB 上的数据读取到 Python 环境中以进行训练

query_api = client.query_api()

query = f'from(bucket:"{bucket}")' \
        ' |> range(start:2016-10-22T00:00:00Z, stop:2018-10-24T23:45:00Z)'\
        ' |> filter(fn: (r) => r._measurement == "Electric usage")' \
        ' |> filter(fn: (r) => r._field == "usage(KWh)")'

在这里，您再次调用 InfluxDB 客户端实例，但这次您选择查询 API，因为您的目标是从数据库中读取数据。然后创建查询。InfluxDB 使用一种称为 Flux 的脚本语言，该语言易于使用。在 Python 中，您在字符串中编写 Flux 查询。在此查询字符串中，您声明要从中读取的存储桶。接下来，您定义您希望查询的时间范围。最后，您声明过滤器以选择在其度量和字段属性中包含指定信息的数据点。在本例中为 "Electric usage" 度量和 "usage(KWh)" 字段

result = query_api.query(org=organization, query=query)

data = {'y': [], 'ds': []}

for table in result:

    for record in table.records:

        data['y'].append(record.get_value())

        data['ds'].append(record.get_time())

    print("here")

df = pd.DataFrame(data=data)

df.to_csv('data/Processed_D202.csv', index=False)

在此代码片段中，您使用 query_api.query 方法以及组织解析查询。然后，您创建存储来保存结果并循环遍历查询结果。当您迭代时，您将在先前定义的存储中整理结果，在本例中，该存储是一个 Python 字典。完成后，处理后的数据将作为 CSV 文件保存在 data/ 目录中。

使用 Prophet 进行预测

从 InfluxDB 加载数据后，下一步是构建预测模型。Facebook 创建并发布了 Prophet，这是一个在 Python 和 R 中用于时间序列预测的名称恰当且性能良好的库。Prophet 旨在处理任何给定时间序列中的异常值、各种变化（季节性、月度和每日等）和缺失数据。它还提供了有助于调整模型以获得更好性能的参数。

Prophet 模型的输入是一个 DataFrame，它由两列组成，y 和 ds。y 表示感兴趣的变量（能源使用量），ds 指的是 datetime 属性。正如您所看到的，上一节中的 DataFrame 列不是随机命名的。

要开始使用，请导入 Prophet 库以及其他用于数据处理和可视化的库

import fbprophet

import matplotlib.pyplot as plt

import pandas as pd

df = pd.read_csv('data/Processed_D202.csv')

df['ds'] = pd.to_datetime(df['ds']).dt.tz_localize(None)

df_copy = df.set_index('ds')

导入所需的库后，将处理后的数据读取到数据框中，将时间戳列转换为 Datetime 对象，并删除时区以避免绘图时出错。然后创建一个数据框的副本，并将时间戳列设置为索引

df_copy.plot(kind='line', 
                        xlabel='Datetime', 
                        ylabel='Energy Consumption (KWh)', 
                        )

plt.title('Household Energy Consumption over Time', fontweight='bold', fontsize=20)

plt.show()

在这里，您使用创建的 DataFrame 副本可视化数据，如下所示。您可以看到，在年底和来年初，能源消耗出现了一个峰值。这是一种季节性变化，您期望任何训练有素的模型都能识别出来

energy_prophet = fbprophet.Prophet(changepoint_prior_scale=0.0005)

energy_prophet.fit(df)

energy_forecast = energy_prophet.make_future_dataframe(periods=365, freq='D')

energy_forecast = energy_prophet.predict(energy_forecast)

energy_prophet.plot(energy_forecast, xlabel = 'Date', ylabel = 'Energy Usage (KWh)') # 0.0005

plt.title('Household Energy Usage')

plt.show()

接下来，实例化 Prophet 模型并将其拟合到数据。在实例化 Prophet 时，您传递 changepoint_prior_scale 参数来控制预测器的灵活性。接下来，您创建一个测试数据框，用于使用 Prophet 进行预测。此 DataFrame 是根据输入数据中的最后一天起 365 天内构建的，间隔与输入中观察到的间隔相同。然后使用 .predict 方法将其解析到模型中进行预测。最后，您绘制结果以查看预测与训练数据的匹配程度。据指出，较小的 changepoint_prior_scale 参数值可以带来更好的预测

在这里，您可以看到预测值大致遵循了前几年发生的趋势。您还可以查看预测的组成趋势，以了解能源消耗如何随一天、一周或一年而变化

energy_prophet.plot_components(energy_forecast)

有了这样的信息，您就可以就能源分配或能源消耗监管做出重要决策。

结论

在本教程中，您了解了时间序列数据和预测的重要性。您还学习了如何通过 Python 客户端与 InfluxDB 交互，以及如何使用 Prophet 构建预测器。

InfluxData 创建了 InfluxDB，这是一种高效的 时间序列数据库，作为您可以用来管理应用程序中的时间序列数据并执行分析的解决方案系统。立即试用 InfluxDB。

作者简介：Fortune 是 Josplay 的 Python 开发者，擅长处理数据和构建智能系统。他也是一名工艺工程师和技术作家。