深入理解 Scikit-learn：机器学习实战的科学之道与避坑指南

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

# 错误示范：在整个数据集上拟合转换器
scaler = StandardScaler().fit(X)  # 泄露测试集信息！
X_scaled = scaler.transform(X)

# 正确方法：严格隔离训练集与测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

scaler = StandardScaler().fit(X_train)  # 仅在训练集拟合
X_train_scaled = scaler.transform(X_train)
X_test_scaled = scaler.transform(X_test)  # 测试集使用训练集的参数转换

from sklearn.impute import SimpleImputer
from sklearn.ensemble import RandomForestRegressor

# 数值型特征：中位数填充 (对异常值稳健)
num_imputer = SimpleImputer(strategy='median')

# 分类特征：众数填充
cat_imputer = SimpleImputer(strategy='most_frequent')

# 高级技巧：模型驱动的填充 (如KNNImputer)
from sklearn.impute import KNNImputer
knn_imputer = KNNImputer(n_neighbors=5)

from sklearn.preprocessing import OneHotEncoder, OrdinalEncoder

# 独热编码 (注意handle_unknown='ignore'防止新类别报错)
ohe = OneHotEncoder(handle_unknown='ignore', sparse_output=False).fit(X_train_cat)
X_train_ohe = ohe.transform(X_train_cat)

# 序数编码 (需指定categories顺序)
education_order = [['高中', '本科', '硕士', '博士']]
ord_enc = OrdinalEncoder(categories=education_order).fit(X_train_edu)
X_train_ord = ord_enc.transform(X_train_edu)

from sklearn.linear_model import Ridge, Lasso, ElasticNet

# Ridge回归：调整alpha控制正则化强度
ridge = Ridge(alpha=1.0).fit(X_train_scaled, y_train)

# Lasso回归：同样调整alpha，特征选择更明显
lasso = Lasso(alpha=0.01, max_iter=10000).fit(X_train_scaled, y_train)  # 常需增加max_iter

# ElasticNet：平衡L1和L2，调整alpha和l1_ratio
en = ElasticNet(alpha=0.1, l1_ratio=0.5).fit(X_train_scaled, y_train)

from sklearn.svm import SVC

# 线性SVM
svm_linear = SVC(kernel='linear', C=0.1).fit(X_train_scaled, y_train)

# RBF核SVM (通常需要特征缩放)
svm_rbf = SVC(kernel='rbf', C=1.0, gamma=0.1).fit(X_train_scaled, y_train)

from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier

# 随机森林
rf = RandomForestClassifier(n_estimators=100, max_depth=5, max_features='sqrt',
                            random_state=42, n_jobs=-1).fit(X_train, y_train)

# 梯度提升树 (sklearn实现)
gb = GradientBoostingClassifier(n_estimators=100, learning_rate=0.1,
                               max_depth=3, subsample=0.8,  # 行采样
                               random_state=42).fit(X_train, y_train)

from sklearn.metrics import confusion_matrix, classification_report, roc_auc_score

# 预测测试集
y_pred = model.predict(X_test)
y_pred_proba = model.predict_proba(X_test)[:, 1]  # 获取正类的概率

# 混淆矩阵
cm = confusion_matrix(y_test, y_pred)
print(cm)

# 详细报告 (Precision, Recall, F1, Support)
print(classification_report(y_test, y_pred))

# AUC-ROC (评估模型排序能力，对不平衡数据敏感度较低)
auc = roc_auc_score(y_test, y_pred_proba)
print(f"AUC-ROC: {auc:.4f}")

from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score

mse = mean_squared_error(y_test, y_pred)
rmse = np.sqrt(mse)
mae = mean_absolute_error(y_test, y_pred)
r2 = r2_score(y_test, y_pred)

print(f"RMSE: {rmse:.2f}, MAE: {mae:.2f}, R²: {r2:.4f}")

from sklearn.model_selection import KFold, StratifiedKFold, cross_val_score

# 标准5折交叉验证 (回归或不平衡不严重的分类)
kf = KFold(n_splits=5, shuffle=True, random_state=42)
scores = cross_val_score(model, X, y, cv=kf, scoring='accuracy')
print(f"CV Accuracy: {scores.mean():.4f} ± {scores.std():.4f}")

# 分层5折交叉验证 (分类，尤其推荐用于不平衡数据)
skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
scores = cross_val_score(model, X, y, cv=skf, scoring='f1_macro')  # 使用F1宏平均
print(f"Stratified CV F1 Macro: {scores.mean():.4f} ± {scores.std():.4f}")

from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
from scipy.stats import randint, uniform

# 定义参数网格/分布
param_grid = {
    'n_estimators': [50, 100, 200],
    'max_depth': [3, 5, 7, None],
    'max_features': ['sqrt', 'log2', 0.8]
}
param_dist = {
    'n_estimators': randint(50, 300),      # 均匀整数分布
    'max_depth': [3, 5, 7, 9, None],
    'learning_rate': uniform(0.01, 0.3),   # 连续均匀分布 [0.01, 0.31)
    'subsample': [0.6, 0.7, 0.8, 0.9, 1.0]
}

# GridSearchCV
grid_search = GridSearchCV(
    estimator=GradientBoostingClassifier(random_state=42),
    param_grid=param_grid,
    cv=5,  # 内部交叉验证折数
    scoring='neg_mean_squared_error',  # 回归常用
    n_jobs=-1,
    verbose=1
)
grid_search.fit(X_train, y_train)
print(f"Best Params (Grid): {grid_search.best_params_}")
print(f"Best MSE: {-grid_search.best_score_:.4f}")  # 注意负号

# RandomizedSearchCV (通常更高效)
random_search = RandomizedSearchCV(
    estimator=GradientBoostingClassifier(random_state=42),
    param_distributions=param_dist,
    n_iter=50,  # 随机尝试的组合数
    cv=5,
    scoring='accuracy',
    n_jobs=-1,
    random_state=42,
    verbose=1
)
random_search.fit(X_train, y_train)
print(f"Best Params (Random): {random_search.best_params_}")
print(f"Best Accuracy: {random_search.best_score_:.4f}")

# 使用找到的最佳参数重新训练最终模型 (或在search对象中best_estimator_已使用全部训练数据训练)
best_model = random_search.best_estimator_

from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer

# 定义数值型和分类型特征列名
numeric_features = ['age', 'income', 'credit_score']
categorical_features = ['gender', 'education', 'city']

# 为数值型特征创建管道：填充中位数 -> 标准化
numeric_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

# 为分类型特征创建管道：填充众数 -> 独热编码
categorical_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False))  # sparse_output=False 返回数组
])

# 使用ColumnTransformer组合不同的转换器，按特征类型应用
preprocessor = ColumnTransformer(
    transformers=[
        ('num', numeric_transformer, numeric_features),
        ('cat', categorical_transformer, categorical_features)
    ],
    remainder='passthrough'  # 处理未被指定的列 (例如，保留ID列或手动处理的特征)
    # 或者 remainder='drop' 丢弃未指定的列
)

# 创建包含预处理和最终模型的完整Pipeline
full_pipeline = Pipeline(steps=[
    ('preprocessor', preprocessor),
    ('classifier', RandomForestClassifier(n_estimators=100, random_state=42, n_jobs=-1))
])

# 训练整个流水线 (在preprocessor上自动调用fit_transform，在classifier上调用fit)
full_pipeline.fit(X_train, y_train)

# 预测 (在preprocessor上自动调用transform，在classifier上调用predict)
y_pred = full_pipeline.predict(X_test)

# 在GridSearchCV/RandomizedSearchCV中使用Pipeline
param_grid = {
    'classifier__n_estimators': [100, 200],
    'classifier__max_depth': [5, 10, None],
    'preprocessor__num__imputer__strategy': ['mean', 'median'],  # 访问嵌套参数
    # ... 其他参数
}
search = GridSearchCV(full_pipeline, param_grid, cv=5, scoring='accuracy')
search.fit(X_train, y_train)

from sklearn.preprocessing import PolynomialFeatures

# 创建多项式特征 (例如 degree=2: 1, a, b, a², ab, b²)
# 通常只应用于数值特征
poly = PolynomialFeatures(degree=2, interaction_only=False, include_bias=False)
X_train_poly = poly.fit_transform(X_train_num)
X_test_poly = poly.transform(X_test_num)  # 注意仅在训练集拟合！

# 在Pipeline中集成
numeric_transformer = Pipeline(steps=[
    ('imputer', ...),
    ('scaler', ...),
    ('poly', PolynomialFeatures(degree=2))
])

from sklearn.preprocessing import KBinsDiscretizer

# 将连续年龄分箱成有序类别
age_binner = KBinsDiscretizer(n_bins=5, encode='ordinal', strategy='quantile').fit(X_train[['age']])
X_train['age_bin'] = age_binner.transform(X_train[['age']])
X_test['age_bin'] = age_binner.transform(X_test[['age']])

# 特征交叉：结合年龄分箱和城市创建新类别特征
X_train['age_city'] = X_train['age_bin'].astype(str) + '_' + X_train['city']
# 测试集同样操作 (注意处理训练集未出现的新组合)

from sklearn.feature_selection import SelectFromModel, RFECV

# 使用RandomForest选择特征
selector = SelectFromModel(
    estimator=RandomForestClassifier(n_estimators=100, random_state=42),
    threshold='median'  # 选择重要性大于中位数的特征
).fit(X_train, y_train)
X_train_selected = selector.transform(X_train)
X_test_selected = selector.transform(X_test)

# 递归特征消除 (带交叉验证)
rfecv = RFECV(
    estimator=LogisticRegression(max_iter=1000, solver='liblinear'),
    step=1,  # 每次迭代移除的特征数
    cv=5,
    scoring='accuracy'
).fit(X_train_scaled, y_train)
X_train_rfecv = rfecv.transform(X_train_scaled)
print(f"Optimal number of features: {rfecv.n_features_}")

import joblib

# 保存整个训练好的Pipeline
joblib.dump(full_pipeline, 'trained_model_pipeline.joblib')

# 加载模型进行预测
loaded_pipeline = joblib.load('trained_model_pipeline.joblib')
new_prediction = loaded_pipeline.predict(new_data)