机器学习实战指南：从数据预处理到模型部署的完整流程

🎯 引言

机器学习已经成为现代数据科学和人工智能领域的核心技术。然而，从理论到实践，从数据到可用的模型，这个过程充满了挑战。本文将带你走过一个完整的机器学习项目流程，从数据预处理开始，到最终的模型部署，提供详细的代码示例和实践经验。

无论你是刚入门的数据科学家，还是希望系统化机器学习工作流程的开发者，这篇文章都将为你提供宝贵的指导。

📊 项目概述与数据理解

1. 项目结构设计

# 项目结构
ml_project/
├── data/
│   ├── raw/                 # 原始数据
│   ├── processed/           # 处理后的数据
│   └── external/            # 外部数据源
├── notebooks/               # Jupyter notebooks
│   ├── 01_data_exploration.ipynb
│   ├── 02_feature_engineering.ipynb
│   ├── 03_model_training.ipynb
│   └── 04_model_evaluation.ipynb
├── src/
│   ├── data/               # 数据处理模块
│   ├── features/           # 特征工程
│   ├── models/             # 模型相关
│   ├── visualization/      # 可视化
│   └── utils/              # 工具函数
├── models/                 # 训练好的模型
├── reports/                # 报告和图表
├── requirements.txt
└── config.yaml

2. 配置管理

# config.py
import yaml
from dataclasses import dataclass
from typing import List, Dict, Any

@dataclass
class DataConfig:
    raw_data_path: str
    processed_data_path: str
    test_size: float
    random_state: int
    target_column: str

@dataclass
class ModelConfig:
    model_type: str
    hyperparameters: Dict[str, Any]
    cv_folds: int
    scoring_metric: str

@dataclass
class TrainingConfig:
    epochs: int
    batch_size: int
    learning_rate: float
    early_stopping_patience: int

@dataclass
class Config:
    data: DataConfig
    model: ModelConfig
    training: TrainingConfig
    
    @classmethod
    def from_yaml(cls, config_path: str):
        with open(config_path, 'r', encoding='utf-8') as f:
            config_dict = yaml.safe_load(f)
        
        return cls(
            data=DataConfig(**config_dict['data']),
            model=ModelConfig(**config_dict['model']),
            training=TrainingConfig(**config_dict['training'])
        )

# config.yaml
data:
  raw_data_path: "data/raw/dataset.csv"
  processed_data_path: "data/processed/"
  test_size: 0.2
  random_state: 42
  target_column: "target"

model:
  model_type: "random_forest"
  hyperparameters:
    n_estimators: 100
    max_depth: 10
    min_samples_split: 5
  cv_folds: 5
  scoring_metric: "accuracy"

training:
  epochs: 100
  batch_size: 32
  learning_rate: 0.001
  early_stopping_patience: 10

3. 数据加载与初步探索

# src/data/data_loader.py
import pandas as pd
import numpy as np
from typing import Tuple, Optional
import logging

class DataLoader:
    def __init__(self, config: Config):
        self.config = config
        self.logger = logging.getLogger(__name__)
    
    def load_data(self, file_path: str) -> pd.DataFrame:
        """加载数据"""
        try:
            if file_path.endswith('.csv'):
                df = pd.read_csv(file_path)
            elif file_path.endswith('.parquet'):
                df = pd.read_parquet(file_path)
            elif file_path.endswith('.json'):
                df = pd.read_json(file_path)
            else:
                raise ValueError(f"Unsupported file format: {file_path}")
            
            self.logger.info(f"Data loaded successfully. Shape: {df.shape}")
            return df
        
        except Exception as e:
            self.logger.error(f"Error loading data: {e}")
            raise
    
    def basic_info(self, df: pd.DataFrame) -> Dict[str, Any]:
        """获取数据基本信息"""
        info = {
            'shape': df.shape,
            'columns': df.columns.tolist(),
            'dtypes': df.dtypes.to_dict(),
            'missing_values': df.isnull().sum().to_dict(),
            'memory_usage': df.memory_usage(deep=True).sum(),
            'duplicate_rows': df.duplicated().sum()
        }
        
        # 数值型列的统计信息
        numeric_columns = df.select_dtypes(include=[np.number]).columns
        if len(numeric_columns) > 0:
            info['numeric_stats'] = df[numeric_columns].describe().to_dict()
        
        # 分类型列的统计信息
        categorical_columns = df.select_dtypes(include=['object', 'category']).columns
        if len(categorical_columns) > 0:
            info['categorical_stats'] = {}
            for col in categorical_columns:
                info['categorical_stats'][col] = {
                    'unique_count': df[col].nunique(),
                    'top_values': df[col].value_counts().head().to_dict()
                }
        
        return info

# 使用示例
config = Config.from_yaml('config.yaml')
loader = DataLoader(config)
df = loader.load_data(config.data.raw_data_path)
data_info = loader.basic_info(df)

print(f"数据形状: {data_info['shape']}")
print(f"缺失值: {data_info['missing_values']}")

🔧 数据预处理与清洗

1. 数据清洗管道

# src/data/preprocessing.py
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler, LabelEncoder, OneHotEncoder
from sklearn.impute import SimpleImputer, KNNImputer
from typing import List, Dict, Any, Tuple
import logging

class DataPreprocessor:
    def __init__(self, config: Config):
        self.config = config
        self.logger = logging.getLogger(__name__)
        self.scalers = {}
        self.encoders = {}
        self.imputers = {}
    
    def handle_missing_values(self, df: pd.DataFrame, strategy: Dict[str, str]) -> pd.DataFrame:
        """处理缺失值"""
        df_processed = df.copy()
        
        for column, method in strategy.items():
            if column not in df_processed.columns:
                continue
                
            if method == 'drop':
                df_processed = df_processed.dropna(subset=[column])
            
            elif method == 'mean':
                if column not in self.imputers:
                    self.imputers[column] = SimpleImputer(strategy='mean')
                    df_processed[column] = self.imputers[column].fit_transform(
                        df_processed[[column]]
                    ).flatten()
                else:
                    df_processed[column] = self.imputers[column].transform(
                        df_processed[[column]]
                    ).flatten()
            
            elif method == 'median':
                if column not in self.imputers:
                    self.imputers[column] = SimpleImputer(strategy='median')
                    df_processed[column] = self.imputers[column].fit_transform(
                        df_processed[[column]]
                    ).flatten()
                else:
                    df_processed[column] = self.imputers[column].transform(
                        df_processed[[column]]
                    ).flatten()
            
            elif method == 'mode':
                if column not in self.imputers:
                    self.imputers[column] = SimpleImputer(strategy='most_frequent')
                    df_processed[column] = self.imputers[column].fit_transform(
                        df_processed[[column]]
                    ).flatten()
                else:
                    df_processed[column] = self.imputers[column].transform(
                        df_processed[[column]]
                    ).flatten()
            
            elif method == 'knn':
                if column not in self.imputers:
                    self.imputers[column] = KNNImputer(n_neighbors=5)
                    df_processed[column] = self.imputers[column].fit_transform(
                        df_processed[[column]]
                    ).flatten()
                else:
                    df_processed[column] = self.imputers[column].transform(
                        df_processed[[column]]
                    ).flatten()
        
        return df_processed
    
    def handle_outliers(self, df: pd.DataFrame, columns: List[str], 
                       method: str = 'iqr', threshold: float = 1.5) -> pd.DataFrame:
        """处理异常值"""
        df_processed = df.copy()
        
        for column in columns:
            if column not in df_processed.columns:
                continue
            
            if method == 'iqr':
                Q1 = df_processed[column].quantile(0.25)
                Q3 = df_processed[column].quantile(0.75)
                IQR = Q3 - Q1
                lower_bound = Q1 - threshold * IQR
                upper_bound = Q3 + threshold * IQR
                
                # 替换异常值为边界值
                df_processed[column] = df_processed[column].clip(
                    lower=lower_bound, upper=upper_bound
                )
            
            elif method == 'zscore':
                z_scores = np.abs((df_processed[column] - df_processed[column].mean()) / 
                                df_processed[column].std())
                df_processed = df_processed[z_scores < threshold]
        
        return df_processed
    
    def encode_categorical_features(self, df: pd.DataFrame, 
                                  encoding_config: Dict[str, str]) -> pd.DataFrame:
        """编码分类特征"""
        df_processed = df.copy()
        
        for column, encoding_type in encoding_config.items():
            if column not in df_processed.columns:
                continue
            
            if encoding_type == 'label':
                if column not in self.encoders:
                    self.encoders[column] = LabelEncoder()
                    df_processed[column] = self.encoders[column].fit_transform(
                        df_processed[column].astype(str)
                    )
                else:
                    df_processed[column] = self.encoders[column].transform(
                        df_processed[column].astype(str)
                    )
            
            elif encoding_type == 'onehot':
                if column not in self.encoders:
                    self.encoders[column] = OneHotEncoder(sparse=False, drop='first')
                    encoded_features = self.encoders[column].fit_transform(
                        df_processed[[column]]
                    )
                    feature_names = [f"{column}_{cat}" for cat in 
                                   self.encoders[column].categories_[0][1:]]
                else:
                    encoded_features = self.encoders[column].transform(
                        df_processed[[column]]
                    )
                    feature_names = [f"{column}_{cat}" for cat in 
                                   self.encoders[column].categories_[0][1:]]
                
                # 添加编码后的特征
                encoded_df = pd.DataFrame(encoded_features, 
                                        columns=feature_names, 
                                        index=df_processed.index)
                df_processed = pd.concat([df_processed.drop(column, axis=1), 
                                        encoded_df], axis=1)
        
        return df_processed
    
    def scale_features(self, df: pd.DataFrame, columns: List[str], 
                      method: str = 'standard') -> pd.DataFrame:
        """特征缩放"""
        df_processed = df.copy()
        
        if method == 'standard':
            scaler_key = f"standard_{'-'.join(columns)}"
            if scaler_key not in self.scalers:
                self.scalers[scaler_key] = StandardScaler()
                df_processed[columns] = self.scalers[scaler_key].fit_transform(
                    df_processed[columns]
                )
            else:
                df_processed[columns] = self.scalers[scaler_key].transform(
                    df_processed[columns]
                )
        
        elif method == 'minmax':
            from sklearn.preprocessing import MinMaxScaler
            scaler_key = f"minmax_{'-'.join(columns)}"
            if scaler_key not in self.scalers:
                self.scalers[scaler_key] = MinMaxScaler()
                df_processed[columns] = self.scalers[scaler_key].fit_transform(
                    df_processed[columns]
                )
            else:
                df_processed[columns] = self.scalers[scaler_key].transform(
                    df_processed[columns]
                )
        
        return df_processed
    
    def preprocess_pipeline(self, df: pd.DataFrame, is_training: bool = True) -> pd.DataFrame:
        """完整的预处理管道"""
        self.logger.info("开始数据预处理...")
        
        # 1. 处理缺失值
        missing_strategy = {
            'age': 'median',
            'income': 'mean',
            'category': 'mode',
            'score': 'knn'
        }
        df = self.handle_missing_values(df, missing_strategy)
        
        # 2. 处理异常值
        numeric_columns = df.select_dtypes(include=[np.number]).columns.tolist()
        if self.config.data.target_column in numeric_columns:
            numeric_columns.remove(self.config.data.target_column)
        df = self.handle_outliers(df, numeric_columns)
        
        # 3. 编码分类特征
        categorical_columns = df.select_dtypes(include=['object', 'category']).columns.tolist()
        if self.config.data.target_column in categorical_columns:
            categorical_columns.remove(self.config.data.target_column)
        
        encoding_config = {}
        for col in categorical_columns:
            unique_count = df[col].nunique()
            if unique_count <= 10:
                encoding_config[col] = 'onehot'
            else:
                encoding_config[col] = 'label'
        
        df = self.encode_categorical_features(df, encoding_config)
        
        # 4. 特征缩放
        numeric_columns = df.select_dtypes(include=[np.number]).columns.tolist()
        if self.config.data.target_column in numeric_columns:
            numeric_columns.remove(self.config.data.target_column)
        df = self.scale_features(df, numeric_columns)
        
        self.logger.info(f"预处理完成。最终数据形状: {df.shape}")
        return df

2. 数据质量检查

# src/data/quality_check.py
import pandas as pd
import numpy as np
from typing import Dict, List, Any
import matplotlib.pyplot as plt
import seaborn as sns

class DataQualityChecker:
    def __init__(self):
        self.quality_report = {}
    
    def check_data_quality(self, df: pd.DataFrame) -> Dict[str, Any]:
        """全面的数据质量检查"""
        report = {
            'basic_info': self._basic_info_check(df),
            'missing_values': self._missing_values_check(df),
            'duplicates': self._duplicates_check(df),
            'data_types': self._data_types_check(df),
            'outliers': self._outliers_check(df),
            'consistency': self._consistency_check(df),
            'completeness': self._completeness_check(df)
        }
        
        self.quality_report = report
        return report
    
    def _basic_info_check(self, df: pd.DataFrame) -> Dict[str, Any]:
        """基本信息检查"""
        return {
            'shape': df.shape,
            'memory_usage_mb': df.memory_usage(deep=True).sum() / 1024 / 1024,
            'column_count': len(df.columns),
            'row_count': len(df)
        }
    
    def _missing_values_check(self, df: pd.DataFrame) -> Dict[str, Any]:
        """缺失值检查"""
        missing_counts = df.isnull().sum()
        missing_percentages = (missing_counts / len(df)) * 100
        
        return {
            'total_missing': missing_counts.sum(),
            'columns_with_missing': missing_counts[missing_counts > 0].to_dict(),
            'missing_percentages': missing_percentages[missing_percentages > 0].to_dict(),
            'columns_mostly_missing': missing_percentages[missing_percentages > 50].index.tolist()
        }
    
    def _duplicates_check(self, df: pd.DataFrame) -> Dict[str, Any]:
        """重复值检查"""
        duplicate_rows = df.duplicated().sum()
        duplicate_percentage = (duplicate_rows / len(df)) * 100
        
        return {
            'duplicate_rows': duplicate_rows,
            'duplicate_percentage': duplicate_percentage,
            'unique_rows': len(df) - duplicate_rows
        }
    
    def _data_types_check(self, df: pd.DataFrame) -> Dict[str, Any]:
        """数据类型检查"""
        type_counts = df.dtypes.value_counts().to_dict()
        
        # 检查可能的类型转换机会
        potential_conversions = {}
        for col in df.columns:
            if df[col].dtype == 'object':
                # 检查是否可以转换为数值型
                try:
                    pd.to_numeric(df[col], errors='raise')
                    potential_conversions[col] = 'numeric'
                except:
                    # 检查是否可以转换为日期型
                    try:
                        pd.to_datetime(df[col], errors='raise')
                        potential_conversions[col] = 'datetime'
                    except:
                        pass
        
        return {
            'type_distribution': type_counts,
            'potential_conversions': potential_conversions
        }
    
    def _outliers_check(self, df: pd.DataFrame) -> Dict[str, Any]:
        """异常值检查"""
        numeric_columns = df.select_dtypes(include=[np.number]).columns
        outliers_info = {}
        
        for col in numeric_columns:
            Q1 = df[col].quantile(0.25)
            Q3 = df[col].quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR
            
            outliers = df[(df[col] < lower_bound) | (df[col] > upper_bound)]
            outliers_info[col] = {
                'count': len(outliers),
                'percentage': (len(outliers) / len(df)) * 100,
                'lower_bound': lower_bound,
                'upper_bound': upper_bound
            }
        
        return outliers_info
    
    def _consistency_check(self, df: pd.DataFrame) -> Dict[str, Any]:
        """一致性检查"""
        consistency_issues = {}
        
        # 检查字符串列的一致性
        string_columns = df.select_dtypes(include=['object']).columns
        for col in string_columns:
            unique_values = df[col].dropna().unique()
            
            # 检查大小写不一致
            lower_values = [str(val).lower() for val in unique_values]
            if len(set(lower_values)) < len(unique_values):
                consistency_issues[f"{col}_case_inconsistency"] = True
            
            # 检查前后空格
            stripped_values = [str(val).strip() for val in unique_values]
            if len(set(stripped_values)) < len(unique_values):
                consistency_issues[f"{col}_whitespace_issues"] = True
        
        return consistency_issues
    
    def _completeness_check(self, df: pd.DataFrame) -> Dict[str, Any]:
        """完整性检查"""
        completeness_score = (1 - df.isnull().sum().sum() / (df.shape[0] * df.shape[1])) * 100
        
        column_completeness = {}
        for col in df.columns:
            column_completeness[col] = (1 - df[col].isnull().sum() / len(df)) * 100
        
        return {
            'overall_completeness': completeness_score,
            'column_completeness': column_completeness,
            'complete_rows': df.dropna().shape[0],
            'complete_rows_percentage': (df.dropna().shape[0] / len(df)) * 100
        }
    
    def generate_quality_report(self, save_path: str = None) -> str:
        """生成数据质量报告"""
        if not self.quality_report:
            raise ValueError("请先运行 check_data_quality 方法")
        
        report_lines = []
        report_lines.append("=" * 50)
        report_lines.append("数据质量检查报告")
        report_lines.append("=" * 50)
        
        # 基本信息
        basic_info = self.quality_report['basic_info']
        report_lines.append(f"\n📊 基本信息:")
        report_lines.append(f"  - 数据形状: {basic_info['shape']}")
        report_lines.append(f"  - 内存使用: {basic_info['memory_usage_mb']:.2f} MB")
        
        # 缺失值
        missing_info = self.quality_report['missing_values']
        report_lines.append(f"\n❌ 缺失值:")
        report_lines.append(f"  - 总缺失值: {missing_info['total_missing']}")
        if missing_info['columns_with_missing']:
            report_lines.append("  - 缺失值列:")
            for col, count in missing_info['columns_with_missing'].items():
                percentage = missing_info['missing_percentages'][col]
                report_lines.append(f"    * {col}: {count} ({percentage:.2f}%)")
        
        # 重复值
        duplicate_info = self.quality_report['duplicates']
        report_lines.append(f"\n🔄 重复值:")
        report_lines.append(f"  - 重复行数: {duplicate_info['duplicate_rows']}")
        report_lines.append(f"  - 重复比例: {duplicate_info['duplicate_percentage']:.2f}%")
        
        # 异常值
        outliers_info = self.quality_report['outliers']
        report_lines.append(f"\n⚠️ 异常值:")
        for col, info in outliers_info.items():
            if info['count'] > 0:
                report_lines.append(f"  - {col}: {info['count']} ({info['percentage']:.2f}%)")
        
        # 完整性
        completeness_info = self.quality_report['completeness']
        report_lines.append(f"\n✅ 完整性:")
        report_lines.append(f"  - 整体完整性: {completeness_info['overall_completeness']:.2f}%")
        report_lines.append(f"  - 完整行数: {completeness_info['complete_rows']}")
        
        report_text = "\n".join(report_lines)
        
        if save_path:
            with open(save_path, 'w', encoding='utf-8') as f:
                f.write(report_text)
        
        return report_text

# 使用示例
quality_checker = DataQualityChecker()
quality_report = quality_checker.check_data_quality(df)
report_text = quality_checker.generate_quality_report('reports/data_quality_report.txt')
print(report_text)

🎨 特征工程

1. 特征创建与选择

# src/features/feature_engineering.py
import pandas as pd
import numpy as np
from sklearn.feature_selection import SelectKBest, f_classif, f_regression, RFE
from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
from sklearn.preprocessing import PolynomialFeatures
from typing import List, Dict, Any, Tuple
import logging

class FeatureEngineer:
    def __init__(self, config: Config):
        self.config = config
        self.logger = logging.getLogger(__name__)
        self.feature_selectors = {}
        self.created_features = []
    
    def create_polynomial_features(self, df: pd.DataFrame, columns: List[str], 
                                 degree: int = 2) -> pd.DataFrame:
        """创建多项式特征"""
        df_processed = df.copy()
        
        poly = PolynomialFeatures(degree=degree, include_bias=False, 
                                interaction_only=False)
        
        # 只对指定列创建多项式特征
        poly_features = poly.fit_transform(df[columns])
        feature_names = poly.get_feature_names_out(columns)
        
        # 移除原始特征（已经包含在多项式特征中）
        new_feature_names = [name for name in feature_names if name not in columns]
        new_features_idx = [i for i, name in enumerate(feature_names) 
                          if name not in columns]
        
        poly_df = pd.DataFrame(
            poly_features[:, new_features_idx],
            columns=new_feature_names,
            index=df.index
        )
        
        df_processed = pd.concat([df_processed, poly_df], axis=1)
        self.created_features.extend(new_feature_names)
        
        self.logger.info(f"创建了 {len(new_feature_names)} 个多项式特征")
        return df_processed
    
    def create_interaction_features(self, df: pd.DataFrame, 
                                  feature_pairs: List[Tuple[str, str]]) -> pd.DataFrame:
        """创建交互特征"""
        df_processed = df.copy()
        
        for feature1, feature2 in feature_pairs:
            if feature1 in df.columns and feature2 in df.columns:
                # 乘法交互
                interaction_name = f"{feature1}_x_{feature2}"
                df_processed[interaction_name] = df[feature1] * df[feature2]
                self.created_features.append(interaction_name)
                
                # 除法交互（避免除零）
                if (df[feature2] != 0).all():
                    ratio_name = f"{feature1}_div_{feature2}"
                    df_processed[ratio_name] = df[feature1] / df[feature2]
                    self.created_features.append(ratio_name)
                
                # 差值特征
                diff_name = f"{feature1}_minus_{feature2}"
                df_processed[diff_name] = df[feature1] - df[feature2]
                self.created_features.append(diff_name)
        
        self.logger.info(f"创建了 {len(feature_pairs) * 3} 个交互特征")
        return df_processed
    
    def create_aggregation_features(self, df: pd.DataFrame, 
                                  group_columns: List[str],
                                  agg_columns: List[str],
                                  agg_functions: List[str]) -> pd.DataFrame:
        """创建聚合特征"""
        df_processed = df.copy()
        
        for group_col in group_columns:
            for agg_col in agg_columns:
                for agg_func in agg_functions:
                    if group_col in df.columns and agg_col in df.columns:
                        feature_name = f"{agg_col}_{agg_func}_by_{group_col}"
                        
                        if agg_func == 'mean':
                            agg_values = df.groupby(group_col)[agg_col].transform('mean')
                        elif agg_func == 'std':
                            agg_values = df.groupby(group_col)[agg_col].transform('std')
                        elif agg_func == 'min':
                            agg_values = df.groupby(group_col)[agg_col].transform('min')
                        elif agg_func == 'max':
                            agg_values = df.groupby(group_col)[agg_col].transform('max')
                        elif agg_func == 'count':
                            agg_values = df.groupby(group_col)[agg_col].transform('count')
                        
                        df_processed[feature_name] = agg_values
                        self.created_features.append(feature_name)
        
        return df_processed
    
    def create_binning_features(self, df: pd.DataFrame, 
                              columns: List[str], 
                              n_bins: int = 5) -> pd.DataFrame:
        """创建分箱特征"""
        df_processed = df.copy()
        
        for col in columns:
            if col in df.columns:
                # 等频分箱
                df_processed[f"{col}_bin_freq"] = pd.qcut(
                    df[col], q=n_bins, labels=False, duplicates='drop'
                )
                
                # 等宽分箱
                df_processed[f"{col}_bin_width"] = pd.cut(
                    df[col], bins=n_bins, labels=False
                )
                
                self.created_features.extend([f"{col}_bin_freq", f"{col}_bin_width"])
        
        return df_processed
    
    def select_features_univariate(self, X: pd.DataFrame, y: pd.Series, 
                                 k: int = 10, task_type: str = 'classification') -> List[str]:
        """单变量特征选择"""
        if task_type == 'classification':
            selector = SelectKBest(score_func=f_classif, k=k)
        else:
            selector = SelectKBest(score_func=f_regression, k=k)
        
        selector.fit(X, y)
        selected_features = X.columns[selector.get_support()].tolist()
        
        self.feature_selectors['univariate'] = selector
        self.logger.info(f"单变量选择了 {len(selected_features)} 个特征")
        
        return selected_features
    
    def select_features_rfe(self, X: pd.DataFrame, y: pd.Series, 
                          n_features: int = 10, task_type: str = 'classification') -> List[str]:
        """递归特征消除"""
        if task_type == 'classification':
            estimator = RandomForestClassifier(n_estimators=50, random_state=42)
        else:
            estimator = RandomForestRegressor(n_estimators=50, random_state=42)
        
        selector = RFE(estimator=estimator, n_features_to_select=n_features)
        selector.fit(X, y)
        
        selected_features = X.columns[selector.get_support()].tolist()
        
        self.feature_selectors['rfe'] = selector
        self.logger.info(f"RFE选择了 {len(selected_features)} 个特征")
        
        return selected_features
    
    def select_features_importance(self, X: pd.DataFrame, y: pd.Series, 
                                 threshold: float = 0.01, 
                                 task_type: str = 'classification') -> List[str]:
        """基于特征重要性的选择"""
        if task_type == 'classification':
            model = RandomForestClassifier(n_estimators=100, random_state=42)
        else:
            model = RandomForestRegressor(n_estimators=100, random_state=42)
        
        model.fit(X, y)
        
        # 获取特征重要性
        importances = model.feature_importances_
        feature_importance_df = pd.DataFrame({
            'feature': X.columns,
            'importance': importances
        }).sort_values('importance', ascending=False)
        
        # 选择重要性大于阈值的特征
        selected_features = feature_importance_df[
            feature_importance_df['importance'] > threshold
        ]['feature'].tolist()
        
        self.feature_selectors['importance'] = {
            'model': model,
            'importances': feature_importance_df
        }
        
        self.logger.info(f"重要性选择了 {len(selected_features)} 个特征")
        
        return selected_features
    
    def get_feature_importance_report(self) -> pd.DataFrame:
        """获取特征重要性报告"""
        if 'importance' not in self.feature_selectors:
            raise ValueError("请先运行 select_features_importance 方法")
        
        return self.feature_selectors['importance']['importances']
    
    def feature_engineering_pipeline(self, df: pd.DataFrame, 
                                   target_column: str) -> Tuple[pd.DataFrame, List[str]]:
        """完整的特征工程管道"""
        self.logger.info("开始特征工程...")
        
        df_processed = df.copy()
        
        # 1. 创建多项式特征
        numeric_columns = df_processed.select_dtypes(include=[np.number]).columns.tolist()
        if target_column in numeric_columns:
            numeric_columns.remove(target_column)
        
        if len(numeric_columns) >= 2:
            # 选择前几个重要的数值特征创建多项式特征
            selected_numeric = numeric_columns[:3]  # 限制数量避免特征爆炸
            df_processed = self.create_polynomial_features(df_processed, selected_numeric, degree=2)
        
        # 2. 创建交互特征
        if len(numeric_columns) >= 2:
            feature_pairs = [(numeric_columns[i], numeric_columns[j]) 
                           for i in range(len(numeric_columns)) 
                           for j in range(i+1, min(i+3, len(numeric_columns)))]  # 限制交互数量
            df_processed = self.create_interaction_features(df_processed, feature_pairs)
        
        # 3. 创建分箱特征
        if len(numeric_columns) > 0:
            df_processed = self.create_binning_features(df_processed, numeric_columns[:5])
        
        # 4. 特征选择
        X = df_processed.drop(target_column, axis=1)
        y = df_processed[target_column]
        
        # 确定任务类型
        task_type = 'classification' if y.dtype == 'object' or y.nunique() < 20 else 'regression'
        
        # 多种特征选择方法
        selected_features_univariate = self.select_features_univariate(X, y, k=min(20, X.shape[1]), task_type=task_type)
        selected_features_importance = self.select_features_importance(X, y, threshold=0.001, task_type=task_type)
        
        # 合并选择的特征
        final_features = list(set(selected_features_univariate + selected_features_importance))
        
        # 确保目标列被保留
        final_columns = final_features + [target_column]
        df_final = df_processed[final_columns]
        
        self.logger.info(f"特征工程完成。最终特征数: {len(final_features)}")
        
        return df_final, final_features

# 使用示例
feature_engineer = FeatureEngineer(config)
df_engineered, selected_features = feature_engineer.feature_engineering_pipeline(
    df_processed, config.data.target_column
)

# 获取特征重要性报告
importance_report = feature_engineer.get_feature_importance_report()
print("Top 10 重要特征:")
print(importance_report.head(10))

2. 特征可视化分析

# src/visualization/feature_analysis.py
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import numpy as np
from typing import List, Tuple
import warnings
warnings.filterwarnings('ignore')

class FeatureVisualizer:
    def __init__(self, figsize: Tuple[int, int] = (12, 8)):
        self.figsize = figsize
        plt.style.use('seaborn-v0_8')
        
    def plot_feature_distributions(self, df: pd.DataFrame, columns: List[str], 
                                 target_column: str = None, ncols: int = 3):
        """绘制特征分布图"""
        nrows = (len(columns) + ncols - 1) // ncols
        fig, axes = plt.subplots(nrows=nrows, ncols=ncols, 
                               figsize=(ncols * 5, nrows * 4))
        axes = axes.flatten() if nrows > 1 else [axes]
        
        for i, col in enumerate(columns):
            ax = axes[i]
            
            if df[col].dtype in ['object', 'category']:
                # 分类变量
                if target_column:
                    sns.countplot(data=df, x=col, hue=target_column, ax=ax)
                else:
                    sns.countplot(data=df, x=col, ax=ax)
                ax.tick_params(axis='x', rotation=45)
            else:
                # 数值变量
                if target_column and df[target_column].dtype in ['object', 'category']:
                    # 按目标变量分组的分布
                    for target_val in df[target_column].unique():
                        subset = df[df[target_column] == target_val]
                        ax.hist(subset[col], alpha=0.7, label=f'{target_column}={target_val}', bins=30)
                    ax.legend()
                else:
                    ax.hist(df[col], bins=30, alpha=0.7)
            
            ax.set_title(f'{col} 分布')
            ax.set_xlabel(col)
        
        # 隐藏多余的子图
        for i in range(len(columns), len(axes)):
            axes[i].set_visible(False)
        
        plt.tight_layout()
        plt.show()
    
    def plot_correlation_matrix(self, df: pd.DataFrame, method: str = 'pearson'):
        """绘制相关性矩阵热图"""
        # 只选择数值型列
        numeric_df = df.select_dtypes(include=[np.number])
        
        if numeric_df.empty:
            print("没有数值型特征可以计算相关性")
            return
        
        correlation_matrix = numeric_df.corr(method=method)
        
        plt.figure(figsize=self.figsize)
        mask = np.triu(np.ones_like(correlation_matrix, dtype=bool))
        
        sns.heatmap(correlation_matrix, mask=mask, annot=True, 
                   cmap='coolwarm', center=0, square=True,
                   linewidths=0.5, cbar_kws={"shrink": .8})
        
        plt.title(f'特征相关性矩阵 ({method.capitalize()})')
        plt.tight_layout()
        plt.show()
    
    def plot_feature_importance(self, importance_df: pd.DataFrame, top_n: int = 20):
        """绘制特征重要性图"""
        top_features = importance_df.head(top_n)
        
        plt.figure(figsize=self.figsize)
        sns.barplot(data=top_features, y='feature', x='importance', palette='viridis')
        plt.title(f'Top {top_n} 特征重要性')
        plt.xlabel('重要性分数')
        plt.ylabel('特征')
        plt.tight_layout()
        plt.show()
    
    def plot_target_analysis(self, df: pd.DataFrame, target_column: str):
        """目标变量分析"""
        fig, axes = plt.subplots(1, 2, figsize=(15, 6))
        
        # 目标变量分布
        if df[target_column].dtype in ['object', 'category']:
            # 分类目标
            value_counts = df[target_column].value_counts()
            axes[0].pie(value_counts.values, labels=value_counts.index, autopct='%1.1f%%')
            axes[0].set_title('目标变量分布')
            
            # 目标变量计数
            sns.countplot(data=df, x=target_column, ax=axes[1])
            axes[1].set_title('目标变量计数')
            axes[1].tick_params(axis='x', rotation=45)
        else:
            # 数值目标
            axes[0].hist(df[target_column], bins=30, alpha=0.7)
            axes[0].set_title('目标变量分布')
            axes[0].set_xlabel(target_column)
            
            # 箱线图
            axes[1].boxplot(df[target_column])
            axes[1].set_title('目标变量箱线图')
            axes[1].set_ylabel(target_column)
        
        plt.tight_layout()
        plt.show()
    
    def plot_feature_vs_target(self, df: pd.DataFrame, features: List[str], 
                             target_column: str, ncols: int = 2):
        """特征与目标变量关系图"""
        nrows = (len(features) + ncols - 1) // ncols
        fig, axes = plt.subplots(nrows=nrows, ncols=ncols, 
                               figsize=(ncols * 6, nrows * 4))
        axes = axes.flatten() if nrows > 1 else [axes]
        
        for i, feature in enumerate(features):
            ax = axes[i]
            
            if df[feature].dtype in ['object', 'category']:
                # 分类特征
                if df[target_column].dtype in ['object', 'category']:
                    # 分类 vs 分类
                    pd.crosstab(df[feature], df[target_column]).plot(kind='bar', ax=ax)
                else:
                    # 分类 vs 数值
                    sns.boxplot(data=df, x=feature, y=target_column, ax=ax)
                ax.tick_params(axis='x', rotation=45)
            else:
                # 数值特征
                if df[target_column].dtype in ['object', 'category']:
                    # 数值 vs 分类
                    for target_val in df[target_column].unique():
                        subset = df[df[target_column] == target_val]
                        ax.scatter(subset[feature], [target_val] * len(subset), 
                                 alpha=0.6, label=target_val)
                    ax.legend()
                else:
                    # 数值 vs 数值
                    ax.scatter(df[feature], df[target_column], alpha=0.6)
                    
                    # 添加趋势线
                    z = np.polyfit(df[feature].dropna(), 
                                 df[target_column][df[feature].notna()], 1)
                    p = np.poly1d(z)
                    ax.plot(df[feature], p(df[feature]), "r--", alpha=0.8)
            
            ax.set_title(f'{feature} vs {target_column}')
            ax.set_xlabel(feature)
            ax.set_ylabel(target_column)
        
        # 隐藏多余的子图
        for i in range(len(features), len(axes)):
            axes[i].set_visible(False)
        
        plt.tight_layout()
        plt.show()

# 使用示例
visualizer = FeatureVisualizer()

# 特征分布分析
numeric_features = df_engineered.select_dtypes(include=[np.number]).columns.tolist()
if config.data.target_column in numeric_features:
    numeric_features.remove(config.data.target_column)

visualizer.plot_feature_distributions(df_engineered, numeric_features[:9], config.data.target_column)

# 相关性分析
visualizer.plot_correlation_matrix(df_engineered)

# 特征重要性
visualizer.plot_feature_importance(importance_report)

# 目标变量分析
visualizer.plot_target_analysis(df_engineered, config.data.target_column)

# 特征与目标关系
visualizer.plot_feature_vs_target(df_engineered, selected_features[:6], config.data.target_column)

🤖 模型训练与评估

1. 模型训练管道

# src/models/model_trainer.py
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV, RandomizedSearchCV
from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor, GradientBoostingClassifier, GradientBoostingRegressor
from sklearn.linear_model import LogisticRegression, LinearRegression, Ridge, Lasso
from sklearn.svm import SVC, SVR
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_score
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
import joblib
from typing import Dict, Any, Tuple, List
import logging

class ModelTrainer:
    def __init__(self, config: Config):
        self.config = config
        self.logger = logging.getLogger(__name__)
        self.models = {}
        self.best_model = None
        self.task_type = None
    
    def prepare_data(self, df: pd.DataFrame) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
        """准备训练数据"""
        X = df.drop(self.config.data.target_column, axis=1)
        y = df[self.config.data.target_column]
        
        # 确定任务类型
        if y.dtype == 'object' or y.nunique() < 20:
            self.task_type = 'classification'
        else:
            self.task_type = 'regression'
        
        # 分割数据
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, 
            test_size=self.config.data.test_size,
            random_state=self.config.data.random_state,
            stratify=y if self.task_type == 'classification' else None
        )
        
        self.logger.info(f"数据分割完成。训练集: {X_train.shape}, 测试集: {X_test.shape}")
        self.logger.info(f"任务类型: {self.task_type}")
        
        return X_train, X_test, y_train, y_test
    
    def get_model_candidates(self) -> Dict[str, Any]:
        """获取候选模型"""
        if self.task_type == 'classification':
            models = {
                'random_forest': RandomForestClassifier(random_state=42),
                'gradient_boosting': GradientBoostingClassifier(random_state=42),
                'logistic_regression': LogisticRegression(random_state=42, max_iter=1000),
                'svm': SVC(random_state=42, probability=True)
            }
        else:
            models = {
                'random_forest': RandomForestRegressor(random_state=42),
                'gradient_boosting': GradientBoostingRegressor(random_state=42),
                'linear_regression': LinearRegression(),
                'ridge': Ridge(random_state=42),
                'lasso': Lasso(random_state=42),
                'svr': SVR()
            }
        
        return models
    
    def get_hyperparameter_grids(self) -> Dict[str, Dict[str, List]]:
        """获取超参数搜索空间"""
        if self.task_type == 'classification':
            param_grids = {
                'random_forest': {
                    'n_estimators': [50, 100, 200],
                    'max_depth': [5, 10, 15, None],
                    'min_samples_split': [2, 5, 10],
                    'min_samples_leaf': [1, 2, 4]
                },
                'gradient_boosting': {
                    'n_estimators': [50, 100, 200],
                    'learning_rate': [0.01, 0.1, 0.2],
                    'max_depth': [3, 5, 7],
                    'subsample': [0.8, 0.9, 1.0]
                },
                'logistic_regression': {
                    'C': [0.1, 1, 10, 100],
                    'penalty': ['l1', 'l2'],
                    'solver': ['liblinear', 'saga']
                },
                'svm': {
                    'C': [0.1, 1, 10],
                    'kernel': ['rbf', 'linear'],
                    'gamma': ['scale', 'auto']
                }
            }
        else:
            param_grids = {
                'random_forest': {
                    'n_estimators': [50, 100, 200],
                    'max_depth': [5, 10, 15, None],
                    'min_samples_split': [2, 5, 10],
                    'min_samples_leaf': [1, 2, 4]
                },
                'gradient_boosting': {
                    'n_estimators': [50, 100, 200],
                    'learning_rate': [0.01, 0.1, 0.2],
                    'max_depth': [3, 5, 7],
                    'subsample': [0.8, 0.9, 1.0]
                },
                'ridge': {
                    'alpha': [0.1, 1, 10, 100]
                },
                'lasso': {
                    'alpha': [0.1, 1, 10, 100]
                },
                'svr': {
                    'C': [0.1, 1, 10],
                    'kernel': ['rbf', 'linear'],
                    'gamma': ['scale', 'auto']
                }
            }
        
        return param_grids
    
    def train_single_model(self, model, X_train: np.ndarray, y_train: np.ndarray,
                          param_grid: Dict = None, use_random_search: bool = False) -> Any:
        """训练单个模型"""
        if param_grid:
            if use_random_search:
                search = RandomizedSearchCV(
                    model, param_grid, 
                    n_iter=20,
                    cv=self.config.model.cv_folds,
                    scoring=self.config.model.scoring_metric,
                    random_state=42,
                    n_jobs=-1
                )
            else:
                search = GridSearchCV(
                    model, param_grid,
                    cv=self.config.model.cv_folds,
                    scoring=self.config.model.scoring_metric,
                    n_jobs=-1
                )
            
            search.fit(X_train, y_train)
            return search.best_estimator_
        else:
            model.fit(X_train, y_train)
            return model
    
    def evaluate_model(self, model, X_test: np.ndarray, y_test: np.ndarray) -> Dict[str, float]:
        """评估模型性能"""
        y_pred = model.predict(X_test)
        
        if self.task_type == 'classification':
            metrics = {
                'accuracy': accuracy_score(y_test, y_pred),
                'precision': precision_score(y_test, y_pred, average='weighted'),
                'recall': recall_score(y_test, y_pred, average='weighted'),
                'f1': f1_score(y_test, y_pred, average='weighted')
            }
            
            # 如果是二分类，添加AUC
            if len(np.unique(y_test)) == 2:
                y_pred_proba = model.predict_proba(X_test)[:, 1]
                metrics['auc'] = roc_auc_score(y_test, y_pred_proba)
        
        else:
            metrics = {
                'mse': mean_squared_error(y_test, y_pred),
                'mae': mean_absolute_error(y_test, y_pred),
                'rmse': np.sqrt(mean_squared_error(y_test, y_pred)),
                'r2': r2_score(y_test, y_pred)
            }
        
        return metrics
    
    def cross_validate_model(self, model, X: np.ndarray, y: np.ndarray) -> Dict[str, float]:
        """交叉验证"""
        cv_scores = cross_val_score(
            model, X, y,
            cv=self.config.model.cv_folds,
            scoring=self.config.model.scoring_metric
        )
        
        return {
            'cv_mean': cv_scores.mean(),
            'cv_std': cv_scores.std(),
            'cv_scores': cv_scores.tolist()
        }
    
    def train_all_models(self, df: pd.DataFrame) -> Dict[str, Dict[str, Any]]:
        """训练所有候选模型"""
        self.logger.info("开始训练所有候选模型...")
        
        # 准备数据
        X_train, X_test, y_train, y_test = self.prepare_data(df)
        
        # 获取候选模型和超参数
        model_candidates = self.get_model_candidates()
        param_grids = self.get_hyperparameter_grids()
        
        results = {}
        
        for model_name, model in model_candidates.items():
            self.logger.info(f"训练模型: {model_name}")
            
            try:
                # 训练模型
                param_grid = param_grids.get(model_name)
                trained_model = self.train_single_model(
                    model, X_train, y_train, param_grid, use_random_search=True
                )
                
                # 评估模型
                test_metrics = self.evaluate_model(trained_model, X_test, y_test)
                cv_metrics = self.cross_validate_model(trained_model, X_train, y_train)
                
                # 保存结果
                results[model_name] = {
                    'model': trained_model,
                    'test_metrics': test_metrics,
                    'cv_metrics': cv_metrics,
                    'best_params': getattr(trained_model, 'get_params', lambda: {})()
                }
                
                self.logger.info(f"{model_name} 训练完成")
                
            except Exception as e:
                self.logger.error(f"训练 {model_name} 时出错: {e}")
                continue
        
        self.models = results
        
        # 选择最佳模型
        self._select_best_model()
        
        return results
    
    def _select_best_model(self):
        """选择最佳模型"""
        if not self.models:
            return
        
        best_score = float('-inf') if self.task_type == 'classification' else float('inf')
        best_model_name = None
        
        for model_name, result in self.models.items():
            if self.task_type == 'classification':
                # 对于分类任务，选择准确率最高的
                score = result['test_metrics']['accuracy']
                if score > best_score:
                    best_score = score
                    best_model_name = model_name
            else:
                # 对于回归任务，选择RMSE最小的
                score = result['test_metrics']['rmse']
                if score < best_score:
                    best_score = score
                    best_model_name = model_name
        
        if best_model_name:
            self.best_model = {
                'name': best_model_name,
                'model': self.models[best_model_name]['model'],
                'metrics': self.models[best_model_name]['test_metrics']
            }
            
            self.logger.info(f"最佳模型: {best_model_name}, 分数: {best_score}")
    
    def save_model(self, model_path: str = None):
        """保存最佳模型"""
        if not self.best_model:
            raise ValueError("没有训练好的模型可以保存")
        
        if not model_path:
            model_path = f"models/best_model_{self.best_model['name']}.joblib"
        
        joblib.dump(self.best_model['model'], model_path)
        self.logger.info(f"模型已保存到: {model_path}")
        
        return model_path
    
    def load_model(self, model_path: str):
        """加载模型"""
        model = joblib.load(model_path)
        self.logger.info(f"模型已从 {model_path} 加载")
        return model
    
    def get_training_report(self) -> str:
        """生成训练报告"""
        if not self.models:
            return "没有训练结果可以报告"
        
        report_lines = []
        report_lines.append("=" * 50)
        report_lines.append("模型训练报告")
        report_lines.append("=" * 50)
        report_lines.append(f"任务类型: {self.task_type}")
        report_lines.append(f"评估指标: {self.config.model.scoring_metric}")
        report_lines.append("")
        
        # 所有模型的性能对比
        report_lines.append("📊 模型性能对比:")
        for model_name, result in self.models.items():
            report_lines.append(f"\n🔸 {model_name}:")
            for metric, value in result['test_metrics'].items():
                report_lines.append(f"  - {metric}: {value:.4f}")
            report_lines.append(f"  - CV平均分: {result['cv_metrics']['cv_mean']:.4f} (±{result['cv_metrics']['cv_std']:.4f})")
        
        # 最佳模型信息
        if self.best_model:
            report_lines.append(f"\n🏆 最佳模型: {self.best_model['name']}")
            report_lines.append("最佳模型性能:")
            for metric, value in self.best_model['metrics'].items():
                report_lines.append(f"  - {metric}: {value:.4f}")
        
        return "\n".join(report_lines)

# 使用示例
trainer = ModelTrainer(config)
training_results = trainer.train_all_models(df_engineered)

# 生成训练报告
training_report = trainer.get_training_report()
print(training_report)

# 保存最佳模型
model_path = trainer.save_model()

2. 模型评估与可视化

# src/models/model_evaluator.py
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import numpy as np
from sklearn.metrics import confusion_matrix, classification_report, roc_curve, auc
from sklearn.metrics import precision_recall_curve, learning_curve
import warnings
warnings.filterwarnings('ignore')

class ModelEvaluator:
    def __init__(self, figsize=(12, 8)):
        self.figsize = figsize
        plt.style.use('seaborn-v0_8')
    
    def plot_confusion_matrix(self, y_true, y_pred, labels=None, normalize=False):
        """绘制混淆矩阵"""
        cm = confusion_matrix(y_true, y_pred)
        
        if normalize:
            cm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]
            fmt = '.2f'
        else:
            fmt = 'd'
        
        plt.figure(figsize=(8, 6))
        sns.heatmap(cm, annot=True, fmt=fmt, cmap='Blues',
                   xticklabels=labels, yticklabels=labels)
        plt.title('混淆矩阵')
        plt.xlabel('预测标签')
        plt.ylabel('真实标签')
        plt.tight_layout()
        plt.show()
    
    def plot_roc_curves(self, models_results, X_test, y_test):
        """绘制ROC曲线对比"""
        plt.figure(figsize=self.figsize)
        
        for model_name, result in models_results.items():
            model = result['model']
            
            # 获取预测概率
            if hasattr(model, 'predict_proba'):
                y_pred_proba = model.predict_proba(X_test)[:, 1]
            elif hasattr(model, 'decision_function'):
                y_pred_proba = model.decision_function(X_test)
            else:
                continue
            
            # 计算ROC曲线
            fpr, tpr, _ = roc_curve(y_test, y_pred_proba)
            roc_auc = auc(fpr, tpr)
            
            plt.plot(fpr, tpr, label=f'{model_name} (AUC = {roc_auc:.3f})')
        
        plt.plot([0, 1], [0, 1], 'k--', label='随机分类器')
        plt.xlim([0.0, 1.0])
        plt.ylim([0.0, 1.05])
        plt.xlabel('假正率 (FPR)')
        plt.ylabel('真正率 (TPR)')
        plt.title('ROC曲线对比')
        plt.legend(loc="lower right")
        plt.grid(True, alpha=0.3)
        plt.tight_layout()
        plt.show()
    
    def plot_precision_recall_curves(self, models_results, X_test, y_test):
        """绘制精确率-召回率曲线"""
        plt.figure(figsize=self.figsize)
        
        for model_name, result in models_results.items():
            model = result['model']
            
            if hasattr(model, 'predict_proba'):
                y_pred_proba = model.predict_proba(X_test)[:, 1]
            elif hasattr(model, 'decision_function'):
                y_pred_proba = model.decision_function(X_test)
            else:
                continue
            
            precision, recall, _ = precision_recall_curve(y_test, y_pred_proba)
            avg_precision = auc(recall, precision)
            
            plt.plot(recall, precision, label=f'{model_name} (AP = {avg_precision:.3f})')
        
        plt.xlim([0.0, 1.0])
        plt.ylim([0.0, 1.05])
        plt.xlabel('召回率')
        plt.ylabel('精确率')
        plt.title('精确率-召回率曲线')
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.tight_layout()
        plt.show()
    
    def plot_learning_curves(self, model, X, y, cv=5):
        """绘制学习曲线"""
        train_sizes, train_scores, val_scores = learning_curve(
            model, X, y, cv=cv, n_jobs=-1,
            train_sizes=np.linspace(0.1, 1.0, 10)
        )
        
        train_mean = np.mean(train_scores, axis=1)
        train_std = np.std(train_scores, axis=1)
        val_mean = np.mean(val_scores, axis=1)
        val_std = np.std(val_scores, axis=1)
        
        plt.figure(figsize=self.figsize)
        plt.plot(train_sizes, train_mean, 'o-', color='blue', label='训练分数')
        plt.fill_between(train_sizes, train_mean - train_std, train_mean + train_std, 
                        alpha=0.1, color='blue')
        
        plt.plot(train_sizes, val_mean, 'o-', color='red', label='验证分数')
        plt.fill_between(train_sizes, val_mean - val_std, val_mean + val_std, 
                        alpha=0.1, color='red')
        
        plt.xlabel('训练样本数')
        plt.ylabel('分数')
        plt.title('学习曲线')
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.tight_layout()
        plt.show()
    
    def plot_residuals(self, y_true, y_pred):
        """绘制残差图（回归任务）"""
        residuals = y_true - y_pred
        
        fig, axes = plt.subplots(1, 2, figsize=(15, 6))
        
        # 残差散点图
        axes[0].scatter(y_pred, residuals, alpha=0.6)
        axes[0].axhline(y=0, color='red', linestyle='--')
        axes[0].set_xlabel('预测值')
        axes[0].set_ylabel('残差')
        axes[0].set_title('残差散点图')
        axes[0].grid(True, alpha=0.3)
        
        # 残差直方图
        axes[1].hist(residuals, bins=30, alpha=0.7)
        axes[1].set_xlabel('残差')
        axes[1].set_ylabel('频次')
        axes[1].set_title('残差分布')
        axes[1].grid(True, alpha=0.3)
        
        plt.tight_layout()
        plt.show()
    
    def plot_prediction_vs_actual(self, y_true, y_pred):
        """绘制预测值vs实际值图（回归任务）"""
        plt.figure(figsize=(8, 8))
        plt.scatter(y_true, y_pred, alpha=0.6)
        
        # 添加完美预测线
        min_val = min(min(y_true), min(y_pred))
        max_val = max(max(y_true), max(y_pred))
        plt.plot([min_val, max_val], [min_val, max_val], 'r--', lw=2)
        
        plt.xlabel('实际值')
        plt.ylabel('预测值')
        plt.title('预测值 vs 实际值')
        plt.grid(True, alpha=0.3)
        
        # 添加R²分数
        from sklearn.metrics import r2_score
        r2 = r2_score(y_true, y_pred)
        plt.text(0.05, 0.95, f'R² = {r2:.3f}', transform=plt.gca().transAxes,
                bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.5))
        
        plt.tight_layout()
        plt.show()
    
    def generate_classification_report(self, y_true, y_pred, target_names=None):
        """生成分类报告"""
        report = classification_report(y_true, y_pred, target_names=target_names)
        print("分类报告:")
        print("=" * 50)
        print(report)
        
        return report

# 使用示例
evaluator = ModelEvaluator()

# 准备测试数据
X_train, X_test, y_train, y_test = trainer.prepare_data(df_engineered)

# 如果是分类任务
if trainer.task_type == 'classification':
    # 混淆矩阵
    best_model = trainer.best_model['model']
    y_pred = best_model.predict(X_test)
    evaluator.plot_confusion_matrix(y_test, y_pred)
    
    # ROC曲线对比
    evaluator.plot_roc_curves(trainer.models, X_test, y_test)
    
    # 精确率-召回率曲线
    evaluator.plot_precision_recall_curves(trainer.models, X_test, y_test)
    
    # 分类报告
    evaluator.generate_classification_report(y_test, y_pred)

else:
    # 回归任务评估
    best_model = trainer.best_model['model']
    y_pred = best_model.predict(X_test)
    
    # 残差分析
    evaluator.plot_residuals(y_test, y_pred)
    
    # 预测vs实际
    evaluator.plot_prediction_vs_actual(y_test, y_pred)

# 学习曲线
evaluator.plot_learning_curves(trainer.best_model['model'], X_train, y_train)

🚀 模型部署

1. Flask API 部署

# src/deployment/flask_api.py
from flask import Flask, request, jsonify
import joblib
import pandas as pd
import numpy as np
from typing import Dict, Any
import logging
import os

app = Flask(__name__)

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ModelAPI:
    def __init__(self, model_path: str, feature_columns: list):
        self.model = joblib.load(model_path)
        self.feature_columns = feature_columns
        logger.info(f"模型已加载: {model_path}")
    
    def preprocess_input(self, data: Dict[str, Any]) -> np.ndarray:
        """预处理输入数据"""
        # 创建DataFrame
        df = pd.DataFrame([data])
        
        # 确保所有特征列都存在
        for col in self.feature_columns:
            if col not in df.columns:
                df[col] = 0  # 或其他默认值
        
        # 选择并排序特征列
        df = df[self.feature_columns]
        
        return df.values
    
    def predict(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """进行预测"""
        try:
            # 预处理输入
            X = self.preprocess_input(data)
            
            # 预测
            prediction = self.model.predict(X)[0]
            
            # 如果是分类任务，获取概率
            if hasattr(self.model, 'predict_proba'):
                probabilities = self.model.predict_proba(X)[0]
                classes = self.model.classes_
                
                return {
                    'prediction': prediction,
                    'probabilities': {
                        str(cls): float(prob) for cls, prob in zip(classes, probabilities)
                    },
                    'status': 'success'
                }
            else:
                return {
                    'prediction': float(prediction),
                    'status': 'success'
                }
        
        except Exception as e:
            logger.error(f"预测错误: {e}")
            return {
                'error': str(e),
                'status': 'error'
            }

# 初始化模型API
model_path = 'models/best_model.joblib'
feature_columns = selected_features  # 从特征工程步骤获取
model_api = ModelAPI(model_path, feature_columns)

@app.route('/health', methods=['GET'])
def health_check():
    """健康检查端点"""
    return jsonify({'status': 'healthy', 'message': 'API is running'})

@app.route('/predict', methods=['POST'])
def predict():
    """预测端点"""
    try:
        # 获取JSON数据
        data = request.get_json()
        
        if not data:
            return jsonify({'error': '没有提供数据', 'status': 'error'}), 400
        
        # 进行预测
        result = model_api.predict(data)
        
        if result['status'] == 'error':
            return jsonify(result), 500
        
        return jsonify(result)
    
    except Exception as e:
        logger.error(f"API错误: {e}")
        return jsonify({'error': str(e), 'status': 'error'}), 500

@app.route('/batch_predict', methods=['POST'])
def batch_predict():
    """批量预测端点"""
    try:
        data = request.get_json()
        
        if not data or 'instances' not in data:
            return jsonify({'error': '没有提供实例数据', 'status': 'error'}), 400
        
        results = []
        for instance in data['instances']:
            result = model_api.predict(instance)
            results.append(result)
        
        return jsonify({'predictions': results, 'status': 'success'})
    
    except Exception as e:
        logger.error(f"批量预测错误: {e}")
        return jsonify({'error': str(e), 'status': 'error'}), 500

@app.route('/model_info', methods=['GET'])
def model_info():
    """模型信息端点"""
    try:
        info = {
            'model_type': type(model_api.model).__name__,
            'feature_count': len(model_api.feature_columns),
            'features': model_api.feature_columns,
            'status': 'success'
        }
        
        # 如果是分类模型，添加类别信息
        if hasattr(model_api.model, 'classes_'):
            info['classes'] = model_api.model.classes_.tolist()
        
        return jsonify(info)
    
    except Exception as e:
        logger.error(f"获取模型信息错误: {e}")
        return jsonify({'error': str(e), 'status': 'error'}), 500

if __name__ == '__main__':
    port = int(os.environ.get('PORT', 5000))
    app.run(host='0.0.0.0', port=port, debug=False)

2. Docker 容器化

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# 复制requirements文件
COPY requirements.txt .

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY src/ ./src/
COPY models/ ./models/
COPY config.yaml .

# 创建非root用户
RUN useradd --create-home --shell /bin/bash app
USER app

# 暴露端口
EXPOSE 5000

# 启动命令
CMD ["python", "src/deployment/flask_api.py"]

# docker-compose.yml
version: '3.8'

services:
  ml-api:
    build: .
    ports:
      - "5000:5000"
    environment:
      - FLASK_ENV=production
      - PORT=5000
    volumes:
      - ./models:/app/models:ro
      - ./logs:/app/logs
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - ml-api
    restart: unless-stopped

3. 模型监控

# src/monitoring/model_monitor.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import logging
from typing import Dict, List, Any
import sqlite3
import json

class ModelMonitor:
    def __init__(self, db_path: str = 'monitoring.db'):
        self.db_path = db_path
        self.logger = logging.getLogger(__name__)
        self._init_database()
    
    def _init_database(self):
        """初始化监控数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 创建预测日志表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS predictions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME,
                input_data TEXT,
                prediction TEXT,
                confidence REAL,
                response_time REAL
            )
        ''')
        
        # 创建性能指标表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS performance_metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME,
                metric_name TEXT,
                metric_value REAL,
                period TEXT
            )
        ''')
        
        # 创建数据漂移表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS data_drift (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME,
                feature_name TEXT,
                drift_score REAL,
                threshold REAL,
                is_drift BOOLEAN
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def log_prediction(self, input_data: Dict, prediction: Any, 
                      confidence: float = None, response_time: float = None):
        """记录预测日志"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO predictions (timestamp, input_data, prediction, confidence, response_time)
            VALUES (?, ?, ?, ?, ?)
        ''', (
            datetime.now(),
            json.dumps(input_data),
            json.dumps(prediction) if not isinstance(prediction, (int, float, str)) else str(prediction),
            confidence,
            response_time
        ))
        
        conn.commit()
        conn.close()
    
    def calculate_drift_score(self, reference_data: np.ndarray, 
                            current_data: np.ndarray) -> float:
        """计算数据漂移分数（使用KS检验）"""
        from scipy.stats import ks_2samp
        
        # 对于多维数据，计算每个特征的KS统计量的平均值
        if reference_data.ndim > 1:
            ks_stats = []
            for i in range(reference_data.shape[1]):
                ks_stat, _ = ks_2samp(reference_data[:, i], current_data[:, i])
                ks_stats.append(ks_stat)
            return np.mean(ks_stats)
        else:
            ks_stat, _ = ks_2samp(reference_data, current_data)
            return ks_stat
    
    def detect_data_drift(self, reference_data: pd.DataFrame, 
                         current_data: pd.DataFrame, 
                         threshold: float = 0.1) -> Dict[str, Any]:
        """检测数据漂移"""
        drift_results = {}
        
        for column in reference_data.columns:
            if column in current_data.columns:
                # 只对数值型特征检测漂移
                if pd.api.types.is_numeric_dtype(reference_data[column]):
                    drift_score = self.calculate_drift_score(
                        reference_data[column].values,
                        current_data[column].values
                    )
                    
                    is_drift = drift_score > threshold
                    
                    drift_results[column] = {
                        'drift_score': drift_score,
                        'is_drift': is_drift,
                        'threshold': threshold
                    }
                    
                    # 记录到数据库
                    self._log_drift_detection(column, drift_score, threshold, is_drift)
        
        return drift_results
    
    def _log_drift_detection(self, feature_name: str, drift_score: float, 
                           threshold: float, is_drift: bool):
        """记录漂移检测结果"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO data_drift (timestamp, feature_name, drift_score, threshold, is_drift)
            VALUES (?, ?, ?, ?, ?)
        ''', (datetime.now(), feature_name, drift_score, threshold, is_drift))
        
        conn.commit()
        conn.close()
    
    def calculate_performance_metrics(self, y_true: List, y_pred: List, 
                                    task_type: str = 'classification') -> Dict[str, float]:
        """计算性能指标"""
        if task_type == 'classification':
            from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
            
            metrics = {
                'accuracy': accuracy_score(y_true, y_pred),
                'precision': precision_score(y_true, y_pred, average='weighted'),
                'recall': recall_score(y_true, y_pred, average='weighted'),
                'f1': f1_score(y_true, y_pred, average='weighted')
            }
        else:
            from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
            
            metrics = {
                'mse': mean_squared_error(y_true, y_pred),
                'mae': mean_absolute_error(y_true, y_pred),
                'rmse': np.sqrt(mean_squared_error(y_true, y_pred)),
                'r2': r2_score(y_true, y_pred)
            }
        
        # 记录到数据库
        for metric_name, metric_value in metrics.items():
            self._log_performance_metric(metric_name, metric_value)
        
        return metrics
    
    def _log_performance_metric(self, metric_name: str, metric_value: float, 
                              period: str = 'daily'):
        """记录性能指标"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO performance_metrics (timestamp, metric_name, metric_value, period)
            VALUES (?, ?, ?, ?)
        ''', (datetime.now(), metric_name, metric_value, period))
        
        conn.commit()
        conn.close()
    
    def get_monitoring_report(self, days: int = 7) -> Dict[str, Any]:
        """生成监控报告"""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        
        conn = sqlite3.connect(self.db_path)
        
        # 预测统计
        prediction_stats = pd.read_sql_query('''
            SELECT COUNT(*) as total_predictions,
                   AVG(response_time) as avg_response_time,
                   AVG(confidence) as avg_confidence
            FROM predictions
            WHERE timestamp BETWEEN ? AND ?
        ''', conn, params=(start_date, end_date))
        
        # 性能趋势
        performance_trend = pd.read_sql_query('''
            SELECT metric_name, AVG(metric_value) as avg_value,
                   MIN(metric_value) as min_value,
                   MAX(metric_value) as max_value
            FROM performance_metrics
            WHERE timestamp BETWEEN ? AND ?
            GROUP BY metric_name
        ''', conn, params=(start_date, end_date))
        
        # 漂移检测结果
        drift_summary = pd.read_sql_query('''
            SELECT feature_name, AVG(drift_score) as avg_drift_score,
                   SUM(CASE WHEN is_drift THEN 1 ELSE 0 END) as drift_count
            FROM data_drift
            WHERE timestamp BETWEEN ? AND ?
            GROUP BY feature_name
        ''', conn, params=(start_date, end_date))
        
        conn.close()
        
        return {
            'period': f'{days} days',
            'prediction_stats': prediction_stats.to_dict('records')[0] if not prediction_stats.empty else {},
            'performance_trend': performance_trend.to_dict('records'),
            'drift_summary': drift_summary.to_dict('records')
        }

# 使用示例
monitor = ModelMonitor()

# 模拟监控使用
# monitor.log_prediction(
#     input_data={'feature1': 1.0, 'feature2': 2.0},
#     prediction=1,
#     confidence=0.85,
#     response_time=0.05
# )

# 生成监控报告
# report = monitor.get_monitoring_report(days=7)
# print(json.dumps(report, indent=2))

🎯 最佳实践与总结

1. 项目管理最佳实践

# 项目配置管理
class MLProjectConfig:
    """机器学习项目配置管理"""
    
    def __init__(self):
        self.project_structure = {
            'data_management': {
                'version_control': 'DVC',
                'storage': 'cloud_storage',
                'backup_strategy': 'automated'
            },
            'experiment_tracking': {
                'tool': 'MLflow',
                'metrics_logging': True,
                'artifact_storage': True
            },
            'code_quality': {
                'linting': 'flake8',
                'formatting': 'black',
                'testing': 'pytest',
                'documentation': 'sphinx'
            },
            'deployment': {
                'containerization': 'docker',
                'orchestration': 'kubernetes',
                'monitoring': 'prometheus'
            }
        }
    
    def get_best_practices(self):
        return {
            '数据管理': [
                '使用版本控制管理数据集',
                '建立数据质量检查流程',
                '实施数据隐私保护措施',
                '建立数据血缘关系追踪'
            ],
            '模型开发': [
                '遵循可复现性原则',
                '实施交叉验证策略',
                '建立基线模型对比',
                '记录实验过程和结果'
            ],
            '代码质量': [
                '编写单元测试',
                '使用类型注解',
                '遵循PEP8规范',
                '编写详细文档'
            ],
            '部署运维': [
                '实施CI/CD流程',
                '建立监控告警机制',
                '实施A/B测试',
                '建立回滚机制'
            ]
        }

# 性能优化建议
class PerformanceOptimizer:
    """性能优化建议"""
    
    @staticmethod
    def get_optimization_tips():
        return {
            '数据处理优化': [
                '使用向量化操作替代循环',
                '合理使用数据类型（如category）',
                '实施数据分块处理',
                '使用并行处理'
            ],
            '模型训练优化': [
                '使用早停机制',
                '实施学习率调度',
                '使用批量处理',
                '考虑模型压缩技术'
            ],
            '推理优化': [
                '模型量化',
                '使用ONNX格式',
                '实施模型缓存',
                '批量推理'
            ],
            '系统优化': [
                '使用GPU加速',
                '优化内存使用',
                '实施负载均衡',
                '使用CDN加速'
            ]
        }

# 常见问题解决方案
class TroubleshootingGuide:
    """常见问题解决指南"""
    
    @staticmethod
    def get_common_issues():
        return {
            '过拟合问题': {
                '症状': ['训练准确率高，验证准确率低', '验证损失上升'],
                '解决方案': [
                    '增加训练数据',
                    '使用正则化技术',
                    '减少模型复杂度',
                    '使用Dropout',
                    '实施早停'
                ]
            },
            '欠拟合问题': {
                '症状': ['训练和验证准确率都很低', '损失下降缓慢'],
                '解决方案': [
                    '增加模型复杂度',
                    '增加特征',
                    '调整学习率',
                    '增加训练轮数'
                ]
            },
            '数据泄露': {
                '症状': ['性能异常好', '特征重要性异常'],
                '解决方案': [
                    '检查特征工程过程',
                    '确保时间序列数据的正确分割',
                    '检查目标变量相关特征'
                ]
            },
            '类别不平衡': {
                '症状': ['某些类别预测效果差', '混淆矩阵不均衡'],
                '解决方案': [
                    '使用重采样技术',
                    '调整类别权重',
                    '使用集成方法',
                    '选择合适的评估指标'
                ]
            }
        }

结语

机器学习项目的成功不仅仅依赖于算法的选择，更重要的是建立一个完整、可靠的工作流程。从数据理解到模型部署，每个环节都需要细致的规划和执行。

通过本文的深入探讨，我们了解了：

1. 数据为王：高质量的数据是成功的基础，数据预处理和质量检查不可忽视
2. 特征工程的艺术：好的特征往往比复杂的算法更重要
3. 模型选择与调优：没有万能的算法，需要根据具体问题选择合适的方法
4. 评估的重要性：全面的模型评估帮助我们理解模型的真实性能
5. 部署与监控：模型的生命周期不止于训练，部署后的监控同样重要

在实际项目中，要始终记住：

• 可复现性：确保实验和结果可以被重现
• 可解释性：理解模型的决策过程
• 可扩展性：考虑系统的扩展能力
• 可维护性：编写清晰、可维护的代码

机器学习是一个快速发展的领域，新的算法和技术不断涌现。但无论技术如何变化，扎实的基础知识和良好的工程实践永远是成功的关键。

希望这篇文章能够帮助你建立完整的机器学习项目思维，在实际工作中取得更好的成果。记住，机器学习不仅仅是科学，也是工程艺术！

---

你在机器学习项目中遇到过哪些挑战？欢迎在评论区分享你的经验和心得！