多变量多标签时序数据异常检测/预测模型构建及Darts库使用求助

核心问题拆解 #

• 处理带相关性的双变量（Flow/Pressure）、双标签（Flow anomaly/Pressure anomaly）秒级时序异常检测
• Darts库无内置多变量Isolation Forest适配支持，需换用更直接的实现方案
• 数据为10小时秒级时序（36000行），已转为Pandas DataFrame

Isolation Forest 多变量适配方案 #

Darts确实没有封装多变量时序版的Isolation Forest，直接结合Scikit-learn的Isolation Forest+时序特征工程是更高效的路径：

1. 提取时序关联特征：把Flow和Pressure的相关性转化为模型可利用的特征，比如滑动窗口统计量、交叉特征
2. 拆分多标签任务：将双标签问题拆为两个独立二分类任务，或用多输出模型统一处理
3. 半监督/有监督结合：先用Isolation Forest做无监督异常得分提取，再把得分作为特征输入到有监督分类器中

代码示例 #

1. 时序特征工程

import pandas as pd
import numpy as np

# 假设你的数据已加载为df，Timestamp为索引
window_size = 60  # 滑动窗口设为60秒，可根据业务调整

# 滑动窗口统计特征
df['flow_mean'] = df['Flow'].rolling(window=window_size).mean()
df['flow_std'] = df['Flow'].rolling(window=window_size).std()
df['pressure_mean'] = df['Pressure'].rolling(window=window_size).mean()
df['pressure_std'] = df['Pressure'].rolling(window=window_size).std()

# Flow与Pressure的关联特征
df['flow_pressure_cov'] = df[['Flow', 'Pressure']].rolling(window=window_size).cov().unstack()['Flow']['Pressure']
df['flow_pressure_ratio'] = df['Flow'] / (df['Pressure'] + 1e-6)  # 避免除零错误

# 移除滑动窗口产生的空值
df = df.dropna()

2. 多标签异常检测流程

from sklearn.ensemble import IsolationForest
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import f1_score

# 特征集与标签集
X = df[['Flow', 'Pressure', 'flow_mean', 'flow_std', 'pressure_mean', 'pressure_std', 'flow_pressure_cov', 'flow_pressure_ratio']]
y_flow = df['Flow anomaly']
y_pressure = df['Pressure anomaly']

# 拆分训练/测试集
X_train, X_test, yf_train, yf_test, yp_train, yp_test = train_test_split(
    X, y_flow, y_pressure, test_size=0.2, random_state=42
)

# 用Isolation Forest提取异常得分
iso_forest = IsolationForest(n_estimators=100, contamination=0.05, random_state=42)
iso_forest.fit(X_train)
X_train['iso_score'] = iso_forest.decision_function(X_train)
X_test['iso_score'] = iso_forest.decision_function(X_test)

# 分别训练Flow/Pressure异常分类器
# Flow异常检测
clf_flow = LogisticRegression()
clf_flow.fit(X_train, yf_train)
print(f"Flow异常检测F1得分: {f1_score(yf_test, clf_flow.predict(X_test)):.4f}")

# Pressure异常检测
clf_pressure = LogisticRegression()
clf_pressure.fit(X_train, yp_train)
print(f"Pressure异常检测F1得分: {f1_score(yp_test, clf_pressure.predict(X_test)):.4f}")

神经网络多标签方案 #

如果倾向用神经网络，可构建多输出LSTM模型直接处理时序序列：

from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, LSTM, Dense, Dropout

# 生成时序序列样本
def create_seq_data(data, y_flow, y_pressure, seq_len):
    X, yf, yp = [], [], []
    for i in range(len(data) - seq_len):
        X.append(data[i:i+seq_len])
        yf.append(y_flow[i+seq_len])
        yp.append(y_pressure[i+seq_len])
    return np.array(X), np.array(yf), np.array(yp)

seq_len = 60
X_seq, yf_seq, yp_seq = create_seq_data(X.values, y_flow.values, y_pressure.values, seq_len)
X_train_seq, X_test_seq, yf_train, yf_test, yp_train, yp_test = train_test_split(
    X_seq, yf_seq, yp_seq, test_size=0.2, random_state=42
)

# 构建多输出LSTM模型
input_layer = Input(shape=(seq_len, X.shape[1]))
lstm = LSTM(64, return_sequences=False)(input_layer)
dropout = Dropout(0.2)(lstm)
flow_output = Dense(1, activation='sigmoid', name='flow_out')(dropout)
pressure_output = Dense(1, activation='sigmoid', name='pressure_out')(dropout)

model = Model(inputs=input_layer, outputs=[flow_output, pressure_output])
model.compile(
    optimizer='adam',
    loss={'flow_out': 'binary_crossentropy', 'pressure_out': 'binary_crossentropy'},
    metrics=['accuracy']
)

# 训练模型
model.fit(
    X_train_seq, [yf_train, yp_train],
    epochs=10, batch_size=32, validation_split=0.1
)

# 评估效果
_, _, flow_acc, pressure_acc = model.evaluate(X_test_seq, [yf_test, yp_test])
print(f"Flow异常检测准确率: {flow_acc:.4f}")
print(f"Pressure异常检测准确率: {pressure_acc:.4f}")

关键注意事项 #

• 滑动窗口大小需结合制造流程异常的时间尺度调整，比如分钟级异常可设为300秒
• Isolation Forest的contamination参数建议设为标注数据中异常样本的实际比例，比默认auto更精准
• 多标签任务也可用MultiOutputClassifier包装Isolation Forest，但无监督模型更适合作为特征提取器配合有监督分类

内容的提问来源于stack exchange，提问作者Perrupi