简介Aiopstools 是一套面向运维开发与 AIOps 入门者的 Python 工具包聚焦用机器学习方法解决日常运维场景中的典型问题适合具备一定 Python 基础、希望把智能算法落地到监控告警链路的工程师学习与二次开发。资源包共 60 个文件以 33 个 py 源码模块为核心辅以 14 个 csv 示例数据、9 个 md 说明文档及少量 txt、authors 等配置说明压缩包约 122KB体量轻便、便于快速导入项目。功能上覆盖异常检测、告警收敛、时间序列预测与告警关联分析四大模块示例脚本与测试文档相互对应可帮助读者理解各算法的输入输出与调用方式。目前已有 339 人学习下载读者可借此掌握从数据样例到模块调用的完整流程并参考 README 与文档说明快速将工具包集成进自有运维平台作为 AIOps 实践的起步脚手架。1. 从“能跑起来”到“能落地”Python AIOps 基本包到底包含什么很多人第一次接触 AIOps是从一份“Python AIOps 基本包”的代码下载开始的。下载完解压发现里面无非是几个.py文件、一份requirements.txt、一个config.yaml跑起来却报一堆依赖冲突于是开始怀疑这东西到底有没有用。我最初也是这么想的直到线上告警把手机震到凌晨三点才真正理解一个“基本包”的价值它不是炫技的算法集合而是把日志采集、指标异常检测、告警收敛、根因初筛这几件事用 Python 串成一条能跑通的最小闭环。这篇文章面向的是运维开发、SRE 和刚转 AIOps 的后端同学目标很明确——让你拿到或自己搭出这样一套基本包后知道每个模块为什么存在、参数怎么调、坑在哪而不是停留在“下载完看一眼就吃灰”。下面按“是什么 → 怎么搭 → 怎么避坑 → 怎么进阶”的顺序展开所有代码都可以直接抄去改。2. 拆解基本包的四个核心模块与选型理由2.1 为什么是采集、检测、收敛、根因这四块AIOps 的论文和商业方案动辄讲几十种算法但落到一个“基本包”里真正必须存在的只有四块。第一块是数据采集与标准化因为运维数据源太杂Prometheus 的指标、ELK 的日志、SkyWalking 的链路格式各不相同不先统一成(timestamp, entity, metric, value)这种扁平结构后面所有算法都是空谈。第二块是异常检测这是 AIOps 区别于传统阈值告警的核心常见做法是用统计方法3-sigma、MAD或轻量机器学习Isolation Forest、Prophet替代固定阈值。第三块是告警收敛线上最痛的不是没告警而是告警风暴一个网络抖动能触发几百条必须做去重、聚合、抑制。第四块是根因初筛哪怕只是基于拓扑和时序相关性的简单排序也能把排查时间从半小时压到几分钟。选型上我一般坚持“能解释优先于能拟合”。基本包里默认用 MAD 做单指标检测、用 Isolation Forest 做多指标联合检测不是因为它们最准而是因为它们的输出能反推——“这个点偏离中位数 6 倍 MAD”运维看得懂敢信。深度模型不是不能上但放进“基本包”里会让依赖膨胀、调参成本陡增不适合作为起点。2.2 用 Python 搭出最小可运行骨架下面这段代码是一个可运行的最小骨架把采集、检测、收敛三步串起来。依赖只有pandas、numpy、scikit-learn、pyyamlrequirements.txt里写死版本能避免大部分“昨天还能跑今天报错”的玄学问题。# aiops_min/package.py import yaml import numpy as np import pandas as pd from sklearn.ensemble import IsolationForest def load_config(pathconfig.yaml): # 读取配置所有阈值和窗口都从这里来避免硬编码 with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def detect_mad(series: pd.Series, threshold: float 3.5): # MAD 检测对尖刺型异常敏感且不受极端值影响 median series.median() mad np.median(np.abs(series - median)) if mad 0: return pd.Series([False] * len(series), indexseries.index) score 0.6745 * (series - median) / mad return score.abs() threshold def detect_iforest(df: pd.DataFrame, contamination: float 0.01): # 多指标联合检测contamination 是预期异常比例 model IsolationForest( n_estimators100, contaminationcontamination, random_state42 ) pred model.fit_predict(df.fillna(0)) return pd.Series(pred -1, indexdf.index) def converge_alerts(alerts: pd.DataFrame, window: str 5min): # 告警收敛同一实体在窗口内只保留最早一条 alerts alerts.sort_values(timestamp) grouped alerts.groupby([entity, pd.Grouper(keytimestamp, freqwindow)]) return grouped.first().reset_index()逻辑说明load_config把阈值外置改参数不用动代码detect_mad里的0.6745是让 MAD 与标准差可比的归一化常数threshold3.5是经验值太松会漏报、太紧会误报detect_iforest的contamination必须根据实际数据调设成 0.01 意味着你预期 1% 的点是异常设大了会把正常波动也标红converge_alerts用pd.Grouper按时间窗口聚合5min是常见起点但要根据业务告警频率调整。参数上最容易翻车的是contamination和 MAD 的threshold。我的血泪经验是先用一周历史数据跑一遍把检测出的异常点画出来人工看确认“这些确实该报”再上线别直接拿默认值怼生产。2.3 配置文件的字段设计与默认值config.yaml不是随便写的字段设计直接决定后面能不能扩展。下面这份是我常用的模板# config.yaml datasource: type: prometheus # 常见做法是 prometheus / elasticsearch endpoint: http://localhost:9090 step: 60s # 采集粒度和 Prometheus scrape_interval 对齐 detection: method: mad # mad / iforest / both mad_threshold: 3.5 iforest_contamination: 0.01 min_points: 30 # 少于 30 个点不做检测避免小样本误判 converge: window: 5min max_alerts_per_entity: 3 # 单实体窗口内最多保留条数 rootcause: enable: true top_k: 5 # 根因候选返回前 5 个字段说明step必须和采集端一致否则时间戳对不齐会导致检测结果错位min_points是我踩过坑之后加的早期没这个限制冷启动阶段数据点太少MAD 直接算出 0整段被误判max_alerts_per_entity是收敛的第二道保险防止窗口内仍然堆积。3. 把基本包接到真实数据源上从本地 CSV 到 Prometheus3.1 先用本地 CSV 验证算法逻辑不要一上来就连生产 Prometheus先用一份 CSV 把算法跑通。CSV 至少要有timestamp、entity、metric、value四列这是基本包约定的最小 schema。# aiops_min/run_local.py import pandas as pd from package import load_config, detect_mad, converge_alerts cfg load_config() df pd.read_csv(sample_metrics.csv, parse_dates[timestamp]) # 按实体和指标分组检测不能全局一起算 results [] for (entity, metric), group in df.groupby([entity, metric]): if len(group) cfg[detection][min_points]: continue flags detect_mad(group[value], cfg[detection][mad_threshold]) abnormal group[flags].copy() abnormal[metric] metric results.append(abnormal) alerts pd.concat(results) if results else pd.DataFrame() converged converge_alerts(alerts, cfg[converge][window]) print(converged[[timestamp, entity, metric, value]])逻辑说明分组检测是关键把不同实体、不同指标混在一起算 MAD 会得到完全错误的基线min_points过滤掉数据不足的分组最后收敛输出。跑通这一步你就能确认算法逻辑没问题再去接真实数据源。3.2 接 Prometheus 的查询与对齐接 Prometheus 最常见的方式是用 HTTP API 拉区间数据然后转成 DataFrame。注意step要和配置一致否则返回的点数和你预期不符。# aiops_min/prom_client.py import requests import pandas as pd from datetime import datetime, timedelta def query_range(endpoint, query, step60s, hours1): end datetime.now() start end - timedelta(hourshours) resp requests.get( f{endpoint}/api/v1/query_range, params{ query: query, start: start.timestamp(), end: end.timestamp(), step: step }, timeout10 ) resp.raise_for_status() data resp.json()[data][result] rows [] for series in data: entity series[metric].get(instance, unknown) for ts, val in series[values]: rows.append({ timestamp: pd.to_datetime(ts, units), entity: entity, metric: query, value: float(val) }) return pd.DataFrame(rows)逻辑说明query_range返回的是多条时间序列每条带自己的 labels这里用instance作为实体标识实际项目里可能要用pod或servicetimeout10是必须的Prometheus 查询慢的时候不设超时会挂死float(val)转换是因为 API 返回的是字符串。参数上hours1决定回看窗口做检测一般至少要看 1 小时做趋势分析要更长。3.3 日志数据接入的字段映射日志接入比指标麻烦因为格式不统一。基本包的做法是先做字段映射把不同来源的日志统一成timestamp、entity、level、message四列再走后续流程。常见做法是用正则或 Grok 提取下面是一个简化版import re import pandas as pd PATTERN re.compile( r(?Ptimestamp\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*? r(?PlevelERROR|WARN|INFO).*? r(?Pentity[\w\-]): (?Pmessage.) ) def parse_logs(lines): rows [] for line in lines: m PATTERN.match(line) if m: rows.append(m.groupdict()) df pd.DataFrame(rows) df[timestamp] pd.to_datetime(df[timestamp]) return df逻辑说明正则里的entity用[\w\-]匹配服务名实际项目里要按你的日志格式改解析失败的日志直接丢弃还是记录取决于你对完整性的要求我一般会单独存一份解析失败的样本方便回头调正则。日志接入后异常检测通常不是对message做而是对日志量、错误率这类聚合指标做这一点新手容易搞混。4. 避坑与排查基本包上线后最容易翻车的五件事4.1 现象检测结果全是异常满屏红原因contamination设得过大或者数据里本身有大量缺失值被fillna(0)填成了 0导致 Isolation Forest 把 0 值区域全判为异常。解决先把contamination降到 0.005 以下试同时检查缺失率缺失超过 30% 的实体直接跳过检测不要硬填。4.2 现象MAD 检测对缓慢漂移完全不报原因MAD 基于中位数对突变敏感但对持续几小时的缓慢上升不敏感因为中位数也跟着漂移了。解决对趋势型指标改用滑动窗口内的差分或 STL 分解后的残差做检测基本包里可以加一个method: diff_mad的分支先做一阶差分再算 MAD。4.3 现象告警收敛后漏掉了真正的根因告警原因收敛策略是“同实体窗口内只留最早一条”如果根因告警比症状告警晚几秒到达就会被丢掉。解决收敛时不要只按时间排序要结合告警级别和指标类型加权根因类指标如 CPU、内存、网络丢包优先级高于症状类如接口超时基本包里可以在converge_alerts前加一列priority再排序。4.4 现象接 Prometheus 后时间戳对不齐检测窗口错位原因Prometheus 返回的时间戳是 UTC本地 CSV 可能是本地时区混用后pd.Grouper分窗全乱。解决所有时间统一转成 UTC 再入库pd.to_datetime(ts, units, utcTrue)展示时再转本地时区。这个坑我踩过两次第二次是因为换了台机器默认时区不同血泪教训。4.5 现象依赖版本冲突昨天能跑今天报错原因requirements.txt里只写了包名没锁版本pip install拉到了不兼容的新版本。解决所有依赖锁死到具体版本并且用虚拟环境隔离。基本包里我一般会附一份requirements.lock用pip freeze生成部署时只装 lock 文件。5. 进阶把基本包做成可复用的检测流水线基本包跑通之后下一步不是加更多算法而是把它变成一条可复用的流水线。我的做法是把检测逻辑抽象成Detector基类不同方法实现fit和predict这样加新算法不用改主流程。# aiops_min/detectors.py from abc import ABC, abstractmethod import numpy as np import pandas as pd class Detector(ABC): abstractmethod def fit(self, series: pd.Series): pass abstractmethod def predict(self, series: pd.Series) - pd.Series: pass class MADDetector(Detector): def __init__(self, threshold3.5): self.threshold threshold self.median None self.mad None def fit(self, series): self.median series.median() self.mad np.median(np.abs(series - self.median)) def predict(self, series): if self.mad 0: return pd.Series([False] * len(series), indexseries.index) score 0.6745 * (series - self.median) / self.mad return score.abs() self.threshold逻辑说明fit和predict分离的好处是可以用历史数据 fit用新数据 predict避免每次检测都重算基线MADDetector把中位数和 MAD 存成实例属性支持增量更新。参数上threshold可以在配置里按指标类型分别设比如 CPU 用 3.0QPS 用 4.0因为不同指标的波动特性不一样。验证方法上我习惯用“回放”来测拿一周历史数据按时间顺序喂给流水线看每天检测出的告警数量和人工标注的故障时间点是否吻合。下面这张表是我常用的验证指标指标含义可接受范围召回率真实故障被检出的比例 0.8误报率正常时段被误报的比例 0.05平均检测延迟故障发生到告警的时间 2 个采集周期收敛比收敛后告警数 / 原始告警数0.1 ~ 0.3最后说一个具体技巧把每次检测的中间结果基线值、偏离分数、是否异常都落库不要只存最终告警。这样当有人质疑“为什么这条没报”时你能直接翻出当时的分数而不是靠猜。这个习惯帮我省了无数次扯皮。做 AIOps 基本包最怕的不是算法不够先进而是出了问题说不清楚。希望帮到你。本文还有配套的精品资源点击获取
Python AIOps 基本包实战：从采集、异常检测到告警收敛的落地指南
简介Aiopstools 是一套面向运维开发与 AIOps 入门者的 Python 工具包聚焦用机器学习方法解决日常运维场景中的典型问题适合具备一定 Python 基础、希望把智能算法落地到监控告警链路的工程师学习与二次开发。资源包共 60 个文件以 33 个 py 源码模块为核心辅以 14 个 csv 示例数据、9 个 md 说明文档及少量 txt、authors 等配置说明压缩包约 122KB体量轻便、便于快速导入项目。功能上覆盖异常检测、告警收敛、时间序列预测与告警关联分析四大模块示例脚本与测试文档相互对应可帮助读者理解各算法的输入输出与调用方式。目前已有 339 人学习下载读者可借此掌握从数据样例到模块调用的完整流程并参考 README 与文档说明快速将工具包集成进自有运维平台作为 AIOps 实践的起步脚手架。1. 从“能跑起来”到“能落地”Python AIOps 基本包到底包含什么很多人第一次接触 AIOps是从一份“Python AIOps 基本包”的代码下载开始的。下载完解压发现里面无非是几个.py文件、一份requirements.txt、一个config.yaml跑起来却报一堆依赖冲突于是开始怀疑这东西到底有没有用。我最初也是这么想的直到线上告警把手机震到凌晨三点才真正理解一个“基本包”的价值它不是炫技的算法集合而是把日志采集、指标异常检测、告警收敛、根因初筛这几件事用 Python 串成一条能跑通的最小闭环。这篇文章面向的是运维开发、SRE 和刚转 AIOps 的后端同学目标很明确——让你拿到或自己搭出这样一套基本包后知道每个模块为什么存在、参数怎么调、坑在哪而不是停留在“下载完看一眼就吃灰”。下面按“是什么 → 怎么搭 → 怎么避坑 → 怎么进阶”的顺序展开所有代码都可以直接抄去改。2. 拆解基本包的四个核心模块与选型理由2.1 为什么是采集、检测、收敛、根因这四块AIOps 的论文和商业方案动辄讲几十种算法但落到一个“基本包”里真正必须存在的只有四块。第一块是数据采集与标准化因为运维数据源太杂Prometheus 的指标、ELK 的日志、SkyWalking 的链路格式各不相同不先统一成(timestamp, entity, metric, value)这种扁平结构后面所有算法都是空谈。第二块是异常检测这是 AIOps 区别于传统阈值告警的核心常见做法是用统计方法3-sigma、MAD或轻量机器学习Isolation Forest、Prophet替代固定阈值。第三块是告警收敛线上最痛的不是没告警而是告警风暴一个网络抖动能触发几百条必须做去重、聚合、抑制。第四块是根因初筛哪怕只是基于拓扑和时序相关性的简单排序也能把排查时间从半小时压到几分钟。选型上我一般坚持“能解释优先于能拟合”。基本包里默认用 MAD 做单指标检测、用 Isolation Forest 做多指标联合检测不是因为它们最准而是因为它们的输出能反推——“这个点偏离中位数 6 倍 MAD”运维看得懂敢信。深度模型不是不能上但放进“基本包”里会让依赖膨胀、调参成本陡增不适合作为起点。2.2 用 Python 搭出最小可运行骨架下面这段代码是一个可运行的最小骨架把采集、检测、收敛三步串起来。依赖只有pandas、numpy、scikit-learn、pyyamlrequirements.txt里写死版本能避免大部分“昨天还能跑今天报错”的玄学问题。# aiops_min/package.py import yaml import numpy as np import pandas as pd from sklearn.ensemble import IsolationForest def load_config(pathconfig.yaml): # 读取配置所有阈值和窗口都从这里来避免硬编码 with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def detect_mad(series: pd.Series, threshold: float 3.5): # MAD 检测对尖刺型异常敏感且不受极端值影响 median series.median() mad np.median(np.abs(series - median)) if mad 0: return pd.Series([False] * len(series), indexseries.index) score 0.6745 * (series - median) / mad return score.abs() threshold def detect_iforest(df: pd.DataFrame, contamination: float 0.01): # 多指标联合检测contamination 是预期异常比例 model IsolationForest( n_estimators100, contaminationcontamination, random_state42 ) pred model.fit_predict(df.fillna(0)) return pd.Series(pred -1, indexdf.index) def converge_alerts(alerts: pd.DataFrame, window: str 5min): # 告警收敛同一实体在窗口内只保留最早一条 alerts alerts.sort_values(timestamp) grouped alerts.groupby([entity, pd.Grouper(keytimestamp, freqwindow)]) return grouped.first().reset_index()逻辑说明load_config把阈值外置改参数不用动代码detect_mad里的0.6745是让 MAD 与标准差可比的归一化常数threshold3.5是经验值太松会漏报、太紧会误报detect_iforest的contamination必须根据实际数据调设成 0.01 意味着你预期 1% 的点是异常设大了会把正常波动也标红converge_alerts用pd.Grouper按时间窗口聚合5min是常见起点但要根据业务告警频率调整。参数上最容易翻车的是contamination和 MAD 的threshold。我的血泪经验是先用一周历史数据跑一遍把检测出的异常点画出来人工看确认“这些确实该报”再上线别直接拿默认值怼生产。2.3 配置文件的字段设计与默认值config.yaml不是随便写的字段设计直接决定后面能不能扩展。下面这份是我常用的模板# config.yaml datasource: type: prometheus # 常见做法是 prometheus / elasticsearch endpoint: http://localhost:9090 step: 60s # 采集粒度和 Prometheus scrape_interval 对齐 detection: method: mad # mad / iforest / both mad_threshold: 3.5 iforest_contamination: 0.01 min_points: 30 # 少于 30 个点不做检测避免小样本误判 converge: window: 5min max_alerts_per_entity: 3 # 单实体窗口内最多保留条数 rootcause: enable: true top_k: 5 # 根因候选返回前 5 个字段说明step必须和采集端一致否则时间戳对不齐会导致检测结果错位min_points是我踩过坑之后加的早期没这个限制冷启动阶段数据点太少MAD 直接算出 0整段被误判max_alerts_per_entity是收敛的第二道保险防止窗口内仍然堆积。3. 把基本包接到真实数据源上从本地 CSV 到 Prometheus3.1 先用本地 CSV 验证算法逻辑不要一上来就连生产 Prometheus先用一份 CSV 把算法跑通。CSV 至少要有timestamp、entity、metric、value四列这是基本包约定的最小 schema。# aiops_min/run_local.py import pandas as pd from package import load_config, detect_mad, converge_alerts cfg load_config() df pd.read_csv(sample_metrics.csv, parse_dates[timestamp]) # 按实体和指标分组检测不能全局一起算 results [] for (entity, metric), group in df.groupby([entity, metric]): if len(group) cfg[detection][min_points]: continue flags detect_mad(group[value], cfg[detection][mad_threshold]) abnormal group[flags].copy() abnormal[metric] metric results.append(abnormal) alerts pd.concat(results) if results else pd.DataFrame() converged converge_alerts(alerts, cfg[converge][window]) print(converged[[timestamp, entity, metric, value]])逻辑说明分组检测是关键把不同实体、不同指标混在一起算 MAD 会得到完全错误的基线min_points过滤掉数据不足的分组最后收敛输出。跑通这一步你就能确认算法逻辑没问题再去接真实数据源。3.2 接 Prometheus 的查询与对齐接 Prometheus 最常见的方式是用 HTTP API 拉区间数据然后转成 DataFrame。注意step要和配置一致否则返回的点数和你预期不符。# aiops_min/prom_client.py import requests import pandas as pd from datetime import datetime, timedelta def query_range(endpoint, query, step60s, hours1): end datetime.now() start end - timedelta(hourshours) resp requests.get( f{endpoint}/api/v1/query_range, params{ query: query, start: start.timestamp(), end: end.timestamp(), step: step }, timeout10 ) resp.raise_for_status() data resp.json()[data][result] rows [] for series in data: entity series[metric].get(instance, unknown) for ts, val in series[values]: rows.append({ timestamp: pd.to_datetime(ts, units), entity: entity, metric: query, value: float(val) }) return pd.DataFrame(rows)逻辑说明query_range返回的是多条时间序列每条带自己的 labels这里用instance作为实体标识实际项目里可能要用pod或servicetimeout10是必须的Prometheus 查询慢的时候不设超时会挂死float(val)转换是因为 API 返回的是字符串。参数上hours1决定回看窗口做检测一般至少要看 1 小时做趋势分析要更长。3.3 日志数据接入的字段映射日志接入比指标麻烦因为格式不统一。基本包的做法是先做字段映射把不同来源的日志统一成timestamp、entity、level、message四列再走后续流程。常见做法是用正则或 Grok 提取下面是一个简化版import re import pandas as pd PATTERN re.compile( r(?Ptimestamp\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*? r(?PlevelERROR|WARN|INFO).*? r(?Pentity[\w\-]): (?Pmessage.) ) def parse_logs(lines): rows [] for line in lines: m PATTERN.match(line) if m: rows.append(m.groupdict()) df pd.DataFrame(rows) df[timestamp] pd.to_datetime(df[timestamp]) return df逻辑说明正则里的entity用[\w\-]匹配服务名实际项目里要按你的日志格式改解析失败的日志直接丢弃还是记录取决于你对完整性的要求我一般会单独存一份解析失败的样本方便回头调正则。日志接入后异常检测通常不是对message做而是对日志量、错误率这类聚合指标做这一点新手容易搞混。4. 避坑与排查基本包上线后最容易翻车的五件事4.1 现象检测结果全是异常满屏红原因contamination设得过大或者数据里本身有大量缺失值被fillna(0)填成了 0导致 Isolation Forest 把 0 值区域全判为异常。解决先把contamination降到 0.005 以下试同时检查缺失率缺失超过 30% 的实体直接跳过检测不要硬填。4.2 现象MAD 检测对缓慢漂移完全不报原因MAD 基于中位数对突变敏感但对持续几小时的缓慢上升不敏感因为中位数也跟着漂移了。解决对趋势型指标改用滑动窗口内的差分或 STL 分解后的残差做检测基本包里可以加一个method: diff_mad的分支先做一阶差分再算 MAD。4.3 现象告警收敛后漏掉了真正的根因告警原因收敛策略是“同实体窗口内只留最早一条”如果根因告警比症状告警晚几秒到达就会被丢掉。解决收敛时不要只按时间排序要结合告警级别和指标类型加权根因类指标如 CPU、内存、网络丢包优先级高于症状类如接口超时基本包里可以在converge_alerts前加一列priority再排序。4.4 现象接 Prometheus 后时间戳对不齐检测窗口错位原因Prometheus 返回的时间戳是 UTC本地 CSV 可能是本地时区混用后pd.Grouper分窗全乱。解决所有时间统一转成 UTC 再入库pd.to_datetime(ts, units, utcTrue)展示时再转本地时区。这个坑我踩过两次第二次是因为换了台机器默认时区不同血泪教训。4.5 现象依赖版本冲突昨天能跑今天报错原因requirements.txt里只写了包名没锁版本pip install拉到了不兼容的新版本。解决所有依赖锁死到具体版本并且用虚拟环境隔离。基本包里我一般会附一份requirements.lock用pip freeze生成部署时只装 lock 文件。5. 进阶把基本包做成可复用的检测流水线基本包跑通之后下一步不是加更多算法而是把它变成一条可复用的流水线。我的做法是把检测逻辑抽象成Detector基类不同方法实现fit和predict这样加新算法不用改主流程。# aiops_min/detectors.py from abc import ABC, abstractmethod import numpy as np import pandas as pd class Detector(ABC): abstractmethod def fit(self, series: pd.Series): pass abstractmethod def predict(self, series: pd.Series) - pd.Series: pass class MADDetector(Detector): def __init__(self, threshold3.5): self.threshold threshold self.median None self.mad None def fit(self, series): self.median series.median() self.mad np.median(np.abs(series - self.median)) def predict(self, series): if self.mad 0: return pd.Series([False] * len(series), indexseries.index) score 0.6745 * (series - self.median) / self.mad return score.abs() self.threshold逻辑说明fit和predict分离的好处是可以用历史数据 fit用新数据 predict避免每次检测都重算基线MADDetector把中位数和 MAD 存成实例属性支持增量更新。参数上threshold可以在配置里按指标类型分别设比如 CPU 用 3.0QPS 用 4.0因为不同指标的波动特性不一样。验证方法上我习惯用“回放”来测拿一周历史数据按时间顺序喂给流水线看每天检测出的告警数量和人工标注的故障时间点是否吻合。下面这张表是我常用的验证指标指标含义可接受范围召回率真实故障被检出的比例 0.8误报率正常时段被误报的比例 0.05平均检测延迟故障发生到告警的时间 2 个采集周期收敛比收敛后告警数 / 原始告警数0.1 ~ 0.3最后说一个具体技巧把每次检测的中间结果基线值、偏离分数、是否异常都落库不要只存最终告警。这样当有人质疑“为什么这条没报”时你能直接翻出当时的分数而不是靠猜。这个习惯帮我省了无数次扯皮。做 AIOps 基本包最怕的不是算法不够先进而是出了问题说不清楚。希望帮到你。本文还有配套的精品资源点击获取
