更新 function.py

function.py

@@ -1,6 +1,6 @@
 # function.py
 import numpy as np
-
+from collections import Counter
 
 """
 挑战说明:
@@ -39,69 +39,120 @@ name = "熊浩燃"
 - int: 预测的分类标签。
 """
 
-def mahalanobis_distance(x1, x2, cov_inv):
-    """
-    计算两点之间的马氏距离。
-    
-    Args:
-    x1 (numpy array): 第一个点。
-    x2 (numpy array): 第二个点。
-    cov_inv (numpy array): 协方差矩阵的逆矩阵。
-    
-    Returns:
-    float: 两点之间的马氏距离。
-    """
-    delta = x1 - x2
-    dist = np.dot(np.dot(delta, cov_inv), delta.T)
-    return dist
-
 def classify0(inX, dataSet, labels, k=3):
-    """
-    使用马氏距离的 KNN 分类算法。
 
-    Args:
-    - inX (numpy array): 输入待分类的样本。
-    - dataSet (numpy array): 训练数据集。
-    - labels (list or numpy array): 训练集对应的标签。
-    - k (int): kNN 算法参数，默认值为 3。
+    if inX.shape[0] != dataSet.shape[1]:
+        raise ValueError(f"待分类样本的特征数量 ({inX.shape[0]}) 与训练数据集的特征数量 ({dataSet.shape[1]}) 不一致。")
 
-    Returns:
-    - int: 预测的分类标签。
-    """
-    # 计算训练数据集的协方差矩阵和均值，避免重复计算
+    # 计算训练数据集的协方差矩阵
     cov_matrix = np.cov(dataSet, rowvar=False)
-    cov_inv = np.linalg.inv(cov_matrix + np.eye(cov_matrix.shape[0]) * 1e-5)  # 加上一个小的单位矩阵以避免奇异矩阵
     
-    # 计算所有训练样本与待分类样本的马氏距离
-    distances = np.array([mahalanobis_distance(inX, x, cov_inv) for x in dataSet])
+    # 计算协方差矩阵的逆矩阵
+    try:
+        inv_cov_matrix = np.linalg.inv(cov_matrix)
+    except np.linalg.LinAlgError:
+        # 如果协方差矩阵是奇异的，添加一个小的值到对角线以进行正则化
+        cov_matrix += np.eye(cov_matrix.shape[0]) * 1e-5
+        try:
+            inv_cov_matrix = np.linalg.inv(cov_matrix)
+        except np.linalg.LinAlgError:
+            raise ValueError("协方差矩阵无法逆转，即使进行了正则化。请检查训练数据集。")
     
-    # 获取与待分类样本距离最近的 k 个样本的索引
-    nearest_indices = distances.argsort()[:k]
+    # 计算马氏距离
+    diff = dataSet - inX
+    left = np.dot(diff, inv_cov_matrix)
+    distances = np.sqrt(np.sum(left * diff, axis=1))
     
-    # 获取 k 个最近样本的标签
-    nearest_labels = [labels[i] for i in nearest_indices]
+    # 检查距离计算结果
+    if distances.size == 0:
+        raise ValueError("距离计算结果为空，请检查数据集和待分类样本。")
     
-    # 返回出现频率最高的标签作为预测结果
-    unique_labels, counts = np.unique(nearest_labels, return_counts=True)
-    return unique_labels[np.argmax(counts)]
+    # 获取距离最小的k个样本的索引
+    sorted_indices = distances.argsort()
+    top_k_indices = sorted_indices[:k]
+    
+    # 检查是否获取到足够的邻居
+    if len(top_k_indices) < k:
+        raise ValueError(f"训练数据集的样本数量 ({len(dataSet)}) 小于 k ({k})。")
+    
+    # 获取k个邻居的标签
+    top_k_labels = []
+    for idx in top_k_indices:
+        if idx >= len(labels):
+            raise IndexError(f"索引 {idx} 超出标签列表的范围 ({len(labels)}).")
+        top_k_labels.append(labels[idx])
+    
+    # 检查 top_k_labels 是否为空
+    if not top_k_labels:
+        raise ValueError("Top K 标签列表为空，无法进行投票。")
+    
+    # 进行多数表决
+    label_count = Counter(top_k_labels)
+    
+    # 检查 label_count 是否为空
+    if not label_count:
+        raise ValueError("标签计数为空，无法进行投票。")
+    
+    predicted_label = label_count.most_common(1)[0][0]
+    
+    return predicted_label
     
 
-def load_data(file_path):
-    """
-    加载 TSV 格式数据文件。
-    """
-    data = np.loadtxt(file_path, delimiter="\t")
-    features = data[:, :-1]
-    labels = data[:, -1].astype(int)
-    return features, labels
+def load_data(filename):
+
+    data = []
+    labels = []
+    with open(filename, 'r') as file:
+        for line_number, line in enumerate(file, start=1):
+            parts = line.strip().split('\t')
+            if len(parts) < 2:
+                print(f"警告: 第 {line_number} 行数据格式不正确，跳过。")
+                continue
+            # 假设最后一列是标签，前面的列是特征
+            *features, label = parts
+            try:
+                data.append([float(x) for x in features])
+                labels.append(int(label))
+            except ValueError as e:
+                print(f"警告: 第 {line_number} 行数据转换失败 ({e})，跳过。")
+                continue
+    data = np.array(data)
+    labels = np.array(labels)
+    print(f"加载的训练数据形状: {data.shape}")
+    print(f"加载的标签数量: {len(labels)}")
+    if data.shape[0] != len(labels):
+        raise ValueError(f"数据集样本数量 ({data.shape[0]}) 与标签数量 ({len(labels)}) 不一致。")
+    return data, labels
+
 
 
 if __name__ == "__main__":
-    # 以下代码可用于测试，可自行随意修改
     train_file = "datingTrainSet2.tsv"
     train_data, train_labels = load_data(train_file)
     
-    # 测试一个样本
-    test_sample = np.array([65915, 3.868359, 0.128078])
+    # 检查训练数据的特征数量
+    num_features = train_data.shape[1]
+    print(f"训练数据的特征数量: {num_features}")
+    
+    # 示例待分类样本，确保其特征数量与训练数据一致
+    test_sample = np.array([65915, 3.868359, 0.128078])  # 示例有3个特征
+    print(f"测试样本的特征数量: {test_sample.shape[0]}")
+    
+    if test_sample.shape[0] != num_features:
+        raise ValueError(f"测试样本的特征数量 ({test_sample.shape[0]}) 与训练数据不匹配 ({num_features})")
+    
+    # 进行分类预测
+    try:
+        predicted_label = classify0(test_sample, train_data, train_labels)
+        print(f"预测的分类标签: {predicted_label}")
+    except Exception as e:
+        print(f"分类过程中发生错误: {e}")
+
+    # 示例待分类样本，确保其特征数量与训练数据一致
+    test_sample = np.array([65915, 3.868359, 0.128078])  # 示例有3个特征
+    if test_sample.shape[0] != num_features:
+        raise ValueError(f"测试样本的特征数量 ({test_sample.shape[0]}) 与训练数据不匹配 ({num_features})")
+    
+    # 进行分类预测
     predicted_label = classify0(test_sample, train_data, train_labels)
-    print(f"The predicted label for the test sample is: {predicted_label}")
+    print(f"预测的分类标签: {predicted_label}")