DAY 31 文件的规范拆分和写法
知识点回顾
- 规范的文件命名
- 规范的文件夹管理
- 机器学习项目的拆分
- 编码格式和类型注解
作业：尝试针对之前的心脏病项目，准备拆分的项目文件，思考下哪些部分可以未来复用。
heart_disease_prediction/
│
├── data/ # 数据文件夹
│ ├── raw/ # 原始数据
│ │ └── heart.csv # <-- 你的原始 heart.csv 文件应该放在这里
│ └── processed/ # 处理后的数据或中间结果 (可选)
│
├── src/ # 项目源代码目录
│ ├── __init__.py # 使 src 成为 Python 包
│ ├── config.py # 项目配置：路径、参数、特征列表等
│ ├── utils.py # 通用工具函数：如对象保存/加载
│ │
│ ├── data/ # 数据处理相关模块
│ │ ├── __init__.py
│ │ ├── loading.py # 数据加载和分割
│ │ ├── preprocessing.py # 数据清洗、缺失值处理、编码、缩放
│ │ └── feature_engineering.py # 特征工程 (可能很简单或不需要，但保留结构)
│ │
│ ├── models/ # 模型相关模块
│ │ ├── __init__.py
│ │ ├── training.py # 模型选择和训练
│ │ └── evaluation.py # 模型评估
│ │
│ └── visualization/ # 可视化相关模块
│ ├── __init__.py
│ └── plots.py # 绘制图表函数
│
├── models/ # 保存训练好的模型和预处理器
│ └── best_model.pkl # 训练好的模型
│ └── preprocessor.pkl # 拟合好的数据预处理器 (重要!)
│
├── notebooks/ # Jupyter/IPython Notebooks 用于探索性分析 (EDA) 或实验
│ └── heart_eda_modeling_exploration.ipynb
│
├── main.py # 项目主入口：运行完整的训练和评估流程
├── predict.py # 独立的预测脚本：加载模型对新数据进行预测
├── requirements.txt # 项目所需的 Python 库列表
└── README.md # 项目说明、设置和使用指南
1 heart_disease_prediction/src/config.py
# -*- coding: utf-8 -*-"""
心脏病预测项目配置中心。
存储文件路径、参数、特征列表等。
"""import os
from typing import List, Dict, Any# 定义项目根目录，通过当前文件路径向上追溯两级
BASE_DIR: str = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))# --- 数据相关配置 ---
RAW_DATA_DIR: str = os.path.join(BASE_DIR, 'data', 'raw')
# 原始数据文件名，请确保你的 heart.csv 放在 data/raw/ 目录下
RAW_DATA_FILE: str = os.path.join(RAW_DATA_DIR, 'heart.csv')
PROCESSED_DATA_DIR: str = os.path.join(BASE_DIR, 'data', 'processed') # 处理后数据保存路径 (可选)# --- 模型保存相关配置 ---
MODELS_DIR: str = os.path.join(BASE_DIR, 'models')
# 训练好的模型保存路径
TRAINED_MODEL_PATH: str = os.path.join(MODELS_DIR, 'best_model.pkl')
# 拟合好的数据预处理器保存路径 (非常重要，预测时需要用同一个预处理器)
PREPROCESSOR_PATH: str = os.path.join(MODELS_DIR, 'preprocessor.pkl')# --- 数据处理参数 ---
TARGET_COLUMN: str = 'target' # 目标变量的列名，heart.csv 通常是 'target'
TEST_SIZE: float = 0.2 # 测试集占总数据的比例
RANDOM_STATE: int = 42 # 随机种子，用于保证数据分割、模型初始化等的可复现性# heart.csv 数据集的特征列表 (请根据你的实际文件核对并修改)
# 数值特征
NUMERICAL_FEATURES: List[str] = ['age', # 年龄'trestbps', # 静息血压'chol', # 血清胆固醇'thalach', # 最大心率'oldpeak' # 运动引起的 ST 段压低# 如果有其他数值特征请添加
]# 类别特征
CATEGORICAL_FEATURES: List[str] = ['sex', # 性别 (0/1)'cp', # 胸痛类型 (1-4)'fbs', # 空腹血糖 > 120 mg/dl (0/1)'restecg', # 静息心电图结果 (0/1/2)'exang', # 运动诱发的心绞痛 (0/1)'slope', # 运动最高峰时段 ST 段的坡度 (0/1/2)'ca', # 主要血管的数量 (0-3)'thal' # 地中海贫血症 (3=正常, 6=固定缺陷, 7=可逆缺陷)# 'ca' 和 'thal' 在某些版本的 heart.csv 中可能包含非数字值 ('?'), 需要在预处理中特别处理或清理# 如果有其他类别特征请添加
]# 预处理参数：缺失值填充策略
# heart.csv 通常没有缺失值，但保留这个配置以备用或用于其他数据集
IMPUTATION_STRATEGY_NUM: str = 'median' # 数值特征缺失值填充策略
IMPUTATION_STRATEGY_CAT: str = 'most_frequent' # 类别特征缺失值填充策略，或使用 'constant' 填充为 'missing'# --- 模型配置 ---
# 选择要使用的模型名称，需要在 src/models/training.py 中实现对应的获取逻辑
SELECTED_MODEL: str = 'logistic_regression' # 可选 'random_forest', 'svm', 'knn' 等# 各模型的超参数字典 (如果选择了其他模型，请在此处添加其参数)
LOGISTIC_REGRESSION_PARAMS: Dict[str, Any] = {'C': 1.0,'solver': 'liblinear', # 适用于小型数据集和二分类'random_state': RANDOM_STATE,'class_weight': 'balanced' # 对于目标类别不平衡的数据集很有用
}RANDOM_FOREST_PARAMS: Dict[str, Any] = {'n_estimators': 200, # 森林中的树数量'max_depth': 8, # 树的最大深度，限制模型复杂度'random_state': RANDOM_STATE,'class_weight': 'balanced' # 对于目标类别不平衡的数据集很有用
}# --- 评估配置 ---
# 需要计算和报告的评估指标列表
METRICS: List[str] = ['accuracy', 'precision', 'recall', 'f1', 'roc_auc']# 其他可能的配置项：
# - 特征工程参数
# - 模型超参数调优的参数 (如交叉验证折数)
# - 日志文件路径等
2 heart_disease_prediction/src/utils.py
# -*- coding: utf-8 -*-"""
通用工具函数：保存和加载 Python 对象、日志记录等。
"""import joblib
import os
from typing import Anydef save_object(obj: Any, filepath: str) -> None:"""使用 joblib 将 Python 对象 (如模型、预处理器) 保存到文件。Args:obj: 要保存的 Python 对象。filepath: 保存文件的完整路径。"""# 确保目标目录存在，如果不存在则创建os.makedirs(os.path.dirname(filepath), exist_ok=True)joblib.dump(obj, filepath)print(f"对象已成功保存到 {filepath}")def load_object(filepath: str) -> Any:"""使用 joblib 从文件加载 Python 对象。Args:filepath: 要加载的文件的完整路径。Returns:从文件加载的 Python 对象。Raises:FileNotFoundError: 如果指定路径的文件不存在。"""if not os.path.exists(filepath):raise FileNotFoundError(f"错误：文件未找到：{filepath}")obj = joblib.load(filepath)print(f"对象已成功从 {filepath} 加载")return obj# 根据需要添加其他通用工具函数，例如：
# def setup_logging(log_filepath: str): ...
# def report_metrics(metrics: Dict[str, float]): ...
3 heart_disease_prediction/src/data/loading.py
# -*- coding: utf-8 -*-"""
负责数据加载和初始分割（特征/目标、训练集/测试集）。
"""import pandas as pd
from sklearn.model_selection import train_test_split
from typing import Tuple# 从 src.config 导入配置
from src.config import TARGET_COLUMN, TEST_SIZE, RANDOM_STATE, RAW_DATA_FILEdef load_data(filepath: str = RAW_DATA_FILE) -> pd.DataFrame:"""从 CSV 文件加载数据集。Args:filepath: CSV 数据文件的路径。Returns:包含数据的 pandas DataFrame。Raises:FileNotFoundError: 如果数据文件未找到。Exception: 读取 CSV 时发生的其他错误。"""print(f"尝试从 {filepath} 加载数据...")try:df = pd.read_csv(filepath)print(f"成功加载数据。数据形状: {df.shape}")return dfexcept FileNotFoundError:print(f"错误：数据文件未找到：{filepath}")raise # 重新抛出异常except Exception as e:print(f"加载数据时发生错误：{e}")raise # 重新抛出异常def split_features_target(df: pd.DataFrame, target_column: str = TARGET_COLUMN) -> Tuple[pd.DataFrame, pd.Series]:"""将 DataFrame 分割为特征 (X) 和目标 (y)。Args:df: 输入 DataFrame。target_column: 目标列的名称。Returns:一个元组，包含特征 DataFrame (X) 和目标 Series (y)。Raises:ValueError: 如果目标列未在 DataFrame 中找到。"""print(f"分割特征和目标变量 (目标列: '{target_column}')...")if target_column not in df.columns:raise ValueError(f"错误：目标列 '{target_column}' 未在 DataFrame 的列中找到。")X = df.drop(columns=[target_column])y = df[target_column]print(f"分割完成。特征形状: {X.shape}, 目标形状: {y.shape}")return X, ydef split_train_test(X: pd.DataFrame, y: pd.Series, test_size: float = TEST_SIZE, random_state: int = RANDOM_STATE) -> Tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:"""将数据分割为训练集和测试集。Args:X: 特征 DataFrame。y: 目标 Series。test_size: 测试集占总数据的比例。random_state: 随机种子。Returns:一个元组，包含 X_train, X_test, y_train, y_test。"""print(f"分割数据为训练集和测试集 (测试集比例: {test_size}, 随机种子: {random_state})...")X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size, random_state=random_state, stratify=y # 分类问题使用 stratify 保证训练集和测试集中目标类别的比例相似)print(f"分割完成。训练集样本数: {len(X_train)}, 测试集样本数: {len(X_test)}")return X_train, X_test, y_train, y_test4 heart_disease_prediction/src/data/preprocessing.py
# -*- coding: utf-8 -*-"""
负责数据预处理（处理缺失值、编码、缩放）的函数。
使用 scikit-learn 的 Pipeline 和 ColumnTransformer 构建预处理流程。
"""import pandas as pd
import numpy as np # 预处理器输出通常是 numpy 数组
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from typing import List, Tuple# 从 src.config 导入配置
from src.config import NUMERICAL_FEATURES, CATEGORICAL_FEATURES, \IMPUTATION_STRATEGY_NUM, IMPUTATION_STRATEGY_CATdef create_preprocessor(numerical_features: List[str] = NUMERICAL_FEATURES,categorical_features: List[str] = CATEGORICAL_FEATURES,num_imputer_strategy: str = IMPUTATION_STRATEGY_NUM,cat_imputer_strategy: str = IMPUTATION_STRATEGY_CAT
) -> ColumnTransformer:"""创建一个 ColumnTransformer，用于对数值和类别特征应用不同的预处理步骤。Args:numerical_features: 数值特征列名列表。categorical_features: 类别特征列名列表。num_imputer_strategy: 数值特征缺失值填充策略。cat_imputer_strategy: 类别特征缺失值填充策略。Returns:一个未拟合的 scikit-learn ColumnTransformer 对象。"""print("创建数据预处理器...")# 1. 数值特征处理管道：缺失值填充 -> 标准化numerical_pipeline = Pipeline(steps=[('imputer', SimpleImputer(strategy=num_imputer_strategy)), # 填充策略 (如 median, mean)('scaler', StandardScaler()) # 标准化 (均值为0，方差为1)])# 2. 类别特征处理管道：缺失值填充 -> One-Hot 编码# 注意：如果 heart.csv 的 'ca'/'thal' 确实有 '?'，# SimpleImputer 需要设置 missing_values='?'categorical_pipeline = Pipeline(steps=[('imputer', SimpleImputer(strategy=cat_imputer_strategy, fill_value='missing')), # 填充策略 (如 most_frequent, constant)('onehot', OneHotEncoder(handle_unknown='ignore')) # One-Hot 编码。handle_unknown='ignore' 在预测时遇到训练集未见过的类别时忽略，而不是报错])# 3. 使用 ColumnTransformer 组合不同类型的特征处理管道preprocessor = ColumnTransformer(transformers=[('num', numerical_pipeline, numerical_features),('cat', categorical_pipeline, categorical_features)],remainder='passthrough' # 对于不在 numerical_features 和 categorical_features 列表中的列，不做任何处理直接通过# 或者设置为 'drop' 来丢弃未指定的列)print("数据预处理器创建完成。")return preprocessordef apply_preprocessing(X: pd.DataFrame, preprocessor: ColumnTransformer) -> np.ndarray:"""将拟合好的预处理器应用于特征 DataFrame。Args:X: 输入特征 DataFrame (例如 X_train, X_test 或新数据)。preprocessor: 拟合好的 scikit-learn ColumnTransformer 对象。Returns:经过预处理的特征的 NumPy 数组。(ColumnTransformer 默认输出 NumPy 数组)"""print("应用数据预处理...")# 注意：这里只调用 transform，fit 应该只在训练数据上调用一次X_processed = preprocessor.transform(X)print(f"数据预处理应用完成。输出形状: {X_processed.shape}")return X_processed# 在 main.py 中，你会先调用 preprocessor.fit_transform(X_train) 来拟合并转换训练数据
# 然后调用 preprocessor.transform(X_test) 来转换测试数据 (只转换，不拟合)
5 heart_disease_prediction/src/data/feature_engineering.py
# -*- coding: utf-8 -*-"""
负责创建新特征的函数。
对于 heart.csv 数据集，通常特征已经比较精简，可能不需要复杂的特征工程。
保留这个文件结构，以便未来扩展或用于其他数据集。
"""import pandas as pd
from typing import List# 可以从 config 导入特征列表，虽然在这个简单示例中可能不需要
# from src.config import NUMERICAL_FEATURES, CATEGORICAL_FEATURESdef create_features(df: pd.DataFrame) -> pd.DataFrame:"""基于现有特征创建新特征。Args:df: 输入 DataFrame (训练集或测试集，包含所有原始列)。Returns:添加了新特征（或原始）的 DataFrame。"""print("进行特征工程 (如果需要)...")# 在这里添加你的特征工程逻辑。例如：# - 创建交互特征 (如 age 和 chol 的乘积)# - 对某些特征进行分箱 (如 age 分为老年/中年/青年)# - 处理 'ca' 和 'thal' 中的特殊字符串/缺失值（如果 load_data 或 preprocessing 没有处理）# 示例：创建一个简单的交互特征 (假设 age 和 chol 列存在)# if 'age' in df.columns and 'chol' in df.columns:# df['age_x_chol'] = df['age'] * df['chol']# print("创建了特征 'age_x_chol'")# 对于标准的 heart.csv 数据集，通常不需要复杂的特征工程，原始特征效果就不错。# 此处保留函数框架，但默认不进行任何特征创建，直接返回原始 DataFrame。# 如果你的具体 heart.csv 版本需要特征工程，请在此处添加代码。print(f"特征工程完成 (可能未创建新特征)。当前形状: {df.shape}")# 如果创建了新特征，**非常重要**：# 1. 确定新特征的类型 (数值或类别)。# 2. **更新 src/config.py 中的 NUMERICAL_FEATURES 或 CATEGORICAL_FEATURES 列表**，以便预处理器能够处理这些新特征。# 或者，设计 preprocessor 更灵活地识别所有数值/类别列。return df# 可以添加特征选择、降维等其他与特征相关的函数
# def select_features(X: pd.DataFrame) -> pd.DataFrame: ...
6 heart_disease_prediction/src/models/training.py
# -*- coding: utf-8 -*-"""
负责模型选择和训练的函数。
"""import pandas as pd # 目标变量 y 通常是 pandas Series
import numpy as np # 特征 X 经过预处理后通常是 numpy 数组
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
# 根据 config.py 中 SELECTED_MODEL 的设置，导入相应的模型类
from sklearn.base import BaseEstimator # 用于类型提示，表示一个 scikit-learn 估计器# 从 src.config 导入配置
from src.config import SELECTED_MODEL, LOGISTIC_REGRESSION_PARAMS, RANDOM_FOREST_PARAMS, RANDOM_STATEdef get_model(model_name: str = SELECTED_MODEL) -> BaseEstimator:"""根据模型名称获取指定的模型实例，并设置预定义参数。Args:model_name: 要实例化模型名称 ('logistic_regression', 'random_forest' 等)。名称应与 config.py 中 SELECTED_MODEL 保持一致，并在本函数中有对应的实现。Returns:一个未拟合的 scikit-learn 模型实例。Raises:ValueError: 如果模型名称不受支持。"""print(f"获取模型：{model_name}...")if model_name == 'logistic_regression':model = LogisticRegression(**LOGISTIC_REGRESSION_PARAMS)elif model_name == 'random_forest':model = RandomForestClassifier(**RANDOM_FOREST_PARAMS)# TODO: 在这里添加对其他模型名称的支持及其参数# elif model_name == 'svm':# from sklearn.svm import SVC# model = SVC(**SVM_PARAMS) # 需要在 config.py 中定义 SVM_PARAMS# elif model_name == 'knn':# from sklearn.neighbors import KNeighborsClassifier# model = KNeighborsClassifier(**KNN_PARAMS) # 需要在 config.py 中定义 KNN_PARAMSelse:raise ValueError(f"错误：模型 '{model_name}' 不受支持。请检查 config.py 或在 get_model 函数中添加其实现。")print(f"模型 {type(model).__name__} 已创建。")return modeldef train_model(model: BaseEstimator, X_train: np.ndarray, y_train: pd.Series) -> BaseEstimator:"""使用训练数据拟合模型。Args:model: 未拟合的 scikit-learn 模型实例。X_train: 训练集的特征（经过预处理的 NumPy 数组）。y_train: 训练集的目标变量（pandas Series）。Returns:拟合好的 scikit-learn 模型实例。"""print(f"训练模型 {type(model).__name__}...")model.fit(X_train, y_train)print("模型训练完成。")return model# 可以添加超参数调优、交叉验证等相关的函数
# def tune_hyperparameters(model: BaseEstimator, X_train, y_train): ...
7 heart_disease_prediction/src/models/evaluation.py
# -*- coding: utf-8 -*-"""
负责模型性能评估的函数。
"""import pandas as pd
import numpy as np
from sklearn.metrics import (accuracy_score,precision_score,recall_score,f1_score,roc_auc_score,confusion_matrix, # 可选用于混淆矩阵roc_curve # 可选用于 ROC 曲线数据
)
from typing import Dict, List, Union# 从 src.config 导入需要报告的评估指标列表
from src.config import METRICSdef evaluate_model(y_true: pd.Series, # 真实标签y_pred: np.ndarray, # 预测标签y_proba: np.ndarray, # 预测概率 (通常是 (n_samples, n_classes) 形状)metrics_list: List[str] = METRICS
) -> Dict[str, float]:"""使用指定的指标列表评估模型性能。Args:y_true: 真实的标签值 (pandas Series)。y_pred: 预测的标签值 (NumPy 数组)。y_proba: 预测的概率分数 (NumPy 数组)。对于二分类，通常是 (n_samples, 2) 形状。第一个维度是每个样本，第二个维度是属于每个类别的概率 [P(class 0), P(class 1)]。metrics_list: 要计算的指标名称列表 ('accuracy', 'precision', 'recall', 'f1', 'roc_auc' 等)。Returns:一个字典，键是指标名称，值是计算出的分数。如果计算失败，值为 float('nan')。"""print("评估模型性能...")metrics: Dict[str, float] = {}# 提取正类的概率 (假设正类是 1)# y_proba 形状通常是 (n_samples, n_classes)。对于二分类，取列索引 1 的概率。y_proba_positive_class = Noneif y_proba.ndim == 2 and y_proba.shape[1] == 2:y_proba_positive_class = y_proba[:, 1]elif y_proba.ndim == 1 and 'roc_auc' in metrics_list:# 如果只请求了 roc_auc 并且 y_proba 是一维的，假设它已经是正类概率y_proba_positive_class = y_probaprint("警告：y_proba 是一维数组，假设其为正类概率，用于 ROC AUC 计算。")elif 'roc_auc' in metrics_list:print(f"警告：y_proba 形状 {y_proba.shape} 异常，无法计算 ROC AUC。")for metric_name in metrics_list:try:if metric_name == 'accuracy':score = accuracy_score(y_true, y_pred)elif metric_name == 'precision':# 对于二分类，默认计算正类 (pos_label=1) 的精度score = precision_score(y_true, y_pred, zero_division=0) # zero_division=0 在没有预测出正类时返回 0elif metric_name == 'recall':# 对于二分类，默认计算正类 (pos_label=1) 的召回率score = recall_score(y_true, y_pred, zero_division=0) # zero_division=0 在真实正类数为 0 时返回 0elif metric_name == 'f1':# 对于二分类，默认计算正类 (pos_label=1) 的 F1 分数score = f1_score(y_true, y_pred, zero_division=0) # zero_division=0 在没有真实正类或没有预测出正类时返回 0elif metric_name == 'roc_auc':if y_proba_positive_class is not None:score = roc_auc_score(y_true, y_proba_positive_class)else:score = float('nan') # 无法计算 ROC AUC# TODO: 在这里添加对其他指标的支持# elif metric_name == 'log_loss':# if y_proba.ndim == 2:# from sklearn.metrics import log_loss# score = log_loss(y_true, y_proba)# else:# score = float('nan')else:print(f"警告：不支持的评估指标 '{metric_name}'。")metrics[metric_name] = float('nan') # 标记为 NaN 或跳过continuemetrics[metric_name] = scoreexcept Exception as e:print(f"计算指标 '{metric_name}' 时发生错误: {e}")metrics[metric_name] = float('nan') # 标记计算失败print("评估完成。指标结果:")# 打印每个指标的结果for metric_name, score in metrics.items():print(f"- {metric_name}: {score:.4f}" if not np.isnan(score) else f"- {metric_name}: 计算失败")return metrics# 可以添加其他评估相关的函数，例如交叉验证分数计算等
8 heart_disease_prediction/src/visualization/plots.py
# -*- coding: utf-8 -*-"""
负责生成各种可视化图表的函数。
"""import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
from sklearn.metrics import roc_curve, confusion_matrix, RocCurveDisplay # 导入 RocCurveDisplay 更方便绘制 ROC# 从 src.config 导入目标列名称
from src.config import TARGET_COLUMNdef plot_feature_distribution(df: pd.DataFrame, feature: str, target: str = TARGET_COLUMN) -> None:"""绘制单个特征的分布图，可选按目标变量分组。Args:df: 输入 DataFrame。feature: 要绘制分布的特征名称。target: 目标列名称 (用于 hue)。"""print(f"绘制特征 '{feature}' 的分布图...")plt.figure(figsize=(10, 6))# 判断特征类型，选择合适的绘图方式if df[feature].dtype in ['int64', 'float64']:# 数值特征使用直方图和 KDE (核密度估计)sns.histplot(data=df, x=feature, hue=target, kde=True, stat='density', common_norm=False)# common_norm=False 使每个类别（心脏病 vs 无心脏病）的 KDE 曲线独立标准化，更易比较分布形状else:# 类别特征使用计数图# order=df[feature].value_counts().index 按照类别数量排序绘制sns.countplot(data=df, x=feature, hue=target, order=df[feature].value_counts().index)plt.xticks(rotation=45, ha='right') # 旋转 x 轴标签避免重叠plt.title(f'特征 "{feature}" 按 "{target}" 分组的分布')plt.xlabel(feature)plt.ylabel('密度' if df[feature].dtype in ['int64', 'float64'] else '数量')plt.tight_layout() # 调整布局，防止元素重叠plt.show()def plot_roc_curve(y_true: pd.Series, y_proba: np.ndarray, model_name: str = "模型") -> None:"""绘制 ROC 曲线。Args:y_true: 真实的标签值 (pandas Series)。y_proba: 预测的概率分数 (NumPy 数组)。对于二分类，应包含正类 (1) 的概率。model_name: 模型名称，用于图表标题/图例。"""print(f"绘制模型 '{model_name}' 的 ROC 曲线...")# 确保 y_proba 是针对正类 (标签为 1) 的概率y_proba_positive_class = Noneif y_proba.ndim == 2 and y_proba.shape[1] == 2:y_proba_positive_class = y_proba[:, 1] # 取第二列作为正类概率 (通常标签 1 是正类)elif y_proba.ndim == 1:y_proba_positive_class = y_proba # 假设输入的一维数组就是正类概率else:print("错误：y_proba 形状异常，无法提取正类概率绘制 ROC 曲线。")return # 无法绘制，直接返回# 使用 RocCurveDisplay 可以简化绘制过程并自动计算 AUC# from_predictions 需要真实标签和预测概率try:roc_display = RocCurveDisplay.from_predictions(y_true, y_proba_positive_class, name=model_name)plt.figure(figsize=(8, 8))roc_display.plot(ax=plt.gca()) # 将绘制结果添加到当前 matplotlib 轴上plt.plot([0, 1], [0, 1], 'k--', label='随机猜测') # 绘制对角线作为随机猜测的参照plt.xlim([0.0, 1.0])plt.ylim([0.0, 1.05])plt.xlabel('假阳性率 (False Positive Rate)')plt.ylabel('真阳性率 (True Positive Rate)')plt.title(f'接收者操作特征 (ROC) 曲线 - {model_name}')plt.legend(loc="lower right")plt.grid(True)plt.show()except Exception as e:print(f"绘制 ROC 曲线时发生错误: {e}")def plot_confusion_matrix(y_true: pd.Series, y_pred: np.ndarray, classes: List[str] = ['无心脏病', '有心脏病']) -> None:"""绘制混淆矩阵。Args:y_true: 真实的标签值 (pandas Series)。y_pred: 预测的标签值 (NumPy 数组)。classes: 类别名称列表，用于混淆矩阵的行和列标签。例如对于 heart.csv，[0, 1] 对应 ['无心脏病', '有心脏病']。"""print("绘制混淆矩阵...")cm = confusion_matrix(y_true, y_pred)plt.figure(figsize=(8, 6))# 使用 seaborn 的 heatmap 绘制混淆矩阵热力图sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=classes, yticklabels=classes)plt.title('混淆矩阵')plt.xlabel('预测标签')plt.ylabel('真实标签')plt.show()# 根据需要添加其他绘图函数（如特征重要性图、校准曲线、数据分布图等）
9 heart_disease_prediction/main.py
# -*- coding: utf-8 -*-"""
主脚本：协调心脏病预测项目的整个训练和评估流程。
依次执行：加载数据 -> 特征工程 -> 分割数据 -> 预处理 -> 模型训练 -> 模型评估 -> 保存结果。
"""import pandas as pd
import numpy as np
import sys
import os# 将项目根目录下的 src 目录添加到 Python 解释器的路径中，
# 这样就可以使用 from src.module import ... 方式导入
# os.path.dirname(__file__) 获取当前文件 (main.py) 的目录
# os.path.abspath(...) 获取绝对路径
# os.path.join(...) 拼接路径
# os.path.dirname(...) 获取父目录
# 所以 os.path.abspath(os.path.join(os.path.dirname(__file__), 'src')) 就是 src 目录的绝对路径
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), 'src')))# 从 src 模块导入所需的函数和配置
from src.config import (RAW_DATA_FILE, # 原始数据文件路径TRAINED_MODEL_PATH, # 模型保存路径PREPROCESSOR_PATH, # 预处理器保存路径SELECTED_MODEL, # 选择的模型名称TARGET_COLUMN, # 目标列名称 (在 split_features_target 内部使用，但导入一下，方便检查)METRICS # 评估指标列表 (在 evaluate_model 内部使用，但导入一下，方便检查)
)
# 从 src.data 子包导入数据处理相关函数
from src.data.loading import load_data, split_features_target, split_train_test
from src.data.preprocessing import create_preprocessor, apply_preprocessing
from src.data.feature_engineering import create_features # 导入特征工程函数
# 从 src.models 子包导入模型相关函数
from src.models.training import get_model, train_model
from src.models.evaluation import evaluate_model
# 从 src.visualization 子包导入绘图函数
from src.visualization.plots import plot_roc_curve, plot_confusion_matrix
# 从 src.utils 导入工具函数
from src.utils import save_object # 导入保存对象的函数def main():print("--- 启动心脏病预测训练流程 ---")# --- 1. 加载数据 ---try:df = load_data(RAW_DATA_FILE)except FileNotFoundError:print("数据加载失败，请检查 config.py 中的 RAW_DATA_FILE 路径和文件是否存在。退出流程。")sys.exit(1) # 数据文件未找到时立即退出脚本except Exception as e:print(f"数据加载时发生意外错误：{e}。退出流程。")sys.exit(1)# --- 2. 特征工程 ---# 注意：通常特征工程在分割之前应用到整个数据集，以保持一致性。# 如果你的特征工程依赖时间顺序或其他只应在训练集上学习的方面，则需要在分割后分别处理训练集和测试集。df_engineered = create_features(df.copy()) # 对数据副本进行操作，避免修改原始 DataFrame# --- 3. 分割特征和目标变量 ---try:X, y = split_features_target(df_engineered)except ValueError as e:print(f"特征/目标分割失败：{e}。请检查 config.py 中的 TARGET_COLUMN。退出流程。")sys.exit(1)# --- 4. 分割训练集和测试集 ---X_train, X_test, y_train, y_test = split_train_test(X, y)# --- 5. 创建并拟合预处理器 ---# 创建预处理器对象preprocessor = create_preprocessor()# **在训练集上**拟合预处理器，并转换训练集X_train_processed = preprocessor.fit_transform(X_train)# 使用**已经拟合好的**预处理器转换测试集 (只调用 transform)X_test_processed = preprocessor.transform(X_test)print("预处理完成。")# --- 6. 获取并训练模型 ---try:model = get_model(SELECTED_MODEL)except ValueError as e:print(f"模型获取失败：{e}。请检查 config.py 中的 SELECTED_MODEL。退出流程。")sys.exit(1)trained_model = train_model(model, X_train_processed, y_train)# --- 7. 评估模型 ---print("\n--- 在测试集上进行模型评估 ---")# 使用训练好的模型对测试集进行预测y_pred = trained_model.predict(X_test_processed)# 获取预测概率，用于计算 ROC AUC 等指标# predict_proba 方法返回一个形状为 (n_samples, n_classes) 的数组y_proba = trained_model.predict_proba(X_test_processed)# 调用评估函数计算指标并打印结果evaluation_metrics = evaluate_model(y_test, y_pred, y_proba, METRICS)# --- 8. 可选：绘制评估图表 ---print("\n--- 绘制评估图表 ---")# 绘制混淆矩阵plot_confusion_matrix(y_test, y_pred, classes=['无心脏病', '有心脏病'])# 绘制 ROC 曲线，确保传递正类概率 (通常是 y_proba[:, 1])plot_roc_curve(y_test, y_proba, model_name=SELECTED_MODEL)# --- 9. 保存训练好的模型和预处理器 ---print("\n--- 保存模型和预处理器 ---")try:save_object(trained_model, TRAINED_MODEL_PATH)save_object(preprocessor, PREPROCESSOR_PATH) # 保存拟合好的预处理器！except Exception as e:print(f"保存模型或预处理器时发生错误：{e}")print("\n--- 心脏病预测训练流程结束 ---")# 最终指标已在 evaluate_model 中打印，这里可以根据需要再次总结或写入日志文件。# print("测试集最终评估指标:", evaluation_metrics)if __name__ == "__main__":# 当 main.py 脚本直接运行时，执行 main() 函数main()
10 heart_disease_prediction/predict.py
# -*- coding: utf-8 -*-"""
独立的预测脚本：
加载之前训练好的模型和预处理器，对新的心脏病数据进行预测。
这个脚本模拟了将模型部署到生产环境进行推理的过程。
"""import pandas as pd
import numpy as np
import sys
import os
from typing import Any # 用于类型提示加载的对象# 将项目根目录下的 src 目录添加到 Python 解释器的路径中
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), 'src')))# 导入所需的组件和配置
from src.config import (TRAINED_MODEL_PATH, # 保存的模型路径PREPROCESSOR_PATH, # 保存的预处理器路径TARGET_COLUMN # 目标列名称 (用于在新数据中检查并移除)
)
from src.data.feature_engineering import create_features # 预测时也需要应用相同的特征工程
from src.utils import load_object # 导入加载对象的函数
# from src.models.evaluation import evaluate_model # 如果新数据包含标签，可以导入用于评估def load_new_data(filepath: str) -> pd.DataFrame:"""加载用于预测的新数据。Args:filepath: 新数据 CSV 文件的路径。Returns:包含新数据的 pandas DataFrame。Raises:FileNotFoundError: 如果数据文件未找到。Exception: 读取 CSV 时发生的其他错误。"""print(f"尝试从 {filepath} 加载新数据用于预测...")try:df = pd.read_csv(filepath)print(f"成功加载新数据。数据形状: {df.shape}")return dfexcept FileNotFoundError:print(f"错误：新数据文件未找到：{filepath}")raise # 重新抛出异常except Exception as e:print(f"加载新数据时发生错误：{e}")raise # 重新抛出异常def make_prediction(data: pd.DataFrame, # 包含新数据的 DataFramemodel: Any, # 加载的、已训练好的模型对象preprocessor: Any # 加载的、已拟合好的预处理器对象
) -> np.ndarray:"""对新数据应用与训练时相同的预处理步骤，并使用加载的模型进行预测。Args:data: 输入的、包含新数据特征的 DataFrame。model: 加载的、已训练好的模型对象 (实现了 .predict() 方法)。preprocessor: 加载的、已拟合好的数据预处理器对象 (实现了 .transform() 方法)。Returns:一个 NumPy 数组，包含每个样本的预测类别标签 (通常是 0 或 1)。"""print("启动预测过程...")# --- 1. 应用与训练时相同的特征工程 ---# 对输入数据创建一个副本，避免修改原始 DataFramedata_for_prediction = create_features(data.copy())# --- 2. 确保目标列不存在 ---# 如果新数据意外地包含了目标列，需要先移除它，因为模型只接受特征作为输入if TARGET_COLUMN in data_for_prediction.columns:print(f"警告：预测数据中检测到目标列 '{TARGET_COLUMN}'。正在移除。")data_for_prediction = data_for_prediction.drop(columns=[TARGET_COLUMN])# --- 3. 应用加载的预处理器 ---# **重要：** 在预测时，只调用 preprocessor 的 `transform()` 方法。# 绝对不要在这里调用 `fit()` 或 `fit_transform()`，因为预处理器必须使用训练集的数据分布进行拟合。print("应用加载的预处理器对新数据进行转换...")try:data_processed = preprocessor.transform(data_for_prediction)print(f"数据预处理完成。转换后形状: {data_processed.shape}")except Exception as e:print(f"应用预处理器时发生错误：{e}。请检查新数据的列是否与训练数据兼容。")raise # 无法继续预测，抛出异常# --- 4. 使用加载的模型进行预测 ---print("使用训练好的模型进行预测...")try:predictions = model.predict(data_processed)print("预测完成。")except Exception as e:print(f"使用模型进行预测时发生错误：{e}。")raise # 预测失败，抛出异常return predictionsdef main():# 定义新数据文件的路径 (请替换为你的实际新数据文件路径)# 假设你的新数据文件名为 new_heart_data.csv，放在 data/raw/ 目录下NEW_DATA_FILE = os.path.join(os.path.dirname(__file__), 'data', 'raw', 'new_heart_data.csv')# --- 1. 加载用于预测的新数据 ---try:new_data_df = load_new_data(NEW_DATA_FILE)except FileNotFoundError:print("加载新数据失败，请检查文件路径。退出预测。")sys.exit(1)except Exception as e:print(f"加载新数据时发生错误：{e}。退出预测。")sys.exit(1)# --- 2. 加载之前训练好的模型和预处理器 ---print("\n--- 加载训练好的模型和预处理器 ---")try:# 从保存的路径加载模型和预处理器对象trained_model = load_object(TRAINED_MODEL_PATH)fitted_preprocessor = load_object(PREPROCESSOR_PATH)except FileNotFoundError:print("错误：未找到保存的模型或预处理器文件。请先运行 main.py 进行训练并保存模型。退出预测。")sys.exit(1)except Exception as e:print(f"加载模型或预处理器时发生错误：{e}。退出预测。")sys.exit(1)# --- 3. 对新数据进行预测 ---try:predictions = make_prediction(new_data_df, trained_model, fitted_preprocessor)except Exception as e:print(f"进行预测时发生错误：{e}。退出预测。")sys.exit(1)# --- 4. 输出或处理预测结果 ---print("\n--- 预测结果 ---")# predictions 是一个 numpy 数组，包含了对应于 new_data_df 中每一行的预测结果# 你可以将预测结果添加到原始新数据 DataFrame 中，或者保存到新的 CSV 文件# 为了演示，我们只打印前10个预测结果和总预测数print("前 10 个预测结果 (0: 无心脏病, 1: 有心脏病):", predictions[:10])print(f"总共为 {len(predictions)} 个样本进行了预测。")# 示例：将预测结果添加到原始新数据 DataFrame 并打印前几行# 如果你想保留原始列，可以这样做new_data_df_with_predictions = new_data_df.copy()new_data_df_with_predictions['predicted_target'] = predictionsprint("\n带预测结果的新数据（前 5 行）：")print(new_data_df_with_predictions.head())# 示例：将带有预测结果的新数据保存到文件# output_filepath = os.path.join(os.path.dirname(__file__), 'data', 'processed', 'heart_predictions.csv')# try:# new_data_df_with_predictions.to_csv(output_filepath, index=False)# print(f"\n带有预测结果的新数据已保存到 {output_filepath}")# except Exception as e:# print(f"保存预测结果到文件时发生错误：{e}")# --- 可选：如果新数据包含真实标签，可以评估预测性能 ---# (这在实际生产环境中通常没有，但如果你有带标签的新的测试集可以用来验证)# if TARGET_COLUMN in new_data_df.columns:# print("\n--- 在新数据上评估预测性能 ---")# y_true_new = new_data_df[TARGET_COLUMN]# # 需要获取新数据对应的预测概率来计算 AUC 等指标# # 注意：这里的 input data 必须是没有目标列、经过特征工程、经过预处理的# new_data_for_proba = create_features(new_data_df.drop(columns=[TARGET_COLUMN])) # 移除目标列，进行特征工程# new_data_for_proba_processed = fitted_preprocessor.transform(new_data_for_proba) # 应用预处理器# y_proba_new = trained_model.predict_proba(new_data_for_proba_processed)# # 导入 evaluate_model 并调用# from src.models.evaluation import evaluate_model# print("评估指标：")# evaluate_model(y_true_new, predictions, y_proba_new)if __name__ == "__main__":# 当 predict.py 脚本直接运行时，执行 main() 函数main()
