generated from LufsX/Experiment-LogReg
This commit is contained in:
+76
-36
@@ -29,7 +29,12 @@ name = "章程鸿"
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
from sklearn.preprocessing import StandardScaler
|
||||
from sklearn.model_selection import KFold
|
||||
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")
|
||||
|
||||
def sigmoid(z):
|
||||
"""Sigmoid 激活函数。"""
|
||||
@@ -54,29 +59,29 @@ def train_logreg(X, y, learning_rate=0.01, iterations=5000, lambda_reg=0.01, bat
|
||||
predictions = sigmoid(z)
|
||||
gradient = np.dot(X_batch.T, (predictions - y_batch)) / X_batch.shape[0]
|
||||
|
||||
# L2正则化
|
||||
# L2 正则化
|
||||
gradient += (lambda_reg / m) * weights
|
||||
|
||||
# 更新权重
|
||||
weights -= learning_rate * gradient
|
||||
|
||||
# 动态调整学习率
|
||||
if i % 1000 == 0: # 每1000次迭代调整一次学习率
|
||||
learning_rate *= 0.9 # 每1000次迭代减小学习率
|
||||
if i % 1000 == 0: # 每 1000 次迭代调整一次学习率
|
||||
learning_rate *= 0.9 # 每 1000 次迭代减小学习率
|
||||
|
||||
if i % 100 == 0: # 每100次迭代输出一次学习率
|
||||
if i % 100 == 0: # 每 100 次迭代输出一次学习率
|
||||
print(f"Iteration {i}, Learning Rate: {learning_rate}")
|
||||
|
||||
return weights
|
||||
|
||||
def predict_logreg(inX, weights, threshold=0.5):
|
||||
def predict_logreg(inX, weights):
|
||||
"""逻辑回归预测函数。"""
|
||||
z = np.dot(inX, weights)
|
||||
prediction_prob = sigmoid(z)
|
||||
return 1 if prediction_prob >= threshold else 0
|
||||
return 1 if prediction_prob >= 0.5 else 0
|
||||
|
||||
def load_data(filepath):
|
||||
"""加载数据并处理缺失值与规划。"""
|
||||
"""加载数据并处理缺失值。"""
|
||||
data = []
|
||||
labels = []
|
||||
|
||||
@@ -85,38 +90,56 @@ def load_data(filepath):
|
||||
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)
|
||||
|
||||
# 合并特征 23 的取值
|
||||
if values[22] == 2 or values[22] == 3: # 特征 23(索引 22)
|
||||
labels.append(0) # 死亡
|
||||
elif values[22] == 1:
|
||||
labels.append(1) # 生存
|
||||
else:
|
||||
continue # 无效标签,丢弃
|
||||
|
||||
# 用0填充缺失值
|
||||
values = [0 if v == 0 else v for v in values] # 直接用0填充
|
||||
data.append(values[:2] + values[3:23]) # 保留特征1, 2 和特征4-23
|
||||
|
||||
# 用均值填充缺失值
|
||||
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
|
||||
|
||||
data = np.array(data)
|
||||
labels = np.array(labels)
|
||||
|
||||
# 数据标准化
|
||||
scaler = StandardScaler()
|
||||
data = scaler.fit_transform(data) # 在这里进行标准化处理
|
||||
|
||||
return data, labels
|
||||
|
||||
def preprocess_data(X):
|
||||
"""特征标准化。"""
|
||||
scaler = StandardScaler()
|
||||
return scaler.fit_transform(X)
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 加载训练集
|
||||
X_train, y_train = load_data("./data/horse-colic.data")
|
||||
|
||||
|
||||
# 加载测试集
|
||||
X_test, y_test = load_data("./data/horse-colic.test")
|
||||
|
||||
# K折交叉验证
|
||||
# 归一化数据
|
||||
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 = []
|
||||
@@ -125,31 +148,48 @@ if __name__ == "__main__":
|
||||
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]
|
||||
|
||||
# 训练逻辑回归模型,设置 batch size
|
||||
batch_size = 32
|
||||
weights = train_logreg(X_train_kf, y_train_kf, learning_rate=0.01, iterations=5000, lambda_reg=0.01, batch_size=batch_size)
|
||||
best_params = None
|
||||
best_score = 0
|
||||
|
||||
# 验证模型
|
||||
val_predictions = [predict_logreg(X_val_kf[i], weights, threshold=0.5) for i in range(len(X_val_kf))]
|
||||
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"K 折交叉验证得分: {accuracies}")
|
||||
print(f"平均准确率: {np.mean(accuracies)}")
|
||||
|
||||
# 在测试集上测试模型
|
||||
test_predictions = [predict_logreg(X_test[i], weights, threshold=0.5) for i in range(len(X_test))]
|
||||
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}")
|
||||
|
||||
# 测试单个数据,输出错误分类
|
||||
for i in range(len(X_test)):
|
||||
y_pred = predict_logreg(X_test[i], weights, threshold=0.5)
|
||||
if y_pred != y_test[i]:
|
||||
print(f"第 {i} 次测试错误,数据 {X_test[i]} 被分类到 {y_pred} 实际是 {y_test[i]}")
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user