基于真实故障数据生成对抗网络算子的实战指南：从数据分布学习到高质量生成

1次阅读

没有评论

共计 4132 个字符，预计需要花费 11 分钟才能阅读完成。

在工业设备故障预测和诊断领域，获取足够且具有代表性的故障数据一直是一个重大挑战。真实故障数据往往存在以下问题：

样本稀缺：设备正常运行时间远大于故障时间，导致故障样本数量极少
分布不均：不同故障类型出现频率差异大，某些罕见故障可能只有个位数样本
采集成本高：主动制造故障进行数据采集可能损坏设备，代价昂贵
隐私限制：某些行业故障数据涉及商业机密，难以获取完整数据集

这些问题导致训练出的故障诊断模型泛化能力差，对罕见故障类型的识别准确率低。数据增强和生成技术成为解决这一痛点的关键。

优点：

能够学习复杂的数据分布
生成样本质量高，细节丰富
可通过条件 GAN 控制生成特定类型故障

缺点：

训练不稳定，容易模式崩溃
需要精心设计网络结构和损失函数
评估生成质量较困难

优点：

训练稳定，有明确的损失函数
提供概率框架，可量化不确定性
隐空间具有良好数学特性

缺点：

生成样本往往较模糊
对复杂分布的建模能力有限
生成样本多样性不足

结合 GAN 和 VAE 的优点，如 VAE-GAN 架构：

使用 VAE 学习数据分布的隐表示
用 GAN 对隐空间进行优化，提高生成质量
平衡生成多样性和样本清晰度

数据清洗
处理缺失值：使用同类设备均值填充或插值
异常值检测：3σ 原则或孤立森林算法
时间对齐：不同采样频率的传感器数据需对齐
特征工程
时域特征：均值、方差、峰度、偏度等
频域特征：FFT 变换后的频谱能量
时频特征：小波变换系数
数据标准化
Min-Max 归一化或 Z -score 标准化
不同类型传感器数据分别处理

采用条件 Wasserstein GAN 架构：

生成器网络：
输入：随机噪声 z + 故障类型标签
结构：5 层全连接网络
激活函数：LeakyReLU(0.2)
输出层：Tanh 激活
判别器网络：
输入：真实 / 生成数据 + 故障类型标签
结构：4 层全连接网络
激活函数：LeakyReLU(0.2)
输出层：线性激活
条件信息注入：
将故障类型标签 embedding 后与噪声 z 拼接
在判别器各层加入条件批量归一化

使用 Wasserstein 距离改进的损失函数：

def generator_loss(fake_output):
    return -tf.reduce_mean(fake_output)

def discriminator_loss(real_output, fake_output):
    return tf.reduce_mean(fake_output) - tf.reduce_mean(real_output)

加入梯度惩罚项实现 Lipschitz 约束：

def gradient_penalty(discriminator, real_samples, fake_samples, labels):
    # 插值样本
    alpha = tf.random.uniform([real_samples.shape[0], 1], 0., 1.)
    interpolates = alpha * real_samples + (1-alpha) * fake_samples
    with tf.GradientTape() as tape:
        tape.watch(interpolates)
        d_interpolates = discriminator([interpolates, labels])
    gradients = tape.gradient(d_interpolates, [interpolates])[0]
    # 计算梯度惩罚项
    slopes = tf.sqrt(tf.reduce_sum(tf.square(gradients), axis=1))
    return tf.reduce_mean((slopes-1.)**2)

训练流程
先训练判别器 5 次，再训练生成器 1 次
使用 Adam 优化器，学习率 0.0001
批量大小 256
稳定训练技巧
权重初始化：Xavier 初始化
标签平滑：真实样本标签设为 0.9
历史平均：跟踪生成器参数的历史平均
早停策略
监控判别器损失和生成样本质量
连续 10 轮无明显改进则停止

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader

# 条件生成器
class Generator(nn.Module):
    def __init__(self, latent_dim, num_classes, feature_dim):
        super().__init__()
        self.label_embedding = nn.Embedding(num_classes, latent_dim)

        self.model = nn.Sequential(nn.Linear(2*latent_dim, 256),
            nn.LeakyReLU(0.2),
            nn.Linear(256, 512),
            nn.LeakyReLU(0.2),
            nn.Linear(512, 1024),
            nn.LeakyReLU(0.2),
            nn.Linear(1024, feature_dim),
            nn.Tanh())

    def forward(self, noise, labels):
        label_embed = self.label_embedding(labels)
        gen_input = torch.cat((label_embed, noise), dim=1)
        return self.model(gen_input)

# 条件判别器
class Discriminator(nn.Module):
    def __init__(self, num_classes, feature_dim):
        super().__init__()
        self.label_embedding = nn.Embedding(num_classes, feature_dim)

        self.model = nn.Sequential(nn.Linear(2*feature_dim, 512),
            nn.LeakyReLU(0.2),
            nn.Linear(512, 256),
            nn.LeakyReLU(0.2),
            nn.Linear(256, 1),
        )

    def forward(self, features, labels):
        label_embed = self.label_embedding(labels)
        dis_input = torch.cat((features, label_embed), dim=1)
        return self.model(dis_input)

# 训练循环
def train_gan(generator, discriminator, dataloader, num_epochs):
    # 初始化
    optimizer_G = optim.Adam(generator.parameters(), lr=0.0001, betas=(0.5, 0.9))
    optimizer_D = optim.Adam(discriminator.parameters(), lr=0.0001, betas=(0.5, 0.9))
    criterion = nn.BCEWithLogitsLoss()

    for epoch in range(num_epochs):
        for i, (real_features, labels) in enumerate(dataloader):
            batch_size = real_features.size(0)

            # 训练判别器
            optimizer_D.zero_grad()

            # 真实数据
            real_validity = discriminator(real_features, labels)
            real_loss = criterion(real_validity, torch.ones(batch_size, 1)*0.9)  # 标签平滑

            # 生成数据
            noise = torch.randn(batch_size, latent_dim)
            gen_features = generator(noise, labels)
            fake_validity = discriminator(gen_features.detach(), labels)
            fake_loss = criterion(fake_validity, torch.zeros(batch_size, 1))

            d_loss = real_loss + fake_loss
            d_loss.backward()
            optimizer_D.step()

            # 训练生成器
            if i % 5 == 0:  # 每 5 批训练一次生成器
                optimizer_G.zero_grad()
                gen_validity = discriminator(gen_features, labels)
                g_loss = criterion(gen_validity, torch.ones(batch_size, 1))
                g_loss.backward()
                optimizer_G.step()