1D CGAN实战：从零构建一维条件生成对抗网络及其在时序数据生成中的应用

1次阅读

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

时序数据生成在金融、物联网等领域具有广泛的应用需求。例如，金融领域需要生成股票价格、交易量等数据用于策略回测；物联网领域则需要生成传感器数据用于设备故障预测。然而，传统方法如 ARIMA 虽然在简单时序预测中表现良好，但在生成复杂模式时序数据时存在明显局限性。

ARIMA 模型假设数据是线性且平稳的，难以捕捉真实世界中的非线性、非平稳特性
传统方法无法生成多样化的数据样本，导致生成结果模式单一
缺乏对特定条件的控制能力，无法按需生成符合特定场景的数据

在时序数据生成领域，除了 CGAN 外，VAE 和传统 GAN 也是常用的方法。以下是它们的主要对比：

VAE（变分自编码器）：
优点：训练稳定，有明确的概率框架
缺点：生成样本通常较为模糊，缺乏锐利的细节
传统 GAN：
优点：生成样本质量高
缺点：训练不稳定，容易出现模式崩溃
1D CGAN：
结合了条件生成和 GAN 的优点
可以通过条件标签控制生成数据的特性
特别适合需要按特定条件生成时序数据的场景

条件生成相比无条件生成的核心优势在于：

可控性：可以通过改变条件标签来控制生成数据的特性
多样性：同一组条件可以生成多个不同的样本
实用性：更符合实际应用场景的需求

使用 PyTorch 构建 1D CGAN 主要包含 Generator 和 Discriminator 两个部分。

import torch
import torch.nn as nn

class Generator(nn.Module):
    def __init__(self, latent_dim, num_classes, seq_len):
        super(Generator, self).__init__()
        self.label_embedding = nn.Embedding(num_classes, num_classes)

        # 初始全连接层
        self.fc = nn.Sequential(nn.Linear(latent_dim + num_classes, 256),
            nn.BatchNorm1d(256),
            nn.ReLU())

        # 1D 转置卷积层
        self.conv_blocks = nn.Sequential(nn.ConvTranspose1d(256, 128, kernel_size=5, stride=2, padding=2, output_padding=1),
            nn.BatchNorm1d(128),
            nn.ReLU(),

            nn.ConvTranspose1d(128, 64, kernel_size=5, stride=2, padding=2, output_padding=1),
            nn.BatchNorm1d(64),
            nn.ReLU(),

            nn.ConvTranspose1d(64, 1, kernel_size=5, stride=1, padding=2),
            nn.Tanh())

        self.seq_len = seq_len

    def forward(self, z, labels):
        # 嵌入标签
        c = self.label_embedding(labels)

        # 拼接噪声和条件
        x = torch.cat([z, c], dim=1)
        x = self.fc(x)

        # 调整形状以便进行转置卷积
        x = x.unsqueeze(2)  # [batch, channels, length]
        x = self.conv_blocks(x)

        # 确保输出序列长度正确
        if x.shape[2] != self.seq_len:
            x = x[:, :, :self.seq_len]

        return x.squeeze(1)  # 移除通道维度，得到[batch, seq_len]

class Discriminator(nn.Module):
    def __init__(self, num_classes, seq_len):
        super(Discriminator, self).__init__()
        self.label_embedding = nn.Embedding(num_classes, seq_len)

        self.conv_blocks = nn.Sequential(nn.Conv1d(2, 64, kernel_size=5, stride=2, padding=2),
            nn.LeakyReLU(0.2),

            nn.Conv1d(64, 128, kernel_size=5, stride=2, padding=2),
            nn.InstanceNorm1d(128),
            nn.LeakyReLU(0.2),

            nn.Conv1d(128, 256, kernel_size=5, stride=2, padding=2),
            nn.InstanceNorm1d(256),
            nn.LeakyReLU(0.2)
        )

        # 计算卷积后的序列长度
        conv_seq_len = seq_len // (2 ** 3)  # 3 层 stride= 2 的卷积

        self.fc = nn.Sequential(nn.Linear(256 * conv_seq_len, 1),
            # 不添加 Sigmoid，因为我们使用 Wasserstein Loss
        )

    def forward(self, x, labels):
        # 嵌入标签并调整形状以匹配输入数据
        c = self.label_embedding(labels).unsqueeze(1)  # [batch, 1, seq_len]

        # 拼接数据和条件
        x = x.unsqueeze(1)  # 添加通道维度 [batch, 1, seq_len]
        x = torch.cat([x, c], dim=1)  # [batch, 2, seq_len]

        # 通过卷积层
        x = self.conv_blocks(x)

        # 展平后通过全连接层
        x = x.view(x.size(0), -1)
        x = self.fc(x)

        return x

卷积核大小：通常选择 3 - 7 之间的奇数，太小难以捕捉特征，太大容易过平滑
步长：下采样通常选择 2，上采样与之对应
隐空间维度：根据数据复杂度选择，通常 64-256
学习率：Generator 和 Discriminator 可以不同，常见配置是 D:1e-4, G:5e-5

Wasserstein Loss 相比原始 GAN 的损失函数能提供更稳定的训练：

def wasserstein_loss(y_pred, y_true):
    """
    y_true: 1 for real samples, -1 for fake samples
    y_pred: discriminator output (no sigmoid!)
    """
    return -torch.mean(y_pred * y_true)

为了保证 Lipschitz 连续性，需要添加梯度惩罚：

def compute_gradient_penalty(D, real_samples, fake_samples, labels, device):
    """Calculates the gradient penalty loss for WGAN-GP"""
    # 随机插值点
    alpha = torch.rand(real_samples.size(0), 1, device=device)
    interpolates = (alpha * real_samples + ((1 - alpha) * fake_samples)).requires_grad_(True)

    d_interpolates = D(interpolates, labels)

    # 计算梯度
    gradients = torch.autograd.grad(
        outputs=d_interpolates,
        inputs=interpolates,
        grad_outputs=torch.ones_like(d_interpolates),
        create_graph=True,
        retain_graph=True,
        only_inputs=True,
    )[0]

    # 计算梯度惩罚
    gradients = gradients.view(gradients.size(0), -1)
    gradient_penalty = ((gradients.norm(2, dim=1) - 1) ** 2).mean()

    return gradient_penalty

可视化是监控训练过程的重要手段：