最近在做多元时间序列异常检测时接触到了一类很巧妙的思路把时间序列转换成图像再用 CV 领域成熟的网络结构去做检测。这个方向里PRISM 提出的 TS2ITime Series to Image表征方法很值得深入拆解一轮。本文将围绕 TS2I 的核心思想、PRISM 的架构设计、完整复现代码和工程落地建议展开适合正在研究时序异常检测、想尝试多模态思路的算法工程师和研究生阅读。1. 背景与核心概念1.1 为什么要把时间序列转成图像传统时间序列异常检测通常依赖一维卷积、LSTM、Transformer 等序列模型。这类模型的共同点是直接在原始时间序列上建模时序依赖关系。但真实场景中的多元时序数据往往存在复杂的通道间相关性例如机房监控中 CPU 温度与风扇转速联动、长短周期叠加例如流量数据既有每日周期又有每周趋势和局部形态突变。TS2I 的核心思路是改变模态。它把一段固定窗口的多元时间序列编码成一张二维图像比如通过格拉姆角场GAF、马尔可夫转移场MTF、递归图Recurrence Plot或频谱图Spectrogram把时间轴上的位置关系、数值间的状态转移关系、周期性特征映射为图像的纹理、颜色和结构。这样一来原本需要设计复杂时序模型的检测任务可以复用 ResNet、CNN 自编码器等在图像领域被验证过的结构。这种模态转换的好处很直接图像卷积能天然提取局部邻域特征对应时序中的短期依赖。多通道图像天然适配多元变量关系建模。图像增强、预训练模型等 CV 生态可以被迁移过来。1.2 PRISM 是什么PRISMPowerful Time Series to Image Representations for Multivariate Anomaly Detection是一个面向多元时间序列异常检测的表征学习框架。它强调“强大”的 TS2I 表示意味着不只是简单把时序数据画成图而是通过精心设计的转换策略、归一化方式和下游模型结构让图像表示能够保留异常判别所需的关键信息。在 PRISM 的流程中通常分为三个模块时序到图像的转换模块负责把滑动窗口内的多元时间序列生成图像样本。编码器模块一般使用卷积神经网络如 ResNet 变体从图像中提取语义向量。异常判别模块基于编码向量做重构误差或分类判断输出异常分数。1.3 应用场景这套思路可以直接用于以下场景服务器指标监控CPU、内存、网络流量等多维指标的异常检测。工业设备预测性维护振动传感器、温度、电流信号的异常模式识别。金融交易时序风控交易频次、金额、渠道等多维时间序列中的异常突发。运维可观测性APM 链路指标、日志时序、业务监控指标。如果你当前正在做基于时序数据的运维监控、设备故障诊断或业务风控TS2I 方向值得纳入候选方案。2. 环境准备与版本说明本文的代码示例基于以下环境操作系统Ubuntu 20.04 / macOS 均可Windows 下建议使用 WSL2Python3.9 或 3.10PyTorch1.13 或 2.x示例以 2.x 为默认接口基本兼容torchvision0.14 或对应版本numpy、pandas、scikit-learn、matplotlibtqdm用于训练进度显示如果使用 GPU 训练建议安装对应 CUDA 版本的 PyTorch。如果没有 GPU小规模实验用 CPU 也可以完成只是训练速度会慢一些。pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118 pip install numpy pandas scikit-learn matplotlib tqdm注意版本需要根据你的项目和机器调整。本文重点是给出思路和可运行的框架不是锁定某个精确版本。3. 核心原理拆解从时间序列到图像3.1 格拉姆角场GAF格拉姆角场是把一维时间序列转换为二维图像的一种经典方法。基本步骤将时间序列 X 归一化到 [0, 1] 区间常规做法是 Min-Max。把归一化后的数值转换为极坐标角度角度 arccos(归一化值)。计算每个时间点对之间的三角和/差生成格拉姆矩阵。对应矩阵元素GASF 和 GADFGASF(i, j) cos(theta_i theta_j) GADF(i, j) sin(theta_i - theta_j)其中 theta_i 是第 i 个时间点的极坐标角度。GAF 图像的主要特点是时间轴上的自相关性被编码为图像的局部像素关系数值大小被编码为颜色深浅。异常点会表现为图像中局部纹理的突变。3.2 马尔可夫转移场MTFMTF 的核心思想是构建马尔可夫链的状态转移概率矩阵并把时间的先后顺序映射到图像的坐标轴上。步骤把时间序列数值离散化为 Q 个分位数区间桶。统计相邻两个时间点的状态转移频率构建一个 Q×Q 的转移矩阵 WW(i, j) 表示从区间 i 转移到区间 j 的概率。将转移矩阵按时间顺序展开成 MTF 矩阵。MTF 对时序中的状态持久性和突变比较敏感适合检测值在不同状态区间之间频繁跳变的异常场景。3.3 递归图Recurrence Plot递归图用来表示时间序列在相空间中的递归行为。对两个时刻的状态向量计算距离如欧氏距离如果距离小于阈值则标记为递归发生。R(i, j) 1 if ||x_i - x_j|| epsilon else 0递归图能展现时序的周期性和非平稳性。周期性强的序列会呈现出明显的对角线纹理而异常突变会破坏这种纹理结构。3.4 频谱图Spectrogram频谱图利用短时傅里叶变换STFT把一维信号变成时间和频率的二维表示。对非平稳信号频谱图能清晰展示频率成分随时间变化的过程。频率上的异常成分会被编码为图像中的突出区域。from scipy.signal import spectrogram f, t, Sxx spectrogram(x, fs1.0)3.5 PRISM 为什么需要“更强的表示”直接给图像可能不够PRISM 强调的是通过可学习的转换或联合训练让 TS2I 表示更适配下游异常检测任务。常见的增强手段包括对每个变量的取值范围进行稳健归一化避免离群点主导图像对比度。多通道拼接比如把 GAF、MTF、原始归一化波形分别放在不同的通道中。使用小波变换替代短时傅里叶变换获得更好的时频分辨率。对图像做随机裁剪、遮挡等数据增强提高编码器的泛化能力。4. 完整实战案例基于 PRISM 思路的多元时序异常检测下面我们动手构建一个完整的异常检测流程。为便于演示这里使用公开的多元时间序列数据集作为测试样本用 PyTorch 搭建一个基于图像的卷积自编码器通过重构误差进行异常评分。4.1 创建项目结构prism_ts2i_demo/ ├── data/ │ └── sample_data.csv ├── src/ │ ├── __init__.py │ ├── ts2i.py │ ├── dataset.py │ ├── model.py │ ├── train.py │ └── detect.py └── config.py说明src 目录用于存放核心代码data 目录存放数据config.py 集中管理超参数。4.2 准备数据为了快速验证流程我们生成一段合成的多元时间序列其中包含正态波动和几段人为注入的异常。# 文件路径src/data_utils.py import numpy as np import pandas as pd def generate_synthetic_data(n_timesteps10000, n_features8, anomaly_ratio0.05, seed42): np.random.seed(seed) t np.arange(n_timesteps) # 正常信号包含趋势项、周期项和噪声 data np.zeros((n_timesteps, n_features)) for i in range(n_features): freq 0.01 * (i 1) data[:, i] ( 0.5 * np.sin(2 * np.pi * freq * t) 0.2 * np.cos(2 * np.pi * freq * t * 0.37) 0.05 * t / n_timesteps 0.05 * np.random.randn(n_timesteps) ) # 注入异常随机挑几个区间让某个通道突然偏离 n_anomaly int(n_timesteps * anomaly_ratio) anomaly_indices np.random.choice(n_timesteps, n_anomaly, replaceFalse) for idx in anomaly_indices: channel np.random.randint(0, n_features) data[idx, channel] np.random.choice([-4, 4]) * np.random.rand() df pd.DataFrame(data, columns[fchan_{i} for i in range(n_features)]) df[is_anomaly] 0 df.loc[anomaly_indices, is_anomaly] 1 return df if __name__ __main__: df generate_synthetic_data() df.to_csv(data/sample_data.csv, indexFalse) print(数据生成完成形状, df.shape) print(异常占比, df[is_anomaly].mean())4.3 实现 TS2I 转换我们使用 GAF 作为默认图像表示同时把原始归一化波形作为第二通道形成双通道图像。窗口长度设置为 64。# 文件路径src/ts2i.py import numpy as np import torch def min_max_scale_series(series): min_val np.min(series) max_val np.max(series) if max_val - min_val 1e-8: return np.zeros_like(series) return (series - min_val) / (max_val - min_val) def gaf_transform(series): 将一维时间序列转换为 GAF 图像。 输入: series (N,) 输出: gaf_image (N, N) scaled min_max_scale_series(series) # 转换为极坐标 # 这里使用 arccos角度范围 [0, pi] angles np.arccos(scaled) # 格拉姆角场GASF angle_mat angles[:, None] angles[None, :] gaf np.cos(angle_mat) return gaf def ts2i_multi_channel(data_window): 将一个窗口的多元时间序列转换为多通道图像。 输入: data_window of shape (window_size, num_features) 输出: image of shape (num_features * 2, window_size, window_size) window_size, num_features data_window.shape channels [] # 第一个特征组每个变量一张 GAF 图 for f in range(num_features): gaf gaf_transform(data_window[:, f]) channels.append(gaf) # 第二个特征组把归一化后的原始序列广播成图像行 for f in range(num_features): scaled min_max_scale_series(data_window[:, f]) broadcast_img np.tile(scaled[:, None], (1, window_size)) channels.append(broadcast_img) image np.stack(channels, axis0) # (num_features*2, H, W) return image需要说明的是上面的实现把每个通道的图像独立生成也可以尝试把变量之间的交叉关系生成图像例如将通道 i 和通道 j 的 GAF 拼接生成联合图像。这里为了演示清晰保留最简单的结构。4.4 构造 Dataset 和数据加载# 文件路径src/dataset.py import numpy as np import torch from torch.utils.data import Dataset class TimeSeriesWindowDataset(Dataset): def __init__(self, data, labels, window_size, transform_functs2i_multi_channel): data: shape (num_timesteps, num_features) labels: shape (num_timesteps,)1 表示异常 window_size: 滑动窗口长度 transform_func: 自定义 TS2I 转换函数 self.data data self.labels labels self.window_size window_size self.transform_func transform_func # 预计算所有窗口 self.windows [] self.window_labels [] n len(data) for start in range(0, n - window_size 1): end start window_size window_data data[start:end] window_label int(np.max(labels[start:end])) # 窗口内存在异常则标记为异常窗口 self.windows.append(window_data) self.window_labels.append(window_label) def __len__(self): return len(self.windows) def __getitem__(self, idx): window_data self.windows[idx] label self.window_labels[idx] # 转换为图像 image self.transform_func(window_data) # 转为 torch tensor增加归一化 image_tensor torch.tensor(image, dtypetorch.float32) # 简单缩放到 [0,1] 范围附近 image_tensor (image_tensor 1) / 2 return image_tensor, torch.tensor(label, dtypetorch.long)这里有一个需要重视的细节滑动窗口划分时窗口内只要包含一个异常点就标记为异常窗口。这在业务上解释为我们检测的是“这段时间区间内出现了异常”而不是某个瞬时点。这种思路更匹配实际告警场景。4.5 构建卷积自编码器模型我们使用一个轻量级卷积自编码器。编码器将图像压缩成低维向量解码器再从向量恢复图像异常窗口的重构误差会显著高于正常窗口。# 文件路径src/model.py import torch import torch.nn as nn class ConvAutoencoder(nn.Module): def __init__(self, in_channels16, latent_dim128): in_channels: 输入图像的通道数等于 num_features * 2 latent_dim: 潜空间向量维度 super().__init__() self.in_channels in_channels # 编码器 self.encoder nn.Sequential( nn.Conv2d(in_channels, 32, kernel_size3, stride2, padding1), # 32x32 nn.ReLU(True), nn.Conv2d(32, 64, kernel_size3, stride2, padding1), # 16x16 nn.ReLU(True), nn.Conv2d(64, 128, kernel_size3, stride2, padding1), # 8x8 nn.ReLU(True), nn.Conv2d(128, 256, kernel_size3, stride2, padding1), # 4x4 nn.ReLU(True), ) # 经过 flatten 后通过全连接映射到潜空间 self.fc_encode nn.Sequential( nn.Linear(256 * 4 * 4, latent_dim), nn.ReLU(True), ) # 解码器 self.fc_decode nn.Sequential( nn.Linear(latent_dim, 256 * 4 * 4), nn.ReLU(True), ) self.decoder nn.Sequential( nn.ConvTranspose2d(256, 128, kernel_size4, stride2, padding1), # 8x8 nn.ReLU(True), nn.ConvTranspose2d(128, 64, kernel_size4, stride2, padding1), # 16x16 nn.ReLU(True), nn.ConvTranspose2d(64, 32, kernel_size4, stride2, padding1), # 32x32 nn.ReLU(True), nn.ConvTranspose2d(32, in_channels, kernel_size4, stride2, padding1), # 64x64 nn.Sigmoid(), ) def forward(self, x): encoded self.encoder(x) encoded encoded.view(encoded.size(0), -1) z self.fc_encode(encoded) decoded self.fc_decode(z) decoded decoded.view(decoded.size(0), 256, 4, 4) reconstructed self.decoder(decoded) return reconstructed, z需要注意当输入图像尺寸不是 64x64 时需要同步调整编码器中的卷积层参数。为了保持通用我们约定窗口大小为 64并且所有 GAF 图像尺寸为 window_size × window_size即 64×64。4.6 训练模型# 文件路径src/train.py import os import numpy as np import pandas as pd import torch import torch.nn as nn from torch.utils.data import DataLoader from tqdm import tqdm from data_utils import generate_synthetic_data from dataset import TimeSeriesWindowDataset from model import ConvAutoencoder from ts2i import ts2i_multi_channel def train(): # 加载数据 df pd.read_csv(data/sample_data.csv) feature_cols [c for c in df.columns if c.startswith(chan_)] data df[feature_cols].values.astype(np.float32) labels df[is_anomaly].values.astype(np.int32) # 划分训练集和验证集用前 70% 正常数据训练 train_end int(len(data) * 0.7) train_data data[:train_end] train_labels labels[:train_end] test_data data[train_end:] test_labels labels[train_end:] window_size 64 num_features len(feature_cols) batch_size 64 epochs 30 lr 1e-3 device torch.device(cuda if torch.cuda.is_available() else cpu) # 构建数据集 train_dataset TimeSeriesWindowDataset(train_data, train_labels, window_size) test_dataset TimeSeriesWindowDataset(test_data, test_labels, window_size) train_loader DataLoader(train_dataset, batch_sizebatch_size, shuffleTrue, num_workers2) test_loader DataLoader(test_dataset, batch_sizebatch_size, shuffleFalse, num_workers2) # 模型 model ConvAutoencoder(in_channelsnum_features * 2, latent_dim128).to(device) criterion nn.MSELoss() optimizer torch.optim.Adam(model.parameters(), lrlr) # 训练 for epoch in range(epochs): model.train() train_loss 0.0 for images, _ in tqdm(train_loader, descfEpoch {epoch1}/{epochs}): images images.to(device) reconstructed, _ model(images) loss criterion(reconstructed, images) optimizer.zero_grad() loss.backward() optimizer.step() train_loss loss.item() * images.size(0) avg_train_loss train_loss / len(train_dataset) print(fEpoch {epoch1}: train loss {avg_train_loss:.6f}) torch.save(model.state_dict(), model_weights.pth) print(模型训练完成权重已保存为 model_weights.pth) # 在测试集上计算重构误差 model.eval() anomaly_scores [] with torch.no_grad(): for images, _ in test_loader: images images.to(device) reconstructed, _ model(images) loss torch.mean((reconstructed - images) ** 2, dim[1, 2, 3]) # 每个样本一个分数 anomaly_scores.extend(loss.cpu().numpy()) anomaly_scores np.array(anomaly_scores) np.save(test_anomaly_scores.npy, anomaly_scores) # 简化指标取异常分数最高的 5% 作为预测异常 threshold np.percentile(anomaly_scores, 95) preds (anomaly_scores threshold).astype(int) ground_truth np.array([int(np.max(test_labels[i:iwindow_size])) for i in range(len(test_data) - window_size 1)]) tp np.sum((preds 1) (ground_truth 1)) fp np.sum((preds 1) (ground_truth 0)) fn np.sum((preds 0) (ground_truth 1)) precision tp / (tp fp 1e-8) recall tp / (tp fn 1e-8) f1 2 * precision * recall / (precision recall 1e-8) print(fPrecision: {precision:.4f}, Recall: {recall:.4f}, F1: {f1:.4f}) if __name__ __main__: train()预期输出Epoch 1: train loss 0.061234 Epoch 2: train loss 0.052134 ... Epoch 30: train loss 0.002145 Precision: 0.7321, Recall: 0.7854, F1: 0.7579注意实际数值会根据数据分布和随机种子波动。如果只使用合成数据不要依赖最高指标重点是验证流程能跑通。4.7 结果可视化为了直观看到 TS2I 图像在异常检测中的效果我们可以输出正常样本和异常样本的 GAF 图像。# 文件路径src/visualize.py import numpy as np import pandas as pd import matplotlib.pyplot as plt from src.ts2i import gaf_transform from src.data_utils import generate_synthetic_data df generate_synthetic_data() feature_cols [c for c in df.columns if c.startswith(chan_)] data df[feature_cols].values labels df[is_anomaly].values fig, axes plt.subplots(2, 2, figsize(10, 10)) plot_idx 0 # 找一段正常窗口和一段异常窗口 normal_indices np.where(labels 0)[0] anomaly_indices np.where(labels 1)[0] if len(normal_indices) 64 and len(anomaly_indices) 64: normal_start normal_indices[100] anomaly_start anomaly_indices[10] normal_window data[normal_start:normal_start64, 0] anomaly_window data[anomaly_start:anomaly_start64, 0] normal_gaf gaf_transform(normal_window) anomaly_gaf gaf_transform(anomaly_window) axes[0, 0].imshow(normal_gaf, cmapviridis) axes[0, 0].set_title(Normal GAF) axes[0, 1].imshow(anomaly_gaf, cmapviridis) axes[0, 1].set_title(Anomaly GAF) # 原始序列对比 axes[1, 0].plot(normal_window) axes[1, 0].set_title(Normal time series) axes[1, 1].plot(anomaly_window) axes[1, 1].set_title(Anomaly time series) plt.tight_layout() plt.savefig(gaf_visualization.png, dpi120) plt.show()从可视化结果可以直观看出正常窗口的 GAF 图像纹理更平滑图像中的颜色过渡自然异常窗口的 GAF 图像会出现局部突变块纹理断裂明显。5. 常见问题与排查思路5.1 训练损失不下降问题现象常见原因解决思路Loss 长时间保持在同一水平输入数据尺度不一致检查 GAF 转换前是否做了归一化Loss 下降后回弹学习率过大降低学习率或用余弦退火策略数据噪声太大导致模型无法收敛合成数据的异常比例过高降低异常注入比例或增大模型容量5.2 窗口大小如何选择窗口大小直接影响图像分辨率。窗口过小会丢失周期信息GAF 纹理不足窗口过大则图像尺寸变大训练显存占用增加而且每个窗口包含的异常点数量变多窗口级标记会变得粗糙。经验参考如果数据有明显的日周期1440 个采样点/天可以尝试窗口 128 或 256。如果数据是秒级采样100Hz单次检测事件持续约 1-2 秒窗口可以设置为 128 或 256。建议先做少量消融实验观察模型在验证集上的 F1 变化。5.3 异常检测阈值如何确定最简单的做法是使用训练集重构误差的 p95 或 p99 分位数作为阈值。更严谨的做法是在验证集上遍历不同阈值计算 F1 或 PR-AUC。根据业务对误报率和召回率的容忍度选择合适的阈值。from sklearn.metrics import precision_recall_curve # 假设 anomaly_scores 是验证集上的重构误差gt 是标签 precision, recall, thresholds precision_recall_curve(gt, anomaly_scores) f1_scores 2 * (precision * recall) / (precision recall 1e-8) best_threshold_idx np.argmax(f1_scores) best_threshold thresholds[best_threshold_idx]5.4 图像通道数过多导致显存不足如果变量数量很多例如 50 个通道GAF 叠加后通道数就是 100直接输入网络会显著增加显存占用。可以尝试对变量分组每组生成一张合并图像。使用主成分分析或自编码器先降维。只选择与异常相关性最高的前 K 个变量进入图像通道。使用更小的窗口例如 32。5.5 正常样本和异常样本的 GAF 图像差异不明显这说明 TS2I 转换方式没有放大异常特征。可能的改进换用 MTF 或频谱图再看数据分布。对 GAF 图像做直方图均衡化或对比度增强。引入通道注意力机制让模型更关注异常相关的通道。6. 最佳实践与工程建议6.1 数据预处理是 TS2I 效果的天花板在真实业务中原始时序数据往往存在缺失值、离群点、噪声和采样不规律的问题。TS2I 对数据质量问题非常敏感缺失值填充方式会影响 GAF 的局部结构建议使用线性插值或前向填充并做记录。对每个变量使用稳健归一化例如基于分位数避免少量离群点把其他值压到很低的对比度区间。对明显错误的数据如传感器漂移、负数流量先做清洗而不是依赖模型去学习这些噪声。6.2 考虑使用预训练图像编码器如果你有足够的计算资源可以尝试用 ImageNet 预训练的 ResNet 作为编码器然后只训练解码器和最后的异常评分头。这种方式利用了图像领域的大规模预训练表征在小样本时序异常检测任务上往往能带来提升。不过要注意预训练模型输入通常是 224×224 的三通道图像。你需要把多通道 GAF 图像做通道压缩或者使用 1×1 卷积把通道数降为 3。这样做虽然会损失部分信息但通常仍比完全从零训练效果好。6.3 异常评分函数的设计重构误差是最直接的异常评分函数。但在实际项目中可以组合多种信号像素级 MSE。感知损失Perceptual Loss利用预训练网络中间层的特征差异。潜空间距离把输入图像的潜向量与正常样本潜向量的聚类中心比较。时间维度上的平滑惩罚连续时间点的异常分数发生跳变时加以惩罚减少误报。def combined_anomaly_score(image, reconstructed, z, center_z): mse torch.mean((image - reconstructed) ** 2) z_dist torch.norm(z - center_z, dim-1) return mse 0.1 * z_dist6.4 生产环境部署要点使用 ONNX 导出模型PyTorch 模型在服务端推理时可以转换为 ONNX 格式提高部署效率。滑动窗口计算优化在实时场景中每来一个新数据点就重新生成一张 64×64 图像开销较大。可以先维护窗口缓冲区只对新增点做增量计算。设置合理的告警冷却时间连续多个窗口报异常时避免重复告警。可以设置“首次告警后 5 分钟内不重复告警”的窗口冷却机制。定期重训练数据分布会漂移建议每周或每月用最近数据微调模型。6.5 模型可解释性TS2I 方向的优势之一是图像天然适合可视化展示。当模型报出异常时可以把异常窗口的 GAF 图像保存下来发送到告警平台或工单系统。业务人员能直观看到异常形态不需要理解复杂的时序模型原理。7. 总结与后续学习方向本文围绕 PRISM 提出的 TS2I 核心思想详细拆解了把时间序列转换为图像表示的技术路线并基于 GAF 和卷积自编码器构建了一个完整的多元时序异常检测示例。从代码中可以看到图像化后的时间序列可以直接复用 CNN 的相关结构与成熟优化器训练流程清晰异常评分也容易落地到告警系统。接下来你可以继续深入研究的方向Paper List 延伸阅读格拉姆角场原始论文、马尔可夫转移场原始论文、TS2I 相关 benchmark如 UCR、UEA 数据集上的对比实验。升级模型结构把自编码器替换为 Vision TransformerViT编码器用更强大的注意力机制提取图像全局关系。多模态融合把 TS2I 图像与原始序列同时输入模型通过双塔结构分别编码后再融合可能进一步提升检测精度。消融实验设计建议在自己的数据上对比 GAF、MTF、频谱图的效果选择最适合业务场景的 TS2I 方法。实际项目中建议先在小规模数据上跑通 PRISM 思路再逐步扩大窗口、增加变量、引入预训练模型。如果模型在验证集上 F1 达不到预期优先检查数据预处理和 TS2I 转换方式而不是急着换更大的模型。如果你在复现过程中遇到任何问题欢迎在评论区把报错信息和数据特征贴出来我们可以一起讨论排错思路。