06 模型验证

import numpy as np
import matplotlib.pyplot as plt
import matplotlib as mpl

# 设置中文字体
# 注意：这需要在您的环境中安装支持中文的字体，例如 "SimHei"
# 如果没有，matplotlib会回退到默认字体，中文可能显示为方框
mpl.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'Arial Unicode MS']
mpl.rcParams['axes.unicode_minus'] = False # 解决负号显示问题



x = np.linspace(0.1, 1, 100)
bias_sq = 0.8 / (x * 10)
variance = x * 0.4
noise = np.full_like(x, 0.1)
total_error = bias_sq + variance + noise

min_error_idx = np.argmin(total_error)
optimal_complexity = x[min_error_idx]

fig, ax = plt.subplots(figsize=(10, 6))
ax.plot(x, bias_sq, 'r-', label='偏差平方 (Bias²)')
ax.plot(x, variance, 'b--', label='方差 (Variance)')
ax.plot(x, noise, 'orange', linestyle=':', label='噪声 (Irreducible Error)')
ax.plot(x, total_error, 'k-.', label='总误差 (Total Error)', linewidth=2)
ax.axvline(optimal_complexity, color='grey', linestyle='--', label='最佳复杂度')

ax.fill_between(x, 0, 1.2, where=x < optimal_complexity, color='lightblue', alpha=0.3, label='欠拟合区域')
ax.fill_between(x, 0, 1.2, where=x > optimal_complexity, color='lightcoral', alpha=0.3, label='过拟合区域')

ax.set_xlabel('模型复杂度', fontsize=14)
ax.set_ylabel('误差', fontsize=14)
ax.set_title('偏差-方差权衡', fontsize=16)
ax.legend(fontsize=12)
ax.set_ylim(0, 1.2)
ax.set_xlim(0, 1)

ax.text(optimal_complexity - 0.25, 1.1, '欠拟合\n(Underfitting)', ha='center', fontsize=12)
ax.text(optimal_complexity + 0.25, 1.1, '过拟合\n(Overfitting)', ha='center', fontsize=12)
ax.plot(optimal_complexity, total_error[min_error_idx], 'ko', markersize=8)
ax.annotate('平衡点',
            xy=(optimal_complexity, total_error[min_error_idx]),
            xytext=(optimal_complexity, total_error[min_error_idx] + 0.2),
            arrowprops=dict(facecolor='black', shrink=0.05),
            ha='center', fontsize=12)

plt.show()

import matplotlib.pyplot as plt
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import PolynomialFeatures
from sklearn.linear_model import LinearRegression

# --- 训练集上的拟合 ---
plt.figure(figsize=(14, 5))
plt.suptitle('模型在训练集上的表现', fontsize=16)
for i, degree in enumerate(degrees):
    ax = plt.subplot(1, len(degrees), i + 1)
    plt.setp(ax, xticks=(), yticks=())

    polynomial_features = PolynomialFeatures(degree=degree, include_bias=False)
    linear_regression = LinearRegression()
    pipeline = Pipeline([("poly", polynomial_features), ("lr", linear_regression)])
    pipeline.fit(X_train[:, np.newaxis], y_train)

    X_plot = np.linspace(0, 1, 100)
    plt.plot(X_plot, pipeline.predict(X_plot[:, np.newaxis]), label="模型")
    plt.scatter(X_train, y_train, edgecolor='b', s=20, label="训练样本")
    plt.xlabel("x")
    plt.ylabel("y")
    plt.xlim((0, 1))
    plt.ylim((-2, 2))
    plt.legend(loc="best")
    if degree == 1:
        plt.title(f'欠拟合 (d={degree})')
    elif degree == 4:
        plt.title(f'适当拟合 (d={degree})')
    else:
        plt.title(f'过拟合 (d={degree})')
plt.tight_layout(rect=[0, 0, 1, 0.95])
plt.show()

# --- 验证集上的表现 ---
plt.figure(figsize=(14, 5))
plt.suptitle('模型在验证集上的表现', fontsize=16)
for i, degree in enumerate(degrees):
    ax = plt.subplot(1, len(degrees), i + 1)
    plt.setp(ax, xticks=(), yticks=())

    polynomial_features = PolynomialFeatures(degree=degree, include_bias=False)
    linear_regression = LinearRegression()
    pipeline = Pipeline([("poly", polynomial_features), ("lr", linear_regression)])
    pipeline.fit(X_train[:, np.newaxis], y_train)

    X_plot = np.linspace(0, 1, 100)
    plt.plot(X_plot, pipeline.predict(X_plot[:, np.newaxis]), label="模型")
    plt.scatter(X_test, y_test, edgecolor='r', s=20, label="验证样本")
    plt.xlabel("x")
    plt.ylabel("y")
    plt.xlim((0, 1))
    plt.ylim((-2, 2))
    plt.legend(loc="best")
    if degree == 1:
        plt.title(f'欠拟合 (高误差)')
    elif degree == 4:
        plt.title(f'适当拟合 (低误差)')
    else:
        plt.title(f'过拟合 (高误差)')
plt.tight_layout(rect=[0, 0, 1, 0.95])
plt.show()

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error

# 为了示例的可重复性，我们在这里创建模拟数据
# 设定种子以确保结果可复现
np.random.seed(42)
train_size = 100
test_size = 50

# 假设的真实关系: eps_basic ~ 2*log(pps) + 3*bm - 1*roa + noise

# 创建训练数据
training_data = pd.DataFrame({
    'stkcd': np.random.randint(1, 1000, train_size),
    'date': pd.to_datetime(pd.date_range(start='2010-01-01', periods=train_size, freq='M')),
    'pps': np.random.uniform(5, 50, train_size),
    'bm': np.random.uniform(0.2, 2, train_size),
    'roa': np.random.uniform(0.01, 0.1, train_size)
})
training_data['eps_basic'] = (2 * np.log(training_data['pps']) + 
                              3 * training_data['bm'] - 
                              1 * training_data['roa'] + 
                              np.random.normal(0, 0.5, train_size)) # 训练集噪声较小

# 创建测试数据
testing_data = pd.DataFrame({
    'stkcd': np.random.randint(1, 1000, test_size),
    'date': pd.to_datetime(pd.date_range(start='2018-01-01', periods=test_size, freq='M')),
    'pps': np.random.uniform(5, 50, test_size),
    'bm': np.random.uniform(0.2, 2, test_size),
    'roa': np.random.uniform(0.01, 0.1, test_size)
})
testing_data['eps_basic'] = (2 * np.log(testing_data['pps']) + 
                             3 * testing_data['bm'] - 
                             1 * testing_data['roa'] + 
                             np.random.normal(0, 1.5, test_size)) # 测试集噪声更大，模拟真实情况

print("模拟数据创建完成。")

# 删除不需要的列
training_data_clean = training_data.drop(columns=['date', 'stkcd'])
testing_data_clean = testing_data.drop(columns=['date', 'stkcd'])

# 训练集分割
X_train = training_data_clean.drop('eps_basic', axis=1)
y_train = training_data_clean['eps_basic']

# 测试集(验证集)分割
X_test = testing_data_clean.drop('eps_basic', axis=1)
y_test = testing_data_clean['eps_basic']

print('--- 训练特征维度 ---')
print(X_train.shape)
print('\n--- 测试特征维度 ---')
print(X_test.shape)

# 1. 创建StandardScaler实例
scaler_X = StandardScaler()

# 2. 在训练数据上拟合参数(均值、标准差)并进行转换
X_train_scaled = scaler_X.fit_transform(X_train)

# 3. 使用在训练数据上计算得到的参数来转换测试数据
X_test_scaled = scaler_X.transform(X_test)

print("特征标准化完成，且未发生数据泄露。")

# 创建并训练模型
ols_model = LinearRegression()
ols_model.fit(X_train_scaled, y_train)

# 在训练集上进行预测和评估
train_predictions = ols_model.predict(X_train_scaled)
train_mse = mean_squared_error(y_train, train_predictions)
train_rmse = np.sqrt(train_mse) # 计算RMSE

# 在测试集上进行预测和评估
test_predictions = ols_model.predict(X_test_scaled)
test_mse = mean_squared_error(y_test, test_predictions)
test_rmse = np.sqrt(test_mse) # 计算RMSE

print(f'训练集 RMSE: {train_rmse:.4f}')
print(f'测试集 RMSE: {test_rmse:.4f}')

if test_rmse > train_rmse * 1.5: # 一个简单的判断规则
    print('\n结论: 测试误差显著高于训练误差，模型存在过拟合迹象。')
else:
    print('\n结论: 模型在训练集和测试集上表现接近，泛化能力良好。')

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import StratifiedKFold
from sklearn.metrics import roc_auc_score, roc_curve

# 创建模拟数据
np.random.seed(123)
data_size = 500
data = pd.DataFrame({
    'employmentYear': np.random.randint(1, 10, data_size) + np.random.choice([np.nan, 0], size=data_size, p=[0.1, 0.9]),
    'annualIncome': np.random.lognormal(11, 0.5, data_size) + np.random.choice([np.nan, 0], size=data_size, p=[0.1, 0.9]),
    'dti': np.random.uniform(5, 35, data_size) + np.random.choice([np.nan, 0], size=data_size, p=[0.1, 0.9])
})
# 创建一个不平衡的目标变量 (约20%违约率)
noise = np.random.normal(0, 5, data_size)
y_score = 0.5*data['employmentYear'] - 0.0001*data['annualIncome'] + 0.1*data['dti'] + noise
data['isDefault'] = (y_score > np.percentile(np.nan_to_num(y_score), 80)).astype(int)

# 定义特征和目标
X = data.drop('isDefault', axis=1).values
y = data['isDefault'].values

print("模拟数据创建完成，包含缺失值和不平衡类别。")
print(f"违约样本比例: {np.mean(y):.2%}")

# 初始化逻辑回归模型
logreg = LogisticRegression(solver='liblinear', random_state=42)

# 初始化分层K折交叉验证
# n_splits=5: 分成5折
# shuffle=True: 在分割前打乱数据
# random_state=42: 确保每次运行的随机分割都一样，便于结果复现
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

fig, ax = plt.subplots(figsize=(10, 8))
aucs = []

# 交叉验证循环
for i, (train_index, test_index) in enumerate(cv.split(X, y)):
    # 1. 数据分割
    X_train, X_test = X[train_index], X[test_index]
    y_train, y_test = y[train_index], y[test_index]
    
    # 2. 数据预处理 (在循环内部！)
    imputer = SimpleImputer(strategy='median')
    X_train_imputed = imputer.fit_transform(X_train)
    X_test_imputed = imputer.transform(X_test)
    
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train_imputed)
    X_test_scaled = scaler.transform(X_test_imputed)
    
    # 3. 模型训练
    logreg.fit(X_train_scaled, y_train)
    
    # 4. 预测概率
    y_test_proba = logreg.predict_proba(X_test_scaled)[:, 1]
    
    # 5. 评估与绘图
    fpr, tpr, _ = roc_curve(y_test, y_test_proba)
    auc = roc_auc_score(y_test, y_test_proba)
    aucs.append(auc)
    
    ax.plot(fpr, tpr, alpha=0.6, label=f'Fold {i+1} (AUC = {auc:.3f})')

# 美化图形
ax.plot([0, 1], [0, 1], color='navy', linestyle='--', label='随机猜测')
mean_auc = np.mean(aucs)
ax.set_title(f'k折交叉验证的ROC曲线 (平均AUC = {mean_auc:.3f})', fontsize=16)
ax.set_xlabel('假正例率 (FPR)', fontsize=14)
ax.set_ylabel('真正例率 (TPR)', fontsize=14)
ax.legend(loc='lower right')
plt.grid(True)
plt.show()