Machine Learning: プロジェクト開発 — SalesPredict エンドツーエンド実装ガイド
最終更新：2026-08-26
コードは設計からプロダクトへの架け橋です。このレッスンでは、これまでの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
📖 参照専用
# Function defined successfully
(2) データクリーニングパイプライン
▶ サンプル：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
📖 参照専用
# Function defined successfully
5. Optuna ハイパーパラメータチューニング
▶ サンプル：Optuna による LightGBM チューニング
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
📖 参照専用
# Function defined successfully
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
📖 参照専用
# Function defined successfully
▶ サンプル：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
Q K-Fold ではなく TimeSeriesSplit を使うのはなぜですか？
A SalesPredict は時系列データです。未来のデータで学習して過去を予測するのはデータリークになります。TimeSeriesSplit は学習セットが常にテストセットより前の期間になることを保証します。
Q MLflow の autolog と手動ロギングはどちらを使うべきですか？
A sklearn/xgboost/lightgbm には autolog を使い（パラメータとメトリクスを自動取得）、PyTorch には手動ロギングを使います（autolog のサポートが限定的なため）。両方を組み合わせて使っても問題ありません。
Q Optuna のトライアル数はどれくらいが適切ですか？
A 通常 50〜100 トライアルで良いパラメータが見つかります。データが大きく各トライアルが遅い場合は、まず 20 トライアルで大まかに探索し、その後ベストな領域で 10 トライアル追加して微調整するのが効果的です。
Q モデル比較に MAPE だけでは不十分ですか？
A 不十分です。MAPE（相対誤差、ビジネス直感）、MAE（絶対誤差、コスト定量化）、R²（説明力）の3つの指標を見てください。さらに学習速度と推論レイテンシも考慮しましょう。
Q 本番環境にはどのモデルを選ぶべきですか？
A トレードオフを考慮します。精度（LightGBM ≈ XGBoost > MLP > LR）、速度（LightGBM > XGBoost > MLP）、解釈性（LR > GBDT > MLP）。SalesPredict では LightGBM が最適です。精度が最も高く、速度も最速です。
📖 まとめ
- データパイプライン：生成/読み込み → クリーニング → 特徴量エンジニアリング → 標準化された Pipeline により、コードをモジュール化
- モデル比較フレームワーク：4モデルの学習 + 評価 + MLflow ロギングを統一し、結果の比較と追跡が可能
- Optuna ハイパーパラメータチューニング：50トライアルのベイズ最適化 + 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 のコードを組み合わせてください。
← 前へ：プロジェクト設計 | 次へ：プロジェクトデプロイ →
