时间序列异常检测评测框架：统一接口、阈值校准与可复现评估

简介本资源是一个面向时间序列异常检测算法评估的开源框架适用于计算机、数学、电子信息等专业学生开展课程设计、期末大作业或毕业设计也适合算法工程师快速验证与对比多种检测方法。压缩包共51个文件含40个Python核心模块涵盖数据预处理、算法实现、指标评估、阈值选择、结果可视化等完整流程、7张效果示意图、2份说明文档README与TODO、1个依赖清单及1个Git配置文件整体仅1.31MB轻量易部署。已有439人学习下载体现了其在教学与实践场景中的实用价值。读者可直接运行框架复现KSigma、LSTM、AutoEncoder、Matrix Profile等主流算法并基于统一pipeline完成数据加载、模型训练、预测输出与多维度指标计算含带阈值/无阈值两类评估逻辑目录结构清晰分层模块职责明确便于理解架构设计与二次开发。1. 这不是一个“跑起来就行”的评估脚本而是一套可插拔、可复现、带阈值校准机制的时间序列异常检测算法评测框架你手头那套 LSTM 或 Autoencoder 模型在自己的数据上 AUC 0.92但换到 NAB 数据集上就掉到 0.68不是模型不行而是你缺一套统一的数据预处理管道、标准化的评估协议、可复现的阈值决策逻辑——而这正是这个框架存在的根本理由。它不提供“一键训练”也不封装成黑盒 API它把ksigma.py、lstm.py、matrix_profile.py等 10 种主流检测器抽象为统一接口强制所有算法走detector.fit()→detector.predict()→evaluate.with_threshold()流程并在metric_with_threshold.py中内置了 F1-score、Precision-Recall 曲线、Point-adjusted F1PFA等 7 种工业级指标计算逻辑。适合需要做课程设计、毕设对比实验、或想系统性验证新检测思路的开发者——尤其当你发现不同论文用不同阈值策略导致结果不可比时这个框架的threshold.py模块就是你的校准基准。2. 从源码结构看设计逻辑为什么detector/和evaluate/必须解耦且threshold/不是可选模块这个框架不是把一堆算法脚本堆在一起而是按「责任分离」原则组织detector/负责生成原始异常分值scoreevaluate/负责将 score 映射为二元标签anomaly / normalthreshold/则承担 score→label 的映射决策。这种三层结构直接对应时间序列异常检测的工业实践痛点同一组 score用固定阈值、动态滑动窗口阈值、或基于 ROC 曲线的最优阈值会得出完全不同的 F1 结果。我们先看核心目录关系├── detector/ # 所有算法实现必须继承 base.Detector实现 fit() 和 predict() │ ├── ksigma.py # 返回 raw_score 数组非 0/1 标签 │ ├── lstm.py # 同样只输出 anomaly score不自行判别 │ └── ... ├── evaluate/ # 评估入口接收 detector.predict() 输出 真实标签 y_true │ ├── eval.py # 主评估调度器调用 metric_with_threshold.py │ └── metric_with_threshold.py # 关键含 threshold_search() 和 compute_metrics() ├── threshold/ # 阈值策略实现非装饰器而是独立类 ThresholdSelector │ └── threshold.py # 提供 fixed, percentile, roc 三种策略 └── pipeline/ # 串联流程load → preprocess → detect → threshold → evaluate注意detector.predict()的返回值类型被严格约束为np.ndarray长度与输入时间序列一致每个元素为 float 型异常分值如 reconstruction error、z-score、p-value。任何返回 bool 数组或整数标签的 detector 实现都会在eval.py的validate_detector_output()中抛出ValueError。这是框架强制统一评估口径的第一道防线。2.1detector.py接口契约为什么fit()必须支持X_train和y_train双输入即使无监督算法也需兼容打开detector/detector.py你会看到基类定义class Detector(ABC): abstractmethod def fit(self, X_train: np.ndarray, y_train: Optional[np.ndarray] None) - None: 训练检测器。y_train 仅用于半监督/监督算法 无监督算法如 ksigma必须接受 y_trainNone 并忽略它。 pass abstractmethod def predict(self, X_test: np.ndarray) - np.ndarray: 返回 shape(n_samples,) 的 float array表示每个时间点的异常分值。 注意不是 0/1 标签 pass以ksigma.py为例其fit()方法实际只使用X_train计算均值和标准差但签名仍保留y_train参数# detector/ksigma.py class KSigmaDetector(Detector): def __init__(self, k: float 3.0): self.k k self.mean_ None self.std_ None def fit(self, X_train: np.ndarray, y_train: Optional[np.ndarray] None) - None: # 即使 y_train 是 None也要能通过类型检查 self.mean_ np.mean(X_train) self.std_ np.std(X_train) def predict(self, X_test: np.ndarray) - np.ndarray: # 返回 |x - mu| / sigma即 z-score 绝对值 return np.abs((X_test - self.mean_) / (self.std_ 1e-8))这种设计让框架能统一调用detector.fit(X_train, y_train)无需为每种算法写分支逻辑。如果你自己实现新 detector必须遵守该契约否则pipeline/task.py在run_evaluation()中会因参数不匹配而中断。2.2threshold/threshold.py的三种策略详解为什么roc策略比fixed更可靠且percentile适合长尾分布阈值模块不是简单地score 0.5而是提供三种可配置策略对应不同场景策略名输入参数适用场景内部逻辑说明fixedthreshold_value: float已知业务容忍度如 CPU 使用率 90% 触发告警直接score threshold_value生成 binary labelpercentilep: float0~100数据分布偏斜、存在长尾异常分值取np.percentile(score, p)作为阈值保证恰好 p% 的点被标记为异常roctarget_fpr: float需要控制误报率如监控系统要求 FPR 0.01在验证集上遍历所有可能阈值选择使FPR ≈ target_fpr的阈值再应用于测试集关键代码在threshold/threshold.py的select_threshold()方法def select_threshold( self, scores: np.ndarray, y_true: Optional[np.ndarray] None, strategy: str roc, **kwargs ) - float: if strategy fixed: return kwargs.get(threshold_value, 0.5) elif strategy percentile: p kwargs.get(p, 95.0) return np.percentile(scores, p) elif strategy roc: if y_true is None: raise ValueError(ROC strategy requires y_true for validation) # 计算所有候选阈值下的 FPR thresholds np.unique(scores) fprs [] for t in thresholds: y_pred (scores t).astype(int) tn, fp, fn, tp confusion_matrix(y_true, y_pred).ravel() fpr fp / (fp tn 1e-8) fprs.append(fpr) # 找到最接近 target_fpr 的阈值 target_fpr kwargs.get(target_fpr, 0.01) idx np.argmin(np.abs(np.array(fprs) - target_fpr)) return thresholds[idx] else: raise ValueError(fUnknown strategy: {strategy})提示roc策略必须传入y_true否则会报错。这意味着你在eval.py中调用ThresholdSelector.select_threshold()时需确保验证集标签可用。若只有测试集无标签如真实部署场景则只能选用fixed或percentile。3. 完整运行流程从加载数据、配置 detector、到生成带 PFA 的评估报告框架不依赖 Jupyter全部通过script/grafana.py或命令行驱动。我们以NAB数据集中的realTweets子集为例演示端到端执行链。3.1 数据准备data_prepare/模块如何统一原始时间序列格式原始数据常为 CSV含 timestamp, value, label但框架要求输入为np.ndarray三元组(X, y, timestamps)。data_prepare/dataset.py提供标准化加载器# data_prepare/dataset.py from data_prepare.time_series import TimeSeriesDataset # 加载 NAB 数据集需提前下载到 data/nab/ ds TimeSeriesDataset( root_pathdata/nab/, dataset_namerealTweets, train_ratio0.7, seq_len100 # 用于 LSTM 等需滑窗的模型 ) X_train, y_train, _ ds.get_train_data() # shape: (n_samples, 1) or (n_samples, seq_len) X_test, y_test, timestamps ds.get_test_data() # y_test 含人工标注的 anomaly 标签TimeSeriesDataset内部自动完成时间戳解析支持 ISO、Unix、自定义格式缺失值线性插值preprocess.py中fill_missing_values()归一化Z-score可关闭标签对齐确保y_test[i]对应X_test[i]的时间点3.2 配置并运行单次评估eval.py的run_evaluation()如何串联各模块核心评估函数位于evaluate/eval.py# evaluate/eval.py from detector.lstm import LSTMDetector from threshold.threshold import ThresholdSelector from evaluate.metric_with_threshold import compute_metrics def run_evaluation( detector: Detector, X_train: np.ndarray, X_test: np.ndarray, y_test: np.ndarray, threshold_strategy: str roc, threshold_kwargs: dict None, metrics: list [f1, precision, recall, pfa] # pfa point-adjusted f1 ) - dict: # Step 1: 训练 detector detector.fit(X_train) # Step 2: 获取原始分值 scores detector.predict(X_test) # shape: (len(X_test),) # Step 3: 选择阈值需 y_test 用于 roc selector ThresholdSelector() if threshold_kwargs is None: threshold_kwargs {target_fpr: 0.01} if threshold_strategy roc else {} threshold_value selector.select_threshold( scoresscores, y_truey_test, strategythreshold_strategy, **threshold_kwargs ) # Step 4: 生成二元预测 y_pred (scores threshold_value).astype(int) # Step 5: 计算指标含 PFA results compute_metrics(y_truey_test, y_predy_pred, scoresscores, metricsmetrics) results[threshold] threshold_value return results # 调用示例 lstm_det LSTMDetector(hidden_size64, num_layers2) results run_evaluation( detectorlstm_det, X_trainX_train, X_testX_test, y_testy_test, threshold_strategyroc, threshold_kwargs{target_fpr: 0.005}, metrics[f1, pfa, roc_auc] ) print(results) # 输出: {f1: 0.723, pfa: 0.689, roc_auc: 0.891, threshold: 0.421}compute_metrics()内部调用metric_with_threshold.py其中point_adjust_f1()实现 PFA —— 这是 NAB、UCR 等 benchmark 的核心指标它将连续的异常区间视为一个整体避免单点误报放大 F1 偏差。3.3 批量评估与结果聚合script/grafana.py如何生成多算法对比报告script/grafana.py不是 Grafana 插件而是生成 HTML 报告的脚本名字源于其输出可导入 Grafana 展示python script/grafana.py \ --detectors ksigma,lstm,autoencoder \ --datasets realTweets,ecg,cpu \ --threshold_strategy roc \ --output_dir reports/202406_nab_comparison它会自动遍历algorithm/下指定 detector 模块加载每个dataset_name对应的数据对每组(detector, dataset)运行run_evaluation()将结果存为reports/202406_nab_comparison/results.json并生成index.html表格生成的 HTML 包含每个 detector 在各数据集上的 F1/PFA/ROC-AUC 三指标热力图threshold_value分布直方图验证阈值稳定性scores与y_true的叠加时序图doc/1.png,doc/2.png即此类可视化4. 关键参数调优与常见失败排查为什么seq_len错配会导致 LSTM 报ValueError以及如何修复InfluxDB连接超时框架中多数坑集中在数据维度匹配和阈值策略依赖两个层面。以下是高频问题及解决方案。4.1 LSTM detector 的seq_len必须与TimeSeriesDataset.seq_len严格一致LSTMDetector默认期望输入 shape 为(batch, seq_len, features)而TimeSeriesDataset的get_train_data()若设置seq_len100则返回(n_samples, 100, 1)但若你在LSTMDetector.__init__()中未显式传入seq_len它会使用默认seq_len50导致fit()时维度不匹配# 错误示例detector 期望 (N, 50, 1)但 dataset 提供 (N, 100, 1) lstm LSTMDetector() # 默认 seq_len50 lstm.fit(X_train) # 报错expected input shape (..., 50, 1), got (..., 100, 1)修复方式显式传入seq_len并与 dataset 保持一致ds TimeSeriesDataset(seq_len100) # 设定滑窗长度 X_train, _, _ ds.get_train_data() lstm LSTMDetector( hidden_size128, num_layers2, seq_len100, # 必须等于 ds.seq_len input_size1 ) lstm.fit(X_train) # 成功4.2InfluxDB存储结果时连接超时database/influxdbtool.py的重试与超时配置若使用influxdbtool.py存储评估结果到 InfluxDB需配置连接参数# database/influxdbtool.py from influxdb import InfluxDBClient class InfluxDBTool: def __init__( self, host: str localhost, port: int 8086, username: str admin, password: str admin, database: str ts_anomaly, timeout: int 10, # 连接超时秒 retries: int 3 # 连接失败重试次数 ): self.client InfluxDBClient( hosthost, portport, usernameusername, passwordpassword, databasedatabase, timeouttimeout, retriesretries )常见失败场景InfluxDB 服务未启动或网络延迟高导致首次连接超时。此时retries3会尝试 3 次每次间隔 1 秒。若仍失败则抛出InfluxDBClientError。建议在script/grafana.py中捕获该异常并降级为本地 JSON 存储try: db_tool.write_results(results) except Exception as e: print(fInfluxDB write failed: {e}. Falling back to local JSON.) save_to_json(results, fallback_results.json)4.3metric_without_threshold.py的适用场景何时该用它替代metric_with_threshold.pymetric_without_threshold.py提供auc_score、average_precision_score等无需阈值的指标适用于算法本身输出概率如 VAE 的 reconstruction probability你只想比较 score 的排序质量而非最终二元决策做 ROC 曲线分析需遍历多个阈值调用方式from evaluate.metric_without_threshold import auc_score, average_precision_score # 直接用原始 scores 计算 AUC无需 threshold step auc auc_score(y_truey_test, y_scorescores) # y_score 是 predict() 输出 ap average_precision_score(y_truey_test, y_scorescores)注意auc_score内部调用sklearn.metrics.roc_auc_score要求y_true至少包含两类样本即既有 0 也有 1。若y_test全为 0无异常点会报ValueError: Only one class present in y_true。此时应跳过 AUC 计算改用pfa或precisionk。5. 进阶技巧如何用transform/transform.py实现自定义特征工程以及utils/timer.py的精准耗时统计框架预留了transform/模块用于插入领域知识而timer.py则解决评估中常被忽略的推理延迟指标——这对实时监控系统至关重要。5.1transform.py在 detector 前插入滑动窗口统计特征假设你发现原始时间序列的突变点常伴随方差骤增想为 LSTM 输入额外的「局部方差」特征。transform.py提供BaseTransformer接口# transform/transform.py from abc import ABC, abstractmethod import numpy as np class BaseTransformer(ABC): abstractmethod def transform(self, X: np.ndarray) - np.ndarray: pass class RollingVarianceTransformer(BaseTransformer): def __init__(self, window_size: int 10): self.window_size window_size def transform(self, X: np.ndarray) - np.ndarray: # X shape: (n_samples, 1) → 输出 (n_samples, 2): [value, rolling_var] variances np.array([ np.var(X[max(0, i-self.window_size1):i1]) for i in range(len(X)) ]).reshape(-1, 1) return np.hstack([X, variances])在 pipeline 中启用from transform.transform import RollingVarianceTransformer transformer RollingVarianceTransformer(window_size15) X_train_enhanced transformer.transform(X_train) # shape: (n, 2) X_test_enhanced transformer.transform(X_test) lstm LSTMDetector(input_size2) # 注意input_size 需改为 2 lstm.fit(X_train_enhanced) scores lstm.predict(X_test_enhanced)5.2utils/timer.py精确测量 detector 的fit()和predict()耗时timer.py使用time.perf_counter()纳秒级精度避免time.time()受系统时钟调整影响# utils/timer.py import time from contextlib import contextmanager contextmanager def timer(name: str): start time.perf_counter() yield end time.perf_counter() print(f[{name}] took {end - start:.4f}s) # 在 eval.py 中使用 with timer(LSTM fit): detector.fit(X_train) with timer(LSTM predict): scores detector.predict(X_test)输出示例[LSTM fit] took 12.3456s [LSTM predict] took 0.0872s该耗时会被自动写入results.json的timing字段用于对比不同算法的吞吐量samples/sec和内存占用需配合psutil扩展。5.3TODO.md中的可扩展方向为什么spot.py需要重构为流式接口TODO.md明确列出spot.pySequential Probability Ratio Test当前为批处理模式但实际流式场景需update()接口。其重构要点如下当前SPOTDetector.predict(X_test)一次性处理全部数据目标接口spot.update(new_point: float) - bool返回当前点是否异常关键修改将self.peaks_等状态变量从数组改为滚动 bufferupdate()中只计算新点与历史 tail 的似然比这并非功能增强而是部署形态适配——批处理评估可用但生产环境必须流式。框架已预留detector.py中的update()抽象方法你只需在spot.py中实现即可无缝接入 pipeline。本文还有配套的精品资源点击获取