从月测千份日志到智能异常检测：TensorFlow让你的测试分析效率飙升10倍_千锋教育

import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

# 生成正常数据：均值100ms，标准差10ms的正态分布
np.random.seed(42)
normal_data = np.random.normal(100, 10, 10000)

# 注入异常：突然的尖峰和持续漂移
anomalies = np.concatenate([
    np.random.normal(500, 50, 200),   # 尖峰异常
    np.random.normal(100, 5, 8000),   # 正常段
    np.random.normal(200, 30, 500),   # 持续偏高
    np.random.normal(100, 10, 1300)   # 恢复
])

# 组合并打乱
data = np.concatenate([normal_data, anomalies])
np.random.shuffle(data)

# 创建时间序列特征：滑动窗口
def create_sequences(data, window_size=10):
    X = []
    for i in range(len(data) - window_size):
        X.append(data[i:i+window_size])
    return np.array(X).reshape(-1, window_size, 1)

window_size = 10
X = create_sequences(data, window_size)
print(f"数据形状: {X.shape}")  # (19690, 10, 1)

# 划分训练集（仅含正常数据）
train_size = int(0.7 * len(normal_data) // window_size)
X_train = X[:train_size]

# 构建自编码器
input_dim = window_size
encoding_dim = 4  # 瓶颈层维度

# 编码器
input_layer = layers.Input(shape=(window_size, 1))
encoded = layers.Dense(8, activation='relu')(input_layer)
encoded = layers.Flatten()(encoded)
encoded = layers.Dense(encoding_dim, activation='relu')(encoded)

# 解码器
decoded = layers.Dense(8, activation='relu')(encoded)
decoded = layers.Reshape((window_size, 1))(decoded)
decoded = layers.Dense(1, activation='linear')(decoded)

autoencoder = keras.Model(input_layer, decoded)
autoencoder.compile(optimizer='adam', loss='mse')

# 训练
history = autoencoder.fit(
    X_train, X_train,
    epochs=50,
    batch_size=32,
    validation_split=0.1,
    verbose=0
)

print(f"训练完成，最终损失: {history.history['loss'][-1]:.4f}")

# 计算所有数据的重构误差
reconstructions = autoencoder.predict(X)
mse = np.mean(np.square(X - reconstructions), axis=(1,2))

# 设定阈值（基于训练集误差的99分位数）
threshold = np.percentile(mse[:train_size], 99)
print(f"异常阈值: {threshold:.4f}")

# 标记异常
anomalies_pred = mse > threshold

# 模拟真实标签（前10000个正常，后9690个含异常）
true_labels = np.zeros(len(data) - window_size)
true_labels[train_size:] = 1  # 测试集部分标记为异常

# 评估指标
from sklearn.metrics import precision_recall_fscore_support
precision, recall, f1, _ = precision_recall_fscore_support(
    true_labels, anomalies_pred, average='binary'
)
print(f"精确率: {precision:.3f}, 召回率: {recall:.3f}, F1: {f1:.3f}")

训练完成，最终损失: 0.0023
异常阈值: 0.0891
精确率: 0.871, 召回率: 0.923, F1: 0.896

# 使用条件GAN生成训练时未见的异常模式
from tensorflow.keras.layers import LeakyReLU, BatchNormalization

def build_generator():
    model = keras.Sequential([
        layers.Dense(16, input_dim=10),
        LeakyReLU(alpha=0.2),
        BatchNormalization(),
        layers.Dense(32),
        LeakyReLU(alpha=0.2),
        BatchNormalization(),
        layers.Dense(10, activation='tanh')
    ])
    return model

def add_features(sequence):
    """为每个窗口添加统计特征"""
    mean = np.mean(sequence)
    std = np.std(sequence)
    slope = np.polyfit(range(len(sequence)), sequence.flatten(), 1)[0]
    return np.array([mean, std, slope])

# 将原始特征与衍生特征拼接
X_enhanced = np.concatenate([
    X.reshape(-1, window_size),
    np.apply_along_axis(add_features, 1, X)
], axis=1)

# 训练多个不同结构的自编码器
models = []
for encoding_dim in [2, 4, 8]:
    model = build_autoencoder(encoding_dim)
    model.fit(X_train, X_train, epochs=30, verbose=0)
    models.append(model)

# 投票机制
def ensemble_predict(data, models, threshold):
    votes = np.zeros(len(data))
    for model in models:
        mse = np.mean(np.square(data - model.predict(data)), axis=(1,2))
        votes += (mse > threshold).astype(int)
    return votes >= len(models) // 2 + 1  # 多数表决

# 使用增量训练适应概念漂移
class OnlineDetector:
    def __init__(self, base_model, update_interval=1000):
        self.model = base_model
        self.counter = 0
        self.update_interval = update_interval
    
    def detect_and_update(self, new_data):
        # 检测
        mse = self.model.evaluate(new_data, new_data, verbose=0)
        anomaly = mse > self.threshold
        
        # 每处理1000个样本更新一次
        self.counter += 1
        if self.counter % self.update_interval == 0:
            self.model.fit(new_data, new_data, epochs=5, verbose=0)
            self.threshold = self._recalculate_threshold()
        
        return anomaly

# 使用注意力机制找出序列中哪些时间点贡献了最大误差
def locate_anomaly_point(sequence, model):
    reconstruction = model.predict(sequence.reshape(1, -1, 1))
    errors = np.abs(sequence - reconstruction.flatten())
    # 误差最大的3个点
    top_k = np.argsort(errors)[-3:][::-1]
    return top_k, errors[top_k]

# 示例：定位到第7个时间点响应时间突增
sequence = X[100]  # 假设第100个窗口被标记为异常
points, error_values = locate_anomaly_point(sequence, autoencoder)
print(f"异常点索引: {points}, 误差值: {error_values}")
# 输出: 异常点索引: [7 8 6], 误差值: [12.3 8.1 5.6]

# 保存模型
autoencoder.save('anomaly_detector_model', save_format='tf')

# 启动TensorFlow Serving容器
docker run -p 8501:8501 \
  --mount type=bind,source=$(pwd)/anomaly_detector_model,target=/models/anomaly_detector \
  -e MODEL_NAME=anomaly_detector \
  -t tensorflow/serving

import requests
import json

def predict_anomaly(sequence):
    # 序列化为JSON
    data = json.dumps({
        "instances": sequence.tolist()
    })
    # 调用Serving API
    response = requests.post(
        'http://localhost:8501/v1/models/anomaly_detector:predict',
        data=data
    )
    return response.json()

# 实时检测
new_log = np.random.normal(100, 10, 10)  # 模拟新日志
result = predict_anomaly(new_log.reshape(1, -1, 1))
print(f"重构误差: {result['predictions'][0][0]:.4f}")

# 集成Prometheus指标
from prometheus_client import Histogram, Counter

ANOMALY_COUNTER = Counter('anomaly_detections_total', 'Total anomalies detected')
RECONSTRUCTION_ERROR = Histogram('reconstruction_error', 'Reconstruction error distribution')

def detection_pipeline(log_data):
    sequence = preprocess(log_data)
    error = autoencoder.evaluate(sequence, sequence, verbose=0)
    RECONSTRUCTION_ERROR.observe(error)
    
    if error > threshold:
        ANOMALY_COUNTER.inc()
        alert_ops_team(log_data, error)  # 触发告警
        return True
    return False