欢迎加入专注于财经数据与量化投研的【数据科学实战】知识星球！在这里，您将获取持续更新的《财经数据宝典》和《量化投研宝典》，这两部宝典相辅相成，为您在量化投研道路上提供明确指引。 我们提供了精选的国内外量化投研的 180+ 篇高质量文章，并每日更新最新研究成果，涵盖策略开发、因子分析、风险管理等核心领域。 无论您是量化投资新手还是经验丰富的研究者，星球社区都能帮您少走弯路，事半功倍，共同探索数据驱动的投资世界！

引言

金融市场就像天气一样，有时风平浪静，有时波涛汹涌。对于量化交易者来说，能够准确识别市场处于哪种「状态」（我们称之为「regime」），就像气象预报一样重要——它能帮你决定是该全力出击，还是谨慎观望。

本文将带你深入一个实战项目：使用三种不同的深度学习模型（LGMM、VAE 和 Transformer）来识别标普 500 指数（SPY）的市场状态。我们会从基础概念讲起，逐步构建完整的 Python 实现，并对比三种模型的优劣。

无论你是量化交易新手，还是想提升策略的老手，这篇文章都会给你带来启发。

什么是市场状态检测？

市场状态检测（Market Regime Detection）就是把市场行为划分为几个不同的「状态」或「阶段」，比如：

• 平稳期：低波动，趋势明确
• 震荡期：高波动，均值回归
• 危机期：极端波动，恐慌性下跌

准确识别这些状态后，你可以：

1. 调整仓位大小：在平稳期加大仓位，在危机期减仓
2. 选择交易信号：平稳期用趋势策略，震荡期用均值回归
3. 控制杠杆：根据市场状态动态调整杠杆倍数
4. 风险对冲：在高风险状态加强对冲
5. 优化执行时机：避免在状态转换时激进入场

为什么选择这三个模型？

1. LGMM（潜在高斯混合模型）

特点：简单、可解释、快速

LGMM 把市场数据看作是几个高斯分布的混合，每个分布代表一种状态。它就像把一堆数据点自动分成几个「簇」。

优点：

• 训练速度快，适合实时应用
• 结果容易解释
• 不需要大量数据

缺点：

• 只能处理线性可分的情况
• 忽略时间序列的前后关系
• 对复杂的非线性模式无能为力

2. VAE（变分自编码器）

特点：捕捉非线性结构，概率建模

VAE 是一种深度学习模型，它能把高维特征压缩到低维的「潜在空间」，在这个空间里，相似的市场状态会聚集在一起。

优点：

• 能发现非线性的隐藏模式
• 对异常情况敏感
• 提供概率分布，而不只是硬分类

缺点：

• 需要更多计算资源
• 训练过程需要调参
• 仍然忽略时间序列信息

3. Transformer Encoder

特点：理解时间序列，注意力机制

Transformer 是自然语言处理领域的明星模型，它能「记住」历史信息，通过注意力机制判断哪些历史数据更重要。

优点：

• 捕捉时间依赖关系
• 能识别「序列模式」而不只是「瞬时状态」
• 适合复杂的市场动态

缺点：

• 计算开销最大
• 需要足够长的时间序列数据
• 训练难度较高

完整实现：从数据到模型

环境准备

首先安装所需的 Python 库：

pip install yfinance numpy pandas matplotlib seaborn scikit-learn torch

第一步：数据获取与特征工程

我们使用 SPY（标普 500 ETF）的历史数据，时间跨度从 2005 年到现在。

# 导入必要的库
import yfinance 


    
as yf
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.mixture import GaussianMixture
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.cluster import KMeans
import torch
import torch.nn as nn
from torch.utils.data import TensorDataset, DataLoader
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

# 设置绘图风格
sns.set(style='whitegrid', context='talk')

# 下载 SPY 日线数据
symbol = 'SPY'
start = '2005-01-01'
end = datetime.today().strftime('%Y-%m-%d')

df = yf.download(symbol, start=start, end=end, progress=False)
df = df[('Close', 'SPY')].to_frame(name='close')  # 提取收盘价
df['logret'] = np.log(df['close']).diff()  # 计算对数收益率
df.dropna(inplace=True)
df.reset_index(inplace=True)
df.rename(columns={'index': 'Date'}, inplace=True)

print(df.head())

为什么用对数收益率？

• 对数收益率具有可加性，更适合统计建模
• 对于连续复利，对数收益率更自然

接下来构建三个关键特征：

# 特征工程
window_vol = 21  # 约 1 个月的交易日
window_mom = 5   # 短期动量

df['ret_1d'] = df['logret']  # 日收益率
df['vol_21d'] = df['logret'].rolling(window_vol).std() * np.sqrt(252)  


    
# 年化波动率
df['mom_5d'] = df['close'].pct_change(window_mom)  # 5 日动量

df.dropna(inplace=True)
features = df[['ret_1d', 'vol_21d', 'mom_5d']].copy()

print(features.describe())

特征说明：

• ret_1d：当日收益，反映即时涨跌
• vol_21d：21 日滚动波动率（年化），衡量市场不确定性
• mom_5d：5 日动量，区分趋势和震荡

第二步：探索性数据分析（EDA）

在建模前，先可视化数据分布：

# 绘制特征分布
plt.figure(figsize=(14, 4))

plt.subplot(1, 3, 1)
sns.histplot(features['ret_1d'], kde=True, stat='density', bins=80)
plt.title('日收益率分布')

plt.subplot(1, 3, 2)
sns.histplot(features['vol_21d'], kde=True, stat=


    
'density', bins=60)
plt.title('滚动波动率分布（21 日，年化）')

plt.subplot(1, 3, 3)
sns.histplot(features['mom_5d'].dropna(), kde=True, stat='density', bins=60)
plt.title('5 日动量分布')

plt.tight_layout()
plt.show()

观察结果：

• 收益率：呈现重尾分布，不是标准正态分布
• 波动率：右偏，极端波动（危机）较少但影响大
• 动量：较为对称，但取决于市场状态

查看特征间的相关性：

# 绘制特征配对图
sns.pairplot(features.sample(2000), diag_kind='kde', plot_kws={'alpha': 0.5})
plt.suptitle('特征两两关系（样本）', y=1.02)
plt.show()

print("相关性矩阵：\n", features.corr())

关键发现：

• 波动率和收益率呈弱负相关——波动率飙升时，收益率往往为负
• 动量和收益率有一定正相关

第三步：LGMM 状态聚类

现在开始建模。首先标准化特征：

# 特征标准化
scaler = StandardScaler()
X = scaler.fit_transform(features.values)
dates = features.index

拟合高斯混合模型：

# 拟合 LGMM
n_clusters = 3  # 设定 3 个状态
gmm = GaussianMixture(n_components=n_clusters, covariance_type='full', random_state=42)
gmm.fit(X)
lgmm_labels = gmm.predict(X)  # 预测每个数据点的状态

为什么选 3 个状态？

• 通常市场有三种典型状态：平稳、正常、危机
• 可以通过 BIC 或 AIC 准则自动选择最佳状态数

可视化聚类结果：

# 用 PCA 降维到 2D 以便可视化
pca = PCA(n_components=2)
Xp = pca.fit_transform(X)

plt.figure(figsize=(8, 6))
for k in range(n_clusters):
    plt.scatter(Xp[lgmm_labels == k, 0], Xp[lgmm_labels == k, 1], 
                s=6, alpha=0.7, label=f'LGMM {k}')
plt.title('LGMM 状态聚类（PCA 投影）')
plt.xlabel('主成分 1')
plt.ylabel('主成分 2')
plt.legend()
plt.show()

分析每个状态的统计特征：

# 查看每个状态的特征均值和标准差
df_lgmm = features.copy()
df_lgmm['lgmm'] = lgmm_labels
print(df_lgmm.groupby('lgmm').agg(['mean', 'std']).T)

结果解读（示例）：

• 状态 0：低波动，小幅正收益，正动量 → 平稳期
• 状态 1：中等波动，接近零收益，中性动量 → 过渡期
• 状态 2：高波动，负收益，负动量 → 危机期

第四步：VAE 潜在建模

构建一个简单的 VAE：

# 设置设备和随机种子
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
torch.manual_seed(42)

# 转换为 PyTorch 张量
X_tensor = torch.tensor(X, dtype=torch.float32).to(device)
dataset = TensorDataset(X_tensor)
loader = DataLoader(dataset, batch_size=256, shuffle=True)

# VAE 超参数
latent_dim = 2  # 潜在维度
input_dim = X.shape[1]
hidden_dim = 16
lr = 1e-3
epochs = 80

# 定义 VAE 模型
class VAE(nn.Module):
    def __init__(self, input_dim, hidden_dim, latent_dim):
        super().__init__()
        # 编码器
        self.fc1 = nn.Linear(input_dim, hidden_dim)
        self.fc_mu = nn.Linear(hidden_dim, latent_dim)
        self.fc_logvar = nn.Linear(hidden_dim, latent_dim)
        # 解码器
        self.fc_decode = nn.Linear(latent_dim, hidden_dim)
        self.fc_out = nn.Linear(hidden_dim, input_dim)
        self.act = nn.ReLU()
    
    def encode(self, x):
        h = self.act(self.fc1(x))
        return self.fc_mu(h), self.fc_logvar(h)
    
    def reparameterize(self, mu, logvar):
        # 重参数化技巧
        std = torch.exp(0.5 * logvar)
        eps = torch.randn_like(std)
        return mu + eps * std
    
    def decode(self, z):
        h = self.act(self.fc_decode(z))
        return self.fc_out(h)
    
    def forward(self, x):
        mu, logvar = self.encode(x)
        z = self.reparameterize(mu, logvar)
        recon = self.decode(z)
        return recon, mu, logvar

model = VAE(input_dim, hidden_dim, latent_dim).to(device)
opt = torch.optim.Adam(model.parameters(), lr=lr)

训练 VAE：

# VAE 损失函数：重构损失 + KL 散度
def loss_fn(recon, x, mu, logvar):
    recon_loss = nn.functional.mse_loss(recon, x, reduction='sum')
    # KL 散度：后验 q(z|x) 与先验 N(0,1) 之间的距离
    kl = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp())
    return recon_loss + kl

# 训练循环
model.train()



    
for ep in range(epochs):
    total_loss = 0.0
    for batch in loader:
        x_batch = batch[0]
        opt.zero_grad()
        recon, mu, logvar = model(x_batch)
        loss = loss_fn(recon, x_batch, mu, logvar)
        loss.backward()
        opt.step()
        total_loss += loss.item()
    
    if (ep + 1) % 10 == 0:
        print(f'Epoch {ep+1}/{epochs} - loss: {total_loss/len(X):.4f}')

获取潜在编码并聚类：

# 获取所有数据的潜在编码
model.eval()
with torch.no_grad():
    mu, logvar = model.encode(X_tensor)
    z = mu.cpu().numpy()  # 使用均值作为确定性编码

# 在潜在空间用 KMeans 聚类
kmeans = KMeans(n_clusters=n_clusters, random_state=42).fit(z)
vae_labels = kmeans.labels_

# 可视化潜在空间
plt.figure(figsize=(8, 6))
for k in range(n_clusters):
    plt.scatter(z[vae_labels == k, 0], z[vae_labels == k, 1], 
                s=6, alpha=0.7, label=f'VAE {k}')
plt.title('VAE 潜在空间（2D）聚类')
plt.xlabel('z0')
plt.ylabel('z1')
plt.legend()
plt.show()

VAE 的优势：

• 学习到的潜在空间更平滑，能捕捉非线性关系
• 相似行为的日期在潜在空间中聚集

第五步：Transformer 序列建模

Transformer 需要序列数据，因此我们先构建滚动窗口：

# 创建序列数据（滚动窗口）
seq_len = 16  # 每个样本包含 16 个时间步
X_seq = []
y_seq = []




    
for i in range(seq_len, len(X)):
    seq = X[i - seq_len:i]  # 形状：seq_len x feat_dim
    X_seq.append(seq)
    y_seq.append(vae_labels[i])  # 目标：最后一天的 VAE 状态

X_seq = np.stack(X_seq)
y_seq = np.array(y_seq)

print("序列形状：", X_seq.shape, "目标形状：", y_seq.shape)

定义 Transformer 编码器：

# Transformer 编码器模型
class TransEnc(nn.Module):
    def __init__(self, feat_dim, d_model=32, nhead=4, num_layers=2, n_classes=3):
        super().__init__()
        self.input_proj = nn.Linear(feat_dim, d_model)  # 输入投影
        encoder_layer = nn.TransformerEncoderLayer(d_model=d_model, nhead=nhead, dim_feedforward=64)
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
        # 分类头
        self.classifier = nn.Sequential(
            nn.Linear(d_model, d_model // 2),
            nn.ReLU(),
            nn.Linear(d_model // 2, n_classes)
        )
    
    def forward(self, x):  # x: batch x seq_len x feat_dim
        x = self.input_proj(x)  # -> batch x seq_len x d_model
        x = x.permute(1, 0, 2)  # Transformer 需要 seq_len x batch x d_model
        out = self.transformer(x)
        last = out[-1, :, :]  # 取最后一个时间步的表示
        logits = self.classifier(last)
        return logits

# 准备数据加载器
batch_size = 128
X_seq_t = torch.tensor(X_seq, dtype=torch.float32)
y_seq_t = torch.tensor(y_seq, dtype=torch.long)
dataset_seq = TensorDataset(X_seq_t, y_seq_t)
loader_seq = DataLoader(dataset_seq, batch_size=batch_size, shuffle=True)

# 模型和优化器
model_t = TransEnc(feat_dim=X.shape[1], n_classes=n_clusters).to(device)
opt_t = torch.optim.Adam(model_t.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()

训练 Transformer：

# 训练 Transformer 分类器
epochs_t = 30
model_t.train()

for ep in range(epochs_t):
    total_loss = 0.0
    total_acc = 0.0
    
    for xb, yb in loader_seq:
        xb = xb.to(device)
        yb = yb.to(device)
        
        opt_t.zero_grad()
        logits = model_t(xb)
        loss = criterion(logits, yb)
        loss.backward()
        opt_t.step()
        
        total_loss += loss.item() * xb.size(


    
0)
        total_acc += (logits.argmax(1) == yb).sum().item()
    
    if (ep + 1) % 5 == 0:
        avg_loss = total_loss / len(dataset_seq)
        avg_acc = total_acc / len(dataset_seq)
        print(f'Epoch {ep+1}/{epochs_t} - loss {avg_loss:.4f} acc {avg_acc:.3f}')

获取 Transformer 预测的状态：

# 推理：获取 Transformer 预测的状态
model_t.eval()
with torch.no_grad():
    logits = model_t(X_seq_t.to(device))
    t_labels = logits.argmax(1).cpu().numpy()

# 对齐到完整时间轴（前 seq_len 天没有预测）
t_dates = dates[seq_len:]

第六步：模型对比与分析

状态转换频率

# 构建状态标签数据框
label_df = pd.DataFrame(index=dates)
label_df['LGMM'] = lgmm_labels
label_df['VAE'] = vae_labels
label_df['Transformer'] = np.nan
label_df.loc[t_dates, 'Transformer'] = t_labels

# 计算状态转换次数
def transitions(series):
    s = series.dropna().astype(int)
    return (s.shift(1) != s).sum()

transition_counts = {}
for name in ['LGMM', 'VAE', 'Transformer']:
    total_trans = transitions(label_df[name])
    transition_counts[name] = total_trans
    print(f'{name}：总转换次数 = {total_trans}')

# 可视化
plt.figure(figsize=(6, 4))
plt.bar(transition_counts.keys(), transition_counts.values(), 
        color=['steelblue', 'coral', 'seagreen'])
plt.title('状态转换次数（越低越稳定）')
plt.ylabel('转换次数')

for i, v in enumerate(transition_counts.values()):
    plt.text(i, v + (max(transition_counts.values()) * 0.02), str(v),
             ha='center', fontsize=10)

plt.tight_layout()
plt.show()

结果解读：

• LGMM：转换最少，状态平滑但可能反应滞后
• VAE：转换最频繁，对短期变化高度敏感
• Transformer：介于两者之间，平衡稳定性和响应性

时间轴对比

# 绘制状态时间线（最近 3 年）
start_plot = '2021-01-01'
dfp = label_df.loc[start_plot:]

fig, axes = plt.subplots(3, 1, figsize=(14, 6), sharex=True)
models = ['LGMM', 'VAE', 'Transformer']
colors = ['steelblue', 'coral', 'seagreen']

for ax, model, color in zip(axes, models, colors):
    ax.plot(dfp.index, dfp[model], color=color, lw=1.2)
    ax.set_title(model, loc='left', fontsize=11, fontweight='bold')
    ax.set_yticks([0, 1, 2])
    ax.set_ylabel('状态', fontsize=9)
    ax.grid(True, alpha=0.3)

axes[-1].set_xlabel('日期', fontsize=10)
fig.suptitle('市场状态时间线：LGMM vs VAE vs Transformer', fontsize=14, fontweight='bold')



    
fig.text(0.5, 0.9, '状态含义 → 0 = 平稳 | 1 = 过渡 | 2 = 波动', 
         ha='center', fontsize=10, color='k')

plt.tight_layout(rect=[0, 0, 1, 0.93])
plt.show()

观察重点：

• LGMM 切换较慢，可能错过转折点
• VAE 反应敏锐，但可能过度敏感
• Transformer 有「记忆」，能提前察觉状态变化

风险调整收益

# 计算每个状态的年化收益、波动率和夏普比率
def regime_stats(df_labels, returns, model_name):
    results = []
    for regime in sorted(df_labels[model_name].dropna().unique()):
        mask = df_labels[model_name] == regime
        r = returns[mask]
        
        ann_ret = r.mean() * 252
        ann_vol = r.std() * np.sqrt(252)
        sharpe = ann_ret / ann_vol if ann_vol > 0 else 0
        
        results.append({
            'Model': model_name,
            'Regime': regime,
            'AnnReturn': ann_ret,
            'AnnVol': ann_vol,
            'Sharpe': sharpe
        })
    
    return pd.DataFrame(results)

# 合并所有模型的统计数据
returns = df['logret'].iloc[seq_len:]  # 对齐 Transformer 的时间轴
all_stats = pd.concat([
    regime_stats(label_df.iloc[seq_len:], returns, 'LGMM'),
    regime_stats(label_df.iloc[seq_len:], returns, 'VAE'),
    regime_stats(label_df.iloc[seq_len:], returns, 'Transformer')
])

print(all_stats)

关键发现（示例）：

• 状态 0（平稳期）：夏普比率最高（10+），年化收益 3%
• 状态 1（过渡期）：中等表现，夏普比率 3-5
• 状态 2（波动期）：夏普比率为负（-15 ~ -20），年化收益 -3.5% ~ -4%

结论：

• 所有模型都显示，平稳期风险调整收益最高
• 波动期一致性地损害绩效
• Transformer 在识别高质量状态上表现最佳

实战应用：如何用于交易策略

状态检测不是最终目的，而是决策的输入。以下是几种应用方式：

1. 动态仓位管理

# 示例：根据状态调整仓位
def position_sizing(regime):
    """
    根据检测到的状态返回建议仓位
    """
    if regime == 0:  # 平稳期
        return 1.0  # 满仓
    elif regime == 1:  # 过渡期
        return 0.5  # 半仓
    else:  # 波动期


    

        return 0.2  # 轻仓或观望

# 应用到策略
label_df['position'] = label_df['Transformer'].apply(position_sizing)

2. 信号选择

# 示例：根据状态选择交易信号
def select_signal(regime, momentum, mean_reversion):
    """
    平稳期用动量信号，震荡期用均值回归信号
    """
    if regime == 0:
        return momentum  # 趋势策略
    else:
        return mean_reversion  # 均值回归策略

3. 杠杆控制

# 示例：动态杠杆
def leverage_control(regime):
    if regime == 0:
        return 2.0  # 低波动期可适度加杠杆
    elif regime == 1:
        return 1.0
    else:
        return 0.5  # 高波动期降低杠杆

4. 风险对冲

# 示例：根据状态决定是否对冲
def hedging_strategy(regime):
    if regime == 2:  # 波动期
        return True  # 启动对冲（如买入看跌期权）
    else:
        return False

5. 执行时机

避免在状态转换时激进入场：

# 检测状态变化
label_df['regime_change'] = (label_df['Transformer'] != label_df['Transformer'].shift(1)).astype(int)

# 在状态稳定时才入场
stable_periods = label_df[label_df['regime_change'] == 0]

模型选择建议

总结

本文通过一个完整的实战项目，展示了如何用三种不同的深度学习模型识别市场状态：

1. LGMM：简单高效，适合快速部署，但对非线性模式和时间依赖无能为力
2. VAE：捕捉复杂的潜在结构，但可能过度敏感
3. Transformer：平衡稳定性和响应性，能「记住」历史并提前察觉状态变化

关键要点：

• 市场状态检测是系统化交易的核心组件
• 不同模型有不同的权衡，没有「最好」只有「最合适」
• 状态检测的价值在于指导决策：仓位、信号、杠杆、对冲
• 平稳期风险调整收益最高，波动期需要防守

下一步：

• 尝试更多特征（基本面、市场情绪、另类数据）
• 集成多个模型（ensemble），提高鲁棒性
• 回测完整策略，评估实际 P&L
• 考虑交易成本、滑点等现实因素

希望这篇文章能为你的量化交易之路提供新的思路。记住，市场状态检测不是为了预测未来，而是为了在当下做出更明智的决策。

参考文章

财经数据与量化投研知识社区

核心权益如下：

1. 赠送《财经数据宝典》完整文档，汇集多年财经数据维护经验
2. 赠送《量化投研宝典》完整文档，汇集多年量化投研领域经验
3. 赠送《PyBroker-入门及实战》视频课程，手把手学习量化策略开发
4. 每日分享高质量量化投研文章（已更新 180+篇）、代码和相关资料
5. 定期更新高频财经数据
6. 参与年度不少于 10 次专属直播与录播课程
7. 与核心开发者直接交流，解决实际问题
8. 获取专业微信群交流机会和课程折扣

星球已有丰富内容积累，包括量化投研论文、财经高频数据、 PyBroker 视频教程、定期直播、数据分享和答疑解难。适合对量化投研和财经数据分析有兴趣的学习者及从业者。欢迎加入我们！