Zenn
🗂

[実装編] InfluxDBとPyTorchを活用した時系列データ予測

に公開
InfluxDB
実装
中野哲平
tech

時系列データの予測は、IoTデバイスや金融市場、工業機械の異常検知など、多くの分野で重要な技術です。本記事では、InfluxDBを用いた時系列データの管理と、PyTorchのLSTM（Long Short-Term Memory）を活用した時系列データ予測の実装方法を紹介します。

1. InfluxDBとPyTorchの組み合わせが適切な理由

1.1 InfluxDBの特徴

InfluxDBは、時系列データに特化したデータベースで、高速な書き込みと時系列クエリの最適化が強みです。IoTデバイスのセンサーデータや金融データを扱う際に非常に便利です。

1.2 PyTorchの特徴

PyTorchは、ニューラルネットワークの構築に特化した深層学習フレームワークで、特に時系列データの予測に適したRNNやLSTMを簡単に実装できます。

1.3 組み合わせのメリット

InfluxDBに保存されたリアルタイムデータを、PyTorchのLSTMモデルを用いて学習・予測し、その結果を再びInfluxDBに格納することで、リアルタイムなデータ解析と予測が可能になります。

2. 実装の流れ

2.1 必要な環境の準備

このプロジェクトでは、以下の環境を使用します。

  • Python 3.8+
  • InfluxDB 2.x
  • PyTorch
  • Pandas, NumPy

事前にInfluxDBのセットアップを完了し、必要なPythonライブラリをインストールしてください。

pip install influxdb torch pandas numpy

2.2 InfluxDBからデータ取得

まず、InfluxDBに保存されたセンサーデータを取得します。

from influxdb import InfluxDBClient
import pandas as pd

client = InfluxDBClient(host='localhost', port=8086)
db_name = 'sensor_data'
client.switch_database(db_name)

query = "SELECT time, temperature FROM temperature_measurements ORDER BY time DESC LIMIT 1000"
result = client.query(query)
data_points = list(result.get_points())

df = pd.DataFrame(data_points)
df['time'] = pd.to_datetime(df['time'])
df.set_index('time', inplace=True)

2.3 LSTMモデルの構築とトレーニング

InfluxDBから取得したデータを前処理し、PyTorchのLSTMモデルを用いてトレーニングを行います。

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np

def create_sequences(data, window_size):
    sequences, labels = [], []
    for i in range(len(data) - window_size):
        sequences.append(data[i:i+window_size])
        labels.append(data[i+window_size])
    return np.array(sequences), np.array(labels)

class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(LSTMModel, self).__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)
    
    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        return self.fc(lstm_out[:, -1, :])

データの前処理とモデルの学習を実施します。

window_size = 10
temp_values = df['temperature'].values
temp_values = (temp_values - np.mean(temp_values)) / np.std(temp_values)
X, y = create_sequences(temp_values, window_size)

X = torch.tensor(X, dtype=torch.float32).unsqueeze(-1)
y = torch.tensor(y, dtype=torch.float32).unsqueeze(-1)

model = LSTMModel(input_size=1, hidden_size=50, output_size=1)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

epochs = 50
for epoch in range(epochs):
    model.train()
    optimizer.zero_grad()
    output = model(X)
    loss = criterion(output, y)
    loss.backward()
    optimizer.step()
    print(f'Epoch {epoch+1}/{epochs}, Loss: {loss.item():.6f}')

2.4 予測結果をInfluxDBに保存

学習したモデルを用いて予測し、InfluxDBに保存します。

model.eval()
predictions = model(X).detach().numpy()

json_body = [
    {
        "measurement": "predicted_temperature",
        "tags": {"model": "LSTM"},
        "fields": {"value": float(predictions[i][0])},
        "time": df.index[i + window_size].isoformat()
    }
    for i in range(len(predictions))
]
client.write_points(json_body)

print("予測データをInfluxDBに保存しました！")

3. まとめ

InfluxDBとPyTorchを組み合わせることで、時系列データのリアルタイム処理と高度な予測を実現できます。今回はLSTMモデルを用いましたが、他の時系列予測手法（GRU, Transformerなど）を試すのも良い選択肢です。

この仕組みを活用すれば、製造業の異常検知、エネルギー需要予測、IoTデバイスの予測メンテナンスなど、さまざまな分野での応用が可能になります。ぜひ、自身のプロジェクトにも取り入れてみてください！

Discussion