Python案例如何用Scikit-learn做管道交叉验证

wen python案例 2026-07-16 161

本文目录导读：

基础管道交叉验证
使用GridSearchCV进行超参数优化
结合特征选择和交叉验证
自定义交叉验证策略
回归问题的管道交叉验证
多步骤管道的最佳实践
保存和加载管道模型
关键要点

我来给你详细讲解如何使用Scikit-learn进行管道(Pipeline)交叉验证,包含多个实际案例。

基础管道交叉验证

import numpy as np
import pandas as pd
from sklearn.datasets import load_iris
from sklearn.model_selection import cross_val_score, KFold
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.svm import SVC
# 加载数据
iris = load_iris()
X, y = iris.data, iris.target
# 创建管道
pipeline = Pipeline([
    ('scaler', StandardScaler()),  # 标准化
    ('pca', PCA(n_components=2)),  # PCA降维
    ('svm', SVC(kernel='rbf', C=1.0))  # SVM分类器
])
# 5折交叉验证
cv_scores = cross_val_score(pipeline, X, y, cv=5, scoring='accuracy')
print(f"交叉验证准确率: {cv_scores}")
print(f"平均准确率: {cv_scores.mean():.3f} (+/- {cv_scores.std() * 2:.3f})")

使用GridSearchCV进行超参数优化

from sklearn.model_selection import GridSearchCV
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import MinMaxScaler
# 创建更复杂的管道
pipeline = Pipeline([
    ('scaler', MinMaxScaler()),
    ('classifier', RandomForestClassifier(random_state=42))
])
# 定义超参数网格
param_grid = {
    'classifier__n_estimators': [50, 100, 200],
    'classifier__max_depth': [None, 10, 20],
    'classifier__min_samples_split': [2, 5, 10]
}
# 创建GridSearchCV对象
grid_search = GridSearchCV(
    pipeline, 
    param_grid, 
    cv=5,  # 5折交叉验证
    scoring='accuracy',
    n_jobs=-1,  # 使用所有CPU核心
    verbose=1
)
# 执行搜索
grid_search.fit(X, y)
print("最佳参数:")
print(grid_search.best_params_)
print(f"\n最佳交叉验证得分: {grid_search.best_score_:.3f}")

结合特征选择和交叉验证

from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.linear_model import LogisticRegression
# 特征选择管道
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('feature_selection', SelectKBest(score_func=f_classif)),
    ('classifier', LogisticRegression(random_state=42))
])
# 参数网格包含特征选择
param_grid = {
    'feature_selection__k': [2, 3, 4],
    'classifier__C': [0.1, 1.0, 10.0],
    'classifier__penalty': ['l2']
}
# KFold确保分层抽样
kfold = KFold(n_splits=5, shuffle=True, random_state=42)
grid_search = GridSearchCV(
    pipeline, 
    param_grid, 
    cv=kfold,
    scoring='f1_macro',
    n_jobs=-1
)
grid_search.fit(X, y)
print(f"最佳F1得分: {grid_search.best_score_:.3f}")

自定义交叉验证策略

from sklearn.model_selection import StratifiedKFold, RepeatedKFold
import matplotlib.pyplot as plt
# 创建带有自定义交叉验证的管道
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', LogisticRegression(max_iter=1000))
])
# 不同的交叉验证策略
cv_strategies = {
    'Standard K-Fold': KFold(n_splits=5, shuffle=True, random_state=42),
    'Stratified K-Fold': StratifiedKFold(n_splits=5, shuffle=True, random_state=42),
    'Repeated K-Fold': RepeatedKFold(n_splits=5, n_repeats=3, random_state=42)
}
results = {}
for name, cv in cv_strategies.items():
    scores = cross_val_score(pipeline, X, y, cv=cv, scoring='accuracy')
    results[name] = scores
    print(f"{name}: {scores.mean():.3f} (+/- {scores.std():.3f})")
# 可视化结果
fig, ax = plt.subplots(figsize=(10, 6))
positions = range(len(results))
for i, (name, scores) in enumerate(results.items()):
    ax.boxplot(scores, positions=[i], widths=0.6)
ax.set_xticks(range(len(results)))
ax.set_xticklabels(results.keys())
ax.set_ylabel('Accuracy')
ax.set_title('Cross-validation Strategy Comparison')
plt.show()

回归问题的管道交叉验证

from sklearn.datasets import make_regression
from sklearn.metrics import mean_squared_error, r2_score
from sklearn.ensemble import GradientBoostingRegressor
# 生成回归数据
X_reg, y_reg = make_regression(n_samples=1000, n_features=20, noise=0.1, random_state=42)
# 回归管道
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('feature_selection', SelectKBest(f_regression)),
    ('regressor', GradientBoostingRegressor(random_state=42))
])
# 参数网格
param_grid = {
    'feature_selection__k': [10, 15, 20],
    'regressor__n_estimators': [50, 100],
    'regressor__max_depth': [3, 5]
}
# 使用负均方误差作为评分
grid_search = GridSearchCV(
    pipeline, 
    param_grid, 
    cv=5,
    scoring='neg_mean_squared_error',
    n_jobs=-1
)
grid_search.fit(X_reg, y_reg)
print(f"最佳参数: {grid_search.best_params_}")
print(f"最佳MSE: {-grid_search.best_score_:.3f}")
# 在测试集上评估
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(X_reg, y_reg, test_size=0.2, random_state=42)
grid_search.fit(X_train, y_train)
y_pred = grid_search.predict(X_test)
print(f"测试集 R²: {r2_score(y_test, y_pred):.3f}")
print(f"测试集 RMSE: {np.sqrt(mean_squared_error(y_test, y_pred)):.3f}")

多步骤管道的最佳实践

from sklearn.impute import SimpleImputer
from sklearn.preprocessing import PolynomialFeatures
from sklearn.model_selection import RandomizedSearchCV
from scipy.stats import randint, uniform
# 完整的数据处理管道
pipeline = Pipeline([
    ('imputer', SimpleImputer(strategy='mean')),  # 处理缺失值
    ('scaler', StandardScaler()),
    ('poly_features', PolynomialFeatures(degree=2, include_bias=False)),  # 多项式特征
    ('feature_selection', SelectKBest(f_classif)),
    ('classifier', SVC(random_state=42))
])
# 使用RandomizedSearchCV进行更高效的搜索
param_distributions = {
    'scaler__with_mean': [True, False],
    'poly_features__degree': randint(2, 4),
    'feature_selection__k': randint(5, 15),
    'classifier__C': uniform(0.1, 10.0),
    'classifier__gamma': ['scale', 'auto', 0.1, 1.0]
}
random_search = RandomizedSearchCV(
    pipeline,
    param_distributions,
    n_iter=50,  # 随机搜索次数
    cv=5,
    scoring='accuracy',
    n_jobs=-1,
    random_state=42
)
# 执行搜索
random_search.fit(X, y)
print("最佳参数组合:")
for param, value in random_search.best_params_.items():
    print(f"  {param}: {value}")
print(f"\n最佳得分: {random_search.best_score_:.3f}")
# 管道结构展示
print("\n管道结构:")
for step_name, step_instance in pipeline.named_steps.items():
    print(f"  {step_name}: {type(step_instance).__name__}")

保存和加载管道模型

import joblib
# 训练最佳管道
best_pipeline = random_search.best_estimator_
best_pipeline.fit(X, y)
# 保存管道模型
joblib.dump(best_pipeline, 'best_pipeline.pkl')
# 加载管道模型
loaded_pipeline = joblib.load('best_pipeline.pkl')
# 使用加载的模型进行预测
new_samples = np.array([[5.1, 3.5, 1.4, 0.2],
                        [6.2, 3.4, 5.4, 2.3]])
predictions = loaded_pipeline.predict(new_samples)
print(f"预测类别: {predictions}")

关键要点

数据处理一体化：管道确保所有预处理步骤在交叉验证中正确应用,避免数据泄露。
超参数命名：使用 步骤名__参数名 格式访问管道中的参数。
交叉验证选择：
- 分类问题：使用 StratifiedKFold 保持类别分布
- 回归问题：使用 KFold 或 RepeatedKFold
评分指标：
- 分类：accuracy, f1, precision, recall
- 回归：neg_mean_squared_error, r2
性能优化：使用 n_jobs=-1 并行计算，RandomizedSearchCV 大参数空间

这些案例涵盖了Scikit-learn管道交叉验证的主要应用场景,可以根据具体需求调整使用。

本文地址： https://www.bmjup.com/post/22485.html