从静态阈值到智能体自愈：AIOps 的技术内核与工程落地

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

np.random.seed(42)
n = 600

# 模拟 CPU 时序：带周期性波动 + 一段持续异常 + 若干尖峰
t = np.arange(n)
cpu = 40 + 15 * np.sin(2 * np.pi * t / 144) + np.random.normal(0, 4, n)
cpu[350:380] += 35          # 持续异常区间
for idx in [80, 200, 500]:
    cpu[idx] = 92           # 尖峰噪声

df = pd.DataFrame({"timestamp": pd.date_range("2026-01-01", periods=n, freq="5min"),
                   "cpu": np.clip(cpu, 0, 100)})

# 滑动窗口特征：均值、标准差、环比变化率
window = 12
feat = pd.DataFrame({"timestamp": df["timestamp"], "cpu": df["cpu"]})
feat["cpu_mean"] = df["cpu"].rolling(window, min_periods=window).mean()
feat["cpu_std"]  = df["cpu"].rolling(window, min_periods=window).std()
feat["cpu_pct"]  = df["cpu"].pct_change(periods=window).clip(-5, 5)
feat = feat.dropna().reset_index(drop=True)

# 训练孤立森林（无监督，无需标注异常）
X = feat[["cpu", "cpu_mean", "cpu_std", "cpu_pct"]].values
scaler = StandardScaler().fit(X)
model = IsolationForest(contamination=0.08, random_state=42)
model.fit(scaler.transform(X))

feat["score"] = model.decision_function(scaler.transform(X))
feat["flag"]  = model.predict(scaler.transform(X))  # -1 为异常

# 持续性过滤：连续 3 个窗口异常才触发，抑制瞬时尖峰
feat["sustained"] = (feat["flag"] == -1).rolling(3).sum() >= 3

def auto_repair(row):
    print(f"[AIOps 自愈] {row['timestamp']} 持续异常，CPU 平滑值 {row['cpu_mean']:.1f}% "
          f"→ 触发扩容/缓存清理")

for _, row in feat[feat["sustained"]].iterrows():
    auto_repair(row)

print(f"\n检出异常窗口数（过滤前）: {(feat['flag'] == -1).sum()}")
print(f"持续异常窗口数（过滤后）: {feat['sustained'].sum()}")