generated from LufsX/Experiment-LogReg
This commit is contained in:
+68
-137
@@ -27,169 +27,100 @@ name = "章程鸿"
|
|||||||
|
|
||||||
# 这里写自己的代码实现
|
# 这里写自己的代码实现
|
||||||
import numpy as np
|
import numpy as np
|
||||||
import pandas as pd
|
from sklearn.preprocessing import StandardScaler, MinMaxScaler
|
||||||
from sklearn.preprocessing import StandardScaler
|
from sklearn.model_selection import train_test_split
|
||||||
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):
|
def sigmoid(z):
|
||||||
"""Sigmoid 激活函数。"""
|
return 1.0 / (1 + np.exp(-z))
|
||||||
return 1 / (1 + np.exp(-z))
|
|
||||||
|
|
||||||
def train_logreg(X, y, learning_rate=0.01, iterations=5000, lambda_reg=0.01, batch_size=32):
|
|
||||||
"""逻辑回归训练函数,随机初始化权重,并在每次迭代前打乱数据。"""
|
|
||||||
m, n = X.shape
|
|
||||||
weights = np.random.randn(n) * 0.01 # 随机初始化权重,取较小值
|
|
||||||
|
|
||||||
for i in range(iterations):
|
|
||||||
# 每次迭代打乱数据
|
|
||||||
indices = np.random.permutation(m)
|
|
||||||
X = X[indices]
|
|
||||||
y = 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]
|
|
||||||
|
|
||||||
|
def compute_gradient(X_batch, y_batch, weights, lambda_reg):
|
||||||
z = np.dot(X_batch, weights)
|
z = np.dot(X_batch, weights)
|
||||||
predictions = sigmoid(z)
|
h = sigmoid(z)
|
||||||
gradient = np.dot(X_batch.T, (predictions - y_batch)) / X_batch.shape[0]
|
gradient = np.dot(X_batch.T, (h - y_batch)) / len(y_batch)
|
||||||
|
# 添加L2正则化
|
||||||
|
gradient += (lambda_reg / len(y_batch)) * weights
|
||||||
|
return gradient
|
||||||
|
|
||||||
# L2 正则化
|
def train_logreg(X, y, learning_rate=0.01, iterations=5000, batch_size=64, lambda_reg=0.01):
|
||||||
gradient += (lambda_reg / m) * weights
|
m, n = X.shape
|
||||||
|
weights = np.random.randn(n) * 0.01
|
||||||
|
best_weights = weights.copy()
|
||||||
|
best_accuracy = 0
|
||||||
|
|
||||||
# 更新权重
|
for _ in range(iterations):
|
||||||
|
indices = np.random.permutation(m)
|
||||||
|
X_shuffled = X[indices]
|
||||||
|
y_shuffled = y[indices]
|
||||||
|
|
||||||
|
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
|
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 次迭代调整一次学习率
|
if _ % 100 == 0 and train_accuracy > 0.95: # 达到高准确率时减小学习率
|
||||||
learning_rate *= 0.9 # 每 1000 次迭代减小学习率
|
learning_rate *= 0.9
|
||||||
|
|
||||||
if i % 100 == 0: # 每 100 次迭代输出一次学习率
|
return best_weights
|
||||||
print(f"Iteration {i}, Learning Rate: {learning_rate}")
|
|
||||||
|
|
||||||
return weights
|
def predict_logreg(X, weights):
|
||||||
|
z = np.dot(X, weights)
|
||||||
def predict_logreg(inX, weights):
|
h = sigmoid(z)
|
||||||
"""逻辑回归预测函数。"""
|
return np.where(h >= 0.5, 1, 0)
|
||||||
z = np.dot(inX, weights)
|
|
||||||
prediction_prob = sigmoid(z)
|
|
||||||
return 1 if prediction_prob >= 0.5 else 0
|
|
||||||
|
|
||||||
def load_data(filepath):
|
def load_data(filepath):
|
||||||
"""加载数据并处理缺失值。"""
|
data, labels = [], []
|
||||||
data = []
|
|
||||||
labels = []
|
|
||||||
|
|
||||||
with open(filepath, 'r') as f:
|
with open(filepath, 'r') as f:
|
||||||
for line in f:
|
for line in f:
|
||||||
values = line.strip().replace('?', '0').split()
|
line = line.strip().split()
|
||||||
try:
|
if len(line) >= 1 and line[-1].isdigit():
|
||||||
values = [float(v) for v in values]
|
line_data = [0 if x == '?' else float(x) for x in line[:-1]]
|
||||||
|
data.append(line_data)
|
||||||
# 合并特征 23 的取值
|
labels.append(int(line[-1]))
|
||||||
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
|
|
||||||
|
|
||||||
data = np.array(data)
|
data = np.array(data)
|
||||||
labels = np.array(labels)
|
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):
|
def accuracy(X, y, weights):
|
||||||
"""特征标准化。"""
|
predictions = predict_logreg(X, weights)
|
||||||
scaler = StandardScaler()
|
return np.mean(predictions == y)
|
||||||
return scaler.fit_transform(X)
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
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, y_test = load_data("./data/horse-colic.test")
|
||||||
|
X_test = scaler.transform(X_test)
|
||||||
|
|
||||||
# 归一化数据
|
# 测试模型
|
||||||
X_train = preprocess_data(X_train)
|
final_test_accuracy = accuracy(X_test, y_test, final_weights)
|
||||||
X_test = preprocess_data(X_test)
|
print(f"最终测试集的准确率: {final_test_accuracy:.4f}")
|
||||||
|
|
||||||
# 超参数搜索空间
|
|
||||||
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}")
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user