线性回归详解

import numpy as np class LinearRegressionGD: """从零实现的线性回归（梯度下降法）""" def __init__(self, learning_rate=0.01, n_iterations=1000): self.lr = learning_rate self.n_iter = n_iterations self.weights = None self.bias = None self.cost_history = [] # 记录每次迭代的损失值 def fit(self, X, y): n_samples, n_features = X.shape # 初始化参数为 0 self.weights = np.zeros(n_features) self.bias = 0 for i in range(self.n_iter): # 前向传播：计算预测值 y_pred = np.dot(X, self.weights) + self.bias # 计算损失 (MSE) cost = np.mean((y - y_pred) ** 2) self.cost_history.append(cost) # 计算梯度 dw = (-2 / n_samples) * np.dot(X.T, (y - y_pred)) db = (-2 / n_samples) * np.sum(y - y_pred) # 更新参数 self.weights -= self.lr * dw self.bias -= self.lr * db return self def predict(self, X): return np.dot(X, self.weights) + self.bias def score(self, X, y): """计算 R² 决定系数""" y_pred = self.predict(X) ss_res = np.sum((y - y_pred) ** 2) ss_tot = np.sum((y - np.mean(y)) ** 2) return 1 - (ss_res / ss_tot) # ===== 使用示例 ===== # 生成模拟数据 np.random.seed(42) X = 2 * np.random.rand(100, 1) # 100 个样本, 1 个特征 y = 4 + 3 * X.squeeze() + np.random.randn(100) * 0.5 # y = 3x + 4 + noise # 特征标准化（加速收敛） X_mean, X_std = X.mean(axis=0), X.std(axis=0) X_scaled = (X - X_mean) / X_std # 训练模型 model = LinearRegressionGD(learning_rate=0.1, n_iterations=500) model.fit(X_scaled, y) print(f"R² = {model.score(X_scaled, y):.4f}") print(f"最终 MSE: {model.cost_history[-1]:.4f}")

def normal_equation(X, y): """正规方程解析解: beta = (X^T X)^(-1) X^T y""" # 添加偏置列 (全1列) X_b = np.c_[np.ones((X.shape[0], 1)), X] # 计算最优参数 beta = np.linalg.inv(X_b.T @ X_b) @ X_b.T @ y return beta # beta[0] = intercept, beta[1:] = weights beta = normal_equation(X, y) print(f"Intercept: {beta[0]:.4f}, Weight: {beta[1]:.4f}") # 输出接近 4.0 和 3.0（我们设定的真实值）

from sklearn.linear_model import LinearRegression from sklearn.model_selection import train_test_split from sklearn.metrics import mean_squared_error, r2_score, mean_absolute_error import numpy as np # 准备数据 np.random.seed(42) X = np.random.rand(200, 3) # 200 样本, 3 个特征 y = 5 + 2*X[:, 0] - 3*X[:, 1] + 1.5*X[:, 2] + np.random.randn(200) * 0.3 # 划分训练集/测试集 (80/20) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 ) # 训练模型 model = LinearRegression() model.fit(X_train, y_train) # 查看学到的参数 print(f"Intercept (b): {model.intercept_:.4f}") print(f"Coefficients (w): {model.coef_}") # 应接近 [2.0, -3.0, 1.5] # 在测试集上预测 y_pred = model.predict(X_test) # 评估模型 print(f"\nR² = {r2_score(y_test, y_pred):.4f}") print(f"MSE = {mean_squared_error(y_test, y_pred):.4f}") print(f"MAE = {mean_absolute_error(y_test, y_pred):.4f}")

from sklearn.linear_model import Ridge from sklearn.preprocessing import StandardScaler from sklearn.pipeline import Pipeline # Ridge 需要特征标准化 ridge_pipe = Pipeline([ ('scaler', StandardScaler()), ('ridge', Ridge(alpha=1.0)) # alpha = 正则化强度 ]) ridge_pipe.fit(X_train, y_train) print(f"Ridge R²: {ridge_pipe.score(X_test, y_test):.4f}") # 用交叉验证选择最优 alpha from sklearn.linear_model import RidgeCV ridge_cv = RidgeCV(alphas=[0.01, 0.1, 1.0, 10.0, 100.0], cv=5) ridge_cv.fit(X_train, y_train) print(f"Best alpha: {ridge_cv.alpha_}")

from sklearn.linear_model import Lasso, LassoCV # Lasso 回归 lasso = Pipeline([ ('scaler', StandardScaler()), ('lasso', Lasso(alpha=0.1)) ]) lasso.fit(X_train, y_train) print(f"Lasso R²: {lasso.score(X_test, y_test):.4f}") # 查看哪些特征被选中（权重不为0） lasso_model = lasso.named_steps['lasso'] for i, coef in enumerate(lasso_model.coef_): status = "保留" if abs(coef) > 1e-6 else "丢弃" print(f" Feature {i}: w={coef:.4f} ({status})")

from sklearn.linear_model import ElasticNet, ElasticNetCV # 用交叉验证同时搜索 alpha 和 l1_ratio enet_cv = ElasticNetCV( l1_ratio=[0.1, 0.5, 0.7, 0.9, 0.95, 1.0], alphas=[0.001, 0.01, 0.1, 1.0], cv=5 ) enet_cv.fit(X_train, y_train) print(f"Best alpha: {enet_cv.alpha_:.4f}") print(f"Best l1_ratio: {enet_cv.l1_ratio_:.2f}") print(f"ElasticNet R²: {enet_cv.score(X_test, y_test):.4f}")

from sklearn.preprocessing import PolynomialFeatures # 原始特征: [x₁, x₂] # degree=2 后: [1, x₁, x₂, x₁², x₁x₂, x₂²] poly = PolynomialFeatures(degree=2, include_bias=False) X_poly = poly.fit_transform(X_train) print(f"原始特征数: {X_train.shape[1]}") print(f"多项式特征数: {X_poly.shape[1]}") # 用多项式特征训练线性回归 from sklearn.pipeline import make_pipeline poly_model = make_pipeline( PolynomialFeatures(degree=2, include_bias=False), StandardScaler(), Ridge(alpha=1.0) # 多项式特征容易过拟合，建议加正则化 ) poly_model.fit(X_train, y_train) print(f"Polynomial R²: {poly_model.score(X_test, y_test):.4f}")

from sklearn.preprocessing import StandardScaler, MinMaxScaler # 标准化: 均值=0, 标准差=1 (推荐用于线性回归) scaler = StandardScaler() X_scaled = scaler.fit_transform(X_train) # 归一化: 缩放到 [0, 1] 区间 minmax = MinMaxScaler() X_minmax = minmax.fit_transform(X_train) # 重要: 测试集使用训练集的 scaler（不能重新fit） X_test_scaled = scaler.transform(X_test) # transform only, no fit!

import pandas as pd from sklearn.preprocessing import OneHotEncoder from sklearn.compose import ColumnTransformer # 示例数据 df = pd.DataFrame({ 'area': [80, 120, 60, 200], 'rooms': [2, 3, 1, 4], 'city': ['Beijing', 'Shanghai', 'Beijing', 'Shenzhen'], 'price': [300, 500, 200, 800] }) # 自动处理数值+分类特征 preprocessor = ColumnTransformer(transformers=[ ('num', StandardScaler(), ['area', 'rooms']), ('cat', OneHotEncoder(drop='first'), ['city']) # drop='first' 避免虚拟变量陷阱 ]) X = preprocessor.fit_transform(df[['area', 'rooms', 'city']]) print(f"编码后特征维度: {X.shape}")

from sklearn.metrics import ( r2_score, mean_squared_error, mean_absolute_error ) import numpy as np def evaluate_regression(y_true, y_pred, n_features=None): """打印所有回归评估指标""" r2 = r2_score(y_true, y_pred) mse = mean_squared_error(y_true, y_pred) rmse = np.sqrt(mse) mae = mean_absolute_error(y_true, y_pred) print(f"R² = {r2:.4f}") print(f"MSE = {mse:.4f}") print(f"RMSE = {rmse:.4f}") print(f"MAE = {mae:.4f}") if n_features is not None: n = len(y_true) adj_r2 = 1 - (1 - r2) * (n - 1) / (n - n_features - 1) print(f"Adjusted R² = {adj_r2:.4f}") return {"r2": r2, "mse": mse, "rmse": rmse, "mae": mae} # 使用示例 metrics = evaluate_regression(y_test, y_pred, n_features=X_train.shape[1])

import matplotlib.pyplot as plt import numpy as np from sklearn.linear_model import LinearRegression # 生成示例数据 np.random.seed(42) X = 2 * np.random.rand(80, 1) y = 4 + 3 * X.squeeze() + np.random.randn(80) * 0.8 # 训练模型 model = LinearRegression().fit(X, y) # 绘图 fig, ax = plt.subplots(figsize=(8, 5)) ax.scatter(X, y, alpha=0.6, color='#6c63ff', label='Data points') # 绘制回归线 X_line = np.linspace(0, 2, 100).reshape(-1, 1) y_line = model.predict(X_line) ax.plot(X_line, y_line, color='#ff6b6b', linewidth=2, label=f'y = {model.coef_[0]:.2f}x + {model.intercept_:.2f}') ax.set_xlabel('X') ax.set_ylabel('y') ax.set_title('Linear Regression Fit') ax.legend() plt.tight_layout() plt.savefig('regression_fit.png', dpi=150) plt.show()

fig, axes = plt.subplots(1, 2, figsize=(12, 4)) y_pred = model.predict(X) residuals = y - y_pred # 残差 vs 预测值 axes[0].scatter(y_pred, residuals, alpha=0.6, color='#6c63ff') axes[0].axhline(y=0, color='#ff6b6b', linestyle='--') axes[0].set_xlabel('Predicted') axes[0].set_ylabel('Residuals') axes[0].set_title('Residuals vs Predicted') # 残差直方图（检查正态性） axes[1].hist(residuals, bins=20, color='#6c63ff', alpha=0.7, edgecolor='white') axes[1].set_xlabel('Residual') axes[1].set_ylabel('Frequency') axes[1].set_title('Residual Distribution') plt.tight_layout() plt.savefig('residual_analysis.png', dpi=150) plt.show() # 理想情况：残差图无明显模式，直方图近似正态分布

from sklearn.model_selection import learning_curve train_sizes, train_scores, val_scores = learning_curve( LinearRegression(), X, y, cv=5, train_sizes=np.linspace(0.1, 1.0, 10), scoring='r2' ) fig, ax = plt.subplots(figsize=(8, 5)) ax.plot(train_sizes, train_scores.mean(axis=1), 'o-', label='Training R²') ax.plot(train_sizes, val_scores.mean(axis=1), 'o-', label='Validation R²') ax.fill_between(train_sizes, train_scores.mean(axis=1) - train_scores.std(axis=1), train_scores.mean(axis=1) + train_scores.std(axis=1), alpha=0.1) ax.fill_between(train_sizes, val_scores.mean(axis=1) - val_scores.std(axis=1), val_scores.mean(axis=1) + val_scores.std(axis=1), alpha=0.1) ax.set_xlabel('Training Set Size') ax.set_ylabel('R² Score') ax.set_title('Learning Curve') ax.legend() plt.tight_layout() plt.show()

from statsmodels.stats.outliers_influence import variance_inflation_factor import pandas as pd def check_vif(X, feature_names): """计算各特征的 VIF 值""" vif_data = pd.DataFrame() vif_data['Feature'] = feature_names vif_data['VIF'] = [ variance_inflation_factor(X, i) for i in range(X.shape[1]) ] vif_data = vif_data.sort_values('VIF', ascending=False) print(vif_data.to_string(index=False)) # VIF > 10: 严重共线性, 考虑移除 # VIF 5-10: 中度共线性, 需关注 # VIF < 5: 可接受 return vif_data

from sklearn.linear_model import RANSACRegressor, HuberRegressor # RANSAC: 随机采样一致性，忽略异常值 ransac = RANSACRegressor(random_state=42) ransac.fit(X_train, y_train) inlier_mask = ransac.inlier_mask_ print(f"Inliers: {inlier_mask.sum()}/{len(inlier_mask)}") # Huber: 对小误差用平方损失，大误差用绝对损失 huber = HuberRegressor(epsilon=1.35) huber.fit(X_train, y_train) print(f"Huber R²: {huber.score(X_test, y_test):.4f}")