3：L的无监督异常检测：蓝队的未知威胁猎手

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

# 准备网络流量数据
# 假设我们有正常流量和异常流量数据
np.random.seed(42)
# 生成正常流量数据
normal_data = np.random.normal(0, 1, (1000, 5))
# 生成异常流量数据
anomaly_data = np.random.normal(5, 1, (100, 5))
# 合并数据
X = np.vstack([normal_data, anomaly_data])

# 数据标准化
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# 训练Isolation Forest模型
model = IsolationForest(contamination=0.1, random_state=42)
model.fit(X_scaled)

# 预测异常
predictions = model.predict(X_scaled)
# 计算异常分数
scores = model.score_samples(X_scaled)

# 分析结果
print("异常检测结果:")
print(f"正常样本预测为正常: {np.sum((predictions[:1000] == 1))}")
print(f"异常样本预测为异常: {np.sum((predictions[1000:] == -1))}")

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense

# 准备数据
np.random.seed(42)
# 生成正常流量数据
normal_data = np.random.normal(0, 1, (1000, 10))
# 生成异常流量数据
anomaly_data = np.random.normal(3, 1, (100, 10))

# 构建自编码器模型
input_dim = 10
encoding_dim = 5

inputs = Input(shape=(input_dim,))
encoded = Dense(encoding_dim, activation='relu')(inputs)
decoded = Dense(input_dim, activation='sigmoid')(encoded)

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

# 训练自编码器（只使用正常数据）
autoencoder.fit(normal_data, normal_data, epochs=50, batch_size=32, shuffle=True, validation_split=0.2)

# 计算重构误差
normal_reconstruction_error = np.mean(np.square(autoencoder.predict(normal_data) - normal_data), axis=1)
anomaly_reconstruction_error = np.mean(np.square(autoencoder.predict(anomaly_data) - anomaly_data), axis=1)

# 设置阈值
threshold = np.percentile(normal_reconstruction_error, 95)
print(f"异常检测阈值: {threshold}")

# 检测异常
normal_pred = normal_reconstruction_error > threshold
anomaly_pred = anomaly_reconstruction_error > threshold

print(f"正常样本被误判为异常: {np.sum(normal_pred)}")
print(f"异常样本被正确检测: {np.sum(anomaly_pred)}")

import numpy as np
from sklearn.ensemble import IsolationForest

class DynamicBaseline:
    def __init__(self, window_size=1000, contamination=0.1):
        self.window_size = window_size
        self.contamination = contamination
        self.data_window = []
        self.model = None
    
    def add_data(self, new_data):
        """添加新数据并更新基线"""
        self.data_window.extend(new_data)
        # 保持窗口大小
        if len(self.data_window) > self.window_size:
            self.data_window = self.data_window[-self.window_size:]
        # 更新模型
        self.update_model()
    
    def update_model(self):
        """更新异常检测模型"""
        if len(self.data_window) >= 100:  # 确保有足够的数据
            self.model = IsolationForest(contamination=self.contamination, random_state=42)
            self.model.fit(self.data_window)
    
    def detect_anomalies(self, data):
        """检测异常"""
        if self.model is None:
            return np.zeros(len(data), dtype=bool)
        predictions = self.model.predict(data)
        return predictions == -1

# 测试动态基线
np.random.seed(42)
db = DynamicBaseline(window_size=1000, contamination=0.1)

# 模拟流式数据
for i in range(10):
    # 生成正常数据
    normal_batch = np.random.normal(0, 1, (100, 5))
    # 每隔5批次添加一些异常数据
    if i % 5 == 4:
        anomaly_batch = np.random.normal(5, 1, (10, 5))
        batch = np.vstack([normal_batch, anomaly_batch])
    else:
        batch = normal_batch
    
    # 添加数据并更新基线
    db.add_data(batch.tolist())
    
    # 检测异常
    anomalies = db.detect_anomalies(batch)
    print(f"批次 {i+1}: 检测到 {np.sum(anomalies)} 个异常")