【PaddlePaddle】基础理论教程 - 机器学习理论实践

# 导入必要库import paddlefrom sklearn.datasets import fetch_california_housingimport pandas as pd# 加载加州房价数据集housing = fetch_california_housing(as_frame=True)# 转换为DataFramedata = housing.frame# 显示数据集基本信息print("数据集基本信息：")print(data.info())# 显示前几行样本print("数据集示例：")print(data.head())# 查看目标变量分布print("目标变量分布（房价中位数）：")print(data['MedHouseVal'].describe())

# 检查缺失值missing_values = data.isnull().sum()print("各列缺失值情况：\n", missing_values)# 若存在缺失值，可以选择删除或填充# 删除缺失值样本cleaned_data = data.dropna()# 或者使用均值填充filled_data = data.fillna(data.mean())print("清洗后的数据集基本信息：")print(cleaned_data.info())

# 统计每列的上下四分位数和异常值范围for col in data.columns[:-1]:  # 遍历所有特征列
    Q1 = data[col].quantile(0.25)
    Q3 = data[col].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR    
    # 过滤异常值
    outliers = data[(data[col] < lower_bound) | (data[col] > upper_bound)]    print(f"{col} 列异常值数量：{len(outliers)}")    
    # 去除异常值（可选）
    data = data[(data[col] >= lower_bound) & (data[col] <= upper_bound)]

!pip install seaborn

import matplotlib.pyplot as pltimport seaborn as sns# 设置画图风格sns.set(style="whitegrid")# 选择部分特征绘制箱线图selected_features = ['MedInc', 'HouseAge', 'AveRooms', 'AveBedrms', 'Population']# 绘制箱线图plt.figure(figsize=(12, 6))
sns.boxplot(data=data[selected_features])
plt.title("Boxplot of dataset feature.", fontsize=16)
plt.xticks(fontsize=12)
plt.ylabel("value", fontsize=12)
plt.xlabel("feature", fontsize=12)
plt.show()

# # 使用最大最小归一化import numpy as npfrom sklearn.model_selection import train_test_split# 提取特征与目标变量features = data.iloc[:, :-1].values
target = data.iloc[:, -1].values# # 最小-最大归一化实现# def min_max_scaling(features):#     min_vals = np.min(features, axis=0)#     max_vals = np.max(features, axis=0)#     return (features - min_vals) / (max_vals - min_vals)# # # 应用归一化# # normalized_features = min_max_scaling(features)# # 数据集划分# X_train, X_test, y_train, y_test = train_test_split(normalized_features, target, test_size=0.2, random_state=42)

# 标准化from sklearn.preprocessing import StandardScaler# 标准化实现scaler = StandardScaler()
standardized_features = scaler.fit_transform(features)# 数据集划分X_train, X_test, y_train, y_test = train_test_split(standardized_features, target, test_size=0.2, random_state=42)

import paddleimport paddle.nn.functional as F# 将数据转换为Paddle张量X_train_tensor = paddle.to_tensor(X_train, dtype='float32')
X_test_tensor = paddle.to_tensor(X_test, dtype='float32')# 使用Paddle实现标准化mean = paddle.mean(X_train_tensor, axis=0)
std = paddle.std(X_train_tensor, axis=0)
X_train_normalized = (X_train_tensor - mean) / std
X_test_normalized = (X_test_tensor - mean) / stdprint("标准化后的训练数据：", X_train_normalized.numpy()[:5])

import matplotlib.pyplot as plt

plt.figure(figsize=(12, 6))
plt.boxplot(features, labels=data.columns[:-1])
plt.title("Feature distribution before normalization") # 归一化前特征分布plt.xticks(rotation=45)
plt.show()

plt.figure(figsize=(12, 6))
plt.boxplot(standardized_features)
plt.title("Feature distribution after normalization") # 归一化后特征分布（Min-Max）plt.show()

import paddleimport paddle.nn as nnimport paddle.nn.functional as F# 自定义Linear算子class Linear(nn.Layer):
    def __init__(self, in_features, out_features):
        super(Linear, self).__init__()        # 初始化权重和偏置
        self.weight = self.create_parameter(
            shape=[in_features, out_features],
            default_initializer=nn.initializer.XavierUniform())
        self.bias = self.create_parameter(
            shape=[out_features],
            default_initializer=nn.initializer.Constant(0.0))    def forward(self, x):
        # 线性变换公式：y = Wx + b
        return paddle.matmul(x, self.weight) + self.bias

# Runner类定义class Runner:
    def __init__(self, input_dim, regularization_lambda=0.01):
        """
        初始化线性回归模型以及其他必需的参数
        """
        self.model = Linear(in_features=input_dim, out_features=1)
        self.loss_fn = F.mse_loss  # 均方误差损失
        self.optimizer = paddle.optimizer.SGD(parameters=self.model.parameters(), learning_rate=0.01)
        self.regularization_lambda = regularization_lambda  # L2正则化项

    def prepare_model(self, X_train, y_train):
        """
        使用正规方程（带L2正则化）来求解模型的权重和偏置
        """
        # 转换为Paddle张量
        X_train_tensor = paddle.to_tensor(X_train, dtype='float32')
        y_train_tensor = paddle.to_tensor(y_train, dtype='float32')        # 添加偏置项到 X_train
        X_train_with_bias = paddle.concat([X_train_tensor, paddle.ones([X_train_tensor.shape[0], 1], dtype='float32')], axis=1)        # 正规方程解析解：W = (X^T X + λI)^-1 X^T y
        X_transpose = paddle.transpose(X_train_with_bias, perm=[1, 0])        
        # 加上正则化项 (λI)
        identity_matrix = paddle.eye(X_train_with_bias.shape[1])
        regularization_matrix = self.regularization_lambda * identity_matrix        
        # 计算正规方程解
        weights = paddle.matmul(
            paddle.inverse(paddle.matmul(X_transpose, X_train_with_bias) + regularization_matrix),
            paddle.matmul(X_transpose, y_train_tensor)
        )        # 拆分权重和偏置
        self.model.weight.set_value(weights[:-1].reshape([X_train.shape[1], 1]))  # 权重调整为二维张量
        self.model.bias.set_value(weights[-1:].reshape([1]))  # 偏置调整为一维张量

        print("解析解求得的模型参数：")        print("权重：", self.model.weight.numpy())        print("偏置：", self.model.bias.numpy())    def evaluate_model(self, X_test, y_test):
        """
        使用测试集进行模型评价
        """
        X_test_tensor = paddle.to_tensor(X_test, dtype='float32')
        y_test_tensor = paddle.to_tensor(y_test, dtype='float32')        # 进行预测
        predictions = self.model(X_test_tensor)
        loss = self.loss_fn(predictions, y_test_tensor)        print("模型评价 - 测试集均方误差：", loss.numpy())        return predictions.numpy()    def predict(self, X_new):
        """
        对新数据进行预测
        """
        X_new_tensor = paddle.to_tensor(X_new, dtype='float32')
        predictions = self.model(X_new_tensor)        return predictions.numpy()

# 加载数据X_train, X_test, y_train, y_test = X_train_normalized, X_test_normalized, y_train, y_test# 创建 Runner 实例runner = Runner(input_dim=X_train.shape[1])# 使用最小二乘法解析解求解模型参数runner.prepare_model(X_train, y_train)# 评价模型predictions = runner.evaluate_model(X_test, y_test)# 对新数据进行预测X_new = X_test[:5]  # 假设我们使用测试集的前5条数据进行预测predicted_prices = runner.predict(X_new)print("预测结果（前5条）：", predicted_prices)

import paddle# 定义 Logistic 函数def Logistic(x):
    return 1 / (1 + paddle.exp(-x))# 创建一组从 -10 到 10 的数值x = paddle.linspace(-10, 10, 10000)# 计算 Logistic 函数值y = Logistic(x)# 打印结果的前几项print(y[:10])  # 输出前10项# 可视化 Logistic 函数import matplotlib.pyplot as plt

plt.plot(x.tolist(), y.tolist(),color = 'red')
plt.title('Sigmoid Function (Logistic)')
plt.xlabel('z')
plt.ylabel('σ(z)')
plt.grid(True)
plt.show()

# 使用 `sklearn.datasets.make_moons` 来生成一个二分类的数据集。随机抽取其中的 1000 个样本，import numpy as npimport matplotlib.pyplot as pltfrom sklearn.datasets import make_moonsfrom sklearn.preprocessing import StandardScaler# 生成 Moon 数据集X, y = make_moons(n_samples=1000, noise=0.1, random_state=42)# 数据标准化scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)# 可视化数据分布plt.figure(figsize=(8, 6))
plt.scatter(X_scaled[y == 0][:, 0], X_scaled[y == 0][:, 1], color='b', label='Class 0', s=10)
plt.scatter(X_scaled[y == 1][:, 0], X_scaled[y == 1][:, 1], color='r', label='Class 1', s=10)
plt.title('Moon1000 Dataset Visualization')
plt.xlabel('Feature 1')
plt.ylabel('Feature 2')
plt.legend()
plt.grid(True)
plt.show()

# 拆分数据集（训练集、验证集、测试集）from sklearn.model_selection import train_test_split# 第一次拆分：将数据集分为训练集（60%）和临时集（40%）X_train, X_temp, y_train, y_temp = train_test_split(X_scaled, y, test_size=0.4, random_state=42)# 第二次拆分：将临时集分为验证集（20%）和测试集（20%）X_val, X_test, y_val, y_test = train_test_split(X_temp, y_temp, test_size=0.5, random_state=42)# 打印数据集形状print(f"训练集特征 X_train 形状: {X_train.shape}")print(f"验证集特征 X_val 形状: {X_val.shape}")print(f"测试集特征 X_test 形状: {X_test.shape}")print(f"训练集标签 y_train 形状: {y_train.shape}")print(f"验证集标签 y_val 形状: {y_val.shape}")print(f"测试集标签 y_test 形状: {y_test.shape}")

import paddleimport paddle.nn as nnimport paddle.optimizer as optimimport paddle.nn.functional as F# 自定义 Logistic 回归模型（仅用 paddle.matmul）class LogisticRegression(nn.Layer):
    def __init__(self, input_dim):
        super(LogisticRegression, self).__init__()        # 定义线性层
        self.params = {}
        self.params['w'] = paddle.zeros([input_dim, 1])
        self.params['b'] = paddle.zeros(shape = [1])
        self.fc = nn.Linear(input_dim, 1)    def forward(self, x):
        # 使用 paddle.matmul 计算加权和
        logits = paddle.matmul(x, self.params['w']) + self.params['b']
        outputs = Logistic(logits)        return outputs  # 这里没有使用激活函数 sigmoid

paddle.seed(0)input = paddle.randn([3, 4])print(input)

model = LogisticRegression(4)
output = model(input)print(output)

import paddleimport paddle.nn as nnclass CrossEntropyLoss(nn.Layer):
    def __init__(self):
        super(CrossEntropyLoss, self).__init__()    def forward(self, logits, labels):
        """
        计算交叉熵损失
        :param logits: 预测值（通常是模型的输出，没有经过 softmax 的原始值）
        :param labels: 真实标签，通常是 one-hot 编码
        :return: 交叉熵损失
        """
        # 使用 PaddlePaddle 的函数计算交叉熵损失
        loss = nn.functional.cross_entropy(logits, labels)        return loss# 示例：使用自定义交叉熵损失函数logits = paddle.to_tensor([[1.2, 0.3, -0.5], [0.7, 1.4, -1.0]], dtype='float32')  # 模拟模型输出labels = paddle.to_tensor([0, 1], dtype='int64')  # 假设有两个样本，标签分别为类别 0 和类别 1loss_fn = CrossEntropyLoss()
loss = loss_fn(logits, labels)print("交叉熵损失：", loss.numpy())

import paddleimport paddle.nn as nn# 自定义 Logistic 回归模型（仅用 paddle.matmul）class LogisticRegression(nn.Layer):
    def __init__(self, input_dim):
        super(LogisticRegression, self).__init__()        # 定义线性层
        self.params = {}
        self.params['w'] = paddle.zeros([input_dim, 1])  # 初始化权重
        self.params['b'] = paddle.zeros(shape=[1])  # 初始化偏置
        self.fc = nn.Linear(input_dim, 1)    def forward(self, x):
        # 使用 paddle.matmul 计算加权和
        logits = paddle.matmul(x, self.params['w']) + self.params['b']
        outputs = self.sigmoid(logits)        return outputs    def sigmoid(self, x):
        # Sigmoid 激活函数
        return 1 / (1 + paddle.exp(-x))    def backward(self, x, y):
        """
        计算损失函数对模型参数的梯度，并将其存放在 grads 属性中。
        :param x: 输入特征
        :param y: 真实标签
        :return: None
        """
        # 计算预测输出
        logits = paddle.matmul(x, self.params['w']) + self.params['b']
        y_pred = self.sigmoid(logits)        # 计算损失函数对权重的偏导数
        dw = paddle.matmul(paddle.transpose(x, [1, 0]), (y_pred - y)) / x.shape[0]
        db = paddle.sum(y_pred - y) / x.shape[0]        # 将梯度存放在 grads 属性中
        self.grads = {}
        self.grads['w'] = dw
        self.grads['b'] = db        # 输出计算的梯度
        print("梯度 w：", dw.numpy())        print("梯度 b：", db.numpy())

# 使用 backward 函数进行梯度计算# 模型创建与输入数据生成input_dim = 2model = LogisticRegression(input_dim)# 随机生成输入数据和标签x_sample = paddle.to_tensor([[0.5, 1.0], [1.5, -1.0], [1.0, 2.0]], dtype='float32')
y_sample = paddle.to_tensor([[1], [0], [1]], dtype='float32')# 前向计算outputs = model(x_sample)print("模型输出：", outputs.numpy())# 计算梯度model.backward(x_sample, y_sample)# 查看存储的梯度print("存储在 grads 中的权重梯度：", model.grads['w'].numpy())print("存储在 grads 中的偏置梯度：", model.grads['b'].numpy())

import paddleimport paddle.nn as nn# 自定义 Logistic 回归模型（仅用 paddle.matmul）class LogisticRegression(nn.Layer):
    def __init__(self, input_dim):
        super(LogisticRegression, self).__init__()        # 定义线性层
        self.params = {}
        self.params['w'] = paddle.zeros([input_dim, 1])  # 初始化权重
        self.params['b'] = paddle.zeros(shape=[1])  # 初始化偏置
        self.fc = nn.Linear(input_dim, 1)    def forward(self, x):
        # 使用 paddle.matmul 计算加权和
        logits = paddle.matmul(x, self.params['w']) + self.params['b']
        outputs = self.sigmoid(logits)        return outputs    def sigmoid(self, x):
        # Sigmoid 激活函数
        return 1 / (1 + paddle.exp(-x))    def backward(self, x, y):
        """
        计算损失函数对模型参数的梯度，并将其存放在 grads 属性中。
        :param x: 输入特征
        :param y: 真实标签
        :return: None
        """
        # 计算预测输出
        logits = paddle.matmul(x, self.params['w']) + self.params['b']
        y_pred = self.sigmoid(logits)        # 计算损失函数对权重的偏导数
        dw = paddle.matmul(paddle.transpose(x, [1, 0]), (y_pred - y)) / x.shape[0]
        db = paddle.sum(y_pred - y) / x.shape[0]        # 将梯度存放在 grads 属性中
        self.grads = {}
        self.grads['w'] = dw
        self.grads['b'] = db        # 输出计算的梯度
        # print("梯度 w：", dw.numpy())
        # print("梯度 b：", db.numpy())

    def update_params(self, lr=0.01):
        """
        使用梯度下降法更新模型的权重和偏置
        :param lr: 学习率
        :return: None
        """
        # 更新权重和偏置
        self.params['w'] = self.params['w'] - lr * self.grads['w']
        self.params['b'] = self.params['b'] - lr * self.grads['b']        # print("更新后的权重 w：", self.params['w'].numpy())
        # print("更新后的偏置 b：", self.params['b'].numpy())

# 将数据集转换为 Paddle 张量X_train_tensor = paddle.to_tensor(X_train, dtype='float32')
y_train_tensor = paddle.to_tensor(y_train.reshape(-1, 1), dtype='float32')  # 确保标签形状匹配# 构建模型input_dim = X_train.shape[1]  # 输入特征的维度model = LogisticRegression(input_dim)# 训练过程epochs = 10000learning_rate = 0.01for epoch in range(epochs):    # 前向计算
    outputs = model(X_train_tensor)    
    # 计算梯度
    model.backward(X_train_tensor, y_train_tensor)    
    # 更新参数
    model.update_params(lr=learning_rate)    
    # 每200回合打印一次损失值
    if (epoch + 1) % 200 == 0:        # 计算预测值与实际标签的损失
        logits = paddle.matmul(X_train_tensor, model.params['w']) + model.params['b']
        predictions = model.sigmoid(logits)
        loss = paddle.mean(-y_train_tensor * paddle.log(predictions) - (1 - y_train_tensor) * paddle.log(1 - predictions))        print(f"Epoch {epoch + 1}/{epochs}, Loss: {loss.numpy()}")

# 模型评估def evaluate_model(model, X_test, y_test):
    # 前向计算
    logits = paddle.matmul(X_test, model.params['w']) + model.params['b']
    predictions = model.sigmoid(logits)    
    # 计算交叉熵损失
    loss = paddle.mean(-y_test * paddle.log(predictions) - (1 - y_test) * paddle.log(1 - predictions))    print(f"测试集损失: {loss.numpy()}")    
    # 预测标签
    predicted_labels = (predictions > 0.5).astype('float32')  # 将 predicted_labels 转为 float32
    
    # 计算准确率
    accuracy = paddle.mean(paddle.cast(predicted_labels == y_test, dtype='float32'))    
    print(f"测试集准确率: {accuracy.numpy()}")    return predicted_labels.numpy()# 将数据转换为 Paddle 张量X_test_tensor = paddle.to_tensor(X_test, dtype='float32')
y_test_tensor = paddle.to_tensor(y_test, dtype='float32')

predicted_labels = evaluate_model(model, X_test_tensor, y_test_tensor)# print("预测标签:", predicted_labels)

<br/>

# 将数据转换为 Paddle 张量X_test_tensor = paddle.to_tensor(X_test, dtype='float32')
y_test_tensor = paddle.to_tensor(y_test, dtype='float32')# 测试阶段的预测predictions = model(X_test_tensor)  # 概率值predicted_labels = (predictions > 0.5).astype('int32')  # 转为二分类标签# # 检查预测结果分布# print("预测概率分布:", predictions.numpy().flatten())# print("预测标签分布:", predicted_labels.numpy().flatten())# 可视化对比plt.figure(figsize=(10, 6))# 实际标签可视化plt.subplot(1, 2, 1)
plt.scatter(X_test[:, 0], X_test[:, 1], c=y_test.flatten(), cmap='bwr', alpha=0.6)
plt.title('Actual Labels')
plt.xlabel('Feature 1')
plt.ylabel('Feature 2')# 预测标签可视化plt.subplot(1, 2, 2)
plt.scatter(X_test[:, 0], X_test[:, 1], c=predicted_labels.numpy().flatten(), cmap='bwr', alpha=0.6)
plt.title('Predicted Labels')
plt.xlabel('Feature 1')
plt.ylabel('Feature 2')

plt.tight_layout()
plt.show()