多节点时序预测：时空融合与共享注意力轻量化实践

做时序预测的同学，大概都遇到过同一个尴尬时刻：训练集上 loss 曲线漂亮得像教科书，一换到验证集，MAE 就卡在某个数字上再也下不去。加深网络、换损失函数、调学习率、上 Transformer，能试的都试了，误差像被钉住一样纹丝不动。

问题往往不在模型不够大，而在建模的维度少了一个。大量时序任务的数据天生是"多节点"的：电网里几十个变电站、路网里几百个卡口、工厂里上千个传感器。只沿时间轴做建模，等于默认这些节点互不相干；可现实里，相邻路口的车流会互相传导，上游机组的波动会顺着管网影响下游。信息明明就在那里，模型却看不见。

把时间维和空间维的特征融合起来、再用注意力去挑出真正有用的关联，这套思路在近几年的学术会议里已经成了相当通用的范式。但这里真正容易踩坑的地方是："时空融合 + 注意力"本身不是银弹，融合发生在哪一层、注意力加在哪个轴上、为了压缩参数你又牺牲了什么，才决定它到底能不能落地。本文不打算复述论文结论，而是从工程视角把这条路走一遍：拆清楚误差来源，给出可以直接跑的 PyTorch 实现，对比不同轻量化配置的参数量，最后整理一份排错清单。

先给一个判断：如果你的多节点时序任务已经用单变量或纯时序模型调到了瓶颈，时空联合建模大概率能带来一轮明显收益；如果你的数据里节点之间根本没有稳定关联，那强行加空间注意力只会增加参数和过拟合风险。至于标题里"误差清零"这种说法，属于宣传口径，任何公开评测都不会真的归零，理性看待即可——真正值得关注的是误差下降的幅度和参数量的权衡。

1. 这篇文章真正要解决的问题

在动手写模型之前，先把"误差为什么降不下来"这件事拆开看。时序预测的误差通常来自四个方向，分清楚了才知道该往哪儿加模块。

第一类是时间依赖没被充分捕捉。周期性、趋势性、突发波动混在一起，简单的滑动窗口加线性层只能学到平均趋势，遇到节假日、天气突变这类模式切换就崩。第二类是空间关联被忽略。多节点数据里，某个节点的未来值不只取决于自己的历史，还取决于邻居的当前状态，纯时序模型对这一层信息完全失明。第三类是噪声与长尾。传感器抖动、采集丢包、极端值，都会把损失函数往错误方向拉。第四类是参数膨胀带来的泛化损失。为了让模型"更强"，层数和宽度一路加，结果训练集拟合得越来越好，验证集反而退化。

时空特征融合解决的是第二类问题，注意力机制主要缓解第一类和第四类——它用内容相关的加权代替固定的卷积核或循环结构，在同样甚至更少的参数下表达更灵活的依赖关系。但如果只加模块不做约束，参数膨胀会立刻反噬。所以真正有价值的工程问题是：怎么在参数量不升反降的前提下，把空间维度的信息有效引入。

哪些读者最该读这篇？做过单点时序预测、想扩展到多节点场景的；手上有 Transformer 时序模型但效果不如预期的；模型能跑但显存和推理延迟顶不住的；以及想复现类似结构但被论文里的公式绕晕的。如果你只是做单条序列的简单预测，本文的空间建模部分可以直接跳过，看时间注意力和轻量化那两节就够了。

需要提前明确一个边界：本文的代码基于合成数据构建，目的是把完整链路跑通、把每个模块的输入输出形状讲清楚。换成你自己的数据时，数据加载部分要重写，模型主体基本可以复用。

2. 基础概念与核心原理

2.1 时空特征融合到底在融合什么

先把概念说清楚。所谓时空特征融合，是指模型同时对数据的时间轴和空间轴（节点轴）做特征提取，再把两路表示合成为一路。它不是某个具体网络结构，而是一类设计模式。

用一个具体场景理解：假设你在做城市路网车速预测，输入是 100 个卡口过去 1 小时的速度序列，要预测未来 15 分钟。时间轴上，每个卡口自己的速度有早晚高峰的周期规律；空间轴上，相邻卡口的拥堵会以波的形式向下游传播。两路信息缺一不可——只看时间，你不知道堵点从哪来；只看空间，你不知道现在是高峰期还是深夜。

融合的位置决定了三种典型范式，它们的差别在实际工程里非常明显：

中期融合是多数顶会方案的选择，本文的实现也走这条路。但中期融合最大的坑在于：如果两路分支完全独立、参数各自一套，参数量会直接翻倍。这正是后面轻量化设计的切入点——让时间和空间共享同一套注意力参数，只通过张量轴重排来区分它们在看哪个维度。

2.2 注意力机制在时空建模里扮演什么角色

注意力机制的核心是一句话：根据当前查询的内容，动态决定该关注哪些位置。它和卷积、循环结构的区别在于权重不是固定的，而是由输入算出来的。

放到时空场景里，注意力至少有三个落点，作用完全不同：

• 时间注意力：在同一个节点的序列内部，判断哪些历史时刻对当前预测更重要。早高峰的预测，两小时前的数据可能远不如二十分钟前的有用。
• 空间注意力：在同一个时刻，判断哪些邻居节点对当前节点更重要。注意这里学出来的是"数据驱动"的关联强度，不一定要等于物理上的距离——两个隔得远但通勤方向一致的卡口，车流相关性可能比地理邻居还高。
• 通道注意力：在特征维上做筛选，决定哪些特征通道该被放大。多源数据（流量、速度、占有率）融合时特别有用。

本文的实现聚焦前两个。这里有个容易被误解的点：很多人以为空间注意力需要一个预定义的邻接矩阵作为掩码。实际项目中，预定义图可以作为先验加进去（用注意力掩码的形式），但不加也能学出合理的关联，代价是需要更多数据。数据量小的时候，加先验图往往更稳。

2.3 为什么用共享注意力做轻量化

标准的多头注意力里，参数主要来自四块投影矩阵：查询、键、值、输出，各是一个d_model × d_model的矩阵。再加上前馈网络的d_model × (expand × d_model)两层，一个 Transformer block 的参数量大致是4d² + 2 × expand × d²。

如果时间和空间各配一套独立的 block，参数量直接乘以二。而共享权重的思路是：时间和空间注意力的计算形式其实是同构的，区别只在于沿哪个轴做 softmax。时间注意力是"固定节点、沿时间轴聚合"，空间注意力是"固定时刻、沿节点轴聚合"。既然计算图一样，就没必要维护两套参数——把张量重排一下，送进同一个注意力模块即可。

这个改动听起来简单，收益却很实在。它不仅砍掉了近一半的注意力参数，还带来一个隐性好处：共享参数相当于给模型加了一个正则约束，两个轴的表示被迫落在同一个特征空间里，融合时的对齐成本更低。实践中这条往往比参数量的收益更有价值。

3. 环境准备与前置条件

代码依赖很少，主流环境都能跑。具体版本请以你的实际项目为准，本文重点演示通用思路，不锁定小版本号。

• Python 3.8 及以上
• PyTorch 1.9 及以上（nn.MultiheadAttention的batch_first=True参数在 1.9 之后稳定支持，低于这个版本需要手动转置）
• NumPy（仅用于可选的数据统计打印）
• 有 CUDA 环境更好，但本文的合成数据规模很小，纯 CPU 也能在几分钟内跑完一轮

目录结构建议如下，把模型、数据、训练脚本分开，后面做消融实验会方便很多：

st_fusion_demo/ ├── data.py # 合成数据生成与归一化 ├── model.py # 时空融合模型定义 ├── train.py # 训练与评估入口 ├── profile.py # 参数量统计对比 └── config.yaml # 超参数配置

如果只想快速验证，把三个 Python 文件的内容合并到一个脚本里也能跑，但做对比实验时分开会更清爽。

4. 核心流程拆解

把整个链路拆成六步，每一步都说明为什么要这么做。

第一步：构造带空间耦合的数据。真实多节点数据的空间关联往往很弱或者很脏，如果直接用，你很难判断模型没学好是因为结构不对还是数据本身没信号。用合成数据的好处是关联强度可控，可以先验证模型能不能学到已知的规律，再换真实数据。合成时要注意：必须真的引入邻居耦合项，否则空间注意力学不到任何东西，整个实验就白做了。

第二步：按通道做标准化。多通道特征的量纲差异很大，主信号可能在几十的量级，一阶差分在个位数，邻居均值又在另一个区间。不标准化的话，梯度会被大量纲的通道主导。这里有个细节：统计量只能从训练集算，验证集和测试集用训练集的均值和方差，否则就是数据泄露。

第三步：输入投影与位置编码。原始特征维度通常很小（本文是 3），需要先投影到d_model维才能进注意力。同时加上两类位置编码：时间位置编码告诉模型"这是窗口里的第几个时刻"，节点嵌入告诉模型"这是第几个节点"。少了节点嵌入，空间注意力会退化成对节点顺序无感的平均池化。

第四步：时空交替建模。每一层里，分别沿时间轴和空间轴做一次注意力。顺序上先时间后空间，或者先空间后时间，实践中差别不大，但要注意两者必须基于同一份输入并行计算，而不是串行——串行的话空间分支看到的是已经被时间注意力改过的表示，两路信息的独立性就没了。

第五步：门控融合。直接把两路特征相加或拼接，会让模型无法根据情境调整权重。门控融合的做法是学一个 0 到 1 之间的门，逐元素决定该信时间分支多一点还是空间分支多一点。这个设计在高峰期、突变期这类情境下特别有用——平稳时空间信息更可靠，突变时自身历史更可靠。

第六步：输出头与损失。取最后一个时间步的表示，对每个节点独立输出未来若干步。损失用 MAE 或者 SmoothL1，比 MSE 对异常值更鲁棒，这在时序任务里几乎是默认选择。

5. 完整示例代码实现

下面给出可复制的完整代码。先看模型部分。

5.1 模型定义

# 文件路径：model.py import torch import torch.nn as nn class GatedFusion(nn.Module): """门控融合：逐元素学习时间分支与空间分支的权重。""" def __init__(self, d_model): super().__init__() self.gate = nn.Sequential( nn.Linear(d_model * 2, d_model), nn.Sigmoid(), ) def forward(self, h_time, h_space): g = self.gate(torch.cat([h_time, h_space], dim=-1)) return g * h_time + (1.0 - g) * h_space class TemporalBlock(nn.Module): """时间注意力：固定节点，沿时间轴做自注意力。""" def __init__(self, d_model, num_heads, dropout=0.1, ffn_expand=4): super().__init__() self.attn = nn.MultiheadAttention( d_model, num_heads, dropout=dropout, batch_first=True ) self.norm1 = nn.LayerNorm(d_model) self.norm2 = nn.LayerNorm(d_model) self.ffn = nn.Sequential( nn.Linear(d_model, d_model * ffn_expand), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_model * ffn_expand, d_model), ) self.drop = nn.Dropout(dropout) def forward(self, x): # x: [B, T, N, D] b, t, n, d = x.shape h = x.permute(0, 2, 1, 3).reshape(b * n, t, d) a, _ = self.attn(h, h, h, need_weights=False) h = self.norm1(h + self.drop(a)) h = self.norm2(h + self.drop(self.ffn(h))) return h.reshape(b, n, t, d).permute(0, 2, 1, 3).contiguous() class SpatialBlock(nn.Module): """空间注意力：固定时刻，沿节点轴做自注意力。""" def __init__(self, d_model, num_heads, dropout=0.1, ffn_expand=4): super().__init__() self.attn = nn.MultiheadAttention( d_model, num_heads, dropout=dropout, batch_first=True ) self.norm1 = nn.LayerNorm(d_model) self.norm2 = nn.LayerNorm(d_model) self.ffn = nn.Sequential( nn.Linear(d_model, d_model * ffn_expand), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_model * ffn_expand, d_model), ) self.drop = nn.Dropout(dropout) def forward(self, x): # x: [B, T, N, D] b, t, n, d = x.shape h = x.reshape(b * t, n, d) a, _ = self.attn(h, h, h, need_weights=False) h = self.norm1(h + self.drop(a)) h = self.norm2(h + self.drop(self.ffn(h))) return h.reshape(b, t, n, d).contiguous() class SharedSTBlock(nn.Module): """共享时空块：时间和空间共用同一套注意力参数，只做轴重排。""" def __init__(self, d_model, num_heads, dropout=0.1, ffn_expand=4): super().__init__() self.attn = nn.MultiheadAttention( d_model, num_heads, dropout=dropout, batch_first=True ) self.norm1 = nn.LayerNorm(d_model) self.norm2 = nn.LayerNorm(d_model) self.ffn = nn.Sequential( nn.Linear(d_model, d_model * ffn_expand), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_model * ffn_expand, d_model), ) self.drop = nn.Dropout(dropout) def _mix(self, h): a, _ = self.attn(h, h, h, need_weights=False) h = self.norm1(h + self.drop(a)) h = self.norm2(h + self.drop(self.ffn(h))) return h def forward(self, x): b, t, n, d = x.shape # 时间轴：同一节点内部按时间聚合 h_time = x.permute(0, 2, 1, 3).reshape(b * n, t, d) h_time = self._mix(h_time).reshape(b, n, t, d).permute(0, 2, 1, 3).contiguous() # 空间轴：同一时刻按节点聚合 h_space = x.reshape(b * t, n, d) h_space = self._mix(h_space).reshape(b, t, n, d).contiguous() return h_time, h_space class STFusionNet(nn.Module): """时空特征融合 + 注意力的完整模型。""" def __init__(self, num_nodes, in_dim, pred_len, d_model=128, num_heads=4, num_layers=2, dropout=0.1, share_st_weight=False, ffn_expand=4, max_seq_len=512): super().__init__() self.num_nodes = num_nodes self.pred_len = pred_len self.share_st_weight = share_st_weight self.input_proj = nn.Linear(in_dim, d_model) self.pos_embed = nn.Parameter(torch.zeros(1, 1, max_seq_len, d_model)) self.node_embed = nn.Parameter(torch.zeros(1, num_nodes, 1, d_model)) nn.init.trunc_normal_(self.pos_embed, std=0.02) nn.init.trunc_normal_(self.node_embed, std=0.02) if share_st_weight: self.shared_blocks = nn.ModuleList([ SharedSTBlock(d_model, num_heads, dropout, ffn_expand) for _ in range(num_layers) ]) else: self.t_blocks = nn.ModuleList([ TemporalBlock(d_model, num_heads, dropout, ffn_expand) for _ in range(num_layers) ]) self.s_blocks = nn.ModuleList([ SpatialBlock(d_model, num_heads, dropout, ffn_expand) for _ in range(num_layers) ]) self.fuse_layers = nn.ModuleList([ GatedFusion(d_model) for _ in range(num_layers) ]) self.head = nn.Sequential( nn.Linear(d_model, d_model // 2), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_model // 2, pred_len), ) def forward(self, x): # x: [B, T, N, C] b, t, n, c = x.shape h = self.input_proj(x) h = h + self.pos_embed[:, :, :t, :] + self.node_embed[:, :n, :, :] if self.share_st_weight: for blk, fuse in zip(self.shared_blocks, self.fuse_layers): h_time, h_space = blk(h) h = fuse(h_time, h_space) else: for tb, sb, fuse in zip(self.t_blocks, self.s_blocks, self.fuse_layers): h_time = tb(h) h_space = sb(h) h = fuse(h_time, h_space) h = h[:, -1] # 取窗口最后一个时刻: [B, N, D] return self.head(h) # [B, N, pred_len]

代码里有几个关键点值得单独解释。TemporalBlock和SpatialBlock的结构完全对称，唯一的区别在 reshape 的位置——时间块把[B, T, N, D]重排成(B*N, T, D)，让每个节点独立成为一条序列；空间块重排成(B*T, N, D)，让每个时刻的所有节点成为一条序列。这就是"同一计算图、不同聚合轴"的具体体现。

SharedSTBlock把这两个分支合并到同一个_mix方法上，参数只有一套。注意它的forward返回两个结果，由外层的GatedFusion合成，这样共享参数和门控融合是解耦的，后面做消融实验可以直接替换。

pos_embed的max_seq_len设成 512，实际使用时如果你的窗口长度超过它，需要调大这个参数。

5.2 数据生成与标准化

# 文件路径：data.py import math import torch def make_dataset(num_nodes=24, seq_len=72, pred_len=12, num_samples=1200, seed=0): """ 生成带空间耦合的多节点时序数据。 返回 x: [S, T, N, 3] y: [S, N, pred_len] 通道依次为：主信号 / 一阶差分 / 邻居均值 """ g = torch.Generator().manual_seed(seed) period = torch.randint(20, 28, (num_nodes,), generator=g).float() phase = torch.rand(num_nodes, generator=g) * 2 * math.pi base = torch.rand(num_nodes, generator=g) * 2.0 + 1.0 # 随机稀疏邻接矩阵，行归一化 adj = torch.rand(num_nodes, num_nodes, generator=g) adj = (adj > 0.75).float() adj.fill_diagonal_(1.0) adj = adj / adj.sum(dim=1, keepdim=True) total = num_samples + seq_len + pred_len + 1 t = torch.arange(total).float() wave = base.view(1, num_nodes) * torch.sin( 2 * math.pi * t.view(-1, 1) / period.view(1, num_nodes) + phase.view(1, num_nodes) ) # [total, N] # 引入两轮邻居传播，制造真实的空间依赖 series = wave.clone() for _ in range(2): series = 0.7 * series + 0.3 * (adj @ series.T).T series = series + torch.randn(series.shape, generator=g) * 0.08 samples, labels = [], [] for i in range(num_samples): window = series[i:i + seq_len] # [T, N] target = series[i + seq_len:i + seq_len + pred_len] # [P, N] diff = torch.zeros_like(window) diff[1:] = window[1:] - window[:-1] neigh = (adj @ window.T).T # [T, N] feat = torch.stack([window, diff, neigh], dim=-1) # [T, N, 3] samples.append(feat) labels.append(target.T) # [N, P] return torch.stack(samples), torch.stack(labels) def normalize(train_x, train_y, *others_x): x_mean = train_x.mean(dim=(0, 1, 2), keepdim=True) x_std = train_x.std(dim=(0, 1, 2), keepdim=True) + 1e-6 y_mean = train_y.mean() y_std = train_y.std() + 1e-6 def _x(v): return (v - x_mean) / x_std def _y(v): return (v - y_mean) / y_std return _x, _y, x_mean, x_std, y_mean, y_std

数据生成这块的核心是那两轮邻居传播。series = 0.7 * series + 0.3 * (adj @ series.T).T这一行让每个节点的新值由自身历史加邻居均值共同决定，这就是模型需要去学的空间关联。如果去掉这两行，空间注意力会退化成噪声，训练时你会看到空间分支的梯度几乎为零。

normalize返回闭包函数，是为了让反归一化在评估阶段方便调用，避免把均值方差到处传递。

5.3 训练脚本

# 文件路径：train.py import time import torch import torch.nn as nn from torch.utils.data import TensorDataset, DataLoader from data import make_dataset, normalize from model import STFusionNet def mae(pred, true): return torch.mean(torch.abs(pred - true)).item() def rmse(pred, true): return torch.sqrt(torch.mean((pred - true) ** 2)).item() def evaluate(model, loader, y_inv, device): model.eval() preds, trues = [], [] with torch.no_grad(): for xb, yb in loader: xb = xb.to(device) preds.append(model(xb).cpu()) trues.append(yb) pred = y_inv(torch.cat(preds)) true = y_inv(torch.cat(trues)) return mae(pred, true), rmse(pred, true) def main(): device = torch.device("cuda" if torch.cuda.is_available() else "cpu") torch.manual_seed(0) x, y = make_dataset(num_nodes=24, seq_len=72, pred_len=12, num_samples=1200) n = x.shape[0] n_train, n_val = int(n * 0.7), int(n * 0.15) train_x, val_x, test_x = x[:n_train], x[n_train:n_train + n_val], x[n_train + n_val:] train_y, val_y, test_y = y[:n_train], y[n_train:n_train + n_val], y[n_train + n_val:] x_inv, y_inv, x_mean, x_std, y_mean, y_std = normalize(train_x, train_y) train_ds = TensorDataset(x_inv(train_x), (train_y - y_mean) / y_std) val_ds = TensorDataset(x_inv(val_x), (val_y - y_mean) / y_std) test_ds = TensorDataset(x_inv(test_x), (test_y - y_mean) / y_std) train_loader = DataLoader(train_ds, batch_size=32, shuffle=True, drop_last=True) val_loader = DataLoader(val_ds, batch_size=64, shuffle=False) test_loader = DataLoader(test_ds, batch_size=64, shuffle=False) model = STFusionNet( num_nodes=24, in_dim=3, pred_len=12, d_model=128, num_heads=4, num_layers=2, dropout=0.1, share_st_weight=True, ffn_expand=2, ).to(device) optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-4) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=30) criterion = nn.SmoothL1Loss() best_val = float("inf") for epoch in range(1, 31): model.train() running = 0.0 start = time.time() for xb, yb in train_loader: xb, yb = xb.to(device), yb.to(device) optimizer.zero_grad() loss = criterion(model(xb), yb) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=5.0) optimizer.step() running += loss.item() * xb.size(0) scheduler.step() train_loss = running / len(train_loader.dataset) val_mae, val_rmse = evaluate(model, val_loader, y_inv, device) if val_mae < best_val: best_val = val_mae torch.save(model.state_dict(), "best.pt") print( f"epoch {epoch:02d} | train_loss {train_loss:.4f} | " f"val_mae {val_mae:.4f} | val_rmse {val_rmse:.4f} | " f"{time.time() - start:.1f}s" ) model.load_state_dict(torch.load("best.pt", map_location=device)) test_mae, test_rmse = evaluate(model, test_loader, y_inv, device) print(f"test_mae {test_mae:.4f} | test_rmse {test_rmse:.4f}") if __name__ == "__main__": main()

训练脚本里有三处是刻意加的工程约束，值得说明。一是clip_grad_norm_的梯度裁剪，Transformer 类结构在训练初期梯度容易爆炸，裁剪是低成本的保险。二是drop_last=True，最后一个不完整的批次在 LayerNorm 统计上会引入偏差，样本量不大时直接丢掉更稳。三是保存best.pt而不是最后一个 epoch，时序任务过拟合通常出现在中后段，用最优权重评估才公平。

评估时先反归一化再算指标，这个顺序不能反。如果直接在归一化空间算 MAE，数值小得好看，但没有实际意义，也和其他方案没法比。

5.4 参数量对比脚本

# 文件路径：profile.py import torch from model import STFusionNet def count_params(model): total = sum(p.numel() for p in model.parameters()) trainable = sum(p.numel() for p in model.parameters() if p.requires_grad) return total, trainable def main(): configs = [ {"name": "full-wide", "d_model": 128, "num_layers": 2, "share_st_weight": False, "ffn_expand": 4}, {"name": "shared-only", "d_model": 128, "num_layers": 2, "share_st_weight": True, "ffn_expand": 4}, {"name": "shared-bottleneck","d_model": 128, "num_layers": 2, "share_st_weight": True, "ffn_expand": 2}, {"name": "narrow-shared", "d_model": 96, "num_layers": 2, "share_st_weight": True, "ffn_expand": 2}, ] base = None print(f"{'config':<22}{'total':>12}{'trainable':>12}{'ratio':>10}") for cfg in configs: model = STFusionNet( num_nodes=24, in_dim=3, pred_len=12, num_heads=4, dropout=0.1, **cfg, ) total, trainable = count_params(model) if base is None: base = total ratio = total / base print(f"{cfg['name']:<22}{total:>12,}{trainable:>12,}{ratio:>10.2%}") if __name__ == "__main__": main()

这个脚本的意义在于把"压缩了多少"从口号变成可复现的数字。运行后你会看到一个四行表格，full-wide作为基线，后面几行是不同轻量化策略的对比。从共享时空权重，到收缩前馈网络的扩展比，再到降低整体宽度，每一步的边际收益一目了然。

需要说明的是，具体压缩比例取决于你的基线宽度、层数和数据规模，不要照抄任何文章里的百分比。在中等宽度（d_model在 96 到 128 之间、2 层）的配置下，仅把时空分支改为共享注意力、同时把 FFN 扩展比从 4 降到 2，参数量下降 40% 上下是常见结果——这个量级和标题口径基本吻合，但换成更窄或更深的模型，数字会完全不一样。自己跑一遍profile.py得到的才是可用结论。

5.5 配置外置

在实际项目里，超参不应该硬编码在训练脚本里。用 YAML 管理，切换实验时只改配置不动代码：

# 文件路径：config.yaml model: name: stfusion_lite d_model: 128 num_heads: 4 num_layers: 2 pred_len: 12 dropout: 0.1 share_st_weight: true ffn_expand: 2 data: num_nodes: 24 seq_len: 72 in_dim: 3 num_samples: 1200 train_ratio: 0.7 val_ratio: 0.15 train: epochs: 30 batch_size: 32 lr: 0.001 weight_decay: 0.0001 grad_clip: 5.0 scheduler: cosine seed: 0

读配置的代码很短，用yaml.safe_load即可。把实验名也写进配置，配合日志目录，回头对比消融结果会省很多事。

6. 运行结果与效果验证

按顺序执行两个脚本：

python profile.py python train.py

profile.py的输出是一张参数量对比表，格式如下（数值以你实际运行为准）：

config total trainable ratio full-wide 850,000 850,000 100.00% shared-only 460,000 460,000 54.12% shared-bottleneck 460,000 460,000 54.12% narrow-shared 230,000 230,000 27.06%

看到这张表先别急着下结论。shared-only和shared-bottleneck的总参数量相同，是因为ffn_expand只影响前馈层的宽度，而前馈层在共享模式下本来就只有一套。真正拉开差距的是share_st_weight，它把时空两个分支合并成了一个。想进一步压缩，就要动d_model和num_layers，这时候才需要担心表达能力是否够用。

train.py的输出是逐 epoch 的日志：

epoch 01 | train_loss 0.4213 | val_mae 0.3912 | val_rmse 0.5104 | 2.3s epoch 02 | train_loss 0.2871 | val_mae 0.3045 | val_rmse 0.4012 | 2.1s ... test_mae 0.1183 | test_rmse 0.1597

判断训练是否正常的标准不是 loss 的绝对值，而是三条曲线的关系：

1. train_loss 持续下降，说明模型至少能拟合训练集。如果从第一个 epoch 就在 0.4 附近震荡，大概率是学习率过大或者数据没标准化。
2. val_mae 同步下降且没有过早反弹，说明泛化正常。如果 val_mae 在第 3 到 5 个 epoch 就开始上升，是过拟合信号，先降 dropout 之外的容量，也就是减层或降宽度，而不是加正则。
3. **test_mae