机器学习工程实践
第一章：ML 工程概述
1.1 ML 工程师 vs 数据科学家
| 角色 | 重点 | 核心技能 |
| 数据科学家 | 发现洞察、建模 | 统计、可视化、实验设计 |
| ML 工程师 | 部署、扩展、维护 | 软件工程、MLOps、系统设计 |
| 研究科学家 | 新算法、新架构 | 数学、论文阅读、实验 |
1.2 ML 系统生命周期
数据管道 → 特征工程 → 模型训练 → 评估验证 → 部署上线 → 监控运维 → 迭代优化
第二章：特征工程
2.1 数值特征
from sklearn.preprocessing import StandardScaler, MinMaxScaler, PowerTransformer
# 标准化 (z-score)
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X_train)
# 归一化 (min-max)
scaler = MinMaxScaler(feature_range=(0, 1))
X_norm = scaler.fit_transform(X_train)
# Box-Cox 变换（处理偏态分布）
transformer = PowerTransformer(method='box-cox')
X_transformed = transformer.fit_transform(X_train)
2.2 类别特征
from sklearn.preprocessing import OneHotEncoder, OrdinalEncoder, LabelEncoder
# 独热编码（无序类别）
encoder = OneHotEncoder(sparse_output=False, drop='first')
X_encoded = encoder.fit_transform(df[['city', 'department']])
# 有序编码（有序类别）
encoder = OrdinalEncoder(categories=[['低', '中', '高']])
X_ordered = encoder.fit_transform(df[['level']])
# 目标编码（高基数类别）
from category_encoders import TargetEncoder
encoder = TargetEncoder()
X_target = encoder.fit_transform(df['product_category'], y)
2.3 特征选择
from sklearn.feature_selection import (
SelectKBest, f_classif, RFE, mutual_info_classif
)
# 统计检验
selector = SelectKBest(score_func=f_classif, k=10)
X_selected = selector.fit_transform(X, y)
# 递归特征消除
from sklearn.ensemble import RandomForestClassifier
rfe = RFE(RandomForestClassifier(), n_features_to_select=10)
X_rfe = rfe.fit_transform(X, y)
# 互信息
mi_scores = mutual_info_classif(X, y, random_state=42)
第三章：模型训练与调优
3.1 交叉验证
from sklearn.model_selection import (
cross_val_score, StratifiedKFold, GridSearchCV, RandomizedSearchCV
)
# K 折交叉验证
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
scores = cross_val_score(model, X, y, cv=cv, scoring='f1_macro')
print(f"F1: {scores.mean():.3f} ± {scores.std():.3f}")
# 网格搜索
param_grid = {
'n_estimators': [100, 200, 500],
'max_depth': [3, 5, 7, None],
'learning_rate': [0.01, 0.05, 0.1],
}
grid = GridSearchCV(model, param_grid, cv=5, scoring='roc_auc', n_jobs=-1)
grid.fit(X_train, y_train)
print(f"Best params: {grid.best_params_}")
print(f"Best score: {grid.best_score_:.4f}")
# 随机搜索（更高效）
from scipy.stats import uniform
param_dist = {
'n_estimators': [100, 200, 500, 1000],
'learning_rate': uniform(0.01, 0.2),
'subsample': uniform(0.6, 0.4),
}
random_search = RandomizedSearchCV(model, param_dist, n_iter=50, cv=5)
3.2 集成方法
from sklearn.ensemble import (
VotingClassifier, StackingClassifier,
BaggingClassifier, AdaBoostClassifier
)
from xgboost import XGBClassifier
from lightgbm import LGBMClassifier
# 投票集成
voting = VotingClassifier(estimators=[
('rf', RandomForestClassifier()),
('xgb', XGBClassifier()),
('lgb', LGBMClassifier()),
], voting='soft')
# 堆叠集成
stacking = StackingClassifier(estimators=[
('rf', RandomForestClassifier()),
('xgb', XGBClassifier()),
], final_estimator=LogisticRegression(), cv=5)
3.3 深度学习框架
import torch
import torch.nn as nn
# 定义网络
class Classifier(nn.Module):
def __init__(self, input_dim, hidden_dim, num_classes):
super().__init__()
self.net = nn.Sequential(
nn.Linear(input_dim, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(0.3),
nn.Linear(hidden_dim, num_classes)
)
def forward(self, x):
return self.net(x)
# 训练循环
model = Classifier(784, 256, 10)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()
for epoch in range(20):
model.train()
for batch_x, batch_y in train_loader:
pred = model(batch_x)
loss = criterion(pred, batch_y)
optimizer.zero_grad()
loss.backward()
optimizer.step()
第四章：MLOps 实践
4.1 实验追踪
import mlflow
mlflow.set_experiment("my-ml-project")
with mlflow.start_run():
# 记录参数
mlflow.log_params({"lr": 0.01, "epochs": 50, "batch_size": 32})
# 记录指标
for epoch in range(50):
train_loss = train_one_epoch(model)
val_acc = evaluate(model)
mlflow.log_metrics({"train_loss": train_loss, "val_acc": val_acc}, step=epoch)
# 记录模型
mlflow.sklearn.log_model(model, "model")
4.2 模型服务
# FastAPI 模型服务
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
app = FastAPI()
model = joblib.load("model.pkl")
class InputData(BaseModel):
features: list[float]
@app.post("/predict")
def predict(data: InputData):
import numpy as np
X = np.array(data.features).reshape(1, -1)
prediction = model.predict(X)
probability = model.predict_proba(X).max()
return {
"prediction": int(prediction[0]),
"confidence": float(probability),
}
4.3 Docker 部署
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: '3.8'
services:
ml-api:
build: .
ports:
- "8000:8000"
environment:
- MODEL_PATH=/models/latest
volumes:
- ./models:/models
redis:
image: redis:7-alpine
ports:
- "6379:6379"
第五章：模型监控
5.1 数据漂移检测
from scipy.stats import ks_2samp
def detect_drift(reference_data, current_data, threshold=0.05):
"""KS 检验检测数据分布漂移"""
drift_features = []
for col in reference_data.columns:
stat, p_value = ks_2samp(reference_data[col], current_data[col])
if p_value < threshold:
drift_features.append({
'feature': col,
'ks_statistic': stat,
'p_value': p_value
})
return pd.DataFrame(drift_features)
5.2 A/B 测试
from scipy.stats import ttest_ind
def ab_test(control_metrics, treatment_metrics):
"""A/B 测试显著性检验"""
t_stat, p_value = ttest_ind(treatment_metrics, control_metrics)
control_mean = control_metrics.mean()
treatment_mean = treatment_metrics.mean()
lift = (treatment_mean - control_mean) / control_mean
return {
'control_mean': control_mean,
'treatment_mean': treatment_mean,
'lift_pct': lift * 100,
'p_value': p_value,
'significant': p_value < 0.05
}
学习资源