AIGC合成时序数据入门指南：从原理到实战避坑

1次阅读

没有评论

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

在实际项目中，获取高质量的时序数据往往面临三大难题：

数据获取成本高：工业设备传感器数据需要长期部署采集，金融数据需要购买昂贵的数据源
隐私合规风险：医疗健康、用户行为等数据涉及敏感信息，直接使用可能违反 GDPR 等法规
极端场景覆盖不足：设备故障、市场崩盘等罕见事件数据难以获取，但模型训练又必不可少

AIGC 技术通过生成合成数据，可以在保留原始数据统计特性的同时，完美解决上述问题。我最近在一个风电设备故障预测项目中，就通过合成数据将训练样本扩充了 5 倍，使模型召回率提升了 23%。

优势：生成质量高，能捕捉复杂时间依赖关系
劣势：训练不稳定，容易发生模式坍塌
适用场景：高频金融数据、多变量工业传感器数据

优势：训练稳定，有明确概率解释
劣势：生成样本往往过于平滑
适用场景：医疗时序数据、需求预测等平稳序列

优势：最新 SOTA，生成质量极高
劣势：计算成本大，推理速度慢
适用场景：对质量要求极高的场景（如语音合成）

新手建议：从 Conditional GAN 入手，它在生成质量和训练难度间取得了较好平衡。下面分享我的 PyTorch 实现。

import numpy as np
from sklearn.preprocessing import MinMaxScaler

def sliding_window(data, window_size=24, stride=1):
    """
    将时序数据转换为滑动窗口样本
    :param data: 原始时序数据 (T, D)
    :return: (N, window_size, D)
    """
    samples = []
    for i in range(0, len(data)-window_size, stride):
        window = data[i:i+window_size]
        # 每个特征单独归一化到[-1,1]
        scaler = MinMaxScaler(feature_range=(-1, 1))
        window = scaler.fit_transform(window)
        samples.append(window)
    return np.array(samples)

import torch
import torch.nn as nn
from torch.nn.utils import spectral_norm

class Discriminator(nn.Module):
    def __init__(self, input_dim):
        super().__init__()
        self.conv = nn.Sequential(
            # 频谱归一化防止模式坍塌
            spectral_norm(nn.Conv1d(input_dim, 64, kernel_size=5, padding=2)),
            nn.LeakyReLU(0.2),
            spectral_norm(nn.Conv1d(64, 128, kernel_size=5, padding=2)),
            nn.LeakyReLU(0.2),
            # 全局平均池化替代全连接层
            nn.AdaptiveAvgPool1d(1)
        )
        self.fc = spectral_norm(nn.Linear(128, 1))

    def forward(self, x):
        # x 形状: (batch, seq_len, features)
        x = x.permute(0, 2, 1)  # 转换为通道优先
        features = self.conv(x).squeeze()
        return self.fc(features)

def compute_gradient_penalty(D, real_samples, fake_samples):
    """计算梯度惩罚项"""
    # 随机插值样本
    alpha = torch.rand(real_samples.size(0), 1, 1).to(device)
    interpolates = (alpha * real_samples + ((1 - alpha) * fake_samples))
    interpolates.requires_grad_(True)

    d_interpolates = D(interpolates)

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

    # 梯度惩罚项
    gradient_penalty = ((gradients.norm(2, dim=1) - 1) ** 2).mean()
    return gradient_penalty

监控指标：计算 FID（Frechet Inception Distance）

from scipy.linalg import sqrtm

def calculate_fid(real_features, fake_features):
    mu1, sigma1 = real_features.mean(0), np.cov(real_features.T)
    mu2, sigma2 = fake_features.mean(0), np.cov(fake_features.T)

    diff = mu1 - mu2
    covmean = sqrtm(sigma1.dot(sigma2))

    return diff.dot(diff) + np.trace(sigma1 + sigma2 - 2*covmean)

解决方案：
增加 mini-batch 判别层
使用多样性正则化损失

对每个滑动窗口进行 滑动标准化：

def rolling_normalize(x, window=24):
    """滑动窗口标准化"""
    rolled = x.unfold(0, window, 1)  # 滑动窗口展开
    means = rolled.mean(dim=-1)
    stds = rolled.std(dim=-1)

    # 避免除零
    stds[stds < 1e-6] = 1.0

    # 重新对齐维度
    means = means.unsqueeze(-1).expand_as(x)
    stds = stds.unsqueeze(-1).expand_as(x)

    return (x - means) / stds

from dtaidistance import dtw

# 计算真实与合成序列的平均 DTW 距离
distances = []
for real_seq, fake_seq in zip(real_data, fake_data):
    d = dtw.distance(real_seq, fake_seq)
    distances.append(d)

print(f"平均 DTW 距离: {np.mean(distances):.4f}")

import matplotlib.pyplot as plt
from scipy import fft

# 计算 FFT
real_fft = np.abs(fft.fft(real_data[:,0]))
fake_fft = np.abs(fft.fft(fake_data[:,0]))

plt.figure(figsize=(10,4))
plt.plot(real_fft[:100], label='Real')
plt.plot(fake_fft[:100], label='Fake', alpha=0.7)
plt.title('Frequency Domain Comparison')
plt.legend();