半监督学习的工程落地方式：安全攻防中的标签稀缺解决方案

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Dropout
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score

# 1. 数据准备
# 假设我们有少量标注数据和大量未标注数据
X_labeled = np.load('labeled_malware_features.npy')
y_labeled = np.load('labeled_malware_labels.npy')
X_unlabeled = np.load('unlabeled_malware_features.npy')

# 划分训练集和验证集
X_train, X_val, y_train, y_val = train_test_split(X_labeled, y_labeled, test_size=0.2, random_state=42)

# 2. 构建初始模型
def build_model(input_dim):
    model = Sequential([
        Dense(128, activation='relu', input_shape=(input_dim,)),
        Dropout(0.3),
        Dense(64, activation='relu'),
        Dropout(0.3),
        Dense(32, activation='relu'),
        Dense(1, activation='sigmoid')
    ])
    model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
    return model

# 3. 伪标签生成与半监督训练
def semi_supervised_training(X_labeled, y_labeled, X_unlabeled, epochs=10, confidence_threshold=0.95):
    input_dim = X_labeled.shape[1]
    
    # 初始模型训练
    model = build_model(input_dim)
    model.fit(X_labeled, y_labeled, epochs=epochs, batch_size=32, validation_split=0.1, verbose=1)
    
    # 迭代伪标签生成
    for iteration in range(5):
        print(f"\nIteration {iteration+1}:")
        
        # 对未标注数据进行预测
        y_pred_proba = model.predict(X_unlabeled, verbose=0)
        
        # 生成伪标签
        confidence = np.maximum(y_pred_proba, 1 - y_pred_proba).flatten()
        high_confidence_idx = np.where(confidence >= confidence_threshold)[0]
        
        if len(high_confidence_idx) == 0:
            print("No high-confidence predictions found. Stopping iteration.")
            break
        
        # 提取高置信度的伪标签数据
        X_pseudo = X_unlabeled[high_confidence_idx]
        y_pseudo = (y_pred_proba[high_confidence_idx] > 0.5).astype(int).flatten()
        
        print(f"Found {len(high_confidence_idx)} high-confidence samples.")
        print(f"Pseudo-label distribution: Malicious: {np.sum(y_pseudo)}, Benign: {len(y_pseudo) - np.sum(y_pseudo)}")
        
        # 合并标注数据和伪标签数据
        X_combined = np.concatenate([X_labeled, X_pseudo])
        y_combined = np.concatenate([y_labeled, y_pseudo])
        
        # 重新训练模型
        model = build_model(input_dim)
        model.fit(X_combined, y_combined, epochs=epochs, batch_size=32, validation_split=0.1, verbose=1)
    
    return model

# 4. 模型训练
model = semi_supervised_training(X_train, y_train, X_unlabeled, epochs=10, confidence_threshold=0.95)

# 5. 模型评估
y_val_pred = (model.predict(X_val) > 0.5).astype(int).flatten()
accuracy = accuracy_score(y_val, y_val_pred)
f1 = f1_score(y_val, y_val_pred)

print(f"\nValidation Accuracy: {accuracy:.4f}")
print(f"Validation F1 Score: {f1:.4f}")

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense, Dropout
from tensorflow.keras.optimizers import Adam
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report

# 1. 数据准备
X_labeled = np.load('labeled_network_traffic.npy')
y_labeled = np.load('labeled_network_labels.npy')
X_unlabeled = np.load('unlabeled_network_traffic.npy')

# 划分训练集和验证集
X_train, X_val, y_train, y_val = train_test_split(X_labeled, y_labeled, test_size=0.2, random_state=42)

# 2. 数据增强函数（用于一致性正则化）
def augment_data(X, noise_level=0.01):
    # 添加高斯噪声
    noise = np.random.normal(0, noise_level, X.shape)
    return X + noise

# 3. 构建半监督模型（带一致性正则化）
def build_semi_supervised_model(input_dim, num_classes):
    # 共享编码器
    inputs = Input(shape=(input_dim,))
    x = Dense(256, activation='relu')(inputs)
    x = Dropout(0.3)(x)
    x = Dense(128, activation='relu')(x)
    x = Dropout(0.3)(x)
    x = Dense(64, activation='relu')(x)
    encoder = Model(inputs, x, name='encoder')
    
    # 分类器
    classifier_inputs = Input(shape=(64,))
    x = Dense(32, activation='relu')(classifier_inputs)
    outputs = Dense(num_classes, activation='softmax')(x)
    classifier = Model(classifier_inputs, outputs, name='classifier')
    
    # 完整模型
    encoded = encoder(inputs)
    outputs = classifier(encoded)
    model = Model(inputs, outputs, name='semi_supervised_model')
    
    return model, encoder, classifier

# 4. 一致性正则化损失函数
def consistency_loss(y_pred1, y_pred2, temperature=0.5):
    # 温度缩放
    y_pred1 = tf.nn.softmax(y_pred1 / temperature)
    y_pred2 = tf.nn.softmax(y_pred2 / temperature)
    
    # KL散度作为一致性损失
    loss = tf.keras.losses.KLDivergence()(y_pred1, y_pred2)
    return loss

# 5. 半监督训练
def train_semi_supervised(model, encoder, classifier, X_labeled, y_labeled, X_unlabeled, 
                         epochs=20, batch_size=64, consistency_weight=1.0, temperature=0.5):
    
    optimizer = Adam(learning_rate=0.001)
    classification_loss_fn = tf.keras.losses.SparseCategoricalCrossentropy()
    
    # 训练循环
    for epoch in range(epochs):
        print(f"Epoch {epoch+1}/{epochs}")
        
        # 随机打乱数据
        indices = np.random.permutation(len(X_labeled))
        X_labeled_shuffled = X_labeled[indices]
        y_labeled_shuffled = y_labeled[indices]
        
        # 计算批次数量
        num_batches = len(X_labeled) // batch_size
        
        for batch_idx in range(num_batches):
            # 获取标注数据批次
            start_idx = batch_idx * batch_size
            end_idx = start_idx + batch_size
            X_batch_labeled = X_labeled_shuffled[start_idx:end_idx]
            y_batch_labeled = y_labeled_shuffled[start_idx:end_idx]
            
            # 获取未标注数据批次（与标注数据批次大小相同）
            X_batch_unlabeled = X_unlabeled[np.random.choice(len(X_unlabeled), batch_size)]
            
            with tf.GradientTape() as tape:
                # 标注数据的分类损失
                y_pred_labeled = model(X_batch_labeled, training=True)
                cls_loss = classification_loss_fn(y_batch_labeled, y_pred_labeled)
                
                # 未标注数据的一致性损失
                # 对同一未标注数据进行两次不同的增强
                X_aug1 = augment_data(X_batch_unlabeled)
                X_aug2 = augment_data(X_batch_unlabeled)
                
                y_pred1 = model(X_aug1, training=True)
                y_pred2 = model(X_aug2, training=True)
                
                consis_loss = consistency_loss(y_pred1, y_pred2, temperature)
                
                # 总损失
                total_loss = cls_loss + consistency_weight * consis_loss
            
            # 反向传播和优化
            gradients = tape.gradient(total_loss, model.trainable_variables)
            optimizer.apply_gradients(zip(gradients, model.trainable_variables))
        
        # 每个epoch结束后评估模型
        val_loss, val_acc = model.evaluate(X_val, y_val, verbose=0)
        print(f"Validation Loss: {val_loss:.4f}, Validation Accuracy: {val_acc:.4f}")
    
    return model

# 6. 模型训练与评估
num_classes = len(np.unique(y_labeled))
model, encoder, classifier = build_semi_supervised_model(X_labeled.shape[1], num_classes)
model = train_semi_supervised(model, encoder, classifier, X_train, y_train, X_unlabeled, epochs=20, batch_size=64)

# 模型评估
y_val_pred = np.argmax(model.predict(X_val), axis=1)
print("\nClassification Report:")
print(classification_report(y_val, y_val_pred))

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense, Dropout
from sklearn.metrics import roc_auc_score, precision_recall_curve
import matplotlib.pyplot as plt

# 1. 数据准备
# 假设我们有少量正常样本（作为标注数据）和大量未标注样本
X_normal = np.load('normal_traffic_features.npy')  # 少量正常样本
X_unlabeled = np.load('unlabeled_traffic_features.npy')  # 大量未标注样本
X_test = np.load('test_traffic_features.npy')  # 测试集
 y_test = np.load('test_traffic_labels.npy')  # 测试集标签（0:正常, 1:异常）

# 2. 构建半监督自编码器
def build_semi_supervised_autoencoder(input_dim):
    # 编码器
    inputs = Input(shape=(input_dim,))
    x = Dense(128, activation='relu')(inputs)
    x = Dropout(0.2)(x)
    x = Dense(64, activation='relu')(x)
    x = Dropout(0.2)(x)
    latent = Dense(32, activation='relu')(x)
    encoder = Model(inputs, latent, name='encoder')
    
    # 解码器
    latent_inputs = Input(shape=(32,))
    x = Dense(64, activation='relu')(latent_inputs)
    x = Dropout(0.2)(x)
    x = Dense(128, activation='relu')(x)
    x = Dropout(0.2)(x)
    outputs = Dense(input_dim, activation='linear')(x)
    decoder = Model(latent_inputs, outputs, name='decoder')
    
    # 完整自编码器
    encoded = encoder(inputs)
    decoded = decoder(encoded)
    autoencoder = Model(inputs, decoded, name='autoencoder')
    
    # 异常检测模型（基于重构误差）
    def anomaly_score(X):
        X_reconstructed = autoencoder.predict(X)
        return np.mean(np.square(X - X_reconstructed), axis=1)
    
    return autoencoder, encoder, decoder, anomaly_score

# 3. 半监督训练
# 第一步：使用少量正常样本训练自编码器
autoencoder, encoder, decoder, anomaly_score = build_semi_supervised_autoencoder(X_normal.shape[1])
autoencoder.compile(optimizer='adam', loss='mse')
autoencoder.fit(X_normal, X_normal, epochs=20, batch_size=32, validation_split=0.1, verbose=1)

# 第二步：使用未标注数据进行微调（基于熵最小化）
# 计算未标注数据的初始重构误差
initial_recon_error = anomaly_score(X_unlabeled)

# 选择重构误差较小的样本（假设为正常样本）进行微调
low_error_idx = np.argsort(initial_recon_error)[:int(len(X_unlabeled) * 0.3)]  # 选择30%重构误差最小的样本
X_fine_tune = X_unlabeled[low_error_idx]

# 微调自编码器
autoencoder.fit(X_fine_tune, X_fine_tune, epochs=10, batch_size=32, verbose=1)

# 4. 模型评估
# 计算测试集的异常分数
test_anomaly_scores = anomaly_score(X_test)

# 计算ROC-AUC分数
roc_auc = roc_auc_score(y_test, test_anomaly_scores)
print(f"ROC-AUC Score: {roc_auc:.4f}")

# 计算精确率-召回率曲线
precision, recall, thresholds = precision_recall_curve(y_test, test_anomaly_scores)

# 找到最佳阈值
f1_scores = 2 * (precision * recall) / (precision + recall)
best_threshold = thresholds[np.argmax(f1_scores)]
best_f1 = np.max(f1_scores)

print(f"Best F1 Score: {best_f1:.4f} at threshold: {best_threshold:.4f}")

# 5. 异常检测结果
# 使用最佳阈值进行异常检测
y_pred = (test_anomaly_scores >= best_threshold).astype(int)

# 计算检测指标
accuracy = np.mean(y_pred == y_test)
precision_best = precision[np.argmax(f1_scores)]
recall_best = recall[np.argmax(f1_scores)]

print(f"\nAnomaly Detection Results:")
print(f"Accuracy: {accuracy:.4f}")
print(f"Precision: {precision_best:.4f}")
print(f"Recall: {recall_best:.4f}")
print(f"F1 Score: {best_f1:.4f}")