本文目录导读：
我来给你详细讲解如何使用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管道交叉验证的主要应用场景,可以根据具体需求调整使用。
