Python 机器学习实战：模型构建与评估最佳实践

使用示例如下：

if __name__ == "__main__":
    np.random.seed(42)
    X = np.random.randn(1000, 5)
    true_weights = np.array([1.5, -2.0, 0.5, 1.0, -0.5])
    y = np.dot(X, true_weights) + np.random.randn(1000) * 0.1
    split = int(0.8 * len(X))
    X_train, X_test = X[:split], X[split:]
    y_train, y_test = y[:split], y[split:]
    model = CoreAIModel(learning_rate=0.01, epochs=100, batch_size=32)
    model.fit(X_train, y_train)
    train_score = model.score(X_train, y_train)
    test_score = model.score(X_test, y_test)
    print(f"\n训练集 R²: {train_score:.4f}")
    print(f"测试集 R²: {test_score:.4f}")

进阶框架实现

实际项目中通常使用 TensorFlow 或 PyTorch。以下是两种框架的简要对比实现：

# TensorFlow 实现示例
class TensorFlowModel:
    def __init__(self, input_dim: int, hidden_units: list = [64, 32]):
        self.model = self._build_model(input_dim, hidden_units)

    def _build_model(self, input_dim: int, hidden_units: list):
        inputs = keras.Input(shape=(input_dim,))
        x = inputs
        for units in hidden_units:
            x = layers.Dense(units, activation='relu')(x)
            x = layers.BatchNormalization()(x)
            x = layers.Dropout(0.2)(x)
        outputs = layers.Dense(1)(x)
        model = keras.Model(inputs=inputs, outputs=outputs)
        model.compile(optimizer=keras.optimizers.Adam(learning_rate=0.001), loss='mse', metrics=['mae'])
        return model

# PyTorch 实现示例
class PyTorchModel(nn.Module):
    def __init__(self, input_dim: int, hidden_units: list = [64, 32]):
        super(PyTorchModel, self).__init__()
        layers_list = []
        prev_units = input_dim
        for units in hidden_units:
            layers_list.append(nn.Linear(prev_units, units))
            layers_list.append(nn.ReLU())
            layers_list.append(nn.BatchNorm1d(units))
            layers_list.append(nn.Dropout(0.2))
            prev_units = units
        layers_list.append(nn.Linear(prev_units, 1))
        self.network = nn.Sequential(*layers_list)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.network(x)

数据处理流程

数据质量直接决定模型上限。标准流程包括缺失值处理、类别编码和标准化：

class DataProcessor:
    def __init__(self):
        self.scaler = StandardScaler()
        self.label_encoders = {}
        self.imputer = SimpleImputer(strategy='mean')

    def process(self, data: pd.DataFrame, target_col: str, categorical_cols: list = None, test_size: float = 0.2):
        X = data.drop(columns=[target_col])
        y = data[target_col]
        # 处理缺失值
        X_num = X.select_dtypes(include=[np.number])
        X[X_num.columns] = self.imputer.fit_transform(X_num)
        # 编码类别特征
        if categorical_cols:
            for col in categorical_cols:
                if col in X.columns:
                    le = LabelEncoder()
                    X[col] = le.fit_transform(X[col].astype(str))
                    self.label_encoders[col] = le
        # 标准化
        X_scaled = self.scaler.fit_transform(X)
        # 划分数据集
        X_train, X_test, y_train, y_test = train_test_split(
            X_scaled, y, test_size=test_size, random_state=42)
        return X_train, X_test, y_train, y_test

模型评估方法

选择合适的评估指标至关重要。分类问题关注准确率、召回率等，回归问题则关注 MSE、R² 等。

class ModelEvaluator:
    @staticmethod
    def evaluate_classification(y_true, y_pred, y_prob=None):
        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')
        }
        if y_prob is not None:
            metrics['roc_auc'] = roc_auc_score(y_true, y_prob, multi_class='ovr')
        return metrics

    @staticmethod
    def evaluate_regression(y_true, y_pred):
        return {
            'mse': mean_squared_error(y_true, y_pred),
            'rmse': np.sqrt(mean_squared_error(y_true, y_pred)),
            'mae': mean_absolute_error(y_true, y_pred),
            'r2': r2_score(y_true, y_pred)
        }

实践应用与案例

场景分析

主要应用场景包括数据分析挖掘、模型训练优化等。针对不同问题类型推荐不同算法：

实施步骤

1. 环境准备：创建虚拟环境并安装核心库（numpy, pandas, scikit-learn, tensorflow, torch）。
2. 项目结构：遵循标准目录结构（data, notebooks, src, tests, configs）。
3. 开发流程：数据准备 -> 特征工程 -> 模型选择 -> 训练优化 -> 部署上线。

最佳实践

• 代码规范：使用类型注解、编写文档字符串、遵循 PEP8 规范。
• 实验管理：使用版本控制、记录实验参数、保存模型检查点。

案例分析：房价预测

以房价预测为例，展示完整管道构建：

class HousePricePredictor:
    def __init__(self):
        self.model = None
        self.preprocessor = None

    def prepare_data(self, data: pd.DataFrame, target_col: str):
        X = data.drop(columns=[target_col])
        y = data[target_col]
        numeric_features = X.select_dtypes(include=[np.number]).columns.tolist()
        categorical_features = X.select_dtypes(exclude=[np.number]).columns.tolist()
        self.preprocessor = ColumnTransformer(
            transformers=[
                ('num', StandardScaler(), numeric_features),
                ('cat', OneHotEncoder(handle_unknown='ignore'), categorical_features)
            ])
        return train_test_split(X, y, test_size=0.2, random_state=42)

    def train(self, X_train, y_train):
        self.model = Pipeline([
            ('preprocessor', self.preprocessor),
            ('regressor', GradientBoostingRegressor(
                n_estimators=200, learning_rate=0.1, max_depth=5, random_state=42))
        ])
        self.model.fit(X_train, y_train)
        return self

    def evaluate(self, X_test, y_test):
        y_pred = self.model.predict(X_test)
        metrics = {
            'RMSE': np.sqrt(mean_squared_error(y_test, y_pred)),
            'MAE': mean_absolute_error(y_test, y_pred),
            'R2': r2_score(y_test, y_pred)
        }
        return metrics, y_pred

典型效果指标：RMSE 约 25000，R² 可达 0.89。

常见问题与解决

Q1：如何选择合适的模型？ 小样本建议传统 ML，中等样本用集成学习，大样本可尝试深度学习。

Q2：如何处理数据不平衡？ 可使用 SMOTE 过采样、RandomUnderSampler 欠采样或调整类别权重。

Q3：如何避免过拟合？ 增加数据量、使用正则化、添加 Dropout 或早停法。

总结

本文涵盖了 Python 机器学习的关键环节，从基础原理到工程实践。建议读者在理解原理的基础上动手实现，循序渐进地提升技能。保持对新技术的关注，积极参与社区交流，将有助于在 AI 领域持续成长。