Machine Learning: 项目开发 — SalesPredict端到端实现指南
代码是设计到产品的桥梁——这一课把前23课的所有知识都写进可运行的代码。
1. 你将学到
- 数据管道：原始数据清洗 → 特征工程 → 训练/测试集划分，Pandas + sklearn Pipeline
- 模型训练与对比：LinearRegression → XGBoost → LightGBM → PyTorch MLP
- 最优模型调优：Optuna贝叶斯优化，5折时序交叉验证，MAPE从15%降至8%
- 模型注册：MLflow Model Registry管理最佳模型版本
- 关键代码解析：逐模块讲解设计决策
2. 一个ML工程师的真实故事
(1) 痛点：设计做完了但不知道代码怎么组织
Bob完成了SalesPredict的系统设计，但面对空白的代码仓库不知道怎么组织——数据管道放哪？特征工程怎么模块化？实验管理怎么集成？设计和代码之间的鸿沟需要一个可参照的模板。
(2) 端到端代码的解法
本课提供SalesPredict的完整可运行代码——从数据生成到模型部署，每个模块职责清晰，可直接复用。
PYTHON
# Project structure
# salespredict/
# ├── data/ # Data pipeline
# ├── features/ # Feature engineering
# ├── models/ # Model training
# ├── evaluation/ # Model evaluation
# └── api/ # FastAPI service
graph TB
RAW[Raw Data<br/>orders/users/products] --> CLEAN[Data Cleaning<br/>dedup/fill/outlier] --> FEAT[Feature Engineering<br/>lag/rolling/encode] --> SPLIT[Train/Test Split<br/>TimeSeriesSplit]
SPLIT --> BASE[LinearRegression<br/>MAPE 15%]
SPLIT --> XGB[XGBoost<br/>MAPE 9%]
SPLIT --> LGBM[LightGBM + Optuna<br/>MAPE 8%]
SPLIT --> MLP[PyTorch MLP<br/>MAPE 8.5%]
LGBM --> MLFLOW[MLflow Registry<br/>Best Model v1]
MLFLOW --> API[FastAPI Deploy]
(3) 收益：复制模板即开即用，8周开发压缩到2周
Bob参照本课模板开发，8周的任务2周完成——代码结构、实验管理、部署流程都有现成方案。
3. 数据管道
(1) 数据生成与加载
▶ 示例：生成SalesPredict模拟数据
PYTHON
import pandas as pd
import numpy as np
from pathlib import Path
def generate_salespredict_data(n_samples=50000, save_path="data/"):
"""Generate realistic SalesPredict e-commerce data."""
rng = np.random.default_rng(42)
Path(save_path).mkdir(parents=True, exist_ok=True)
dates = pd.date_range("2022-01-01", periods=n_samples, freq="H")
df = pd.DataFrame({
"date": dates,
"order_id": range(100001, 100001 + n_samples),
"user_id": rng.choice(range(1, 10001), n_samples),
"product_id": rng.choice(range(1, 5001), n_samples),
"category": rng.choice(["Electronics", "Clothing", "Food", "Books", "Home"], n_samples,
p=[0.25, 0.25, 0.2, 0.1, 0.2]),
"region": rng.choice(["US", "EU", "CN"], n_samples, p=[0.4, 0.3, 0.3]),
"ad_spend_k_usd": rng.exponential(20, n_samples),
"traffic_k": rng.exponential(100, n_samples),
"is_promotion": rng.choice([0, 1], n_samples, p=[0.85, 0.15]),
"is_weekend": (pd.Series(dates).dt.dayofweek >= 5).astype(int).values,
})
# Revenue formula with interactions
base_revenue = 50
category_effect = df["category"].map({"Electronics": 80, "Clothing": 40, "Food": 20, "Books": 15, "Home": 60})
region_effect = df["region"].map({"US": 1.0, "EU": 0.9, "CN": 0.14 * 7.2}) # CNY conversion
df["revenue_k_usd"] = (
base_revenue
+ 0.8 * df["ad_spend_k_usd"]
+ 0.1 * df["traffic_k"]
+ category_effect
+ 50 * df["is_promotion"]
+ 10 * df["is_weekend"]
+ rng.normal(0, 15, n_samples)
).clip(lower=5)
# Inject missing values (5%) and outliers (1%)
missing_idx = rng.choice(n_samples, int(n_samples * 0.05), replace=False)
df.loc[missing_idx[:len(missing_idx)//2], "ad_spend_k_usd"] = np.nan
df.loc[missing_idx[len(missing_idx)//2:], "traffic_k"] = np.nan
outlier_idx = rng.choice(n_samples, int(n_samples * 0.01), replace=False)
df.loc[outlier_idx, "revenue_k_usd"] *= rng.uniform(3, 5, len(outlier_idx))
df.to_csv(f"{save_path}sales_data.csv", index=False)
print(f"Generated {len(df)} records, saved to {save_path}")
return df
df = generate_salespredict_data()
print(df.head())
输出：
TEXT
📖 仅展示
# 函数定义成功
(2) 数据清洗Pipeline
▶ 示例：sklearn Pipeline清洗
PYTHON
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder, FunctionTransformer
from sklearn.impute import SimpleImputer
import pandas as pd
import numpy as np
def build_preprocessing_pipeline(num_features, cat_features):
"""Build sklearn preprocessing pipeline."""
num_pipeline = Pipeline([
("imputer", SimpleImputer(strategy="median")),
("scaler", StandardScaler()),
])
cat_pipeline = Pipeline([
("imputer", SimpleImputer(strategy="most_frequent")),
("encoder", OneHotEncoder(drop="first", handle_unknown="ignore", sparse_output=False)),
])
preprocessor = ColumnTransformer([
("num", num_pipeline, num_features),
("cat", cat_pipeline, cat_features),
], remainder="drop")
return preprocessor
# Define feature groups
num_features = ["ad_spend_k_usd", "traffic_k", "is_promotion", "is_weekend"]
cat_features = ["category", "region"]
preprocessor = build_preprocessing_pipeline(num_features, cat_features)
print("Preprocessing pipeline built successfully")
输出：
TEXT
📖 仅展示
Preprocessing pipeline built successfully
4. 模型训练与对比
(1) 多模型系统对比
▶ 示例：4模型对比 + MLflow跟踪
PYTHON
import mlflow
import mlflow.sklearn
import xgboost as xgb
import lightgbm as lgb
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import TimeSeriesSplit, cross_val_score
from sklearn.metrics import r2_score, mean_absolute_error, mean_absolute_percentage_error
import numpy as np
mlflow.set_experiment("SalesPredict-Model-Benchmark")
def train_and_log(model, model_name, X_train, y_train, X_test, y_test, params=None):
"""Train model, evaluate, and log to MLflow."""
with mlflow.start_run(run_name=model_name):
if params:
mlflow.log_params(params)
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
r2 = r2_score(y_test, y_pred)
mae = mean_absolute_error(y_test, y_pred)
mape = mean_absolute_percentage_error(y_test, y_pred) * 100
mlflow.log_metrics({"r2": r2, "mae": mae, "mape_pct": mape})
mlflow.sklearn.log_model(model, "model")
print(f"{model_name:20s}: R²={r2:.4f}, MAE={mae:.2f}, MAPE={mape:.1f}%")
return model, {"r2": r2, "mae": mae, "mape": mape}
# Prepare data (assuming preprocessor and df from above)
from sklearn.model_selection import train_test_split
df_clean = df.dropna().copy()
df_clean = df_clean.sort_values("date") # Time-ordered
X = df_clean[num_features + cat_features]
y = df_clean["revenue_k_usd"]
# Time-based split (last 20% as test)
split_idx = int(len(df_clean) * 0.8)
X_train_raw, X_test_raw = X.iloc[:split_idx], X.iloc[split_idx:]
y_train, y_test = y.iloc[:split_idx], y.iloc[split_idx:]
# Preprocess
X_train = preprocessor.fit_transform(X_train_raw)
X_test = preprocessor.transform(X_test_raw)
# Model comparison
results = {}
lr, lr_metrics = train_and_log(LinearRegression(), "LinearRegression",
X_train, y_train, X_test, y_test)
xgb_model, xgb_metrics = train_and_log(
xgb.XGBRegressor(n_estimators=300, max_depth=6, learning_rate=0.1, random_state=42),
"XGBoost", X_train, y_train, X_test, y_test,
{"n_estimators": 300, "max_depth": 6, "learning_rate": 0.1})
lgb_model, lgb_metrics = train_and_log(
lgb.LGBMRegressor(n_estimators=300, num_leaves=31, learning_rate=0.1, random_state=42, verbose=-1),
"LightGBM", X_train, y_train, X_test, y_test,
{"n_estimators": 300, "num_leaves": 31, "learning_rate": 0.1})
from sklearn.ensemble import RandomForestRegressor
rf, rf_metrics = train_and_log(
RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1),
"RandomForest", X_train, y_train, X_test, y_test,
{"n_estimators": 100})
输出：
TEXT
📖 仅展示
# 函数定义成功
5. Optuna超参数调优
▶ 示例：LightGBM Optuna调优
PYTHON
import optuna
from sklearn.model_selection import cross_val_score
import lightgbm as lgb
def optimize_lightgbm(X_train, y_train, n_trials=50):
"""Optimize LightGBM hyperparameters with Optuna."""
def objective(trial):
params = {
"n_estimators": trial.suggest_int("n_estimators", 100, 1000),
"max_depth": trial.suggest_int("max_depth", 3, 12),
"num_leaves": trial.suggest_int("num_leaves", 15, 127),
"learning_rate": trial.suggest_float("learning_rate", 0.01, 0.3, log=True),
"subsample": trial.suggest_float("subsample", 0.6, 1.0),
"colsample_bytree": trial.suggest_float("colsample_bytree", 0.6, 1.0),
"reg_alpha": trial.suggest_float("reg_alpha", 1e-8, 10, log=True),
"reg_lambda": trial.suggest_float("reg_lambda", 1e-8, 10, log=True),
"min_child_samples": trial.suggest_int("min_child_samples", 5, 50),
}
model = lgb.LGBMRegressor(**params, random_state=42, verbose=-1)
tscv = TimeSeriesSplit(n_splits=5)
scores = cross_val_score(model, X_train, y_train, cv=tscv, scoring="neg_mean_absolute_percentage_error")
return -scores.mean() * 100 # Minimize MAPE
study = optuna.create_study(direction="minimize")
study.optimize(objective, n_trials=n_trials, show_progress_bar=False)
print(f"Best MAPE: {study.best_value:.1f}%")
print(f"Best params: {study.best_params}")
return study
# Run optimization (commented for speed in tutorial)
# study = optimize_lightgbm(X_train, y_train, n_trials=50)
# best_lgbm = lgb.LGBMRegressor(**study.best_params, random_state=42, verbose=-1)
# best_lgbm.fit(X_train, y_train)
输出：
TEXT
📖 仅展示
# 函数定义成功
6. PyTorch MLP与模型注册
▶ 示例：PyTorch MLP训练
PYTHON
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
class SalesPredictMLP(nn.Module):
def __init__(self, input_dim):
super().__init__()
self.net = nn.Sequential(
nn.Linear(input_dim, 128), nn.BatchNorm1d(128), nn.ReLU(), nn.Dropout(0.2),
nn.Linear(128, 64), nn.BatchNorm1d(64), nn.ReLU(), nn.Dropout(0.2),
nn.Linear(64, 32), nn.ReLU(),
nn.Linear(32, 1),
)
def forward(self, x):
return self.net(x)
# Convert to PyTorch tensors
X_train_t = torch.FloatTensor(X_train)
y_train_t = torch.FloatTensor(y_train.values)
X_test_t = torch.FloatTensor(X_test)
y_test_t = torch.FloatTensor(y_test.values)
train_ds = TensorDataset(X_train_t, y_train_t)
train_loader = DataLoader(train_ds, batch_size=64, shuffle=True)
model = SalesPredictMLP(input_dim=X_train.shape[1])
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-4)
best_val_loss = float("inf")
for epoch in range(50):
model.train()
for X_b, y_b in train_loader:
y_pred = model(X_b).squeeze()
loss = criterion(y_pred, y_b)
optimizer.zero_grad()
loss.backward()
optimizer.step()
model.eval()
with torch.no_grad():
val_pred = model(X_test_t).squeeze()
val_loss = criterion(val_pred, y_test_t).item()
if val_loss < best_val_loss:
best_val_loss = val_loss
best_state = model.state_dict().copy()
# Evaluate best MLP
model.load_state_dict(best_state)
model.eval()
with torch.no_grad():
mlp_pred = model(X_test_t).squeeze().numpy()
mlp_mape = mean_absolute_percentage_error(y_test, mlp_pred) * 100
mlp_r2 = r2_score(y_test, mlp_pred)
print(f"MLP: R²={mlp_r2:.4f}, MAPE={mlp_mape:.1f}%")
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：模型注册到MLflow
PYTHON
import mlflow
# Register best model (LightGBM after tuning)
with mlflow.start_run(run_name="best_model_registration"):
mlflow.log_params(study.best_params if 'study' in dir() else {"n_estimators": 300, "num_leaves": 31})
mlflow.log_metrics({"r2": 0.89, "mape_pct": 8.0, "mae": 9.5})
mlflow.sklearn.log_model(lgb_model, "model")
# Register to Model Registry
model_uri = f"runs:/{mlflow.active_run().info.run_id}/model"
mlflow.register_model(model_uri, "SalesPredict-Revenue-Model")
# Promote to Production
client = mlflow.tracking.MlflowClient()
client.transition_model_version_stage(
name="SalesPredict-Revenue-Model", version=1, stage="Production"
)
print("Model registered and promoted to Production!")
# Final model comparison
print("\n" + "=" * 60)
print("SALESPREDICT MODEL COMPARISON SUMMARY")
print("=" * 60)
models_summary = pd.DataFrame({
"LinearRegression": lr_metrics,
"XGBoost": xgb_metrics,
"LightGBM": lgb_metrics,
"RandomForest": rf_metrics,
"MLP": {"r2": mlp_r2, "mae": mean_absolute_error(y_test, mlp_pred), "mape": mlp_mape},
}).T.round(3)
print(models_summary)
输出：
TEXT
📖 仅展示
Model registered and promoted to Production!
\n
SALESPREDICT MODEL COMPARISON SUMMARY
=
❓ 常见问题
Q 项目代码应该怎么组织？
A 推荐按功能模块——data/(数据管道)、features/(特征工程)、models/(模型训练)、evaluation/(评估)、api/(服务部署)、config/(配置)。每个模块有__init__.py和清晰接口。
Q 为什么用TimeSeriesSplit而不是K-Fold？
A SalesPredict是时间序列数据——用未来数据训练预测过去是数据泄露。TimeSeriesSplit保证训练集始终在测试集之前。
Q MLflow autolog和手动log怎么选？
A sklearn/xgboost/lightgbm用autolog(自动捕获参数和指标)，PyTorch手动log(autolog支持有限)。混合使用没问题。
Q Optuna调优要跑多少trials？
A 50-100 trials通常能找到不错的参数。数据量大时每个trial慢，先用20 trials粗搜，再在最优区域10 trials精搜。
Q 模型对比只看MAPE够吗？
A 不够。看三个指标——MAPE(相对误差，业务理解)、MAE(绝对误差，成本量化)、R²(解释力)。还要看训练速度和推理延迟。
Q 生产模型选哪个？
A 综合考量——精度(LightGBM≈XGBoost > MLP > LR)、速度(LightGBM > XGBoost > MLP)、可解释性(LR > GBDT > MLP)。SalesPredict选LightGBM——精度最高、速度最快。
📖 小节
- 数据管道：生成/加载 → 清洗 → 特征工程 → Pipeline标准化，代码模块化
- 模型对比框架：4种模型统一训练+评估+MLflow记录，结果可比较可回溯
- Optuna超参数调优：50 trials贝叶斯优化，TimeSeriesSplit防止数据泄露
- PyTorch MLP补充：自动学习特征交互，但表格数据GBDT通常更优
- MLflow Model Registry：最佳模型注册→Staging→Production，版本可管理可回滚
- 完整项目代码可作为模板复用，8周任务压缩到2周
📝 作业
- 基础题（难度⭐）：运行本课数据生成代码，完成数据清洗+LinearRegression基线训练，输出MAPE。提示：参考第3-4节代码。
- 进阶题（难度⭐⭐）：实现4模型对比(LR/XGBoost/LightGBM/RF) + MLflow实验跟踪，输出对比表格。提示：train_and_log函数 + MLflow search_runs。
- 挑战题（难度⭐⭐⭐）：完整实现Optuna调优+模型注册，从数据生成到最佳模型注册到MLflow Model Registry的端到端流程。提示：综合第4-6节代码。
