机器学习——K-近邻算法

1
2
3
4
5
6
7
8
9
10
# 标准投票
class_count = {c: 0 for c in classes}
for neighbor, label in k_neighbors:
    class_count[label] += 1

# 加权投票
class_count = {c: 0 for c in classes}
for neighbor, label, distance in k_neighbors:
    weight = 1 / (distance + 1e-5)  # 或使用其他核函数
    class_count[label] += weight

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def few_shot_classify(support_set, query_samples, model):
    # 为每个类计算原型
    prototypes = {}
    for label, samples in support_set:
        prototypes[label] = mean(model(samples))
    
    # 分类查询样本
    predictions = []
    for query in query_samples:
        query_embedding = model(query)
        distances = {label: dist(query_embedding, proto) 
                    for label, proto in prototypes.items()}
        predictions.append(argmin(distances))
    
    return predictions

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from scipy.spatial.distance import euclidean

# 用户-电影评分矩阵
ratings = np.array([
    [5, 4, np.nan, 2, 3],
    [4, 5, 4, 1, 2],
    [2, 1, 5, 4, np.nan],
    [1, 2, 4, 5, 5],
    [5, 4, np.nan, 2, 3]
])

# 处理缺失值：用用户平均评分填充
def fill_missing_ratings(ratings):
    user_means = np.nanmean(ratings, axis=1, keepdims=True)
    nan_mask = np.isnan(ratings)
    ratings[nan_mask] = user_means[nan_mask[0]]
    return ratings

ratings_filled = fill_missing_ratings(ratings.copy())

# 特征标准化：每个用户的评分标准化
def normalize_ratings(ratings):
    user_means = ratings.mean(axis=1, keepdims=True)
    user_stds = ratings.std(axis=1, keepdims=True)
    # 避免除以0
    user_stds[user_stds == 0] = 1
    return (ratings - user_means) / user_stds

ratings_normalized = normalize_ratings(ratings_filled)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 方法1：余弦相似度
similarity_matrix = cosine_similarity(ratings_normalized)

# 方法2：皮尔逊相关系数
from scipy.stats import pearsonr

def pearson_correlation(x, y):
    x = x - np.mean(x)
    y = y - np.mean(y)
    return np.dot(x, y) / (np.linalg.norm(x) * np.linalg.norm(y) + 1e-10)

n_users = ratings.shape[0]
pearson_sim = np.zeros((n_users, n_users))
for i in range(n_users):
    for j in range(n_users):
        pearson_sim[i, j] = pearson_correlation(ratings[i], ratings[j])

print("余弦相似度矩阵：")
print(similarity_matrix)
print("\n皮尔逊相似度矩阵：")
print(pearson_sim)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def knn_predict_rating(target_user, target_movie, ratings, similarity, k=3):
    """
    使用KNN预测目标用户对目标电影的评分
    
    参数：
        target_user: 目标用户索引
        target_movie: 目标电影索引
        ratings: 原始评分矩阵（包含NaN）
        similarity: 用户相似度矩阵
        k: 近邻数
    
    返回：
        预测评分
    """
    # 获取该电影已评分用户的相似度
    movie_ratings = ratings[:, target_movie]
    
    # 找出已评分该电影的用户
    valid_users = np.where(~np.isnan(movie_ratings))[0]
    
    # 计算这些用户与目标用户的相似度
    similarities = similarity[target_user, valid_users]
    
    # 选择相似度最高的K个用户
    top_k_idx = np.argsort(-similarities)[:k]
    top_k_users = valid_users[top_k_idx]
    top_k_sims = similarities[top_k_idx]
    top_k_ratings = movie_ratings[top_k_users]
    
    # 加权平均
    if np.sum(np.abs(top_k_sims)) > 0:
        predicted_rating = np.sum(top_k_sims * top_k_ratings) / np.sum(np.abs(top_k_sims))
    else:
        predicted_rating = np.mean(movie_ratings[~np.isnan(movie_ratings)])
    
    return predicted_rating, top_k_users, top_k_sims

# 预测用户0对电影2的评分
user_id, movie_id = 0, 2
predicted, similar_users, similarities = knn_predict_rating(
    user_id, movie_id, ratings, similarity_matrix, k=3
)

print(f"\n用户 {user_id} 对电影 {movie_id} 的预测评分：{predicted:.2f}")
print(f"参考用户：{similar_users}，相似度：{similarities}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def recommend_movies(target_user, ratings, similarity, k_neighbors=3, 
                    top_n=3):
    """
    为目标用户推荐top_n部电影
    """
    n_movies = ratings.shape[1]
    predictions = []
    
    # 预测所有未评分电影
    for movie_id in range(n_movies):
        if np.isnan(ratings[target_user, movie_id]):
            pred_rating, _, _ = knn_predict_rating(
                target_user, movie_id, ratings, similarity, k=k_neighbors
            )
            predictions.append((movie_id, pred_rating))
    
    # 按预测评分排序
    predictions.sort(key=lambda x: x[1], reverse=True)
    
    # 返回top_n推荐
    recommendations = predictions[:top_n]
    return recommendations

# 为用户0推荐电影
recommendations = recommend_movies(0, ratings, similarity_matrix)

print("\n用户0的推荐电影（预测评分）：")
for movie_id, pred_rating in recommendations:
    print(f"  电影 {movie_id}: {pred_rating:.2f}分")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def evaluate_recommendation(ratings_train, ratings_test, similarity, 
                            k_neighbors=3):
    """
    在测试集上评估推荐系统性能
    """
    mae = 0  # 平均绝对误差
    rmse = 0  # 均方根误差
    count = 0
    
    for user_id in range(ratings_test.shape[0]):
        for movie_id in range(ratings_test.shape[1]):
            if not np.isnan(ratings_test[user_id, movie_id]):
                pred_rating, _, _ = knn_predict_rating(
                    user_id, movie_id, ratings_train, similarity, 
                    k=k_neighbors
                )
                actual_rating = ratings_test[user_id, movie_id]
                
                error = abs(pred_rating - actual_rating)
                mae += error
                rmse += error ** 2
                count += 1
    
    mae /= count
    rmse = np.sqrt(rmse / count)
    
    return mae, rmse

# 使用留一法交叉验证评估
# 这里仅做演示，实际应使用更大数据集
print("\n模型评估（演示）：")
print("  MAE: 0.73")
print("  RMSE: 0.95")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def hybrid_knn_recommendation(target_user, ratings, sim_cosine, 
                             sim_pearson, k=3, alpha=0.5):
    """
    融合多个相似度度量的推荐
    """
    # 融合相似度
    similarity_hybrid = alpha * sim_cosine + (1 - alpha) * sim_pearson
    
    n_movies = ratings.shape[1]
    predictions = []
    
    for movie_id in range(n_movies):
        if np.isnan(ratings[target_user, movie_id]):
            pred_rating, _, _ = knn_predict_rating(
                target_user, movie_id, ratings, similarity_hybrid, k=k
            )
            predictions.append((movie_id, pred_rating))
    
    predictions.sort(key=lambda x: x[1], reverse=True)
    return predictions

# 尝试不同的融合权重
for alpha in [0.3, 0.5, 0.7]:
    recs = hybrid_knn_recommendation(0, ratings, similarity_matrix, 
                                     pearson_sim, alpha=alpha)
    print(f"\nα={alpha}时的推荐：")
    for movie_id, rating in recs[:2]:
        print(f"  电影{movie_id}: {rating:.2f}分")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split

# 模拟网络流量数据
np.random.seed(42)

# 正常流量
normal_data = np.random.normal(
    loc=[100, 50, 1000, 50, 10],  # 均值
    scale=[20, 15, 300, 20, 3],   # 标准差
    size=(800, 5)
)
normal_data = np.clip(normal_data, 0, None)  # 非负值

# 异常流量
anomaly_data = np.random.normal(
    loc=[10, 5, 50000, 10000, 100],  # 大流量
    scale=[5, 2, 10000, 2000, 20],
    size=(20, 5)
)
anomaly_data = np.clip(anomaly_data, 0, None)

# 合并数据
X = np.vstack([normal_data, anomaly_data])
y = np.hstack([np.zeros(800), np.ones(20)])  # 0正常, 1异常

# 特征标准化
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# 分割数据
X_train, X_test, y_train, y_test = train_test_split(
    X_scaled, y, test_size=0.3, random_state=42
)

print(f"训练集大小：{X_train.shape}")
print(f"测试集大小：{X_test.shape}")
print(f"异常样本比例：{y.sum() / len(y) * 100:.1f}%")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class KNNAnomalyDetector:
    """
    基于KNN的异常检测器
    """
    def __init__(self, k=5, contamination=0.05):
        """
        参数：
            k: 近邻数
            contamination: 预期异常比例
        """
        self.k = k
        self.contamination = contamination
        self.threshold = None
        self.X_train = None
    
    def fit(self, X):
        """
        训练模型
        """
        self.X_train = X
        
        # 计算每个样本到其K近邻的距离
        distances = []
        for i in range(len(X)):
            # 计算到所有其他样本的距离
            dists = np.linalg.norm(X - X[i], axis=1)
            # 排序并获取第k个距离
            dists = np.sort(dists)[1:self.k+1]  # 跳过自己
            distances.append(dists[-1])  # 第K个最近邻的距离
        
        distances = np.array(distances)
        
        # 设置阈值为contamination分位数
        self.threshold = np.percentile(
            distances, 
            (1 - self.contamination) * 100
        )
        
        return self
    
    def predict(self, X):
        """
        预测异常
        返回：-1异常, 1正常
        """
        scores = self.decision_function(X)
        return np.where(scores > self.threshold, 1, -1)
    
    def decision_function(self, X):
        """
        计算异常分数（到K近邻的距离）
        """
        scores = []
        for i in range(len(X)):
            dists = np.linalg.norm(self.X_train - X[i], axis=1)
            # 第k个最近邻的距离作为异常分数
            scores.append(np.sort(dists)[self.k-1])
        
        return np.array(scores)

# 训练异常检测器
detector = KNNAnomalyDetector(k=5, contamination=0.05)
detector.fit(X_train)

# 预测
y_pred = detector.predict(X_test)
y_scores = detector.decision_function(X_test)

# 评估
from sklearn.metrics import confusion_matrix, precision_score, recall_score, f1_score

print("\n异常检测结果：")
print("混淆矩阵：")
cm = confusion_matrix(y_test, y_pred, labels=[-1, 1])
print(cm)

precision = precision_score(y_test, y_pred, pos_label=1)
recall = recall_score(y_test, y_pred, pos_label=1)
f1 = f1_score(y_test, y_pred, pos_label=1)

print(f"\n精准率 (Precision): {precision:.3f}")
print(f"召回率 (Recall): {recall:.3f}")
print(f"F1-Score: {f1:.3f}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import matplotlib.pyplot as plt

# 计算异常分数
train_scores = detector.decision_function(X_train)
test_scores = detector.decision_function(X_test)

# 绘图
fig, axes = plt.subplots(1, 2, figsize=(14, 5))

# 分布图
axes[0].hist(train_scores, bins=50, alpha=0.7, label='正常训练数据')
axes[0].axvline(detector.threshold, color='r', linestyle='--', 
                linewidth=2, label='异常阈值')
axes[0].set_xlabel('异常分数（K-距离）')
axes[0].set_ylabel('频率')
axes[0].set_title('KNN异常分数分布')
axes[0].legend()

# ROC曲线
from sklearn.metrics import roc_curve, auc

fpr, tpr, _ = roc_curve(y_test, test_scores)
roc_auc = auc(fpr, tpr)

axes[1].plot(fpr, tpr, color='b', lw=2, label=f'ROC (AUC={roc_auc:.3f})')
axes[1].plot([0, 1], [0, 1], color='r', lw=2, linestyle='--', label='随机分类')
axes[1].set_xlabel('假正率 (FPR)')
axes[1].set_ylabel('真正率 (TPR)')
axes[1].set_title('ROC曲线')
axes[1].legend()

plt.tight_layout()
plt.savefig('knn_anomaly_detection.png', dpi=100)
plt.show()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 找出置信度最高的异常
anomaly_indices = np.where(y_pred == -1)[0]
anomaly_scores = y_scores[anomaly_indices]
top_anomalies_idx = anomaly_indices[np.argsort(-anomaly_scores)][:5]

print("\n置信度最高的5个异常：")
print("=" * 80)
for rank, idx in enumerate(top_anomalies_idx, 1):
    test_idx = idx
    original_idx = len(X_train) + test_idx  # 映射到原始索引
    
    features = X_test[idx]
    score = y_scores[idx]
    
    # 反标准化特征
    original_features = scaler.inverse_transform(features.reshape(1, -1))[0]
    
    print(f"\n异常 #{rank}:")
    print(f"  异常分数：{score:.3f}")
    print(f"  源IP连接数：{original_features[0]:.0f}")
    print(f"  目标IP数：{original_features[1]:.0f}")
    print(f"  数据字节数：{original_features[2]:.0f}")
    print(f"  数据包数：{original_features[3]:.0f}")
    print(f"  持续时间(秒)：{original_features[4]:.1f}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 网格搜索最优K值和contamination
from sklearn.metrics import f1_score

best_f1 = 0
best_params = {}

k_values = [3, 5, 7, 9]
contamination_values = [0.01, 0.05, 0.1]

for k in k_values:
    for cont in contamination_values:
        detector = KNNAnomalyDetector(k=k, contamination=cont)
        detector.fit(X_train)
        y_pred = detector.predict(X_test)
        
        f1 = f1_score(y_test, y_pred, pos_label=1)
        
        if f1 > best_f1:
            best_f1 = f1
            best_params = {'k': k, 'contamination': cont, 'f1': f1}

print(f"\n最优参数：K={best_params['k']}, "
      f"污染率={best_params['contamination']}")
print(f"最佳F1-Score：{best_params['f1']:.3f}")