如何用Python实现基于GAN的异常检测？生成对抗网络

import tensorflow as tf
from tensorflow.keras import layers, Model, losses
import numpy as np

# 假设我们有一些正常数据，这里用随机数据模拟
# 实际应用中，你需要加载你的正常数据集
normal_data = np.random.rand(1000, 28, 28, 1).astype('float32') # 1000张28x28的灰度图

# 1. 定义生成器
def build_generator():
    model = tf.keras.Sequential([
        layers.Input(shape=(100,)), # 噪声向量维度
        layers.Dense(7*7*256, use_bias=False),
        layers.BatchNormalization(),
        layers.LeakyReLU(),
        layers.Reshape((7, 7, 256)),
        layers.Conv2DTranspose(128, (5, 5), strides=(1, 1), padding='same', use_bias=False),
        layers.BatchNormalization(),
        layers.LeakyReLU(),
        layers.Conv2DTranspose(64, (5, 5), strides=(2, 2), padding='same', use_bias=False),
        layers.BatchNormalization(),
        layers.LeakyReLU(),
        layers.Conv2DTranspose(1, (5, 5), strides=(2, 2), padding='same', use_bias=False, activation='tanh')
    ])
    return model

# 2. 定义判别器
def build_discriminator():
    model = tf.keras.Sequential([
        layers.Input(shape=(28, 28, 1)),
        layers.Conv2D(64, (5, 5), strides=(2, 2), padding='same'),
        layers.LeakyReLU(),
        layers.Dropout(0.3),
        layers.Conv2D(128, (5, 5), strides=(2, 2), padding='same'),
        layers.LeakyReLU(),
        layers.Dropout(0.3),
        layers.Flatten(),
        layers.Dense(1) # 无激活函数，方便计算二元交叉熵
    ])
    return model

# 3. 损失函数和优化器
cross_entropy = losses.BinaryCrossentropy(from_logits=True)

def discriminator_loss(real_output, fake_output):
    real_loss = cross_entropy(tf.ones_like(real_output), real_output)
    fake_loss = cross_entropy(tf.zeros_like(fake_output), fake_output)
    total_loss = real_loss + fake_loss
    return total_loss

def generator_loss(fake_output):
    return cross_entropy(tf.ones_like(fake_output), fake_output)

generator_optimizer = tf.keras.optimizers.Adam(1e-4)
discriminator_optimizer = tf.keras.optimizers.Adam(1e-4)

# 4. 训练步骤
@tf.function
def train_step(images, generator, discriminator):
    noise = tf.random.normal([images.shape[0], 100])

    with tf.GradientTape() as gen_tape, tf.GradientTape() as disc_tape:
        generated_images = generator(noise, training=True)

        real_output = discriminator(images, training=True)
        fake_output = discriminator(generated_images, training=True)

        gen_loss = generator_loss(fake_output)
        disc_loss = discriminator_loss(real_output, fake_output)

    gradients_of_generator = gen_tape.gradient(gen_loss, generator.trainable_variables)
    gradients_of_discriminator = disc_tape.gradient(disc_loss, discriminator.trainable_variables)

    generator_optimizer.apply_gradients(zip(gradients_of_generator, generator.trainable_variables))
    discriminator_optimizer.apply_gradients(zip(gradients_of_discriminator, discriminator.trainable_variables))
    return gen_loss, disc_loss

# 训练循环（简化，实际需要更多epoch和数据批次处理）
generator = build_generator()
discriminator = build_discriminator()

EPOCHS = 50 # 实际可能需要几百甚至上千个epoch
BATCH_SIZE = 64
dataset = tf.data.Dataset.from_tensor_slices(normal_data).shuffle(1000).batch(BATCH_SIZE)

print("开始训练GAN...")
for epoch in range(EPOCHS):
    for image_batch in dataset:
        g_loss, d_loss = train_step(image_batch, generator, discriminator)
    print(f'Epoch {epoch+1}, G Loss: {g_loss:.4f}, D Loss: {d_loss:.4f}')
print("GAN训练完成。")

# 5. 异常检测推理函数
def detect_anomaly(data_point, generator, discriminator, threshold=0.5):
    # 方法1: 基于重构误差
    # 寻找最能重构data_point的噪声向量（这里简化，直接用生成器生成）
    # 实际AnoGAN会通过优化寻找最佳z
    # 这里我们直接用生成器生成一个样本，并计算其与输入数据的距离
    # 更严谨的AnoGAN会反向传播找到最佳z，使得G(z)最接近data_point

    # 简化：直接计算输入数据和它被判别器判断为“真”的程度
    # 以及一个假设的“重构”误差（虽然这里没有真正的反向优化重构）

    # 正常AnoGAN的异常分数通常是：
    # L_recon(x, G(z_x)) + alpha * L_disc(x, D(x), D(G(z_x)))
    # 其中z_x是通过优化G(z)接近x得到的

    # 这里我们采用一个简化的异常分数：判别器对输入数据的“真实性”评分
    # 判别器认为越不像正常数据，分数越高

    # 实际AnoGAN的异常分数计算会复杂一些，通常涉及：
    # 1. 找到一个隐空间向量z，使得G(z)尽可能接近输入x
    # 2. 异常分数 = L_recon(x, G(z)) + alpha * L_disc(D(x), D(G(z)))
    # L_recon是重构误差，L_disc是判别器特征匹配误差

    # 简化版：直接看判别器对原始数据的判断
    score = discriminator(tf.expand_dims(data_point, axis=0), training=False).numpy()[0][0]
    # 注意：判别器的输出是logit，需要sigmoid转换成概率，或者直接用logit比较
    # 如果判别器输出是负值，表示它认为这是假的；正值，表示真的
    # 那么，我们希望异常数据被判别器认为是“假”的，即输出负值
    # 所以，我们可以用 -score 作为异常分数，分数越高越异常
    anomaly_score = -score # 判别器输出越小（越认为是假），异常分数越高

    is_anomaly = anomaly_score > threshold
    return anomaly_score, is_anomaly

# 示例：检测一个“异常”数据点（这里用随机数据模拟一个，假设它和正常数据有显著差异）
# 实际中，你会有新的、未见过的数据来测试
test_anomaly_data = np.random.rand(28, 28, 1).astype('float32') * 2.0 # 假设值域超出正常范围
test_normal_data = normal_data[0] # 取一个正常数据点

# 检测
score_anomaly, is_anomaly_result = detect_anomaly(test_anomaly_data, generator, discriminator, threshold=1.0)
print(f"测试异常数据：异常分数 {score_anomaly:.4f}, 是否异常: {is_anomaly_result}")

score_normal, is_normal_result = detect_anomaly(test_normal_data, generator, discriminator, threshold=1.0)
print(f"测试正常数据：异常分数 {score_normal:.4f}, 是否异常: {is_normal_result}")

z

WGAN-GP

L1重建损失