1
2
3
4
5
6
import numpy as np
import pandas as pd 
import matplotlib.pyplot as plt
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split # 分割数据集，分成训练集和测试集
from sklearn.metrics import mean_squared_error   # 计算MSE

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import numpy as np

def load_and_clean_data(file_path):
    """
    从指定文件中读取数据，清理字符串格式，并返回 NumPy 数组。
    数据格式预期为：[[x], [y]]
    """
    cleaned_data_list = []
    line_number = 0 # 用于跟踪行号

    try:
        with open(file_path, 'r') as f:
            for line in f:
                line_number += 1
                
                # 1. 清理行首和行尾的空白字符
                line_stripped = line.strip()
                
                # 严格跳过空行或仅包含空白字符的行
                if not line_stripped:
                    # 如果需要调试，可以取消注释下面这行
                    # print(f"跳过文件中的空行或空白行 (行号: {line_number})")
                    continue
                    
                # 2. 移除所有方括号 '[]' 和空格，只留下数字和逗号
                # 示例: "[[5.34...], [30.91...]]" -> "5.34...,30.91..."
                clean_value_str = line_stripped.replace('[', '').replace(']', '').replace(' ', '')
                
                # 3. 按逗号分隔，得到 X 和 y 的字符串
                x_str, y_str = clean_value_str.split(',')
                
                # 4. 转换为浮点数并存储
                x_val = float(x_str)
                y_val = float(y_str)
                cleaned_data_list.append([x_val, y_val])
                
    except FileNotFoundError:
        print(f"❌ 错误：未找到文件 {file_path}。请检查文件路径。")
        return None, None
    except ValueError as e:
        print(f"❌ 错误：数据格式不正确，无法转换为浮点数。请检查数据行 (行号: {line_number}): {line_stripped}")
        print(f"具体错误信息: {e}")
        return None, None
    except Exception as e:
        print(f"❌ 发生未知错误: {e}")
        return None, None

    # 5. 转换为最终的 NumPy 数组
    cleaned_data_np = np.array(cleaned_data_list, dtype=np.float64)
    
    # 6. 分离特征 (X) 和目标值 (y)
    # X 必须是二维数组 (n, 1)
    X = cleaned_data_np[:, 0].reshape(-1, 1)
    y = cleaned_data_np[:, 1] # y 是一维数组 (n,)
    
    return X, y

# ----------------- 修改后的使用示例 -----------------
file_name = 'train_data' # 建议使用 .txt 扩展名，以明确文件类型

# **注意:** 请确保您的数据已保存到指定的文件中
X, y = load_and_clean_data(file_name)

if X is not None and y is not None:
    print("--- ✅ 数据读取成功 ---")
    print(f"特征 X 的形状: {X.shape}")
    print(f"目标 y 的形状: {y.shape}")
    print("X 的前 5 个样本:\n", X[:5])
    print("y 的前 5 个样本:\n", y[:5])
    print("\n数据已清理并准备好进行模型训练（未分割数据集）。")
    
    # **********************************************
    # 移除的分割代码原本在此处，现在直接对 X 和 y 进行操作
    # **********************************************
    
    # 现在您可以直接使用整个 X 和 y 进行模型训练（例如，训练集就是整个数据集）
    # from sklearn.linear_model import LinearRegression
    # model = LinearRegression()
    # model.fit(X, y) # 直接使用所有数据进行训练

1
2
3
4
5
6
7
X_train, X_test, y_train, y_test = train_test_split(
    X, y, 
    test_size=0.2, 
    random_state=42 # 设置随机种子以确保结果可重现
)
print(f"训练集大小 (X_train): {X_train.shape}")
print(f"测试集大小 (X_test): {X_test.shape}")

1
2
3
4
5
6
7
model = LinearRegression()
model.fit(X_train, y_train)
# 打印模型参数 (截距和斜率)
# 对于一元线性回归： y = coef * X + intercept
print("\n--- 模型训练结果 ---")
print(f"模型截距 (Intercept): {model.intercept_:.4f}")
print(f"模型系数 (Coefficient/Slope): {model.coef_[0]:.4f}")

1
2
3
4
5
6
7
8
9
10
11
12
# 1. 对训练集进行预测
y_train_pred = model.predict(X_train)
# 2. 对测试集进行预测
y_test_pred = model.predict(X_test)

# 计算 MSE
mse_train = mean_squared_error(y_train, y_train_pred)
mse_test = mean_squared_error(y_test, y_test_pred)

print("\n--- 模型评估 (MSE) ---")
print(f"训练集 MSE: {mse_train:.4f}")
print(f"测试集 MSE: {mse_test:.4f}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
# -----------------------------------------------------------
# 步骤 3: 绘图
# -----------------------------------------------------------

# Matplotlib 绘图设置
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False  # 解决负号显示的问题

# **关键修正步骤：排序**
# 1. 对完整的 X 数组进行排序，以确保拟合线是一条平滑的直线
#    np.argsort 返回排序后的索引
sort_index = np.argsort(X.flatten()) # 注意：X是二维数组，需要先展平
X_sorted = X[sort_index]
# 2. 对模型在 X 上的预测值也使用相同的索引进行排序
y_full_pred_sorted = model.predict(X_sorted)


# --- 绘图要求 1: 所有点和预测线一起展示 (修正) ---
plt.figure(figsize=(10, 6))

# 绘制所有原始数据点 (训练集和测试集)
plt.scatter(X_train, y_train, color='blue', label='训练集数据点', alpha=0.6)
plt.scatter(X_test, y_test, color='green', label='测试集数据点', alpha=0.8)

# 绘制预测线 (拟合线) - 使用排序后的预测数据
# 这将确保红色线是一条平滑的直线
plt.plot(X_sorted, y_full_pred_sorted, color='red', linestyle='-', 
         label='线性回归拟合线 (y = {:.4f}*X + {:.4f})'.format(model.coef_[0], model.intercept_), 
         linewidth=2)

plt.title('所有数据点与线性回归拟合线 (总览)')
plt.xlabel('特征 X')
plt.ylabel('目标 Y')
plt.legend()
plt.grid(True, linestyle='--', alpha=0.6)
plt.show()

# --- 绘图要求 2 & 3: 训练集和测试集分开展示（无需排序，因为点本身就是散点图） ---

fig, axes = plt.subplots(1, 2, figsize=(15, 6), sharey=True)

# 2. 训练集 (Training Set) - 绘制训练集数据和预测线
axes[0].scatter(X_train, y_train, color='blue', label='训练集数据点', alpha=0.6)

# **注意：** 这里的 X_train 依然可能未排序，因此也需要排序以防折叠
sort_index_train = np.argsort(X_train.flatten())
X_train_sorted = X_train[sort_index_train]
y_train_pred_sorted = model.predict(X_train_sorted)

axes[0].plot(X_train_sorted, y_train_pred_sorted, color='red', linestyle='-', label='模型预测线', linewidth=2)
axes[0].set_title('训练集数据与模型拟合')
axes[0].set_xlabel('特征 X')
axes[0].set_ylabel('目标 Y')
axes[0].legend()
axes[0].grid(True, linestyle='--', alpha=0.6)

# 3. 测试集 (Testing Set) - 绘制测试集数据和预测线
axes[1].scatter(X_test, y_test, color='green', label='测试集数据点', alpha=0.8)

# **注意：** 这里的 X_test 也需要排序以防折叠
sort_index_test = np.argsort(X_test.flatten())
X_test_sorted = X_test[sort_index_test]
y_test_pred_sorted = model.predict(X_test_sorted)

axes[1].plot(X_test_sorted, y_test_pred_sorted, color='red', linestyle='-', label='模型预测线', linewidth=2)
axes[1].set_title('测试集数据与模型预测')
axes[1].set_xlabel('特征 X')
axes[1].legend()
axes[1].grid(True, linestyle='--', alpha=0.6)

plt.tight_layout()
plt.show()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# 使用 sklearn 找到最优 w* 和 b*
optimal_model = LinearRegression().fit(X, y)
w_star = optimal_model.coef_[0]
b_star = optimal_model.intercept_
print(f"最优参数 w*: {w_star:.4f}, b*: {b_star:.4f}")

# 定义 MSE 计算函数
def calculate_mse(w, b, X_data, y_data):
    """计算给定 w 和 b 下的均方误差"""
    # 预测值: y_pred = w * X + b
    y_pred = w * X_data + b
    # MSE: mean((y - y_pred)^2)
    return np.mean((y_data - y_pred.flatten())**2)

# -----------------------------------------------------------
# 步骤 2: 绘制 MSE vs W (固定 b = b*)
# -----------------------------------------------------------

# 选择 w 的变化范围 (以最优 w* 为中心)
w_values = np.linspace(w_star - 5, w_star + 5, 100)
# 固定 b 为最优值
b_fixed = b_star

# 计算 MSE 数组
mse_w = [calculate_mse(w, b_fixed, X, y) for w in w_values]

plt.figure(figsize=(8, 6))
plt.plot(w_values, mse_w, color='blue', linewidth=3)
plt.scatter(w_star, calculate_mse(w_star, b_star, X, y), color='red', s=100, label=f'最小MSE点 (w*={w_star:.2f})', zorder=5)

plt.title(f'MSE 随 W (斜率) 的变化 (固定 b = {b_fixed:.2f})')
plt.xlabel('权重 W (斜率)')
plt.ylabel('均方误差 MSE')
plt.grid(True, linestyle='--', alpha=0.6)
plt.legend()
plt.show()

# -----------------------------------------------------------
# 步骤 3: 绘制 MSE vs B (固定 w = w*)
# -----------------------------------------------------------

# 选择 b 的变化范围 (以最优 b* 为中心)
b_values = np.linspace(b_star - 10, b_star + 10, 100)
# 固定 w 为最优值
w_fixed = w_star

# 计算 MSE 数组
mse_b = [calculate_mse(w_fixed, b, X, y) for b in b_values]

plt.figure(figsize=(8, 6))
plt.plot(b_values, mse_b, color='green', linewidth=3)
plt.scatter(b_star, calculate_mse(w_star, b_star, X, y), color='red', s=100, label=f'最小MSE点 (b*={b_star:.2f})', zorder=5)

plt.title(f'MSE 随 B (截距) 的变化 (固定 w = {w_fixed:.2f})')
plt.xlabel('偏置 B (截距)')
plt.ylabel('均方误差 MSE')
plt.grid(True, linestyle='--', alpha=0.6)
plt.legend()
plt.show()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import numpy as np
import matplotlib.pyplot as plt
from sklearn.linear_model import LinearRegression # 用于找到理论最优解

# Matplotlib 绘图设置 (用于显示中文)
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# -----------------------------------------------------------
# 步骤 1: 模拟数据和最优参数 (请用您的实际 X 和 y 替换)
# -----------------------------------------------------------
np.random.seed(42)
# 500 个样本
X = np.linspace(0, 10, 500).reshape(-1, 1)
# 真实模型: y = 5 + 3*X + 噪声
y = 5 + 3 * X.flatten() + np.random.normal(0, 5, 500)
n_samples = X.shape[0]

# 使用 sklearn 找到理论最优解
optimal_model = LinearRegression().fit(X, y)
w_star = optimal_model.coef_[0]
b_star = optimal_model.intercept_
print(f"理论最优参数 w*: {w_star:.4f}, b*: {b_star:.4f}")

# -----------------------------------------------------------
# 步骤 2: 通用梯度下降实现
# -----------------------------------------------------------

def custom_gradient_descent(X, y, learning_rate, n_epochs, batch_size):
    """
    运行梯度下降并记录参数 w 的变化路径。
    batch_size = n_samples -> BGD
    batch_size = 1        -> SGD
    batch_size = 32       -> MBGD
    """
    w = 0.0 # 初始参数 w
    b = 0.0 # 初始参数 b
    w_history = []
    
    for epoch in range(n_epochs):
        # 每次 Epoch 重新打乱数据，对 SGD/MBGD 尤其重要
        if batch_size != n_samples:
            indices = np.random.permutation(n_samples)
            X_shuffled = X[indices]
            y_shuffled = y[indices]
        else:
            X_shuffled = X
            y_shuffled = y
            
        for i in range(0, n_samples, batch_size):
            # 提取当前批次数据
            X_batch = X_shuffled[i:i + batch_size]
            y_batch = y_shuffled[i:i + batch_size]
            m_batch = len(X_batch)
            
            # 预测值
            y_pred = w * X_batch.flatten() + b
            
            # 计算梯度 (仅针对当前批次)
            error = y_pred - y_batch
            dw = (1 / m_batch) * np.sum(error * X_batch.flatten())
            db = (1 / m_batch) * np.sum(error)
            
            # 更新参数
            w = w - learning_rate * dw
            b = b - learning_rate * db
            
            w_history.append(w)
            
    return w_history

# -----------------------------------------------------------
# 步骤 3: 运行三种策略
# -----------------------------------------------------------

# 统一参数设置 (不同的学习率和迭代次数可能需要调整以获得最佳展示效果)
N_EPOCHS = 20
LEARNING_RATE = 0.005 # 适用于 BGD/MBGD
LR_SGD = 0.0005 # SGD 通常需要较小的学习率来减缓震荡

# 1. 批量梯度下降 (BGD)
w_history_bgd = custom_gradient_descent(X, y, learning_rate=LEARNING_RATE, n_epochs=N_EPOCHS, batch_size=n_samples)

# 2. 小批量梯度下降 (MBGD) - 常用 Batch Size = 32
w_history_mbgd = custom_gradient_descent(X, y, learning_rate=LEARNING_RATE, n_epochs=N_EPOCHS, batch_size=32)

# 3. 随机梯度下降 (SGD) - Batch Size = 1
w_history_sgd = custom_gradient_descent(X, y, learning_rate=LR_SGD, n_epochs=N_EPOCHS, batch_size=1)


# -----------------------------------------------------------
# 步骤 4: 画图展示 w 的收敛路径
# -----------------------------------------------------------

plt.figure(figsize=(10, 6))

# 绘制三种策略的 w 历史路径
plt.plot(w_history_bgd, label='BGD (Batch Gradient Descent)', color='blue', linewidth=2)
plt.plot(w_history_mbgd, label='MBGD (Mini-Batch Gradient Descent, Batch=32)', color='orange', alpha=0.7)
plt.plot(w_history_sgd, label='SGD (Stochastic Gradient Descent)', color='red', alpha=0.5)

# 绘制理论最优值 w*
plt.axhline(w_star, color='green', linestyle='--', label=f'理论最优解 $w^*$={w_star:.4f}', linewidth=2)

plt.title(f'不同梯度下降策略下参数 $w$ 的寻优路径 (迭代次数={N_EPOCHS} Epochs)')
plt.xlabel('参数更新次数 (Iterations)')
plt.ylabel('权重参数 $w$ 的值')
plt.legend()
plt.grid(True, linestyle='--', alpha=0.6)
plt.ylim(w_star - 1, w_star + 1) # 聚焦于收敛区域
plt.show()