类别不平衡问题的系统性解决方案：安全视角下的稀有攻击检测实践

import numpy as np
from sklearn.neighbors import NearestNeighbors

class SMOTE:
    def __init__(self, k_neighbors=5, sampling_strategy='auto'):
        self.k_neighbors = k_neighbors
        self.sampling_strategy = sampling_strategy
        self.nn = None
    
    def fit_resample(self, X, y):
        """
        执行SMOTE算法，生成平衡数据集
        X: 特征矩阵，形状为(n_samples, n_features)
        y: 标签向量，形状为(n_samples,)
        返回: (X_resampled, y_resampled)，平衡后的特征矩阵和标签向量
        """
        # 确定少数类和多数类
        classes, counts = np.unique(y, return_counts=True)
        minority_class = classes[np.argmin(counts)]
        majority_class = classes[np.argmax(counts)]
        
        # 计算需要生成的少数类样本数量
        if self.sampling_strategy == 'auto':
            n_minority = counts[np.argmin(counts)]
            n_majority = counts[np.argmax(counts)]
            n_samples_to_generate = n_majority - n_minority
        else:
            n_samples_to_generate = int(self.sampling_strategy * len(X))
        
        # 获取少数类样本
        X_minority = X[y == minority_class]
        y_minority = y[y == minority_class]
        
        # 训练KNN模型
        self.nn = NearestNeighbors(n_neighbors=self.k_neighbors)
        self.nn.fit(X_minority)
        
        # 生成合成样本
        synthetic_samples = []
        for _ in range(n_samples_to_generate):
            # 随机选择一个少数类样本
            idx = np.random.randint(0, len(X_minority))
            sample = X_minority[idx]
            
            # 找到其K个最近邻
            distances, indices = self.nn.kneighbors([sample], n_neighbors=self.k_neighbors+1)
            # 排除样本自身
            indices = indices[0][1:]
            
            # 随机选择一个最近邻
            neighbor_idx = np.random.choice(indices)
            neighbor = X_minority[neighbor_idx]
            
            # 生成合成样本
            alpha = np.random.random()
            synthetic_sample = sample + alpha * (neighbor - sample)
            synthetic_samples.append(synthetic_sample)
        
        # 转换为numpy数组
        synthetic_samples = np.array(synthetic_samples)
        synthetic_labels = np.full(len(synthetic_samples), minority_class)
        
        # 合并原始数据和合成数据
        X_resampled = np.vstack([X, synthetic_samples])
        y_resampled = np.hstack([y, synthetic_labels])
        
        return X_resampled, y_resampled

# 示例用法
if __name__ == "__main__":
    # 生成不平衡数据
    from sklearn.datasets import make_classification
    X, y = make_classification(n_samples=1000, n_features=20, n_informative=15, 
                               n_redundant=5, n_classes=2, weights=[0.9, 0.1], 
                               random_state=42)
    
    print(f"原始数据形状: X={X.shape}, y={y.shape}")
    print(f"原始数据类别分布: {np.bincount(y)}")
    
    # 使用SMOTE生成平衡数据
    smote = SMOTE(k_neighbors=5, sampling_strategy='auto')
    X_resampled, y_resampled = smote.fit_resample(X, y)
    
    print(f"平衡数据形状: X={X_resampled.shape}, y={y_resampled.shape}")
    print(f"平衡数据类别分布: {np.bincount(y_resampled)}")

import torch
import torch.nn as nn
import torch.nn.functional as F

class WeightedCrossEntropyLoss(nn.Module):
    """
    加权交叉熵损失
    weight: 类别权重，形状为(n_classes,)
    """
    def __init__(self, weight=None):
        super(WeightedCrossEntropyLoss, self).__init__()
        self.weight = weight
    
    def forward(self, input, target):
        return F.cross_entropy(input, target, weight=self.weight)

class FocalLoss(nn.Module):
    """
    Focal Loss
    alpha: 类别权重，形状为(n_classes,)
    gamma: 调制因子，默认2
    """
    def __init__(self, alpha=None, gamma=2.0, reduction='mean'):
        super(FocalLoss, self).__init__()
        self.alpha = alpha
        self.gamma = gamma
        self.reduction = reduction
    
    def forward(self, input, target):
        # 计算交叉熵损失
        ce_loss = F.cross_entropy(input, target, weight=self.alpha, reduction='none')
        # 计算预测概率
        p_t = torch.exp(-ce_loss)
        # 计算Focal Loss
        loss = (1 - p_t) ** self.gamma * ce_loss
        
        # 应用reduction
        if self.reduction == 'mean':
            return loss.mean()
        elif self.reduction == 'sum':
            return loss.sum()
        else:
            return loss

# 示例用法
if __name__ == "__main__":
    # 生成示例数据
    input = torch.randn(10, 2, requires_grad=True)
    target = torch.randint(0, 2, (10,))
    
    print(f"输入形状: {input.shape}")
    print(f"目标标签: {target}")
    
    # 计算普通交叉熵损失
    ce_loss = F.cross_entropy(input, target)
    print(f"普通交叉熵损失: {ce_loss.item()}")
    
    # 计算加权交叉熵损失
    # 假设类别0为多数类，类别1为少数类，分配更高权重
    weight = torch.tensor([1.0, 5.0])
    weighted_ce_loss = WeightedCrossEntropyLoss(weight=weight)
    loss = weighted_ce_loss(input, target)
    print(f"加权交叉熵损失: {loss.item()}")
    
    # 计算Focal Loss
    focal_loss = FocalLoss(alpha=weight, gamma=2.0)
    loss = focal_loss(input, target)
    print(f"Focal Loss: {loss.item()}")

import torch
import torch.nn as nn
import torch.nn.functional as F

class ChannelAttention(nn.Module):
    """
    通道注意力模块
    in_channels: 输入通道数
    ratio: 压缩比率，默认16
    """
    def __init__(self, in_channels, ratio=16):
        super(ChannelAttention, self).__init__()
        # 全局平均池化
        self.avg_pool = nn.AdaptiveAvgPool2d(1)
        # 全局最大池化
        self.max_pool = nn.AdaptiveMaxPool2d(1)
        # 共享MLP
        self.mlp = nn.Sequential(
            nn.Conv2d(in_channels, in_channels // ratio, 1, bias=False),
            nn.ReLU(),
            nn.Conv2d(in_channels // ratio, in_channels, 1, bias=False)
        )
        # Sigmoid激活
        self.sigmoid = nn.Sigmoid()
    
    def forward(self, x):
        # 计算平均池化特征
        avg_out = self.mlp(self.avg_pool(x))
        # 计算最大池化特征
        max_out = self.mlp(self.max_pool(x))
        # 相加并经过Sigmoid激活
        out = self.sigmoid(avg_out + max_out)
        # 与输入特征相乘
        return x * out

class ResNetBlock(nn.Module):
    """
    带有通道注意力的ResNet块
    in_channels: 输入通道数
    out_channels: 输出通道数
    stride: 步长，默认1
    """
    def __init__(self, in_channels, out_channels, stride=1):
        super(ResNetBlock, self).__init__()
        # 主分支
        self.conv1 = nn.Conv2d(in_channels, out_channels, kernel_size=3, stride=stride, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(out_channels)
        self.relu = nn.ReLU(inplace=True)
        self.conv2 = nn.Conv2d(out_channels, out_channels, kernel_size=3, stride=1, padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(out_channels)
        
        # 通道注意力
        self.ca = ChannelAttention(out_channels)
        
        # 短路连接
        self.shortcut = nn.Sequential()
        if stride != 1 or in_channels != out_channels:
            self.shortcut = nn.Sequential(
                nn.Conv2d(in_channels, out_channels, kernel_size=1, stride=stride, bias=False),
                nn.BatchNorm2d(out_channels)
            )
    
    def forward(self, x):
        out = self.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        out = self.ca(out)  # 应用通道注意力
        out += self.shortcut(x)
        out = self.relu(out)
        return out

# 示例用法
if __name__ == "__main__":
    # 生成示例数据
    x = torch.randn(1, 64, 32, 32)
    
    print(f"输入形状: {x.shape}")
    
    # 创建通道注意力模块
    ca = ChannelAttention(in_channels=64)
    out = ca(x)
    print(f"通道注意力输出形状: {out.shape}")
    
    # 创建带有通道注意力的ResNet块
    res_block = ResNetBlock(in_channels=64, out_channels=128, stride=2)
    out = res_block(x)
    print(f"ResNet块输出形状: {out.shape}")

# 安装必要的依赖
pip install numpy pandas scikit-learn torch torchvision matplotlib seaborn

# 安装不平衡学习库
pip install imbalanced-learn

# 安装对抗训练库
pip install robustbench

from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier, AdaBoostClassifier, GradientBoostingClassifier
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score
from imblearn.ensemble import BalancedRandomForestClassifier, EasyEnsembleClassifier

# 生成不平衡数据
X, y = make_classification(n_samples=10000, n_features=20, n_informative=15, 
                           n_redundant=5, n_classes=2, weights=[0.99, 0.01], 
                           random_state=42)

# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)

print(f"训练集类别分布: {dict(zip(*np.unique(y_train, return_counts=True)))}")
print(f"测试集类别分布: {dict(zip(*np.unique(y_test, return_counts=True)))}")

# 定义模型列表
models = {
    "Random Forest": RandomForestClassifier(random_state=42),
    "Balanced Random Forest": BalancedRandomForestClassifier(random_state=42),
    "AdaBoost": AdaBoostClassifier(random_state=42),
    "Gradient Boosting": GradientBoostingClassifier(random_state=42),
    "EasyEnsemble": EasyEnsembleClassifier(random_state=42)
}

# 训练和评估模型
results = {}
for name, model in models.items():
    print(f"\n训练模型: {name}")
    model.fit(X_train, y_train)
    
    # 预测
    y_pred = model.predict(X_test)
    y_pred_proba = model.predict_proba(X_test)[:, 1]
    
    # 评估
    report = classification_report(y_test, y_pred, target_names=['正常', '攻击'])
    auc = roc_auc_score(y_test, y_pred_proba)
    cm = confusion_matrix(y_test, y_pred)
    
    results[name] = {
        'report': report,
        'auc': auc,
        'confusion_matrix': cm
    }
    
    print(f"AUC: {auc:.4f}")
    print("分类报告:")
    print(report)
    print("混淆矩阵:")
    print(cm)

# 比较不同模型的AUC
print("\n各模型AUC比较:")
for name, result in results.items():
    print(f"{name}: {result['auc']:.4f}")