表示学习为什么比分类更重要：安全攻防中的通用特征学习范式

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_blobs
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import IsolationForest

# 生成示例数据
X_normal, _ = make_blobs(n_samples=1000, centers=3, n_features=20, random_state=42)
X_anomaly, _ = make_blobs(n_samples=100, centers=1, n_features=20, random_state=24)

# 数据标准化
scaler = StandardScaler()
X_normal_scaled = scaler.fit_transform(X_normal)
X_anomaly_scaled = scaler.transform(X_anomaly)

# 合并正常数据和异常数据
test_data = np.vstack([X_normal_scaled[:300], X_anomaly_scaled])
y_test = np.array([0] * 300 + [1] * 100)  # 0: 正常, 1: 异常

# 1. 传统异常检测方法：Isolation Forest
ios = IsolationForest(contamination=0.25, random_state=42)
ios.fit(X_normal_scaled)
y_pred_iso = (ios.predict(test_data) == -1).astype(int)

# 2. 自监督表示学习方法：主成分分析（PCA）作为简单的表示学习
from sklearn.decomposition import PCA

# 使用PCA学习表示
pca = PCA(n_components=10, random_state=42)
pca.fit(X_normal_scaled)

# 计算重构误差作为异常分数
X_normal_reconstructed = pca.inverse_transform(pca.transform(X_normal_scaled))
train_reconstruction_error = np.sum((X_normal_scaled - X_normal_reconstructed) ** 2, axis=1)

# 设置异常阈值
threshold = np.percentile(train_reconstruction_error, 95)

# 在测试集上进行异常检测
test_reconstruction_error = np.sum((test_data - pca.inverse_transform(pca.transform(test_data))) ** 2, axis=1)
y_pred_pca = (test_reconstruction_error > threshold).astype(int)

# 3. 评估检测效果
from sklearn.metrics import accuracy_score, f1_score, confusion_matrix

accuracy_iso = accuracy_score(y_test, y_pred_iso)
f1_iso = f1_score(y_test, y_pred_iso)

accuracy_pca = accuracy_score(y_test, y_pred_pca)
f1_pca = f1_score(y_test, y_pred_pca)

print("传统Isolation Forest检测效果:")
print(f"准确率: {accuracy_iso:.4f}")
print(f"F1分数: {f1_iso:.4f}")
print("混淆矩阵:")
print(confusion_matrix(y_test, y_pred_iso))

print("\nPCA表示学习检测效果:")
print(f"准确率: {accuracy_pca:.4f}")
print(f"F1分数: {f1_pca:.4f}")
print("混淆矩阵:")
print(confusion_matrix(y_test, y_pred_pca))

# 可视化结果
plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)
plt.hist(train_reconstruction_error, bins=50, alpha=0.8, color='blue')
plt.axvline(threshold, color='red', linestyle='--', label=f'Threshold: {threshold:.4f}')
plt.title('正常数据重构误差分布')
plt.xlabel('重构误差')
plt.ylabel('样本数')
plt.legend()
plt.grid(True)

plt.subplot(1, 2, 2)
plt.hist(test_reconstruction_error[y_test == 0], bins=50, alpha=0.8, color='blue', label='正常样本')
plt.hist(test_reconstruction_error[y_test == 1], bins=50, alpha=0.8, color='red', label='异常样本')
plt.axvline(threshold, color='green', linestyle='--', label=f'Threshold: {threshold:.4f}')
plt.title('测试集重构误差分布')
plt.xlabel('重构误差')
plt.ylabel('样本数')
plt.legend()
plt.grid(True)

plt.tight_layout()
plt.show()

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_moons
from sklearn.preprocessing import StandardScaler
from sklearn.neighbors import KNeighborsClassifier

# 生成示例数据
X, y = make_moons(n_samples=1000, noise=0.1, random_state=42)

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

# 1. 传统KNN分类
knn = KNeighborsClassifier(n_neighbors=5)
knn.fit(X_scaled, y)
y_pred_knn = knn.predict(X_scaled)

# 2. 对比学习：简单的负采样对比损失
class ContrastiveLearning:
    def __init__(self, input_dim, embedding_dim, learning_rate=0.01, epochs=1000):
        self.input_dim = input_dim
        self.embedding_dim = embedding_dim
        self.learning_rate = learning_rate
        self.epochs = epochs
        self.W = np.random.randn(input_dim, embedding_dim) * 0.1
    
    def encode(self, x):
        return np.dot(x, self.W)
    
    def contrastive_loss(self, x_pos, x_neg, margin=1.0):
        # 简单的负采样对比损失
        z_pos = self.encode(x_pos)
        z_neg = self.encode(x_neg)
        
        # 计算相似性
        sim_pos = np.sum(z_pos * z_pos, axis=1)
        sim_neg = np.sum(z_pos * z_neg, axis=1)
        
        # 对比损失
        loss = np.mean(np.maximum(0, margin - sim_pos + sim_neg))
        return loss
    
    def train(self, X, y):
        loss_history = []
        
        for epoch in range(self.epochs):
            # 生成正负样本对
            batch_size = 128
            idx = np.random.choice(len(X), batch_size)
            x_pos = X[idx]
            
            # 生成负样本：随机选择不同类别的样本
            neg_idx = []
            for i in idx:
                # 随机选择不同类别的样本
                possible_neg = np.where(y != y[i])[0]
                neg_idx.append(np.random.choice(possible_neg))
            x_neg = X[neg_idx]
            
            # 计算损失
            loss = self.contrastive_loss(x_pos, x_neg)
            loss_history.append(loss)
            
            # 梯度下降更新
            # 简化的梯度计算
            z_pos = np.dot(x_pos, self.W)
            z_neg = np.dot(x_neg, self.W)
            
            grad = 2 * np.dot(x_pos.T, (z_neg - z_pos)) / batch_size
            self.W -= self.learning_rate * grad
            
            if (epoch + 1) % 100 == 0:
                print(f"Epoch {epoch+1}/{self.epochs}, Loss: {loss:.4f}")
        
        return loss_history
    
    def get_embeddings(self, X):
        return self.encode(X)

# 创建并训练对比学习模型
contrastive_model = ContrastiveLearning(input_dim=2, embedding_dim=10, learning_rate=0.01, epochs=1000)
loss_history = contrastive_model.train(X_scaled, y)

# 获取学习到的表示
X_embeddings = contrastive_model.get_embeddings(X_scaled)

# 使用KNN在学习到的表示上进行分类
knn_emb = KNeighborsClassifier(n_neighbors=5)
knn_emb.fit(X_embeddings, y)
y_pred_emb = knn_emb.predict(X_embeddings)

# 评估效果
accuracy_emb = accuracy_score(y, y_pred_emb)
f1_emb = f1_score(y, y_pred_emb)

print("\n传统KNN分类效果:")
print(f"准确率: {accuracy_score(y, y_pred_knn):.4f}")
print(f"F1分数: {f1_score(y, y_pred_knn):.4f}")

print("\n对比学习+KNN分类效果:")
print(f"准确率: {accuracy_emb:.4f}")
print(f"F1分数: {f1_emb:.4f}")

# 可视化表示空间
plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)
plt.scatter(X_scaled[:, 0], X_scaled[:, 1], c=y, cmap='viridis', alpha=0.8)
plt.title('原始数据空间')
plt.xlabel('特征1')
plt.ylabel('特征2')
plt.grid(True)

plt.subplot(1, 2, 2)
plt.scatter(X_embeddings[:, 0], X_embeddings[:, 1], c=y, cmap='viridis', alpha=0.8)
plt.title('对比学习表示空间')
plt.xlabel('嵌入维度1')
plt.ylabel('嵌入维度2')
plt.grid(True)

plt.tight_layout()
plt.show()

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_classification, make_blobs
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

# 生成源域数据（例如：网络流量数据）
X_source, y_source = make_classification(n_samples=500, n_features=20, n_informative=10, n_redundant=10, random_state=42)

# 生成目标域数据（例如：邮件数据，特征分布不同但类别相似）
X_target, y_target = make_blobs(n_samples=300, n_features=20, centers=2, cluster_std=1.5, random_state=24)
y_target = (y_target > 0).astype(int)

# 数据标准化
scaler_source = StandardScaler()
X_source_scaled = scaler_source.fit_transform(X_source)

scaler_target = StandardScaler()
X_target_scaled = scaler_target.fit_transform(X_target)

# 1. 传统方法：在源域训练，直接在目标域测试
lr = LogisticRegression(random_state=42)
lr.fit(X_source_scaled, y_source)
y_pred_target = lr.predict(X_target_scaled)
accuracy_direct = accuracy_score(y_target, y_pred_target)
print(f"直接迁移准确率: {accuracy_direct:.4f}")

# 2. 表示学习方法：使用PCA学习通用表示
from sklearn.decomposition import PCA

# 在源域学习表示
pca = PCA(n_components=10, random_state=42)
pca.fit(X_source_scaled)

# 在源域表示上训练分类器
X_source_emb = pca.transform(X_source_scaled)
lr_emb = LogisticRegression(random_state=42)
lr_emb.fit(X_source_emb, y_source)

# 在目标域表示上进行预测
X_target_emb = pca.transform(X_target_scaled)
y_pred_target_emb = lr_emb.predict(X_target_emb)
accuracy_emb = accuracy_score(y_target, y_pred_target_emb)
print(f"表示学习迁移准确率: {accuracy_emb:.4f}")

# 3. 领域自适应表示学习：简单的CORAL域自适应
class CORAL:
    def __init__(self):
        pass
    
    def fit(self, X_source, X_target):
        # 计算源域和目标域的协方差矩阵
        cov_source = np.cov(X_source, rowvar=False)
        cov_target = np.cov(X_target, rowvar=False)
        
        # 计算协方差矩阵的平方根和逆平方根
        cov_source_sqrt = np.linalg.cholesky(cov_source)
        cov_source_sqrt_inv = np.linalg.inv(cov_source_sqrt)
        cov_target_sqrt = np.linalg.cholesky(cov_target)
        
        # 计算变换矩阵
        self.T = np.dot(cov_source_sqrt_inv, np.dot(cov_target_sqrt, cov_source_sqrt_inv))
        return self
    
    def transform(self, X):
        return np.dot(X, self.T)

# 使用CORAL进行域自适应
coral = CORAL()
coral.fit(X_source_scaled, X_target_scaled)

# 变换源域数据
X_source_coral = coral.transform(X_source_scaled)

# 在变换后的源域数据上训练分类器
lr_coral = LogisticRegression(random_state=42)
lr_coral.fit(X_source_coral, y_source)

# 在目标域上进行预测
y_pred_target_coral = lr_coral.predict(X_target_scaled)
accuracy_coral = accuracy_score(y_target, y_pred_target_coral)
print(f"CORAL域自适应准确率: {accuracy_coral:.4f}")

# 可视化结果
plt.figure(figsize=(15, 5))

# 可视化源域和目标域的前两个特征
plt.subplot(1, 3, 1)
plt.scatter(X_source_scaled[:, 0], X_source_scaled[:, 1], c=y_source, cmap='viridis', alpha=0.8, label='源域')
plt.scatter(X_target_scaled[:, 0], X_target_scaled[:, 1], c=y_target, cmap='plasma', alpha=0.8, label='目标域')
plt.title('源域和目标域原始特征分布')
plt.xlabel('特征1')
plt.ylabel('特征2')
plt.legend()
plt.grid(True)

# 可视化PCA表示
plt.subplot(1, 3, 2)
plt.scatter(X_source_emb[:, 0], X_source_emb[:, 1], c=y_source, cmap='viridis', alpha=0.8, label='源域')
plt.scatter(X_target_emb[:, 0], X_target_emb[:, 1], c=y_target, cmap='plasma', alpha=0.8, label='目标域')
plt.title('PCA表示空间')
plt.xlabel('嵌入维度1')
plt.ylabel('嵌入维度2')
plt.legend()
plt.grid(True)

# 可视化CORAL变换后的源域
plt.subplot(1, 3, 3)
plt.scatter(X_source_coral[:, 0], X_source_coral[:, 1], c=y_source, cmap='viridis', alpha=0.8, label='变换后源域')
plt.scatter(X_target_scaled[:, 0], X_target_scaled[:, 1], c=y_target, cmap='plasma', alpha=0.8, label='目标域')
plt.title('CORAL变换后特征分布')
plt.xlabel('特征1')
plt.ylabel('特征2')
plt.legend()
plt.grid(True)

plt.tight_layout()
plt.show()

# 安装必要的Python库
pip install numpy pandas scikit-learn matplotlib seaborn