diff --git a/function.py b/function.py index 2cf3dac..cda05dc 100644 --- a/function.py +++ b/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]) - - # 获取与待分类样本距离最近的 k 个样本的索引 - nearest_indices = distances.argsort()[:k] + # 计算协方差矩阵的逆矩阵 + 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_labels = [labels[i] for i in nearest_indices] + # 计算马氏距离 + diff = dataSet - inX + left = np.dot(diff, inv_cov_matrix) + distances = np.sqrt(np.sum(left * diff, axis=1)) + + # 检查距离计算结果 + if distances.size == 0: + raise ValueError("距离计算结果为空,请检查数据集和待分类样本。") + + # 获取距离最小的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 - # 返回出现频率最高的标签作为预测结果 - unique_labels, counts = np.unique(nearest_labels, return_counts=True) - return unique_labels[np.argmax(counts)] +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 -def load_data(file_path): - """ - 加载 TSV 格式数据文件。 - """ - data = np.loadtxt(file_path, delimiter="\t") - features = data[:, :-1] - labels = data[:, -1].astype(int) - return features, labels if __name__ == "__main__": - # 以下代码可用于测试,可自行随意修改 train_file = "datingTrainSet2.tsv" train_data, train_labels = load_data(train_file) + + # 检查训练数据的特征数量 + 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]) + # 示例待分类样本,确保其特征数量与训练数据一致 + 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}") \ No newline at end of file + print(f"预测的分类标签: {predicted_label}") \ No newline at end of file