diff --git a/function.py b/function.py index 8914306..bf27ae3 100644 --- a/function.py +++ b/function.py @@ -27,169 +27,100 @@ name = "章程鸿" # 这里写自己的代码实现 import numpy as np -import pandas as pd -from sklearn.preprocessing import StandardScaler -from sklearn.model_selection import train_test_split, KFold, GridSearchCV -from sklearn.linear_model import LogisticRegression -from sklearn.metrics import confusion_matrix, precision_score, recall_score, f1_score -import warnings - -warnings.filterwarnings("ignore") +from sklearn.preprocessing import StandardScaler, MinMaxScaler +from sklearn.model_selection import train_test_split def sigmoid(z): - """Sigmoid 激活函数。""" - return 1 / (1 + np.exp(-z)) + return 1.0 / (1 + np.exp(-z)) -def train_logreg(X, y, learning_rate=0.01, iterations=5000, lambda_reg=0.01, batch_size=32): - """逻辑回归训练函数，随机初始化权重，并在每次迭代前打乱数据。""" +def compute_gradient(X_batch, y_batch, weights, lambda_reg): + z = np.dot(X_batch, weights) + h = sigmoid(z) + gradient = np.dot(X_batch.T, (h - y_batch)) / len(y_batch) + # 添加L2正则化 + gradient += (lambda_reg / len(y_batch)) * weights + return gradient + +def train_logreg(X, y, learning_rate=0.01, iterations=5000, batch_size=64, lambda_reg=0.01): m, n = X.shape - weights = np.random.randn(n) * 0.01 # 随机初始化权重，取较小值 + weights = np.random.randn(n) * 0.01 + best_weights = weights.copy() + best_accuracy = 0 - for i in range(iterations): - # 每次迭代打乱数据 + for _ in range(iterations): indices = np.random.permutation(m) - X = X[indices] - y = y[indices] + X_shuffled = X[indices] + y_shuffled = y[indices] - for j in range(0, m, batch_size): # 按照 batch size 迭代 - X_batch = X[j:j + batch_size] - y_batch = y[j:j + batch_size] - - z = np.dot(X_batch, weights) - predictions = sigmoid(z) - gradient = np.dot(X_batch.T, (predictions - y_batch)) / X_batch.shape[0] - - # L2 正则化 - gradient += (lambda_reg / m) * weights - - # 更新权重 + for j in range(0, m, batch_size): + X_batch = X_shuffled[j:j + batch_size] + y_batch = y_shuffled[j:j + batch_size] + gradient = compute_gradient(X_batch, y_batch, weights, lambda_reg) weights -= learning_rate * gradient + # 计算训练集准确率 + train_accuracy = accuracy(X, y, weights) + if train_accuracy > best_accuracy: + best_accuracy = train_accuracy + best_weights = weights.copy() + # 动态调整学习率 - if i % 1000 == 0: # 每 1000 次迭代调整一次学习率 - learning_rate *= 0.9 # 每 1000 次迭代减小学习率 + if _ % 100 == 0 and train_accuracy > 0.95: # 达到高准确率时减小学习率 + learning_rate *= 0.9 - if i % 100 == 0: # 每 100 次迭代输出一次学习率 - print(f"Iteration {i}, Learning Rate: {learning_rate}") + return best_weights - return weights - -def predict_logreg(inX, weights): - """逻辑回归预测函数。""" - z = np.dot(inX, weights) - prediction_prob = sigmoid(z) - return 1 if prediction_prob >= 0.5 else 0 +def predict_logreg(X, weights): + z = np.dot(X, weights) + h = sigmoid(z) + return np.where(h >= 0.5, 1, 0) def load_data(filepath): - """加载数据并处理缺失值。""" - data = [] - labels = [] - + data, labels = [], [] with open(filepath, 'r') as f: for line in f: - values = line.strip().replace('?', '0').split() - try: - values = [float(v) for v in values] - - # 合并特征 23 的取值 - if values[22] == 2 or values[22] == 3: # 特征 23（索引 22） - labels.append(0) # 死亡 - elif values[22] == 1: - labels.append(1) # 生存 - else: - continue # 无效标签，丢弃 - - # 用均值填充缺失值 - values = [np.nan if v == 0 else v for v in values] - nan_indices = np.isnan(values) - column_means = np.nanmean(np.array(data), axis=0) - for i, is_nan in enumerate(nan_indices): - if is_nan: - values[i] = column_means[i] - data.append(values[:2] + values[3:23]) # 保留特征 1, 2 和特征 4 - 23 - except ValueError: - continue + line = line.strip().split() + if len(line) >= 1 and line[-1].isdigit(): + line_data = [0 if x == '?' else float(x) for x in line[:-1]] + data.append(line_data) + labels.append(int(line[-1])) data = np.array(data) labels = np.array(labels) + target_variable = np.where((labels == 2) | (labels == 3), 0, labels) - return data, labels + return data, target_variable -def preprocess_data(X): - """特征标准化。""" - scaler = StandardScaler() - return scaler.fit_transform(X) +def accuracy(X, y, weights): + predictions = predict_logreg(X, weights) + return np.mean(predictions == y) if __name__ == "__main__": - # 加载训练集 - X_train, y_train = load_data("./data/horse-colic.data") + # 加载数据 + X, y = load_data("./data/horse-colic.data") + + # 拆分训练集和验证集 + X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42) + + # 数据归一化 + scaler = MinMaxScaler() + X_train = scaler.fit_transform(X_train) + X_val = scaler.transform(X_val) + + # 训练模型 + final_weights = train_logreg(X_train, y_train, learning_rate=0.01, iterations=5000, batch_size=64, lambda_reg=0.1) + + # 验证模型 + val_accuracy = accuracy(X_val, y_val, final_weights) + print(f"验证集的准确率: {val_accuracy:.4f}") # 加载测试集 X_test, y_test = load_data("./data/horse-colic.test") + X_test = scaler.transform(X_test) - # 归一化数据 - X_train = preprocess_data(X_train) - X_test = preprocess_data(X_test) - - # 超参数搜索空间 - param_grid = { - 'learning_rate': [0.001, 0.01, 0.1], - 'iterations': [3000, 5000, 7000], - 'lambda_reg': [0.001, 0.01, 0.1], - 'batch_size': [16, 32, 64] - } - - # K 折交叉验证 - k = 5 - kf = KFold(n_splits=k, shuffle=True, random_state=42) - accuracies = [] - - for train_index, val_index in kf.split(X_train): - X_train_kf, X_val_kf = X_train[train_index], X_train[val_index] - y_train_kf, y_val_kf = y_train[train_index], y_train[val_index] - - best_params = None - best_score = 0 - - for params in GridSearchCV(estimator=LogisticRegression(), param_grid=param_grid, scoring='accuracy', cv=3).fit(X_train_kf, y_train_kf).cv_results_['params']: - weights = train_logreg(X_train_kf, y_train_kf, learning_rate=params['learning_rate'], iterations=params['iterations'], lambda_reg=params['lambda_reg'], batch_size=params['batch_size']) - val_predictions = [predict_logreg(X_val_kf[i], weights) for i in range(len(X_val_kf))] - score = np.mean(val_predictions == y_val_kf) - if score > best_score: - best_score = score - best_params = params - - print(f"Best parameters for fold: {best_params}") - - # 使用最佳参数重新训练 - weights = train_logreg(X_train_kf, y_train_kf, learning_rate=best_params['learning_rate'], iterations=best_params['iterations'], lambda_reg=best_params['lambda_reg'], batch_size=best_params['batch_size']) - val_predictions = [predict_logreg(X_val_kf[i], weights) for i in range(len(X_val_kf))] - accuracy = np.mean(val_predictions == y_val_kf) - accuracies.append(accuracy) - - # 输出交叉验证的准确率和平均准确率 - print(f"K 折交叉验证得分: {accuracies}") - print(f"平均准确率: {np.mean(accuracies)}") - - # 在测试集上测试模型 - best_params = None - best_score = 0 - - for params in GridSearchCV(estimator=LogisticRegression(), param_grid=param_grid, scoring='accuracy', cv=3).fit(X_train, y_train).cv_results_['params']: - weights = train_logreg(X_train, y_train, learning_rate=params['learning_rate'], iterations=params['iterations'], lambda_reg=params['lambda_reg'], batch_size=params['batch_size']) - test_predictions = [predict_logreg(X_test[i], weights) for i in range(len(X_test))] - score = np.mean(test_predictions == y_test) - if score > best_score: - best_score = score - best_params = params - - print(f"Best parameters for test set: {best_params}") - - weights = train_logreg(X_train, y_train, learning_rate=best_params['learning_rate'], iterations=best_params['iterations'], lambda_reg=best_params['lambda_reg'], batch_size=best_params['batch_size']) - test_predictions = [predict_logreg(X_test[i], weights) for i in range(len(X_test))] - test_accuracy = np.mean(test_predictions == y_test) - print(f"测试集准确率: {test_accuracy}") - + # 测试模型 + final_test_accuracy = accuracy(X_test, y_test, final_weights) + print(f"最终测试集的准确率: {final_test_accuracy:.4f}")