Python入门（18）--实战项目

机器学习实战项目指南 🤖

项目概览 🌐

本项目是一个综合性的机器学习入门实战指南，通过实际案例展示机器学习项目的完整生命周期，包括数据处理、模型训练、评估和部署等关键环节。

1. 系统架构 🏗️

1.1 核心组件

数据处理模块
- 数据清洗与预处理
- 特征工程
- 数据集划分
- 数据增强
- 特征选择
模型训练模块
- 模型选择
- 参数调优
- 交叉验证
- 模型集成
- 训练监控
评估与分析模块
- 性能指标计算
- 结果可视化
- 模型解释
- 错误分析
- 对比实验
预测系统模块
- 模型部署
- API服务
- 批量预测
- 结果输出
- 监控反馈

2. 技术栈 🛠️

核心框架:
- scikit-learn
- tensorflow
- pytorch
- xgboost
数据处理:
- pandas
- numpy
- scipy
可视化:
- matplotlib
- seaborn
- plotly
部署:
- flask
- fastapi
- docker
监控:
- mlflow
- wandb
- tensorboard

3. 系统详细设计 📋

3.1 数据处理流水线

from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.model_selection import train_test_split
import pandas as pd
import numpy as np

class DataPipeline:
    def __init__(self, config: dict):
        self.config = config
        self.scaler = StandardScaler()
        self.label_encoder = LabelEncoder()
        
    def load_data(self, file_path: str) -> pd.DataFrame:
        """加载数据集并进行基础检查"""
        df = pd.read_csv(file_path)
        print(f"数据集形状: {df.shape}")
        print(f"缺失值统计:\n{df.isnull().sum()}")
        return df
    
    def preprocess_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """特征预处理"""
        # 处理缺失值
        for col in df.columns:
            if df[col].isnull().sum() > 0:
                if df[col].dtype in ['int64', 'float64']:
                    df[col].fillna(df[col].mean(), inplace=True)
                else:
                    df[col].fillna(df[col].mode()[0], inplace=True)
        
        # 特征编码
        categorical_cols = df.select_dtypes(include=['object']).columns
        for col in categorical_cols:
            df[col] = self.label_encoder.fit_transform(df[col])
        
        # 特征缩放
        numerical_cols = df.select_dtypes(include=['int64', 'float64']).columns
        df[numerical_cols] = self.scaler.fit_transform(df[numerical_cols])
        
        return df
    
    def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """特征工程"""
        # 示例：创建交互特征
        for col1 in df.columns:
            for col2 in df.columns:
                if col1 != col2 and df[col1].dtype in ['int64', 'float64'] and df[col2].dtype in ['int64', 'float64']:
                    df[f'{col1}_{col2}_interaction'] = df[col1] * df[col2]
        
        return df

3.2 模型训练管理器

from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import GridSearchCV
from typing import Dict, Any
import mlflow
import joblib

class ModelTrainer:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.models = {
            'random_forest': RandomForestClassifier(),
            'gradient_boosting': GradientBoostingClassifier(),
            'logistic_regression': LogisticRegression()
        }
        self.param_grids = {
            'random_forest': {
                'n_estimators': [100, 200, 300],
                'max_depth': [None, 10, 20, 30],
                'min_samples_split': [2, 5, 10]
            },
            'gradient_boosting': {
                'n_estimators': [100, 200],
                'learning_rate': [0.01, 0.1],
                'max_depth': [3, 5, 7]
            },
            'logistic_regression': {
                'C': [0.1, 1.0, 10.0],
                'penalty': ['l1', 'l2']
            }
        }
    
    def train_model(self, X_train, y_train, model_name: str):
        """训练模型并进行参数调优"""
        mlflow.start_run()
        
        # 记录训练参数
        mlflow.log_params(self.config)
        
        # 网格搜索最优参数
        model = self.models[model_name]
        grid_search = GridSearchCV(
            model,
            self.param_grids[model_name],
            cv=5,
            scoring='accuracy',
            n_jobs=-1
        )
        
        grid_search.fit(X_train, y_train)
        
        # 记录最优参数和性能指标
        mlflow.log_params(grid_search.best_params_)
        mlflow.log_metric('best_score', grid_search.best_score_)
        
        # 保存模型
        mlflow.sklearn.log_model(grid_search.best_estimator_, "model")
        
        mlflow.end_run()
        return grid_search.best_estimator_

3.3 模型评估器

from sklearn.metrics import accuracy_score, precision_recall_fscore_support
from sklearn.metrics import confusion_matrix, roc_curve, auc
import matplotlib.pyplot as plt
import seaborn as sns
from typing import Dict, Any

class ModelEvaluator:
    def __init__(self):
        self.metrics = {}
    
    def evaluate_model(self, model, X_test, y_test) -> Dict[str, Any]:
        """综合评估模型性能"""
        # 获取预测结果
        y_pred = model.predict(X_test)
        y_pred_proba = model.predict_proba(X_test)[:, 1]
        
        # 计算各项指标
        accuracy = accuracy_score(y_test, y_pred)
        precision, recall, f1, _ = precision_recall_fscore_support(
            y_test, y_pred, average='weighted'
        )
        
        # 计算ROC曲线
        fpr, tpr, _ = roc_curve(y_test, y_pred_proba)
        roc_auc = auc(fpr, tpr)
        
        self.metrics = {
            'accuracy': accuracy,
            'precision': precision,
            'recall': recall,
            'f1': f1,
            'roc_auc': roc_auc
        }
        
        return self.metrics
    
    def plot_results(self, y_test, y_pred):
        """可视化评估结果"""
        # 混淆矩阵
        plt.figure(figsize=(10, 8))
        cm = confusion_matrix(y_test, y_pred)
        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues')
        plt.title('Confusion Matrix')
        plt.ylabel('True Label')
        plt.xlabel('Predicted Label')
        plt.show()
        
        # 性能指标柱状图
        plt.figure(figsize=(10, 6))
        metrics_df = pd.DataFrame(
            list(self.metrics.items()),
            columns=['Metric', 'Value']
        )
        sns.barplot(x='Metric', y='Value', data=metrics_df)
        plt.title('Model Performance Metrics')
        plt.xticks(rotation=45)
        plt.show()

3.4 预测系统服务

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import numpy as np
import joblib
from typing import List, Dict

class PredictionRequest(BaseModel):
    features: List[float]

class PredictionService:
    def __init__(self, model_path: str):
        self.model = joblib.load(model_path)
        self.app = FastAPI()
        self.setup_routes()
    
    def setup_routes(self):
        @self.app.post("/predict")
        async def predict(request: PredictionRequest):
            try:
                features = np.array(request.features).reshape(1, -1)
                prediction = self.model.predict(features)
                probability = self.model.predict_proba(features)
                
                return {
                    "prediction": int(prediction[0]),
                    "probability": float(probability[0][1]),
                    "status": "success"
                }
            except Exception as e:
                raise HTTPException(
                    status_code=500,
                    detail=f"预测失败: {str(e)}"
                )
    
    def start(self, host: str = "0.0.0.0", port: int = 8000):
        import uvicorn
        uvicorn.run(self.app, host=host, port=port)

4. 部署与优化 🚀

4.1 Docker部署配置

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "main.py"]

4.2 性能优化建议

数据处理优化
- 使用增量学习
- 特征选择优化
- 数据采样策略
- 并行处理
模型优化
- 模型压缩
- 量化技术
- 模型蒸馏
- 超参数优化
部署优化
- 批量预测
- 模型缓存
- 负载均衡
- 资源管理

5. 实践案例 📊

5.1 信用卡欺诈检测

# 完整示例代码
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split

# 生成示例数据
X, y = make_classification(
    n_samples=10000,
    n_features=20,
    n_informative=15,
    n_redundant=5,
    random_state=42
)

# 数据集划分
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# 初始化组件
data_pipeline = DataPipeline(config={})
model_trainer = ModelTrainer(config={})
evaluator = ModelEvaluator()

# 数据预处理
X_train_processed = data_pipeline.preprocess_features(pd.DataFrame(X_train))
X_test_processed = data_pipeline.preprocess_features(pd.DataFrame(X_test))

# 模型训练
model = model_trainer.train_model(
    X_train_processed,
    y_train,
    'random_forest'
)

# 模型评估
metrics = evaluator.evaluate_model(model, X_test_processed, y_test)
evaluator.plot_results(y_test, model.predict(X_test_processed))

# 部署预测服务
prediction_service = PredictionService('model.joblib')
prediction_service.start()