基于离散化与掩码扩散模型的时间序列缺失值插补实战

在时间序列分析的实际项目中，我们常常面临一个棘手的问题：如何处理那些因传感器故障、网络中断或人为遗漏而产生的缺失值？传统的插补方法，如均值填充或线性插值，在处理复杂、非线性的时间序列模式时往往力不从心。近期，结合深度学习，特别是扩散模型（Diffusion Models）的时序数据插补方法，展现出了强大的潜力。然而，一个核心的挑战在于：扩散模型最初是为图像等连续数据设计的，如何将其高效地应用于连续值的时间序列？

本文将深入探讨一种关键技术方案：“离散化连续时间序列以进行基于掩码扩散训练的插补”。我们将从核心概念入手，逐步拆解为何需要离散化、如何实现，并提供一个完整的、可运行的 PyTorch 实战案例，演示如何使用掩码扩散模型（Masked Diffusion）来学习时间序列的分布并完成高质量的缺失值填充。无论你是数据科学新手，还是希望将前沿深度学习应用于时序问题的开发者，都能从本文获得一套从理论到实践的闭环解决方案。

1. 背景与核心概念：为什么需要离散化？

在深入代码之前，我们首先要理解问题的本质和所涉及的核心概念。

1.1 时间序列插补的挑战

时间序列数据是按时间顺序排列的数据点序列，常见于金融、物联网、医疗等领域。数据缺失会破坏序列的连续性，导致下游分析（如预测、分类）性能下降。理想的插补方法不仅要填充缺失值，还应尽可能保持数据的原始分布、时序依赖关系和内在模式。

1.2 扩散模型简介

扩散模型是一类生成模型，其核心思想是通过一个前向过程逐步向数据中添加噪声，直至数据变成纯噪声；然后训练一个神经网络学习反向过程，从噪声中逐步重建出原始数据。它在图像、音频生成上取得了巨大成功。其优势在于能建模复杂的数据分布，生成高质量、多样化的样本。

1.3 连续与离散的鸿沟

标准的扩散模型通常假设数据是连续值（如图像像素值归一化到[-1, 1]）。然而，许多时间序列是连续取值的（如温度、股价）。直接对连续值应用扩散模型会遇到一些问题：

1. 训练不稳定：连续值的噪声尺度需要精细调整。
2. 量化误差：生成的结果是连续值，但实际观测数据可能具有特定的精度或分布。
3. 与分类扩散的衔接：一些先进的扩散模型变体（如掩码扩散）在处理离散标签（如文本）时表现更好，将连续值离散化后，可以借鉴这些思路。

1.4 离散化与掩码扩散训练

• 离散化：将连续的时间序列值映射到有限的、离散的“桶”或“词汇表”中。例如，将温度值划分为“低温”、“中温”、“高温”几个区间，并用整数索引代表。
• 掩码扩散训练：这是一种训练策略，灵感来自自然语言处理中的掩码语言模型（如 BERT）。在训练时，我们随机“掩码”（即遮盖）输入序列的一部分（用特殊标记如[MASK]代替），然后让模型去预测被掩码的原始值。结合扩散模型，我们可以将“添加噪声”的过程视为一种更复杂的“掩码”，模型学习从被噪声破坏或部分掩码的序列中恢复原始序列。

核心思路：先将连续时间序列离散化，将其转化为一个“符号序列”。然后，在这个符号序列上应用掩码扩散训练。模型学习到符号序列的联合分布后，就可以用于插补：将缺失位置视为被掩码的部分，让模型预测出最可能的一组离散符号，最后再将其映射回连续的数值空间。

2. 环境准备与版本说明

本实战将使用 Python 和 PyTorch 框架。请确保你的环境已满足以下要求。

操作系统：Windows 10/11, macOS, 或 Linux (Ubuntu 20.04+) 均可。Python：>= 3.8核心库：

• torch: 深度学习框架
• numpy: 数值计算
• pandas: 数据处理（用于示例数据加载）
• scikit-learn: 用于数据预处理（如归一化）
• matplotlib: 结果可视化（可选）

版本建议：

# 使用 conda 或 venv 创建虚拟环境后，安装以下包 pip install torch==2.0.1 numpy==1.24.3 pandas==2.0.3 scikit-learn==1.3.0 matplotlib==3.7.2

注意：本文重点在于演示方法流程，代码设计上力求清晰。实际项目中，你可能需要根据 PyTorch 和 CUDA 版本进行调整。模型结构、离散化桶的数量等均为超参数，需根据具体数据集调优。

项目结构预览：

time_series_imputation/ ├── data/ │ └── sample.csv # 示例时间序列数据 ├── utils/ │ ├── discretizer.py # 离散化工具类 │ └── dataset.py # 自定义 Dataset ├── models/ │ └── masked_diffusion.py # 掩码扩散模型定义 ├── train.py # 训练脚本 ├── impute.py # 插补脚本 └── config.yaml # 配置文件

3. 核心原理与步骤拆解

我们将整个流程拆解为四个关键步骤：数据预处理与离散化、模型架构设计、掩码扩散训练、以及插补推理。

3.1 数据预处理与离散化

这是将连续信号转化为模型可处理格式的第一步。

1. 归一化：将原始时间序列归一化到 [0, 1] 区间，消除量纲影响。

   from sklearn.preprocessing import MinMaxScaler scaler = MinMaxScaler(feature_range=(0, 1)) normalized_series = scaler.fit_transform(original_series.reshape(-1, 1)).flatten()

2. 离散化：将归一化后的连续值分桶。
   • 等宽分桶：将 [0, 1] 区间均匀划分为 N 个桶。
   • 等频分桶：根据数据分布划分，使每个桶内数据点数量大致相等。
   • 聚类分桶：使用 K-Means 等算法聚类。 本文采用等宽分桶进行演示，因其简单且可复现。

   import numpy as np def discretize(continuous_array, n_bins=100): """将连续数组离散化为 [0, n_bins-1] 的整数索引。""" # 等宽分桶边界 bins = np.linspace(0, 1, n_bins+1) # digitize 返回桶索引（从1开始），我们调整为从0开始 discrete_indices = np.digitize(continuous_array, bins) - 1 # 处理边界情况：最大值归于最后一个桶 discrete_indices = np.clip(discrete_indices, 0, n_bins-1) return discrete_indices

   离散化后，一个时间序列变成了一个由整数（如 0,1,2,...,99）组成的序列，类似于一个句子中的单词索引。

3.2 掩码扩散模型架构

我们需要一个能够处理序列数据的模型。Transformer 编码器因其强大的序列建模能力成为自然选择。

模型工作流程：

1. 输入嵌入：将离散的索引通过一个嵌入层（Embedding Layer）转换为稠密向量。
2. 位置编码：加入位置信息，让模型感知时间顺序。
3. Transformer 编码器：多层 Transformer 块，用于捕获序列中复杂的依赖关系。
4. 输出层：一个线性层，将编码后的向量映射到每个“桶”的 logits（未归一化的概率），其维度为[序列长度, 桶数量]。

扩散/掩码过程：

• 在训练时，我们模拟扩散模型的前向过程。对于输入序列x，我们随机选择一定比例的时间步进行“掩码/破坏”。
• 破坏方式可以是：
  • 随机替换：用随机的一个其他桶索引替换。
  • 特殊掩码标记：用一个预定义的、超出正常桶范围的索引（如n_bins）替换。
  • 添加噪声：更接近标准扩散，对嵌入向量添加高斯噪声，但本文为简化，采用掩码标记策略。
• 模型的目标是，给定被部分破坏的序列，预测出原始序列每个位置上的桶索引分布。

3.3 训练目标

训练目标类似于分类任务。对于序列中的每个位置，模型输出一个大小为n_bins的 logits 向量。我们使用标准的交叉熵损失（CrossEntropyLoss），但只计算那些在扩散前向过程中被“破坏”或“掩码”的位置的损失。未被破坏的位置，模型只需学会复制输入，这部分损失可以忽略或降低权重。

3.4 插补推理

当模型训练好后，我们可以用它来填充缺失值：

1. 将待插补的序列中，缺失值位置用特殊的[MASK]索引填充。
2. 将整个序列输入模型。
3. 模型会输出序列每个位置的概率分布。
4. 对于缺失位置，我们取概率最高的那个桶索引作为预测结果。
5. 最后，将预测出的桶索引通过离散化的逆过程，映射回归一化的连续值，再通过归一化的逆变换得到最终的插补值。

4. 完整实战案例

下面我们构建一个完整的、可运行的示例。我们将使用一个合成的正弦波加噪声的时间序列数据。

4.1 创建项目结构与生成数据

首先，创建项目目录和示例数据。

# 文件：create_sample_data.py import numpy as np import pandas as pd import os # 创建目录 os.makedirs('data', exist_ok=True) # 生成合成时间序列：正弦波 + 噪声 np.random.seed(42) time_steps = 1000 t = np.linspace(0, 20, time_steps) # 基础信号 signal = np.sin(t) + 0.1 * np.sin(5*t) # 添加噪声 noise = np.random.normal(0, 0.1, time_steps) series = signal + noise # 模拟缺失：随机将20%的数据点设为NaN mask = np.random.rand(time_steps) < 0.2 series_with_missing = series.copy() series_with_missing[mask] = np.nan # 保存数据 df = pd.DataFrame({ 'timestamp': np.arange(time_steps), 'value_original': series, 'value_with_missing': series_with_missing }) df.to_csv('data/sample.csv', index=False) print("示例数据已保存至 data/sample.csv") print(f"原始序列长度：{len(series)}， 缺失值数量：{pd.isna(series_with_missing).sum()}")

4.2 实现离散化工具类

我们将离散化和逆离散化封装成一个类。

# 文件：utils/discretizer.py import numpy as np from sklearn.preprocessing import MinMaxScaler class TimeSeriesDiscretizer: def __init__(self, n_bins=100, method='uniform'): """ 初始化离散化器。 Args: n_bins: 离散桶的数量。 method: 离散化方法，'uniform'（等宽）或 'quantile'（等频）。 """ self.n_bins = n_bins self.method = method self.scaler = MinMaxScaler(feature_range=(0, 1)) self.bin_edges_ = None # 用于等宽 self.quantiles_ = None # 用于等频 self.fitted = False def fit(self, continuous_series): """拟合数据，计算归一化参数和分桶边界。""" # 1. 归一化 series_2d = continuous_series.reshape(-1, 1) self.scaler.fit(series_2d) normalized = self.scaler.transform(series_2d).flatten() # 2. 计算分桶边界 if self.method == 'uniform': self.bin_edges_ = np.linspace(0, 1, self.n_bins + 1) elif self.method == 'quantile': # 计算分位数点 self.quantiles_ = np.percentile(normalized, np.linspace(0, 100, self.n_bins + 1)) else: raise ValueError(f"Unsupported method: {self.method}") self.fitted = True return self def transform(self, continuous_series): """将连续序列离散化为索引。""" if not self.fitted: raise RuntimeError("Discretizer must be fitted before transform.") # 归一化 series_2d = continuous_series.reshape(-1, 1) normalized = self.scaler.transform(series_2d).flatten() # 离散化 if self.method == 'uniform': indices = np.digitize(normalized, self.bin_edges_) - 1 else: # quantile indices = np.digitize(normalized, self.quantiles_) - 1 # 确保索引在有效范围内 [0, n_bins-1] indices = np.clip(indices, 0, self.n_bins - 1) return indices.astype(np.int64) def inverse_transform(self, discrete_indices): """将离散索引转换回连续的归一化值（取桶中心）。""" if not self.fitted: raise RuntimeError("Discretizer must be fitted before inverse_transform.") if self.method == 'uniform': bin_centers = (self.bin_edges_[:-1] + self.bin_edges_[1:]) / 2 else: # 对于等频分桶，我们用分位数的中点作为桶的代表值 bin_centers = (self.quantiles_[:-1] + self.quantiles_[1:]) / 2 # 根据索引选择桶中心值 continuous_normalized = bin_centers[discrete_indices] # 逆归一化 series_2d = continuous_normalized.reshape(-1, 1) original_scale = self.scaler.inverse_transform(series_2d).flatten() return original_scale @property def mask_token_id(self): """返回掩码标记的ID，我们约定为 n_bins。""" return self.n_bins

4.3 实现数据集类

我们需要一个 PyTorch Dataset 来加载数据，并生成训练时所需的“被破坏”的序列。

# 文件：utils/dataset.py import torch from torch.utils.data import Dataset import numpy as np class MaskedTimeSeriesDataset(Dataset): def __init__(self, discrete_sequences, seq_len, mask_ratio=0.15): """ Args: discrete_sequences: 列表，每个元素是一个离散化后的序列（1D numpy数组）。 seq_len: 模型输入的固定序列长度。 mask_ratio: 训练时随机掩码的比例。 """ self.sequences = discrete_sequences self.seq_len = seq_len self.mask_ratio = mask_ratio # 计算所有序列的总长度，用于生成样本索引 self.valid_starts = [] for seq in self.sequences: if len(seq) >= self.seq_len: # 可以从此序列的多个起始点截取 self.valid_starts.extend([(seq_idx, start) for start in range(len(seq) - self.seq_len + 1)]) self.num_samples = len(self.valid_starts) def __len__(self): return self.num_samples def __getitem__(self, idx): seq_idx, start = self.valid_starts[idx] seq = self.sequences[seq_idx] # 截取固定长度片段 segment = seq[start: start + self.seq_len] # 创建掩码：随机选择一部分位置进行掩码 mask = torch.rand(self.seq_len) < self.mask_ratio # 创建被破坏的输入：将掩码位置替换为 mask_token_id corrupted_segment = segment.copy() # 假设 mask_token_id 是 n_bins，通过外部传入或全局定义，这里先写死逻辑 # 实际使用时，mask_token_id 应从 discretizer 获取 mask_token_id = 100 # 示例，应与 discretizer.n_bins 一致 corrupted_segment[mask.numpy()] = mask_token_id # 转换为张量 segment_tensor = torch.from_numpy(segment).long() corrupted_tensor = torch.from_numpy(corrupted_segment).long() mask_tensor = mask.long() # 1 表示被掩码，0 表示未掩码 return { 'corrupted': corrupted_tensor, # 模型输入：被部分掩码的序列 'target': segment_tensor, # 训练目标：原始完整序列 'mask': mask_tensor # 掩码位置指示器 }

4.4 实现掩码扩散模型

这里我们实现一个基于 Transformer 编码器的简单模型。

# 文件：models/masked_diffusion.py import torch import torch.nn as nn import math class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=5000): super().__init__() pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) pe = pe.unsqueeze(0) # (1, max_len, d_model) self.register_buffer('pe', pe) def forward(self, x): # x: (batch_size, seq_len, d_model) return x + self.pe[:, :x.size(1), :] class MaskedDiffusionModel(nn.Module): def __init__(self, vocab_size, d_model=128, nhead=4, num_layers=3, seq_len=100): """ Args: vocab_size: 离散词汇表大小，即 n_bins + 1（+1 用于掩码标记）。 d_model: 嵌入维度。 nhead: Transformer 多头注意力头数。 num_layers: Transformer 编码器层数。 seq_len: 输入序列长度。 """ super().__init__() self.vocab_size = vocab_size self.embedding = nn.Embedding(vocab_size, d_model) self.pos_encoder = PositionalEncoding(d_model, max_len=seq_len) encoder_layer = nn.TransformerEncoderLayer(d_model=d_model, nhead=nhead, batch_first=True) self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers) self.output_layer = nn.Linear(d_model, vocab_size) def forward(self, x): # x: (batch_size, seq_len) 包含普通token和mask token的索引 # 1. 嵌入 x_emb = self.embedding(x) # (batch_size, seq_len, d_model) # 2. 位置编码 x_emb = self.pos_encoder(x_emb) # 3. Transformer 编码 # 注意：标准 Transformer 需要 src_key_padding_mask 来处理变长，这里我们固定长度，暂不需要。 encoded = self.transformer_encoder(x_emb) # 4. 输出 logits logits = self.output_layer(encoded) # (batch_size, seq_len, vocab_size) return logits

4.5 编写训练脚本

现在，我们将所有组件串联起来进行训练。

# 文件：train.py import torch import torch.nn as nn from torch.utils.data import DataLoader import numpy as np import yaml from utils.discretizer import TimeSeriesDiscretizer from utils.dataset import MaskedTimeSeriesDataset from models.masked_diffusion import MaskedDiffusionModel def load_config(config_path='config.yaml'): with open(config_path, 'r') as f: config = yaml.safe_load(f) return config def prepare_data(config): """加载数据，拟合离散化器，并创建数据集。""" # 1. 加载原始数据（这里用合成数据示例） # 假设我们有一个完整的、无缺失的序列用于训练 t = np.linspace(0, 20, 10000) train_series = np.sin(t) + 0.1 * np.sin(5*t) + np.random.normal(0, 0.1, 10000) # 2. 离散化 discretizer = TimeSeriesDiscretizer( n_bins=config['data']['n_bins'], method=config['data']['discretize_method'] ) discretizer.fit(train_series) discrete_seq = discretizer.transform(train_series) # 3. 创建数据集（将长序列分割成多个样本） # 这里简单地将长序列切成多个固定长度的子序列 seq_len = config['model']['seq_len'] sub_sequences = [] for i in range(0, len(discrete_seq) - seq_len + 1, seq_len // 2): # 使用滑动窗口，步长为一半长度 sub_sequences.append(discrete_seq[i:i+seq_len]) print(f"Created {len(sub_sequences)} training subsequences.") dataset = MaskedTimeSeriesDataset( discrete_sequences=sub_sequences, seq_len=seq_len, mask_ratio=config['training']['mask_ratio'] ) # 注意：需要将 mask_token_id 传递给 dataset，这里简化处理，在 dataset 内部使用与 discretizer 一致的 n_bins # 更严谨的做法是将 discretizer 传入 dataset。 return dataset, discretizer def train(): config = load_config() device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') print(f"Using device: {device}") # 准备数据 dataset, discretizer = prepare_data(config) dataloader = DataLoader(dataset, batch_size=config['training']['batch_size'], shuffle=True) # 初始化模型 vocab_size = discretizer.n_bins + 1 # +1 for mask token model = MaskedDiffusionModel( vocab_size=vocab_size, d_model=config['model']['d_model'], nhead=config['model']['nhead'], num_layers=config['model']['num_layers'], seq_len=config['model']['seq_len'] ).to(device) # 定义损失函数和优化器 criterion = nn.CrossEntropyLoss(reduction='none') # 先对每个位置计算损失 optimizer = torch.optim.Adam(model.parameters(), lr=config['training']['lr']) # 训练循环 num_epochs = config['training']['num_epochs'] for epoch in range(num_epochs): model.train() total_loss = 0 for batch in dataloader: corrupted = batch['corrupted'].to(device) target = batch['target'].to(device) mask = batch['mask'].to(device) optimizer.zero_grad() logits = model(corrupted) # (batch, seq_len, vocab_size) # 计算损失，但只考虑被掩码的位置 loss_per_token = criterion(logits.view(-1, vocab_size), target.view(-1)) loss_per_token = loss_per_token.view(target.shape) # 恢复形状 (batch, seq_len) # 只对 mask==1 的位置求平均损失 loss = (loss_per_token * mask).sum() / (mask.sum() + 1e-8) loss.backward() optimizer.step() total_loss += loss.item() * corrupted.size(0) avg_loss = total_loss / len(dataset) if (epoch + 1) % 10 == 0: print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {avg_loss:.4f}') # 保存模型和离散化器 torch.save(model.state_dict(), 'checkpoints/masked_diffusion_model.pth') import pickle with open('checkpoints/discretizer.pkl', 'wb') as f: pickle.dump(discretizer, f) print("Training finished. Model and discretizer saved.") if __name__ == '__main__': train()

4.6 配置文件

# 文件：config.yaml data: n_bins: 100 discretize_method: 'uniform' # 'uniform' or 'quantile' model: seq_len: 100 d_model: 128 nhead: 4 num_layers: 3 training: batch_size: 32 mask_ratio: 0.15 lr: 0.001 num_epochs: 100

4.7 编写插补推理脚本

模型训练好后，我们可以用它来对真实缺失数据进行插补。

# 文件：impute.py import torch import numpy as np import pandas as pd import pickle from models.masked_diffusion import MaskedDiffusionModel from utils.discretizer import TimeSeriesDiscretizer def impute_missing_values(raw_series_with_nan, model_path, discretizer_path, config): """ 对含有NaN的原始序列进行插补。 Args: raw_series_with_nan: 一维numpy数组，包含NaN值。 model_path: 训练好的模型权重路径。 discretizer_path: 保存的离散化器路径。 config: 配置字典。 Returns: imputed_series: 插补完成的连续序列。 discrete_predictions: 预测的离散索引（用于调试）。 """ device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') # 1. 加载离散化器 with open(discretizer_path, 'rb') as f: discretizer = pickle.load(f) # 2. 加载模型 vocab_size = discretizer.n_bins + 1 model = MaskedDiffusionModel( vocab_size=vocab_size, d_model=config['model']['d_model'], nhead=config['model']['nhead'], num_layers=config['model']['num_layers'], seq_len=config['model']['seq_len'] ).to(device) model.load_state_dict(torch.load(model_path, map_location=device)) model.eval() # 3. 预处理输入序列 # 我们需要一个完整的序列（包括已知值和NaN）进行归一化，但归一化器不能处理NaN。 # 策略：用已知值的均值和标准差进行近似归一化，或简单用0填充NaN进行归一化再替换。 # 这里采用一个简单方法：用已知值的线性插值临时填充NaN，用于归一化。 from scipy import interpolate known_mask = ~np.isnan(raw_series_with_nan) known_indices = np.where(known_mask)[0] known_values = raw_series_with_nan[known_mask] if len(known_indices) < 2: raise ValueError("Not enough known points to perform interpolation.") # 线性插值填充所有位置（包括已知点，为了得到连续序列） interp_func = interpolate.interp1d(known_indices, known_values, kind='linear', fill_value='extrapolate') filled_for_norm = interp_func(np.arange(len(raw_series_with_nan))) # 用填充后的序列进行归一化和离散化 continuous_normalized = discretizer.scaler.transform(filled_for_norm.reshape(-1, 1)).flatten() discrete_full = discretizer.transform(filled_for_norm) # 这只是初始离散序列 # 4. 创建掩码输入：将原始NaN对应的位置标记为MASK mask_token_id = discretizer.mask_token_id discrete_input = discrete_full.copy() discrete_input[~known_mask] = mask_token_id # 5. 模型推理（需要按固定长度滑动窗口处理长序列） seq_len = config['model']['seq_len'] all_predicted_indices = discrete_full.copy() # 初始化，用初始离散值填充已知部分 with torch.no_grad(): # 滑动窗口处理 for start in range(0, len(discrete_input) - seq_len + 1, seq_len // 2): # 重叠滑动 end = start + seq_len segment = discrete_input[start:end] segment_tensor = torch.from_numpy(segment).long().unsqueeze(0).to(device) # (1, seq_len) logits = model(segment_tensor) # (1, seq_len, vocab_size) predictions = torch.argmax(logits, dim=-1).squeeze(0).cpu().numpy() # (seq_len,) # 只更新窗口中原本是MASK的位置的预测结果 window_original_mask = ~known_mask[start:end] all_predicted_indices[start:end][window_original_mask] = predictions[window_original_mask] # 6. 将预测的离散索引转换回连续值 imputed_continuous = discretizer.inverse_transform(all_predicted_indices) return imputed_continuous, all_predicted_indices if __name__ == '__main__': # 加载配置 import yaml with open('config.yaml', 'r') as f: config = yaml.safe_load(f) # 加载有缺失的测试数据 df_test = pd.read_csv('data/sample.csv') series_with_missing = df_test['value_with_missing'].to_numpy() # 执行插补 imputed_series, _ = impute_missing_values( raw_series_with_nan=series_with_missing, model_path='checkpoints/masked_diffusion_model.pth', discretizer_path='checkpoints/discretizer.pkl', config=config ) # 保存结果 df_test['value_imputed'] = imputed_series df_test.to_csv('data/result_imputed.csv', index=False) print("插补完成，结果已保存至 data/result_imputed.csv") # 简单评估：计算在缺失位置上的误差（如果有真实值） original_series = df_test['value_original'].to_numpy() missing_mask = np.isnan(series_with_missing) if missing_mask.any(): mae = np.mean(np.abs(original_series[missing_mask] - imputed_series[missing_mask])) print(f"在缺失位置上的平均绝对误差(MAE)为: {mae:.4f}")

4.8 运行与验证

1. 生成数据：运行python create_sample_data.py。
2. 训练模型：运行python train.py。确保已创建checkpoints目录。
3. 执行插补：运行python impute.py。
4. 可视化结果（可选）：

   import matplotlib.pyplot as plt import pandas as pd df = pd.read_csv('data/result_imputed.csv') plt.figure(figsize=(12, 4)) plt.plot(df['timestamp'], df['value_original'], label='Original', alpha=0.7) plt.plot(df['timestamp'], df['value_with_missing'], 'o', label='Observed (with NaN)', markersize=2, alpha=0.5) plt.plot(df['timestamp'], df['value_imputed'], '--', label='Imputed (Diffusion)', linewidth=1.5) plt.xlabel('Time Step') plt.ylabel('Value') plt.legend() plt.title('Time Series Imputation with Masked Diffusion') plt.tight_layout() plt.savefig('imputation_result.png', dpi=150) plt.show()

   你应该能看到，模型生成的插补值（虚线）能够较好地拟合原始信号（实线）的趋势，即使在缺失段。

5. 常见问题与排查思路

在实际应用上述流程时，你可能会遇到以下问题：

6. 最佳实践与工程建议

将离散化掩码扩散模型应用于生产环境的时间序列插补，需要考虑以下工程细节：

1. 数据预处理是关键：

   • 异常值处理：在离散化前，必须处理异常值。极端值会扭曲归一化区间，导致大部分数据被压缩在少数几个桶内。可以使用缩尾处理或基于标准差的方法。
   • 缺失模式模拟：训练数据应尽可能模拟真实场景的缺失模式。除了随机点缺失，还应加入连续块缺失（模拟传感器长时间离线）进行训练，使模型更鲁棒。

2. 离散化策略选择：

   • 等宽分桶：简单快速，适用于分布相对均匀的数据。
   • 等频分桶：使每个桶内数据点数量相等，能更好地处理长尾分布，但对新数据的泛化可能稍弱。
   • 模型化分桶：对于特别复杂的数据，可以考虑使用 VQ-VAE 等模型学习离散表示，但这会引入额外的训练复杂度。

3. 模型架构优化：

   • 位置编码：对于时间序列，绝对位置编码可能不够。可以尝试相对位置编码或添加可学习的位置嵌入。
   • 注意力机制：标准 Transformer 的自注意力复杂度是序列长度的平方。对于超长序列，可以考虑使用稀疏注意力、线性注意力或 Informer 等专门为时序设计的模型。
   • 条件生成：在插补时，可以利用缺失点周围的已知值作为强条件。确保模型架构能充分接收到这些已知信息的上下文。

4. 训练技巧：

   • 课程学习：开始时使用较低的掩码比例，随着训练进行逐渐增加，帮助模型先学习简单模式。
   • 损失函数加权：对于连续缺失的块，可以给予更高的损失权重，因为填补连续空缺比填补孤立点更难。
   • 验证集：务必使用一个独立的验证集来监控模型性能，防止过拟合到训练序列的特定模式上。

5. 部署与性能：

   • 模型量化：训练完成后，可以考虑对模型进行量化（如使用 PyTorch Quantization），以降低推理时的内存占用和延迟，便于边缘部署。
   • 缓存：对于固定的、周期性重复的查询模式，可以将插补结果缓存起来，避免重复计算。
   • 不确定性估计：扩散模型的一个优势是可以提供生成的不确定性。你可以从模型的输出分布中采样多次，得到多个可能的插补值，通过其方差来估计该位置插补结果的可信度。

6. 领域知识融合：

   • 纯粹的数据驱动方法可能违反物理或业务约束。例如，温度值不会瞬间跳变，库存数量不能为负。在将模型预测的离散索引映射回连续值后，可以加入后处理规则，对结果进行平滑或约束，使其符合领域常识。

离散化结合掩码扩散训练，为时间序列插补提供了一个强大而灵活的范式。它巧妙地将连续回归问题转化为离散分类问题，利用了 Transformer 在序列建模和扩散模型在分布学习上的优势。通过本文提供的完整代码框架，你可以快速上手，并将其适配到自己的时序数据问题上。记住，没有放之四海而皆准的模型，成功的应用离不开对业务数据的深入理解、耐心的调参和严谨的评估。