GAN与金融数据增强

概述

生成对抗网络 (Generative Adversarial Network, GAN) 通过生成器 (Generator) 与判别器 (Discriminator) 的对抗博弈学习数据分布，为金融领域的数据增强 (Data Augmentation)、市场模拟 (Market Simulation) 和压力测试 (Stress Testing) 提供了全新工具。本文介绍 GAN 的基础架构、WGAN 改进、金融时间序列生成及评估方法。

GAN 基础架构

原始 GAN 目标函数

生成器 \(G\) 将噪声 \(z \sim p_z\) 映射为合成数据，判别器 \(D\) 区分真实数据与合成数据：

\[\min_G \max_D V(D, G) = \mathbb{E}_{x \sim p_{\text{data}}}[\log D(x)] + \mathbb{E}_{z \sim p_z}[\log(1 - D(G(z)))]\]

在均衡状态下，\(p_G = p_{\text{data}}\)，判别器输出 \(D(x) = 0.5\)。

训练不稳定性

原始 GAN 存在训练不稳定 (Training Instability)、模式崩塌 (Mode Collapse) 和梯度消失等问题。在金融数据上，这些问题因数据维度高、分布复杂（厚尾、偏斜）而更加突出。

Wasserstein GAN (WGAN)

WGAN 使用 Wasserstein 距离（Earth Mover's Distance）替代 JS 散度，提供更平滑的梯度信号：

\[W(p_{\text{data}}, p_G) = \inf_{\gamma \in \Pi(p_{\text{data}}, p_G)} \mathbb{E}_{(x,y) \sim \gamma}[\|x - y\|]\]

通过 Kantorovich-Rubinstein 对偶：

\[W(p_{\text{data}}, p_G) = \sup_{\|f\|_L \leq 1} \left\{ \mathbb{E}_{x \sim p_{\text{data}}}[f(x)] - \mathbb{E}_{x \sim p_G}[f(x)] \right\}\]

WGAN-GP 通过梯度惩罚 (Gradient Penalty) 实施 Lipschitz 约束：

\[\mathcal{L}_{\text{GP}} = \lambda \mathbb{E}_{\hat{x}}[(\|\nabla_{\hat{x}} D(\hat{x})\|_2 - 1)^2]\]

其中 \(\hat{x} = \epsilon x + (1 - \epsilon) G(z)\)，\(\epsilon \sim U[0, 1]\)。

import torch
import torch.nn as nn

class WGANGPTrainer:
    def __init__(self, generator, critic, lr=1e-4, gp_weight=10):
        self.G = generator
        self.D = critic  # WGAN 中称为 Critic
        self.opt_G = torch.optim.Adam(self.G.parameters(), lr=lr, betas=(0.0, 0.9))
        self.opt_D = torch.optim.Adam(self.D.parameters(), lr=lr, betas=(0.0, 0.9))
        self.gp_weight = gp_weight

    def gradient_penalty(self, real, fake):
        eps = torch.rand(real.size(0), 1, 1).to(real.device)
        interpolated = eps * real + (1 - eps) * fake
        interpolated.requires_grad_(True)
        d_inter = self.D(interpolated)
        grad = torch.autograd.grad(
            outputs=d_inter, inputs=interpolated,
            grad_outputs=torch.ones_like(d_inter),
            create_graph=True
        )[0]
        grad_norm = grad.view(grad.size(0), -1).norm(2, dim=1)
        return ((grad_norm - 1) ** 2).mean()

    def train_step(self, real_data, n_critic=5):
        # 训练 Critic（多步）
        for _ in range(n_critic):
            z = torch.randn(real_data.size(0), self.G.latent_dim).to(real_data.device)
            fake = self.G(z).detach()
            loss_D = self.D(fake).mean() - self.D(real_data).mean()
            gp = self.gradient_penalty(real_data, fake)
            total_D = loss_D + self.gp_weight * gp
            self.opt_D.zero_grad()
            total_D.backward()
            self.opt_D.step()
        # 训练 Generator
        z = torch.randn(real_data.size(0), self.G.latent_dim).to(real_data.device)
        fake = self.G(z)
        loss_G = -self.D(fake).mean()
        self.opt_G.zero_grad()
        loss_G.backward()
        self.opt_G.step()

金融时间序列生成

时序 GAN 架构

金融时间序列具有时序依赖性，生成器和判别器需采用序列模型：

class TimeSeriesGenerator(nn.Module):
    def __init__(self, latent_dim=32, hidden_dim=64, seq_len=252,
                 n_features=5):
        super().__init__()
        self.latent_dim = latent_dim
        self.seq_len = seq_len
        self.lstm = nn.LSTM(latent_dim, hidden_dim, num_layers=2,
                            batch_first=True)
        self.fc = nn.Linear(hidden_dim, n_features)
        self.tanh = nn.Tanh()

    def forward(self, z):
        # z: (batch, seq_len, latent_dim)
        h, _ = self.lstm(z)
        return self.tanh(self.fc(h))  # (batch, seq_len, n_features)

class TimeSeriesCritic(nn.Module):
    def __init__(self, n_features=5, hidden_dim=64):
        super().__init__()
        self.lstm = nn.LSTM(n_features, hidden_dim, num_layers=2,
                            batch_first=True)
        self.fc = nn.Linear(hidden_dim, 1)

    def forward(self, x):
        h, _ = self.lstm(x)
        return self.fc(h[:, -1, :])  # 取最后时刻输出

条件生成 (Conditional Generation)

在特定市场状态条件下生成时间序列，用于场景分析 (Scenario Analysis)：

class ConditionalGenerator(nn.Module):
    def __init__(self, latent_dim, cond_dim, hidden_dim, seq_len, n_features):
        super().__init__()
        self.embed_cond = nn.Linear(cond_dim, hidden_dim)
        self.lstm = nn.LSTM(latent_dim + hidden_dim, hidden_dim,
                            num_layers=2, batch_first=True)
        self.fc = nn.Linear(hidden_dim, n_features)

    def forward(self, z, condition):
        # condition: 市场状态 (如 VIX 水平、利率环境)
        cond_embed = self.embed_cond(condition).unsqueeze(1).expand(-1, z.size(1), -1)
        combined = torch.cat([z, cond_embed], dim=-1)
        h, _ = self.lstm(combined)
        return self.fc(h)

应用场景

压力测试：生成极端市场条件下的模拟路径
数据增强：为稀有事件（如金融危机）生成更多样本
隐私保护：生成统计性质相似但不包含真实交易数据的合成数据集

市场模拟 (Market Simulation)

高保真的市场模拟需要复现以下统计特征 (Stylized Facts)：

厚尾分布 (Fat Tails)：收益率的峰度 \(\kappa > 3\)
波动率聚集 (Volatility Clustering)：\(|r_t|\) 的自相关性
杠杆效应 (Leverage Effect)：\(\text{Corr}(r_t, \sigma_{t+1}) < 0\)
非对称性 (Asymmetry)：下跌幅度大于上涨幅度

评估指标 (Evaluation Metrics)

指标	公式/方法	评估维度
分布距离	KL散度、Wasserstein距离	边际分布
ACF 差异	\(\sum_k \\|\hat{\rho}_k^{\text{real}} - \hat{\rho}_k^{\text{fake}}\\|\)	时序依赖
统计矩匹配	均值、方差、偏度、峰度	分布形状
最大均值差异	MMD: \(\\|\mu_P - \mu_Q\\|_{\mathcal{H}}^2\)	分布整体差异
判别分数	Train-on-Synthetic-Test-on-Real	实际可用性

def evaluate_generated_series(real, fake):
    """综合评估生成时间序列的质量"""
    metrics = {}
    # 统计矩匹配
    for name, func in [('mean', np.mean), ('std', np.std),
                        ('skew', scipy.stats.skew),
                        ('kurtosis', scipy.stats.kurtosis)]:
        metrics[f'{name}_diff'] = abs(func(real) - func(fake))
    # ACF 差异
    from statsmodels.tsa.stattools import acf
    acf_real = acf(real, nlags=20)
    acf_fake = acf(fake, nlags=20)
    metrics['acf_rmse'] = np.sqrt(np.mean((acf_real - acf_fake) ** 2))
    return metrics

TSTR 评估范式

Train on Synthetic, Test on Real (TSTR) 是评估合成数据实用价值的金标准：在合成数据上训练下游模型，在真实数据上评估。如果合成数据质量高，TSTR 性能应接近 TRTR (Train on Real, Test on Real)。

小结

GAN 为金融数据的合成与增强提供了强大工具，尤其在极端事件模拟、隐私保护数据共享和策略压力测试等场景中具有独特价值。WGAN-GP 的训练稳定性改进使其更适合金融时间序列的复杂分布。然而，生成质量的评估仍是开放问题，需要结合统计检验与下游任务表现综合判断。