zzy1213/A-Share-Quant-Radar
0
1"""2model.py — A 股多模态量化分析深度学习模型（v2 强化版）3 4承接 data_fetcher.py 的数据：5 - 时序分支：1D-CNN 提取局部模式 → Transformer Encoder 捕捉长程依赖6 - 文本分支：多条新闻嵌入经 Self-Attention 动态加权聚合7 - 回归输出：预测明日最高价 & 明日最低价8"""9 10import torch11import torch.nn as nn12import torch.nn.functional as F13import math14 15 16# ============================================================17# 1. 时序分支：CNN + Transformer 混合架构18# ============================================================19class TimeSeriesBranch(nn.Module):20 """21 混合时序编码器：22 1) 多尺度 1D-CNN 提取局部特征（3日/5日/7日窗口）23 2) Transformer Encoder 建模全局时序依赖24 3) 均值池化输出固定维度向量25 """26 27 def __init__(28 self,29 in_features: int = 10,30 cnn_channels: int = 64,31 d_model: int = 128,32 nhead: int = 4,33 ff_dim: int = 256,34 dropout: float = 0.1,35 ):36 super().__init__()37 38 # --- 多尺度 1D-CNN（并行三组不同 kernel_size） ---39 # Conv1d 输入格式: (B, C_in, L)，所以 forward 中需要 transpose40 self.conv3 = nn.Conv1d(in_features, cnn_channels, kernel_size=3, padding=1)41 self.conv5 = nn.Conv1d(in_features, cnn_channels, kernel_size=5, padding=2)42 self.conv7 = nn.Conv1d(in_features, cnn_channels, kernel_size=7, padding=3)43 self.bn = nn.BatchNorm1d(cnn_channels * 3)44 45 # --- 将 CNN 输出映射到 Transformer 的 d_model ---46 self.proj = nn.Linear(cnn_channels * 3, d_model)47 48 # --- 位置编码 ---49 self.pos_encoding = PositionalEncoding(d_model=d_model, dropout=dropout)50 51 # --- Transformer Encoder（2 层） ---52 encoder_layer = nn.TransformerEncoderLayer(53 d_model=d_model,54 nhead=nhead,55 dim_feedforward=ff_dim,56 dropout=dropout,57 batch_first=True,58 )59 self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=2)60 self.out_dim = d_model61 62 def forward(self, x: torch.Tensor) -> torch.Tensor:63 """64 Args:65 x: (B, S, F) — batch, seq_len, features66 Returns:67 (B, d_model)68 """69 # Conv1d 需要 (B, F, S)70 x_t = x.transpose(1, 2) # (B, F, S)71 72 # 多尺度卷积 + 拼接73 c3 = F.relu(self.conv3(x_t)) # (B, cnn_ch, S)74 c5 = F.relu(self.conv5(x_t)) # (B, cnn_ch, S)75 c7 = F.relu(self.conv7(x_t)) # (B, cnn_ch, S)76 x_cnn = torch.cat([c3, c5, c7], dim=1) # (B, cnn_ch*3, S)77 x_cnn = self.bn(x_cnn)78 79 # 转回 (B, S, cnn_ch*3)，映射到 d_model80 x_cnn = x_cnn.transpose(1, 2) # (B, S, cnn_ch*3)81 x_proj = self.proj(x_cnn) # (B, S, d_model)82 83 # Transformer 编码84 x_enc = self.pos_encoding(x_proj)85 x_enc = self.transformer(x_enc) # (B, S, d_model)86 87 # 均值池化88 return x_enc.mean(dim=1) # (B, d_model)89 90 91# ============================================================92# 2. 文本分支：Self-Attention 动态聚合多条新闻93# ============================================================94class NewsSelfAttention(nn.Module):95 """96 对多条新闻的嵌入向量进行自注意力加权聚合。97 98 输入: (B, num_news, embed_dim) — 每个样本有 num_news 条新闻99 输出: (B, hidden_dim) — 加权聚合后的单一新闻表征100 """101 102 def __init__(self, embed_dim: int = 768, hidden_dim: int = 128, dropout: float = 0.1):103 super().__init__()104 105 # 先将每条新闻投影到 hidden_dim106 self.proj = nn.Linear(embed_dim, hidden_dim)107 108 # 自注意力：标准 scaled dot-product，单头即可109 self.query = nn.Linear(hidden_dim, hidden_dim)110 self.key = nn.Linear(hidden_dim, hidden_dim)111 self.value = nn.Linear(hidden_dim, hidden_dim)112 self.scale = math.sqrt(hidden_dim)113 114 # 聚合后的输出层115 self.out_mlp = nn.Sequential(116 nn.Linear(hidden_dim, hidden_dim),117 nn.ReLU(),118 nn.Dropout(dropout),119 )120 self.out_dim = hidden_dim121 122 def forward(self, x: torch.Tensor) -> torch.Tensor:123 """124 Args:125 x: (B, N, embed_dim) — N 条新闻的嵌入126 Returns:127 (B, hidden_dim)128 """129 h = F.relu(self.proj(x)) # (B, N, hidden_dim)130 131 Q = self.query(h) # (B, N, hidden_dim)132 K = self.key(h) # (B, N, hidden_dim)133 V = self.value(h) # (B, N, hidden_dim)134 135 # Scaled dot-product attention136 attn_scores = torch.bmm(Q, K.transpose(1, 2)) / self.scale # (B, N, N)137 attn_weights = F.softmax(attn_scores, dim=-1) # (B, N, N)138 attended = torch.bmm(attn_weights, V) # (B, N, hidden_dim)139 140 # 对所有新闻维度求均值，得到聚合表征141 pooled = attended.mean(dim=1) # (B, hidden_dim)142 return self.out_mlp(pooled) # (B, hidden_dim)143 144 145# ============================================================146# 3. 多模态量化模型（v2 — 回归输出明日最高/最低价）147# ============================================================148class MultimodalQuantModel(nn.Module):149 """150 双分支多模态网络 v2：151 - 时序分支：多尺度 1D-CNN + Transformer Encoder152 - 文本分支：Self-Attention 聚合多条新闻153 - 融合层：拼接 → MLP → 输出 2 个连续值（明日最高价、明日最低价）154 """155 156 def __init__(157 self,158 ts_features: int = 10,159 cnn_channels: int = 64,160 ts_d_model: int = 128,161 ts_nhead: int = 4,162 ts_ff_dim: int = 256,163 text_embedding_dim: int = 768,164 text_hidden: int = 128,165 num_news: int = 10,166 fusion_hidden: int = 128,167 num_outputs: int = 2, # 明日最高价 + 明日最低价168 dropout: float = 0.1,169 ):170 super().__init__()171 172 # ---------- 时序分支 ----------173 self.ts_branch = TimeSeriesBranch(174 in_features=ts_features,175 cnn_channels=cnn_channels,176 d_model=ts_d_model,177 nhead=ts_nhead,178 ff_dim=ts_ff_dim,179 dropout=dropout,180 )181 182 # ---------- 文本分支 ----------183 self.text_branch = NewsSelfAttention(184 embed_dim=text_embedding_dim,185 hidden_dim=text_hidden,186 dropout=dropout,187 )188 189 # ---------- 融合回归头 ----------190 fused_dim = self.ts_branch.out_dim + self.text_branch.out_dim191 self.fusion = nn.Sequential(192 nn.Linear(fused_dim, fusion_hidden),193 nn.ReLU(),194 nn.Dropout(dropout),195 nn.Linear(fusion_hidden, fusion_hidden // 2),196 nn.ReLU(),197 nn.Linear(fusion_hidden // 2, num_outputs),198 )199 200 def forward(self, ts_input: torch.Tensor, text_input: torch.Tensor) -> torch.Tensor:201 """202 Args:203 ts_input: (B, seq_len, ts_features) — 日线量价序列204 text_input: (B, num_news, text_embedding_dim) — 多条新闻嵌入205 206 Returns:207 predictions: (B, 2) — [明日预测最高价, 明日预测最低价]208 """209 x_ts = self.ts_branch(ts_input) # (B, ts_d_model)210 x_text = self.text_branch(text_input) # (B, text_hidden)211 212 fused = torch.cat([x_ts, x_text], dim=1) # (B, ts_d_model + text_hidden)213 return self.fusion(fused) # (B, 2)214 215 216# ============================================================217# 4. 正弦位置编码218# ============================================================219class PositionalEncoding(nn.Module):220 """标准正弦-余弦位置编码，batch_first=True。"""221 222 def __init__(self, d_model: int, dropout: float = 0.1, max_len: int = 500):223 super().__init__()224 self.dropout = nn.Dropout(p=dropout)225 226 pe = torch.zeros(max_len, d_model)227 position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)228 div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))229 pe[:, 0::2] = torch.sin(position * div_term)230 pe[:, 1::2] = torch.cos(position * div_term)231 pe = pe.unsqueeze(0)232 self.register_buffer("pe", pe)233 234 def forward(self, x: torch.Tensor) -> torch.Tensor:235 x = x + self.pe[:, :x.size(1), :]236 return self.dropout(x)237 238 239# ============================================================240# 5. 自定义 Huber Loss — 应对 A 股极端跳空噪声241# ============================================================242class HuberLoss(nn.Module):243 """244 自定义 Huber Loss（Smooth L1 的推广形式）。245 246 |error| <= delta → 0.5 * error²（MSE，对小误差敏感）247 |error| > delta → delta * |error| - 0.5 * delta²（MAE，容忍极端跳空）248 249 适用于 A 股涨跌停、极端跳空场景，有效抑制离群点主导梯度。250 """251 252 def __init__(self, delta: float = 1.0):253 super().__init__()254 self.delta = delta255 256 def forward(self, y_pred: torch.Tensor, y_true: torch.Tensor) -> torch.Tensor:257 residual = torch.abs(y_true - y_pred)258 quadratic = torch.clamp(residual, max=self.delta)259 linear = residual - quadratic260 loss = 0.5 * quadratic ** 2 + self.delta * linear261 return loss.mean()262 263 264# ============================================================265# 6. 测试代码 — 验证维度对齐266# ============================================================267if __name__ == "__main__":268 # ----- 超参设定 -----269 BATCH_SIZE = 4270 SEQ_LEN = 10 # ~10 个交易日271 TS_FEATURES = 10 # open/close/high/low/volume/amount/amplitude/pct_change/change/turnover272 NUM_NEWS = 10 # 每个样本 10 条新闻273 TEXT_EMBEDDING_DIM = 768 # 预训练模型嵌入维度274 275 print("=" * 60)276 print(" MultimodalQuantModel v2 结构测试")277 print("=" * 60)278 279 # ----- 实例化模型 -----280 model = MultimodalQuantModel(281 ts_features=TS_FEATURES,282 num_news=NUM_NEWS,283 text_embedding_dim=TEXT_EMBEDDING_DIM,284 )285 286 total_params = sum(p.numel() for p in model.parameters())287 trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)288 print(f"\n[模型参数量] 总计: {total_params:,} | 可训练: {trainable_params:,}")289 print(f"\n[模型结构]\n{model}\n")290 291 # ----- 生成 dummy data -----292 dummy_ts = torch.randn(BATCH_SIZE, SEQ_LEN, TS_FEATURES)293 dummy_text = torch.randn(BATCH_SIZE, NUM_NEWS, TEXT_EMBEDDING_DIM)294 295 print("[输入维度]")296 print(f" 时序输入 (ts_input): {dummy_ts.shape} ← (batch, seq_len, features)")297 print(f" 文本输入 (text_input): {dummy_text.shape} ← (batch, num_news, embed_dim)")298 299 # ----- 前向传播 -----300 model.eval()301 with torch.no_grad():302 preds = model(dummy_ts, dummy_text)303 304 print(f"\n[输出维度]")305 print(f" predictions: {preds.shape} ← (batch, 2) = [明日最高价, 明日最低价]")306 print(f"\n[示例输出（第 1 条样本）]")307 print(f" 预测明日最高价: {preds[0][0].item():.4f}")308 print(f" 预测明日最低价: {preds[0][1].item():.4f}")309 310 # ----- 测试 Huber Loss -----311 print(f"\n{'=' * 60}")312 print(" HuberLoss 回归测试")313 print("=" * 60)314 315 criterion = HuberLoss(delta=1.0)316 fake_pred = torch.tensor([[15.2, 14.0], [20.5, 19.1]])317 fake_true = torch.tensor([[15.0, 13.8], [25.0, 18.5]])318 loss = criterion(fake_pred, fake_true)319 print(f" pred: {fake_pred.tolist()}")320 print(f" true: {fake_true.tolist()}")321 print(f" Huber Loss: {loss.item():.4f}")322 323 # ----- 维度断言 -----324 assert preds.shape == (BATCH_SIZE, 2), f"输出维度错误: {preds.shape}"325 print(f"\n{'=' * 60}")326 print(" 所有维度检查通过！v2 模型骨架就绪。")327 print("=" * 60)328