线性回归:画出数据里的最优直线

文档摘要

线性回归:画出数据里的最优直线本节摘要:线性回归(Linear Regression)画出穿过数据的最优直线,是机器学习的「Hello World」。它把房价随面积的变化拟合成一条 ,让你代入任意面积得到价格预测。本节更重要之处在于:它引入了整个 ML 训练循环——定义模型、定义代价函数、优化参数,每一个 ML 算法都遵循这套范式。你将亲手推导均方误差(MSE)的梯度下降更新规则,从零实现单变量线性回归、正规方程闭式解、多元线性回归(含特征标准化)、多项式回归,以及岭回归(L2 正则化)防止过拟合的机制。掌握这些,你就能在任何算法里认出这个「最小化代价函数」的骨架,并用 R² 评估拟合好坏。学习目标阅读完本节,你应当能够: 推导均方误差的梯度下降更新规则,并从零实现线性回归。

线性回归:画出数据里的最优直线

本节摘要:线性回归(Linear Regression)画出穿过数据的最优直线,是机器学习的「Hello World」。它把房价随面积的变化拟合成一条 y = wx + b,让你代入任意面积得到价格预测。本节更重要之处在于:它引入了整个 ML 训练循环——定义模型、定义代价函数、优化参数,每一个 ML 算法都遵循这套范式。你将亲手推导均方误差(MSE)的梯度下降更新规则,从零实现单变量线性回归、正规方程闭式解、多元线性回归(含特征标准化)、多项式回归,以及岭回归(L2 正则化)防止过拟合的机制。掌握这些,你就能在任何算法里认出这个「最小化代价函数」的骨架,并用 R² 评估拟合好坏。

学习目标

阅读完本节,你应当能够:

推导均方误差的梯度下降更新规则,并从零实现线性回归。
对比梯度下降与正规方程的计算复杂度,知道各自何时用。
构建带特征标准化的多元线性回归模型,并解读学到的权重。
解释岭回归(Ridge / L2 正则化)如何通过惩罚大权重来防止过拟合。

一、问题与直觉

你手上有数据:房屋面积和它们的成交价。你想根据一套新房的面积预测价格。你可以在散点图上目测,但你需要一个公式,需要一条最贴合数据的直线,这样代入任意面积就能给出价格预测。

线性回归给你的就是这条直线。更重要的是,它引入了整套 ML 训练循环:定义模型、定义代价函数、优化参数。每个 ML 算法都走这套流程。在最简单的情况下吃透它,你以后到处都能认出它。

它不只用于简单问题。生产系统里,线性回归用于需求预测、A/B 测试分析、金融建模,并作为每个回归任务的基线。

模型

线性回归假设输入(x)与输出(y)之间是线性关系:


y = wx + b

w(权重/斜率):x 增加 1 时 y 变化多少
b(偏置/截距):x = 0 时 y 的值

对多个输入(特征),扩展为:


y = w1*x1 + w2*x2 + ... + wn*xn + b

或写成向量形式:y = w^T * x + b

目标:找到 w 和 b 的值,使预测 y 在所有训练样本上尽可能接近真实 y。

代价函数(均方误差)

怎么衡量「尽可能接近」?你需要一个单一数字来概括预测错得有多离谱。最常用的是均方误差(Mean Squared Error, MSE):


MSE = (1/n) * sum((y_predicted - y_actual)^2)

为什么平方?两个原因。第一,它对大误差的惩罚远大于小误差(误差为 10 比误差为 1 糟 100 倍,不是 10 倍)。第二,平方函数处处光滑可导,让优化变简单。

代价函数构成一个曲面。对单个权重 w 和偏置 b,MSE 曲面像一个碗(凸抛物面)。碗底就是 MSE 最小处。训练就是找到那个底。

梯度下降

梯度下降靠一步步下坡来找到碗底。

梯度告诉你两件事:每个参数往哪个方向移、移多少。

对 y_hat = wx + b 的 MSE:


dMSE/dw = (2/n) * sum((y_hat - y) * x)
dMSE/db = (2/n) * sum(y_hat - y)

更新规则:


w = w - learning_rate * dMSE/dw
b = b - learning_rate * dMSE/db

学习率控制步长。太大:越过最小值发散;太小:训练无穷无尽。典型初值:0.01、0.001 或 0.0001。

正规方程(闭式解)

仅对线性回归,有一个直接公式,无需任何迭代就给出最优权重:


w = (X^T * X)^(-1) * X^T * y

它通过一次矩阵求逆解出 w。对小数据集完美。对大数据集(百万行或数千特征),由于矩阵求逆在特征数上是 O(n³),梯度下降更优。

多元线性回归

特征变多时,模型变成:


y = w1*x1 + w2*x2 + ... + wn*xn + b

一切照旧:MSE 仍是代价函数,梯度下降同时更新所有权重。唯一区别是你在拟合一个超平面而非直线。

这里特征缩放很关键。如果一个特征范围是 01,另一个是 01000000,梯度下降会很吃力,因为代价曲面被拉长。训练前先标准化特征(减均值、除标准差)。

多项式回归

关系不是线性怎么办?你仍可用线性回归,只需造多项式特征:


y = w1*x + w2*x^2 + w3*x^3 + b

这仍是「线性」回归,因为模型对权重(w1, w2, w3)是线性的,你只是用了 x 的非线性特征。

高次多项式能拟合更复杂的曲线,但有过拟合风险。10 次多项式能穿过 10 个数据点的每一个,但在新数据上预测很差。

R² 评分

MSE 告诉你错多少,但数字依赖 y 的尺度。R²(R-squared)给出与尺度无关的度量:


R^2 = 1 - (残差平方和) / (对均值的离差平方和)
    = 1 - SS_res / SS_tot

R² = 1.0:完美预测
R² = 0.0:模型不比每次都预测均值强
R² < 0.0:模型比预测均值还差

正则化预览(岭回归)

特征很多时,模型可能因赋予大权重而过拟合。岭回归(Ridge,L2 正则化)加一个惩罚项:


Cost = MSE + lambda * sum(w_i^2)

惩罚项抑制大权重。超参数 lambda 控制权衡:lambda 越大权重越小、正则越强。这会在后续章节深入。现在只需知道它存在,以及为什么有用。

二、从零实现

第 1 步:生成样本数据


import random
import math

random.seed(42)

TRUE_W = 3.0
TRUE_B = 7.0
N_SAMPLES = 100

X = [random.uniform(0, 10) for _ in range(N_SAMPLES)]
y = [TRUE_W * x + TRUE_B + random.gauss(0, 2.0) for x in X]

print(f"Generated {N_SAMPLES} samples")
print(f"True relationship: y = {TRUE_W}x + {TRUE_B} (+ noise)")
print(f"First 5 points: {[(round(X[i], 2), round(y[i], 2)) for i in range(5)]}")

第 2 步:梯度下降从零实现线性回归


class LinearRegression:
    def __init__(self, learning_rate=0.01):
        self.w = 0.0
        self.b = 0.0
        self.lr = learning_rate
        self.cost_history = []

    def predict(self, X):
        return [self.w * x + self.b for x in X]

    def compute_cost(self, X, y):
        predictions = self.predict(X)
        n = len(y)
        cost = sum((pred - actual) ** 2 for pred, actual in zip(predictions, y)) / n
        return cost

    def compute_gradients(self, X, y):
        predictions = self.predict(X)
        n = len(y)
        dw = (2 / n) * sum((pred - actual) * x for pred, actual, x in zip(predictions, y, X))
        db = (2 / n) * sum(pred - actual for pred, actual in zip(predictions, y))
        return dw, db

    def fit(self, X, y, epochs=1000, print_every=200):
        for epoch in range(epochs):
            dw, db = self.compute_gradients(X, y)
            self.w -= self.lr * dw
            self.b -= self.lr * db
            cost = self.compute_cost(X, y)
            self.cost_history.append(cost)
            if epoch % print_every == 0:
                print(f"  Epoch {epoch:4d} | Cost: {cost:.4f} | w: {self.w:.4f} | b: {self.b:.4f}")
        return self

    def r_squared(self, X, y):
        predictions = self.predict(X)
        y_mean = sum(y) / len(y)
        ss_res = sum((actual - pred) ** 2 for actual, pred in zip(y, predictions))
        ss_tot = sum((actual - y_mean) ** 2 for actual in y)
        return 1 - (ss_res / ss_tot)


print("=== Training Linear Regression (Gradient Descent) ===")
model = LinearRegression(learning_rate=0.005)
model.fit(X, y, epochs=1000, print_every=200)
print(f"\nLearned: y = {model.w:.4f}x + {model.b:.4f}")
print(f"True:    y = {TRUE_W}x + {TRUE_B}")
print(f"R-squared: {model.r_squared(X, y):.4f}")

第 3 步:正规方程(闭式解)


class LinearRegressionNormal:
    def __init__(self):
        self.w = 0.0
        self.b = 0.0

    def fit(self, X, y):
        n = len(X)
        x_mean = sum(X) / n
        y_mean = sum(y) / n
        numerator = sum((X[i] - x_mean) * (y[i] - y_mean) for i in range(n))
        denominator = sum((X[i] - x_mean) ** 2 for i in range(n))
        self.w = numerator / denominator
        self.b = y_mean - self.w * x_mean
        return self

    def predict(self, X):
        return [self.w * x + self.b for x in X]

    def r_squared(self, X, y):
        predictions = self.predict(X)
        y_mean = sum(y) / len(y)
        ss_res = sum((actual - pred) ** 2 for actual, pred in zip(y, predictions))
        ss_tot = sum((actual - y_mean) ** 2 for actual in y)
        return 1 - (ss_res / ss_tot)


print("\n=== Normal Equation (Closed-Form) ===")
model_normal = LinearRegressionNormal()
model_normal.fit(X, y)
print(f"Learned: y = {model_normal.w:.4f}x + {model_normal.b:.4f}")
print(f"R-squared: {model_normal.r_squared(X, y):.4f}")

第 4 步:多元线性回归


class MultipleLinearRegression:
    def __init__(self, n_features, learning_rate=0.01):
        self.weights = [0.0] * n_features
        self.bias = 0.0
        self.lr = learning_rate
        self.cost_history = []

    def predict_single(self, x):
        return sum(w * xi for w, xi in zip(self.weights, x)) + self.bias

    def predict(self, X):
        return [self.predict_single(x) for x in X]

    def compute_cost(self, X, y):
        predictions = self.predict(X)
        n = len(y)
        return sum((pred - actual) ** 2 for pred, actual in zip(predictions, y)) / n

    def fit(self, X, y, epochs=1000, print_every=200):
        n = len(y)
        n_features = len(X[0])
        for epoch in range(epochs):
            predictions = self.predict(X)
            errors = [pred - actual for pred, actual in zip(predictions, y)]
            for j in range(n_features):
                grad = (2 / n) * sum(errors[i] * X[i][j] for i in range(n))
                self.weights[j] -= self.lr * grad
            grad_b = (2 / n) * sum(errors)
            self.bias -= self.lr * grad_b
            cost = self.compute_cost(X, y)
            self.cost_history.append(cost)
            if epoch % print_every == 0:
                print(f"  Epoch {epoch:4d} | Cost: {cost:.4f}")
        return self

    def r_squared(self, X, y):
        predictions = self.predict(X)
        y_mean = sum(y) / len(y)
        ss_res = sum((actual - pred) ** 2 for actual, pred in zip(y, predictions))
        ss_tot = sum((actual - y_mean) ** 2 for actual in y)
        return 1 - (ss_res / ss_tot)


random.seed(42)
N = 100
X_multi = []
y_multi = []
for _ in range(N):
    size = random.uniform(500, 3000)
    bedrooms = random.randint(1, 5)
    age = random.uniform(0, 50)
    price = 50 * size + 10000 * bedrooms - 1000 * age + 50000 + random.gauss(0, 20000)
    X_multi.append([size, bedrooms, age])
    y_multi.append(price)


def standardize(X):
    n_features = len(X[0])
    means = [sum(X[i][j] for i in range(len(X))) / len(X) for j in range(n_features)]
    stds = []
    for j in range(n_features):
        variance = sum((X[i][j] - means[j]) ** 2 for i in range(len(X))) / len(X)
        stds.append(variance ** 0.5)
    X_scaled = []
    for i in range(len(X)):
        row = [(X[i][j] - means[j]) / stds[j] if stds[j] > 0 else 0 for j in range(n_features)]
        X_scaled.append(row)
    return X_scaled, means, stds


y_mean_val = sum(y_multi) / len(y_multi)
y_std_val = (sum((yi - y_mean_val) ** 2 for yi in y_multi) / len(y_multi)) ** 0.5
y_scaled = [(yi - y_mean_val) / y_std_val for yi in y_multi]

X_scaled, x_means, x_stds = standardize(X_multi)

print("\n=== Multiple Linear Regression (3 features) ===")
print("Features: house size, bedrooms, age")
multi_model = MultipleLinearRegression(n_features=3, learning_rate=0.01)
multi_model.fit(X_scaled, y_scaled, epochs=1000, print_every=200)

print(f"\nWeights (standardized): {[round(w, 4) for w in multi_model.weights]}")
print(f"Bias (standardized): {multi_model.bias:.4f}")
print(f"R-squared: {multi_model.r_squared(X_scaled, y_scaled):.4f}")

第 5 步:多项式回归


class PolynomialRegression:
    def __init__(self, degree, learning_rate=0.01):
        self.degree = degree
        self.weights = [0.0] * degree
        self.bias = 0.0
        self.lr = learning_rate

    def make_features(self, X):
        return [[x ** (d + 1) for d in range(self.degree)] for x in X]

    def predict(self, X):
        features = self.make_features(X)
        return [sum(w * f for w, f in zip(self.weights, row)) + self.bias for row in features]

    def fit(self, X, y, epochs=1000, print_every=200):
        features = self.make_features(X)
        n = len(y)
        for epoch in range(epochs):
            predictions = [sum(w * f for w, f in zip(self.weights, row)) + self.bias for row in features]
            errors = [pred - actual for pred, actual in zip(predictions, y)]
            for j in range(self.degree):
                grad = (2 / n) * sum(errors[i] * features[i][j] for i in range(n))
                self.weights[j] -= self.lr * grad
            grad_b = (2 / n) * sum(errors)
            self.bias -= self.lr * grad_b
            if epoch % print_every == 0:
                cost = sum(e ** 2 for e in errors) / n
                print(f"  Epoch {epoch:4d} | Cost: {cost:.6f}")
        return self

    def r_squared(self, X, y):
        predictions = self.predict(X)
        y_mean = sum(y) / len(y)
        ss_res = sum((actual - pred) ** 2 for actual, pred in zip(y, predictions))
        ss_tot = sum((actual - y_mean) ** 2 for actual in y)
        return 1 - (ss_res / ss_tot)


random.seed(42)
X_poly = [x / 10.0 for x in range(0, 50)]
y_poly = [0.5 * x ** 2 - 2 * x + 3 + random.gauss(0, 1.0) for x in X_poly]

x_max = max(abs(x) for x in X_poly)
X_poly_norm = [x / x_max for x in X_poly]
y_poly_mean = sum(y_poly) / len(y_poly)
y_poly_std = (sum((yi - y_poly_mean) ** 2 for yi in y_poly) / len(y_poly)) ** 0.5
y_poly_norm = [(yi - y_poly_mean) / y_poly_std for yi in y_poly]

print("\n=== Polynomial Regression (degree 2 vs degree 5) ===")
print("True relationship: y = 0.5x^2 - 2x + 3")

print("\nDegree 2:")
poly2 = PolynomialRegression(degree=2, learning_rate=0.1)
poly2.fit(X_poly_norm, y_poly_norm, epochs=2000, print_every=500)
print(f"  R-squared: {poly2.r_squared(X_poly_norm, y_poly_norm):.4f}")

print("\nDegree 5:")
poly5 = PolynomialRegression(degree=5, learning_rate=0.1)
poly5.fit(X_poly_norm, y_poly_norm, epochs=2000, print_every=500)
print(f"  R-squared: {poly5.r_squared(X_poly_norm, y_poly_norm):.4f}")

print("\nDegree 2 fits the true curve well. Degree 5 fits training data slightly better")
print("but risks overfitting on new data.")

第 6 步:岭回归(L2 正则化)


class RidgeRegression:
    def __init__(self, n_features, learning_rate=0.01, alpha=1.0):
        self.weights = [0.0] * n_features
        self.bias = 0.0
        self.lr = learning_rate
        self.alpha = alpha

    def predict_single(self, x):
        return sum(w * xi for w, xi in zip(self.weights, x)) + self.bias

    def predict(self, X):
        return [self.predict_single(x) for x in X]

    def fit(self, X, y, epochs=1000, print_every=200):
        n = len(y)
        n_features = len(X[0])
        for epoch in range(epochs):
            predictions = self.predict(X)
            errors = [pred - actual for pred, actual in zip(predictions, y)]
            mse = sum(e ** 2 for e in errors) / n
            reg_term = self.alpha * sum(w ** 2 for w in self.weights)
            cost = mse + reg_term
            for j in range(n_features):
                grad = (2 / n) * sum(errors[i] * X[i][j] for i in range(n))
                grad += 2 * self.alpha * self.weights[j]
                self.weights[j] -= self.lr * grad
            grad_b = (2 / n) * sum(errors)
            self.bias -= self.lr * grad_b
            if epoch % print_every == 0:
                print(f"  Epoch {epoch:4d} | Cost: {cost:.4f} | L2 penalty: {reg_term:.4f}")
        return self


print("\n=== Ridge Regression (L2 Regularization) ===")
print("Same data as multiple regression, with alpha=0.1")
ridge = RidgeRegression(n_features=3, learning_rate=0.01, alpha=0.1)
ridge.fit(X_scaled, y_scaled, epochs=1000, print_every=200)
print(f"\nRidge weights: {[round(w, 4) for w in ridge.weights]}")
print(f"Plain weights: {[round(w, 4) for w in multi_model.weights]}")
print("Ridge weights are smaller (shrunk toward zero) due to the L2 penalty.")

三、框架对比

同样的东西,用 scikit-learn 写,生产里你真正会用的版本:


from sklearn.linear_model import LinearRegression as SklearnLR
from sklearn.linear_model import Ridge
from sklearn.preprocessing import PolynomialFeatures, StandardScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error, r2_score
import numpy as np

np.random.seed(42)
X_sk = np.random.uniform(0, 10, (100, 1))
y_sk = 3.0 * X_sk.squeeze() + 7.0 + np.random.normal(0, 2.0, 100)

X_train, X_test, y_train, y_test = train_test_split(X_sk, y_sk, test_size=0.2, random_state=42)

lr = SklearnLR()
lr.fit(X_train, y_train)
y_pred = lr.predict(X_test)

print("=== Scikit-learn Linear Regression ===")
print(f"Coefficient (w): {lr.coef_[0]:.4f}")
print(f"Intercept (b): {lr.intercept_:.4f}")
print(f"R-squared (test): {r2_score(y_test, y_pred):.4f}")
print(f"MSE (test): {mean_squared_error(y_test, y_pred):.4f}")

poly = PolynomialFeatures(degree=2, include_bias=False)
X_poly_sk = poly.fit_transform(X_train)
X_poly_test = poly.transform(X_test)

lr_poly = SklearnLR()
lr_poly.fit(X_poly_sk, y_train)
print(f"\nPolynomial degree 2 R-squared: {r2_score(y_test, lr_poly.predict(X_poly_test)):.4f}")

scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

ridge = Ridge(alpha=1.0)
ridge.fit(X_train_scaled, y_train)
print(f"Ridge R-squared: {r2_score(y_test, ridge.predict(X_test_scaled)):.4f}")
print(f"Ridge coefficient: {ridge.coef_[0]:.4f}")

维度	手写实现	scikit-learn
结果	与 sklearn 几乎一致	工业级稳定
关键差异	仅用于理解原理	处理边界情况、数值稳定、性能优化
适用场景	教学、看清每一步	生产环境

四、可复用产物

本节产出:

outputs/skill-regression.md —— 一个根据问题特点选择合适回归方法的技能文档。

五、练习

实现批量梯度下降、随机梯度下降(SGD)、小批量梯度下降,在同一数据集上比较收敛速度。哪个最快?哪个代价曲线最平滑?
从三次函数(y = ax³ + bx² + cx + d + 噪声)生成数据,拟合 1、3、10 次多项式,比较训练 R² 和测试 R²。从几次开始过拟合变得明显?
实现 Lasso 回归(L1 正则化:惩罚 = alpha * sum(|w_i|)),在多特征房价数据上训练,对比哪些权重归零,与 Ridge 有何不同。为什么 L1 产生稀疏解而 L2 不会?

本节要点回顾

线性回归是 ML 的骨架:模型 y=wx+b、代价函数 MSE、梯度下降优化,这套三段式贯穿所有 ML 算法。
MSE 平方有理:大误差被指数级惩罚,且处处可导便于优化;代价曲面是凸碗,碗底即最优。
梯度下降三件套:算预测、算梯度、按学习率更新参数,循环至代价收敛。
正规方程一步到位:w = (X^T X)^(-1) X^T y,小数据集快,大数据集因 O(n³) 求逆败给梯度下降。
多元靠特征缩放:不同尺度特征会拉长代价曲面,标准化(减均值除标准差)是收敛前提。
多项式仍是线性:对 x 造高次特征,但模型对权重线性;高次易过拟合。
R² 与尺度无关:1.0 完美、0.0 等于预测均值、负值比预测均值还差。
岭回归(L2)抑制大权重:Cost = MSE + lambda*sum(w_i²),lambda 越大权重越被往零压,防过拟合。
手写 vs sklearn 结果一致:库赢在边界处理、数值稳定、性能,理解用前者、生产用后者。

下一节,我们把回归的「连续输出」换成「离散类别」,进入逻辑回归——用 sigmoid 把线性组合压成概率,开启分类世界。