Project Development
Code is the bridge from design to product—this lesson turns everything from the previous 23 lessons into runnable code.
1. What You'll Learn
- Data pipeline: raw data cleaning → feature engineering → train/test split, using Pandas + sklearn Pipeline
- Model training and comparison: LinearRegression → XGBoost → LightGBM → PyTorch MLP
- Hyperparameter tuning: Optuna Bayesian optimization with 5-fold time-series cross-validation, reducing MAPE from 15% to 8%
- Model registration: managing the best model version with MLflow Model Registry
- Key code walkthrough: module-by-module explanation of design decisions
2. A Real ML Engineer's Story
(1) The Pain Point: Design Is Done, but How Do You Organize the Code?
Bob finished the system design for SalesPredict, but stared at an empty repository with no idea how to structure it—where does the data pipeline go? How do you modularize feature engineering? How do you integrate experiment tracking? The gap between design and code demands a reference template.
(2) The End-to-End Code Solution
This lesson provides complete, runnable code for SalesPredict—from data generation to model deployment, with each module having a clear responsibility that you can reuse directly.
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) The Payoff: Copy the Template and Ship Fast—8 Weeks Compressed to 2
Bob followed this lesson's template and finished an 8-week task in just 2 weeks—code structure, experiment management, and deployment workflow all had ready-made solutions.
3. Data Pipeline
(1) Data Generation and Loading
▶ Example: Generating SalesPredict Simulated Data
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())
Output:
TEXT
# Function defined successfully
(2) Data Cleaning Pipeline
▶ Example: sklearn Pipeline for Cleaning
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")
Output:
TEXT
Preprocessing pipeline built successfully
4. Model Training and Comparison
(1) Systematic Multi-Model Comparison
▶ Example: 4-Model Comparison + MLflow Tracking
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})
Output:
TEXT
# Function defined successfully
5. Optuna Hyperparameter Tuning
▶ Example: LightGBM Tuning with 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)
Output:
TEXT
# Function defined successfully
6. PyTorch MLP and Model Registration
▶ Example: PyTorch MLP Training
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}%")
Output:
TEXT
# Function defined successfully
▶ Example: Registering the Model with 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)
Output:
TEXT
Model registered and promoted to Production!
\n
SALESPREDICT MODEL COMPARISON SUMMARY
=
❓ FAQ
Q How should I organize project code?
A Structure it by functional module—data/ (data pipeline), features/ (feature engineering), models/ (model training), evaluation/ (evaluation), api/ (service deployment), config/ (configuration). Each module should have an init
Q Why use TimeSeriesSplit instead of K-Fold?
A SalesPredict is time-series data—training on future data to predict the past is data leakage. TimeSeriesSplit guarantees the training set always precedes the test set.
Q Should I use MLflow autolog or manual logging?
A Use autolog for sklearn/xgboost/lightgbm (it automatically captures parameters and metrics), and manual logging for PyTorch (autolog support is limited). Mixing both approaches is perfectly fine.
Q How many trials should I run with Optuna?
A 50–100 trials usually finds good parameters. When data is large and each trial is slow, start with 20 trials for a coarse search, then run 10 more trials in the best region for fine-tuning.
Q Is MAPE alone enough for model comparison?
A No. Look at three metrics—MAPE (relative error, business intuition), MAE (absolute error, cost quantification), and R² (explanatory power). Also consider training speed and inference latency.
Q Which model should I choose for production?
A Weigh the trade-offs—accuracy (LightGBM ≈ XGBoost > MLP > LR), speed (LightGBM > XGBoost > MLP), interpretability (LR > GBDT > MLP). For SalesPredict, LightGBM wins—highest accuracy and fastest speed.
📖 Summary
- Data pipeline: generation/loading → cleaning → feature engineering → standardized Pipeline, with modularized code
- Model comparison framework: unified training + evaluation + MLflow logging across 4 models, with comparable and traceable results
- Optuna hyperparameter tuning: 50-trial Bayesian optimization with TimeSeriesSplit to prevent data leakage
- PyTorch MLP as a complement: automatically learns feature interactions, though GBDTs typically outperform on tabular data
- MLflow Model Registry: register the best model → Staging → Production, with version management and rollback capability
- The complete project code serves as a reusable template, compressing an 8-week effort into 2 weeks
📝 Exercises
- Basic (Difficulty ⭐): Run the data generation code from this lesson, complete data cleaning + LinearRegression baseline training, and output the MAPE. Hint: refer to the code in Sections 3–4.
- Intermediate (Difficulty ⭐⭐): Implement the 4-model comparison (LR/XGBoost/LightGBM/RF) + MLflow experiment tracking, and output a comparison table. Hint: use the train_and_log function + MLflow search_runs.
- Challenge (Difficulty ⭐⭐⭐): Fully implement Optuna tuning + model registration—the end-to-end flow from data generation to registering the best model in MLflow Model Registry. Hint: combine the code from Sections 4–6.
← Previous: Project Design | Next: Project Deployment →
