@浙大疏锦行
# 导入基础库 import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns import time # 导入 time 库 import warnings # 忽略警告 warnings.filterwarnings("ignore") # 设置中文字体和负号正常显示 plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False # 导入 Pipeline 和相关预处理工具 from sklearn.pipeline import Pipeline # 用于创建机器学习工作流 from sklearn.compose import ColumnTransformer # 用于将不同的预处理应用于不同的列 from sklearn.preprocessing import OrdinalEncoder, OneHotEncoder, StandardScaler # 用于数据预处理（有序编码、独热编码、标准化） from sklearn.impute import SimpleImputer # 用于处理缺失值 # 导入机器学习模型和评估工具 from sklearn.ensemble import RandomForestClassifier # 随机森林分类器 from sklearn.metrics import classification_report, confusion_matrix # 用于评估分类器性能 from sklearn.model_selection import train_test_split # 用于划分训练集和测试集 # --- 加载原始数据 --- # 我们加载原始数据，不对其进行任何手动预处理 data = pd.read_csv('data.csv') print("原始数据加载完成，形状为:", data.shape) # print(data.head()) # 可以打印前几行看看原始数据 # --- 分离特征和标签 (使用原始数据) --- y = data['Credit Default'] # 标签 X = data.drop(['Credit Default'], axis=1) # 特征 (axis=1 表示按列删除) print("\n特征和标签分离完成。") print("特征 X 的形状:", X.shape) print("标签 y 的形状:", y.shape) # --- 划分训练集和测试集 (在任何预处理之前划分) --- # 按照8:2划分训练集和测试集 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) # 80%训练集，20%测试集 print("\n数据集划分完成 (预处理之前)。") print("X_train 形状:", X_train.shape) print("X_test 形状:", X_test.shape) print("y_train 形状:", y_train.shape) print("y_test 形状:", y_test.shape) # --- 定义不同列的类型和它们对应的预处理步骤 --- # 这些定义是基于原始数据 X 的列类型来确定的 # 识别原始的 object 列 (对应你原代码中的 discrete_features 在预处理前) object_cols = X.select_dtypes(include=['object']).columns.tolist() # 识别原始的非 object 列 (通常是数值列) numeric_cols = X.select_dtypes(exclude=['object']).columns.tolist() # 有序分类特征 (对应你之前的标签编码) # 注意：OrdinalEncoder默认编码为0, 1, 2... 对应你之前的1, 2, 3...需要在模型解释时注意 # 这里的类别顺序需要和你之前映射的顺序一致 ordinal_features = ['Home Ownership', 'Years in current job', 'Term'] # 定义每个有序特征的类别顺序，这个顺序决定了编码后的数值大小 ordinal_categories = [ ['Own Home', 'Rent', 'Have Mortgage', 'Home Mortgage'], # Home Ownership 的顺序 (对应1, 2, 3, 4) ['< 1 year', '1 year', '2 years', '3 years', '4 years', '5 years', '6 years', '7 years', '8 years', '9 years', '10+ years'], # Years in current job 的顺序 (对应1-11) ['Short Term', 'Long Term'] # Term 的顺序 (对应0, 1) ] # 构建处理有序特征的 Pipeline: 先填充缺失值，再进行有序编码 ordinal_transformer = Pipeline(steps=[ ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充分类特征的缺失值 ('encoder', OrdinalEncoder(categories=ordinal_categories, handle_unknown='use_encoded_value', unknown_value=-1)) # 进行有序编码 ]) print("有序特征处理 Pipeline 定义完成。") # 标称分类特征 (对应你之前的独热编码) nominal_features = ['Purpose'] # 使用原始列名 # 构建处理标称特征的 Pipeline: 先填充缺失值，再进行独热编码 nominal_transformer = Pipeline(steps=[ ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充分类特征的缺失值 imputer 是约定俗成的命名，仅用于标识该步骤（后续可通过 nominal_transformer.named_steps['imputer'] 调用该实例） ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False)) # 进行独热编码, sparse_output=False 使输出为密集数组 ]) print("标称特征处理 Pipeline 定义完成。") # 连续特征 (对应你之前的众数填充 + 添加标准化) # 从所有列中排除掉分类特征，得到连续特征列表 # continuous_features = X.columns.difference(object_cols).tolist() # 原始X中非object类型的列 # 也可以直接从所有列中排除已知的有序和标称特征 continuous_features = [f for f in X.columns if f not in ordinal_features + nominal_features] # 构建处理连续特征的 Pipeline: 先填充缺失值，再进行标准化 continuous_transformer = Pipeline(steps=[ ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充缺失值 (复现你的原始逻辑) ('scaler', StandardScaler()) # 标准化，一个好的实践 (如果你严格复刻原代码，可以移除这步) ]) print("连续特征处理 Pipeline 定义完成。") # --- 构建 ColumnTransformer --- # 将不同的预处理应用于不同的列子集，构造一个完备的转化器 # ColumnTransformer 接收一个 transformers 列表，每个元素是 (名称, 转换器对象, 列名列表) preprocessor = ColumnTransformer( transformers=[ ('ordinal', ordinal_transformer, ordinal_features), # 对 ordinal_features 列应用 ordinal_transformer ('nominal', nominal_transformer, nominal_features), # 对 nominal_features 列应用 nominal_transformer ('continuous', continuous_transformer, continuous_features) # 对 continuous_features 列应用 continuous_transformer ], remainder='passthrough' # 如何处理没有在上面列表中指定的列。 # 'passthrough' 表示保留这些列，不做任何处理。 # 'drop' 表示丢弃这些列。 ) print("\nColumnTransformer (预处理器) 定义完成。") # print(preprocessor) # 可以打印 preprocessor 对象看看它的结构 import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns import time # 导入 time 库 import warnings warnings.filterwarnings("ignore") plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False # 防止负号显示问题 # 导入 Pipeline 和相关预处理工具 from sklearn.pipeline import Pipeline # 用于创建机器学习工作流 from sklearn.compose import ColumnTransformer # 用于将不同的预处理应用于不同的列，之前是对datafame的某一列手动处理，如果在pipeline中直接用standardScaler等函数就会对所有列处理，所以要用到这个工具 from sklearn.preprocessing import OrdinalEncoder, OneHotEncoder, StandardScaler # 用于数据预处理 from sklearn.impute import SimpleImputer # 用于处理缺失值 # 机器学习相关库 from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import classification_report, confusion_matrix, accuracy_score, precision_score, recall_score, f1_score from sklearn.model_selection import train_test_split # 只导入 train_test_split # --- 加载原始数据 --- data = pd.read_csv('data.csv') # Pipeline 将直接处理分割后的原始数据 X_train, X_test # 原手动预处理步骤 (将被Pipeline替代): # Home Ownership 标签编码 # Years in current job 标签编码 # Purpose 独热编码 # Term 0 - 1 映射并重命名 # 连续特征用众数补全 # --- 分离特征和标签 (使用原始数据) --- y = data['Credit Default'] X = data.drop(['Credit Default'], axis=1) # --- 划分训练集和测试集 (在任何预处理之前划分) --- # X_train 和 X_test 现在是原始数据中划分出来的部分，不包含你之前的任何手动预处理结果 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) # --- 定义不同列的类型和它们对应的预处理步骤 (这些将被放入 Pipeline 的 ColumnTransformer 中) --- # 这些定义是基于原始数据 X 的列类型来确定的 # 识别原始的 object 列 (对应你原代码中的 discrete_features 在预处理前) object_cols = X.select_dtypes(include=['object']).columns.tolist() # 有序分类特征 (对应你之前的标签编码) # 注意：OrdinalEncoder默认编码为0, 1, 2... 对应你之前的1, 2, 3...需要在模型解释时注意 # 这里的类别顺序需要和你之前映射的顺序一致 ordinal_features = ['Home Ownership', 'Years in current job', 'Term'] # 定义每个有序特征的类别顺序，这个顺序决定了编码后的数值大小 ordinal_categories = [ ['Own Home', 'Rent', 'Have Mortgage', 'Home Mortgage'], # Home Ownership 的顺序 (对应1, 2, 3, 4) ['< 1 year', '1 year', '2 years', '3 years', '4 years', '5 years', '6 years', '7 years', '8 years', '9 years', '10+ years'], # Years in current job 的顺序 (对应1-11) ['Short Term', 'Long Term'] # Term 的顺序 (对应0, 1) ] # 先用众数填充分类特征的缺失值，然后进行有序编码 ordinal_transformer = Pipeline(steps=[ ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充分类特征的缺失值 ('encoder', OrdinalEncoder(categories=ordinal_categories, handle_unknown='use_encoded_value', unknown_value=-1)) ]) # 分类特征 nominal_features = ['Purpose'] # 使用原始列名 # 先用众数填充分类特征的缺失值，然后进行独热编码 nominal_transformer = Pipeline(steps=[ ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充分类特征的缺失值 ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False)) # sparse_output=False 使输出为密集数组 ]) # 连续特征 # 从X的列中排除掉分类特征，得到连续特征列表 continuous_features = X.columns.difference(object_cols).tolist() # 原始X中非object类型的列 # 先用众数填充缺失值，然后进行标准化 continuous_transformer = Pipeline(steps=[ ('imputer', SimpleImputer(strategy='most_frequent')), # 用众数填充缺失值 (复现你的原始逻辑) ('scaler', StandardScaler()) # 标准化，一个好的实践 ]) # --- 构建 ColumnTransformer --- # 将不同的预处理应用于不同的列子集，构造一个完备的转化器 preprocessor = ColumnTransformer( transformers=[ ('ordinal', ordinal_transformer, ordinal_features), ('nominal', nominal_transformer, nominal_features), ('continuous', continuous_transformer, continuous_features) ], remainder='passthrough' # 保留没有在transformers中指定的列（如果存在的话），或者 'drop' 丢弃 ) # --- 构建完整的 Pipeline --- # 将预处理器和模型串联起来 # 使用你原代码中 RandomForestClassifier 的默认参数和 random_state，这里的参数用到了元组这个数据结构 pipeline = Pipeline(steps=[ ('preprocessor', preprocessor), # 第一步：应用所有的预处理 (ColumnTransformer) ('classifier', RandomForestClassifier(random_state=42)) # 第二步：随机森林分类器 ]) # --- 1. 使用 Pipeline 在划分好的训练集和测试集上评估 --- print("--- 1. 默认参数随机森林 (训练集 -> 测试集) ---") start_time = time.time() # 记录开始时间 # 在原始的 X_train 上拟合整个Pipeline # Pipeline会自动按顺序执行preprocessor的fit_transform(X_train)，然后用处理后的数据拟合classifier pipeline.fit(X_train, y_train) # 在原始的 X_test 上进行预测 # Pipeline会自动按顺序执行preprocessor的transform(X_test)，然后用处理后的数据进行预测 pipeline_pred = pipeline.predict(X_test) end_time = time.time() # 记录结束时间 print(f"训练与预测耗时: {end_time - start_time:.4f} 秒") # 使用你原代码的输出格式 print("\n默认随机森林 在测试集上的分类报告：") # 使用你原代码的输出文本 print(classification_report(y_test, pipeline_pred)) print("默认随机森林 在测试集上的混淆矩阵：") # 使用你原代码的输出文本 print(confusion_matrix(y_test, pipeline_pred)) import pandas as pd import numpy as np import time import warnings import matplotlib.pyplot as plt import seaborn as sns from typing import Dict, List, Union, Optional from sklearn.pipeline import Pipeline from sklearn.compose import ColumnTransformer from sklearn.preprocessing import ( OrdinalEncoder, OneHotEncoder, StandardScaler, MinMaxScaler, RobustScaler ) from sklearn.impute import SimpleImputer from sklearn.model_selection import train_test_split from sklearn.metrics import ( accuracy_score, precision_score, recall_score, f1_score, classification_report, confusion_matrix, mean_squared_error, mean_absolute_error, r2_score, mean_absolute_percentage_error ) from sklearn.linear_model import LogisticRegression # 适配原代码的逻辑回归模型 # 全局配置 warnings.filterwarnings("ignore") plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False # ===================== 适配心脏病数据集的核心逻辑 ===================== def heart_disease_pipeline_demo(): """心脏病数据集 + 通用Pipeline 演示""" # 1. 加载数据（替换为你的数据路径） data = pd.read_csv(r'D:\PythonStudy\python60-days-challenge-master\heart.csv') # 2. 列名重命名（完全复用原代码的中文映射） feature_name_mapping = { 'age': '年龄', 'sex': '性别', 'cp': '胸痛类型', 'trestbps': '静息血压', 'chol': '血清胆固醇', 'fbs': '空腹血糖', 'restecg': '静息心电图结果', 'thalach': '最大心率', 'exang': '运动诱发心绞痛', 'oldpeak': 'ST段压低', 'slope': 'ST段斜率', 'ca': '荧光检查染色血管数', 'thal': '地中海贫血', 'target': '患病状态' } data = data.rename(columns=feature_name_mapping) # 3. 定义特征类型（核心：匹配原代码的预处理逻辑） ## 3.1 名义分类特征（原代码做了独热编码的列） nominal_features = ['胸痛类型', 'ST段斜率', '静息心电图结果', '地中海贫血'] ## 3.2 连续/数值特征（原代码无额外处理的列） # 排除名义分类特征和标签，剩余为连续/数值特征 continuous_features = [col for col in data.columns if col not in nominal_features + ['患病状态']] ## 3.3 连续特征配置（原代码无缺失值填充、无标准化，故scaler=None） continuous_config = { "features": continuous_features, "imputer_strategy": "most_frequent", # 原代码无填充，此处兼容可能的缺失值 "scaler": None # 原代码未做标准化/归一化，保持原始数值 } # 4. 初始化通用Pipeline（分类任务 + 逻辑回归模型） # 模型使用逻辑回归（适配原代码的模型选择） model = LogisticRegression(random_state=42, max_iter=1000) # 增加max_iter避免收敛警告 heart_pipeline = MLGeneralPipeline( task_type="classification", # 心脏病预测是二分类任务 model=model, ordinal_config={"features": []}, # 该数据集无有序分类特征 nominal_features=nominal_features, continuous_config=continuous_config, remainder="drop" # 无未指定特征，直接丢弃 ) # 5. 分离特征和标签 X = data.drop(['患病状态'], axis=1) y = data['患病状态'] # 6. 训练模型（自动划分训练/测试集，比例8:2，随机种子42） heart_pipeline.train(X, y, test_size=0.2, random_state=42) # 7. 评估模型（输出分类报告、混淆矩阵） metrics = heart_pipeline.evaluate(print_report=True) print(f"\n心脏病预测模型最终指标: {metrics}") # # 8. 新数据预测示例（可选） # # 构造一条新数据（列名需和训练集一致） # X_new = pd.DataFrame({ # '年龄': [50], # '性别': [1], # '胸痛类型': [2], # '静息血压': [120], # '血清胆固醇': [240], # '空腹血糖': [0], # '静息心电图结果': [1], # '最大心率': [150], # '运动诱发心绞痛': [0], # 'ST段压低': [1.0], # 'ST段斜率': [2], # '荧光检查染色血管数': [0], # '地中海贫血': [2] # }) # y_new_pred = heart_pipeline.predict(X_new) # print(f"\n新数据预测结果（0=未患病，1=患病）: {y_new_pred[0]}") # ===================== 运行演示 ===================== if __name__ == "__main__": heart_disease_pipeline_demo()
