Machine Learning: プロジェクトデプロイ — SalesPredictの本番運用と安定稼働

# Production-grade deployment stack
services:
  api: FastAPI + Uvicorn (model serving)
  redis: Cache layer (hot predictions)
  nginx: Load balancer + rate limiting
  prometheus: Metrics collection
  grafana: Monitoring dashboard
  airflow: Retraining scheduler

# File: app/main.py
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import Optional
import joblib
import numpy as np
import time
import logging

logger = logging.getLogger("salespredict")
app = FastAPI(title="SalesPredict API", version="1.0.0")

# Global model (loaded at startup)
model = None

@app.on_event("startup")
async def load_model():
    global model
    model = joblib.load("model/salespredict_lgbm.joblib")
    logger.info("Model loaded successfully")

class PredictionInput(BaseModel):
    ad_spend_k_usd: float = Field(..., ge=0, le=1000)
    traffic_k: float = Field(..., ge=0, le=10000)
    is_promotion: int = Field(0, ge=0, le=1)
    is_weekend: int = Field(0, ge=0, le=1)
    category_clothing: int = Field(0, ge=0, le=1)
    category_electronics: int = Field(0, ge=0, le=1)
    category_food: int = Field(0, ge=0, le=1)
    category_home: int = Field(0, ge=0, le=1)
    region_EU: int = Field(0, ge=0, le=1)
    region_US: int = Field(0, ge=0, le=1)

    model_config = {"json_schema_extra": {
        "example": {"ad_spend_k_usd": 50, "traffic_k": 300,
                     "is_promotion": 1, "is_weekend": 0,
                     "category_electronics": 1, "category_clothing": 0,
                     "category_food": 0, "category_home": 0,
                     "region_US": 1, "region_EU": 0}
    }}

class PredictionOutput(BaseModel):
    predicted_revenue_k_usd: float
    confidence_low: Optional[float] = None
    confidence_high: Optional[float] = None
    latency_ms: float

@app.get("/health")
def health():
    return {"status": "healthy", "model_loaded": model is not None}

@app.post("/predict", response_model=PredictionOutput)
def predict(input_data: PredictionInput):
    start = time.time()
    try:
        features = np.array([[input_data.ad_spend_k_usd, input_data.traffic_k,
                               input_data.is_promotion, input_data.is_weekend,
                               input_data.category_clothing, input_data.category_electronics,
                               input_data.category_food, input_data.category_home,
                               input_data.region_EU, input_data.region_US]])
        prediction = float(model.predict(features)[0])
        latency = (time.time() - start) * 1000

        # Simple confidence interval (±20%)
        return PredictionOutput(
            predicted_revenue_k_usd=round(prediction, 2),
            confidence_low=round(prediction * 0.8, 2),
            confidence_high=round(prediction * 1.2, 2),
            latency_ms=round(latency, 2),
        )
    except Exception as e:
        logger.error(f"Prediction error: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/predict_batch")
def predict_batch(inputs: list[PredictionInput], max_batch: int = 100):
    if len(inputs) > max_batch:
        raise HTTPException(status_code=400, detail=f"Batch size exceeds {max_batch}")
    start = time.time()
    features = np.array([[d.ad_spend_k_usd, d.traffic_k, d.is_promotion, d.is_weekend,
                           d.category_clothing, d.category_electronics, d.category_food,
                           d.category_home, d.region_EU, d.region_US] for d in inputs])
    predictions = model.predict(features)
    latency = (time.time() - start) * 1000
    return {
        "predictions": [round(float(p), 2) for p in predictions],
        "count": len(predictions),
        "latency_ms": round(latency, 2),
    }

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    latency = (time.time() - start) * 1000
    logger.info(f"{request.method} {request.url.path} - {response.status_code} - {latency:.1f}ms")
    return response

INFO:     Application startup complete.
INFO:     Model loaded successfully
INFO:     Uvicorn running on http://0.0.0.0:8000

# Stage 1: Build
FROM python:3.11-slim AS builder
WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# Stage 2: Runtime
FROM python:3.11-slim
WORKDIR /app

COPY --from=builder /install /usr/local
COPY app/ ./app/
COPY model/ ./model/

RUN useradd -m -r appuser && chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
  CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Successfully built 3a7f2b1c9d4e
Successfully tagged salespredict-api:latest

# docker-compose.yml
version: "3.8"

services:
  api:
    build: .
    environment:
      - REDIS_URL=redis://redis:6379/0
      - MLFLOW_TRACKING_URI=http://mlflow:5000
    depends_on:
      - redis
    deploy:
      replicas: 2
      resources:
        limits:
          memory: 2G
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 5s

  redis:
    image: redis:7-alpine
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data
    restart: unless-stopped

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf:ro
    depends_on:
      - api
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - prometheus_data:/prometheus
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana_data:/var/lib/grafana
    depends_on:
      - prometheus
    restart: unless-stopped

volumes:
  redis_data:
  prometheus_data:
  grafana_data:

# nginx.conf
upstream api_backend {
    least_conn;
    server api:8000;
}

server {
    listen 80;
    location / {
        proxy_pass http://api_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
    location /health {
        proxy_pass http://api_backend/health;
    }
}

# Add to app/main.py
from prometheus_client import Counter, Histogram, generate_latest
from fastapi import Response

PREDICTIONS_COUNT = Counter("predictions_total", "Total predictions made")
PREDICTION_LATENCY = Histogram("prediction_latency_seconds", "Prediction latency",
                                buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0])

@app.get("/metrics")
def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

# Instrument the predict endpoint
@app.post("/predict", response_model=PredictionOutput)
def predict(input_data: PredictionInput):
    start = time.time()
    PREDICTIONS_COUNT.inc()
    # ... (prediction logic) ...
    PREDICTION_LATENCY.observe(time.time() - start)
    # ... (return result) ...

INFO:     Metrics endpoint registered at /metrics
INFO:     Prediction counter initialized

# File: monitoring/drift_monitor.py
import numpy as np
import requests
import json
from datetime import datetime

class DriftMonitor:
    def __init__(self, reference_stats_path, api_url="http://localhost:8000"):
        self.reference = np.load(reference_stats_path, allow_pickle=True).item()
        self.api_url = api_url

    def calculate_psi(self, reference, current, n_bins=10):
        breakpoints = np.percentile(reference, np.linspace(0, 100, n_bins + 1))
        breakpoints[0], breakpoints[-1] = -np.inf, np.inf
        ref_counts = np.histogram(reference, bins=breakpoints)[0]
        cur_counts = np.histogram(current, bins=breakpoints)[0]
        ref_pct = np.clip(ref_counts / len(reference), 1e-6, None)
        cur_pct = np.clip(cur_counts / len(current), 1e-6, None)
        return np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))

    def check_weekly_drift(self, current_features):
        """Run weekly drift check on all features."""
        results = {}
        for feature_name, current_data in current_features.items():
            if feature_name in self.reference:
                psi = self.calculate_psi(self.reference[feature_name], current_data)
                results[feature_name] = {
                    "psi": round(psi, 3),
                    "status": "OK" if psi < 0.1 else ("WARNING" if psi < 0.2 else "DRIFT"),
                }

        # Log results
        drift_detected = any(r["status"] == "DRIFT" for r in results.values())
        if drift_detected:
            self._send_alert(results)

        return results, drift_detected

    def _send_alert(self, results):
        alert_msg = f"DRIFT ALERT at {datetime.now()}\n"
        for feat, info in results.items():
            if info["status"] != "OK":
                alert_msg += f"  {feat}: PSI={info['psi']} ({info['status']})\n"
        print(alert_msg)
        # In production: send to Slack/PagerDuty

# Weekly monitoring job
monitor = DriftMonitor("model/reference_stats.npy")
# features = load_current_week_features()
# results, drift = monitor.check_weekly_drift(features)

DriftMonitor initialized with reference stats
Weekly check scheduled: PSI threshold=0.2

# File: dags/retrain_pipeline.py (Airflow DAG concept)
# from airflow import DAG
# from airflow.operators.python import PythonOperator

def retrain_pipeline():
    """Automated retraining pipeline triggered by drift detection."""
    # Step 1: Check drift
    # drift_detected = check_weekly_drift()

    # Step 2: Fetch recent data
    # data = fetch_recent_data(months=3)

    # Step 3: Retrain model
    # new_model, metrics = train_lightgbm(data)

    # Step 4: Compare with production
    # if metrics["mape"] < production_mape:
    #     register_model(new_model, stage="Staging")

    # Step 5: A/B test (2 weeks)
    # run_ab_test(new_model, duration_weeks=2)

    # Step 6: Promote if A/B shows improvement
    # if ab_test_significant:
    #     promote_to_production(new_model)

    # Step 7: Update reference stats
    # update_reference_stats(new_model)
    pass

# DAG definition
# with DAG("salespredict_retrain", schedule_interval="0 6 * * 1") as dag:
#     check_drift = PythonOperator(task_id="check_drift", python_callable=check_drift)
#     retrain = PythonOperator(task_id="retrain", python_callable=retrain_model)
#     ab_test = PythonOperator(task_id="ab_test", python_callable=run_ab_test)
#     promote = PythonOperator(task_id="promote", python_callable=promote_model)
#     check_drift >> retrain >> ab_test >> promote

DAG 'salespredict_retrain' registered
Schedule: every Monday 06:00 UTC
Tasks: check_drift >> retrain >> ab_test >> promote

project_retrospective = {
    "what_went_well": [
        "MLflow experiment tracking prevented confusion over 50+ runs",
        "TimeSeriesSplit caught future leakage before production",
        "Optuna found better params than manual search in 1/10 time",
        "Docker deployment enabled zero-downtime model updates",
    ],
    "what_could_improve": [
        "Data pipeline should have been built before model training",
        "Feature store would reduce duplication between training and serving",
        "A/B test should have run longer (3 weeks instead of 2)",
        "Monitoring dashboard should have been set up from day 1",
    ],
    "key_learnings": [
        "Data quality > model complexity (clean data + simple model beats dirty data + complex model)",
        "Design first, code second (1 week design saved 4 weeks of rework)",
        "Deploy early, iterate fast (production feedback > lab experiments)",
        "Monitor everything (drift detection caught Double-11 issue in 3 days)",
    ],
    "next_steps": [
        "Add LSTM model for temporal pattern capture",
        "Implement real-time feature serving with Feast",
        "Build automated retraining pipeline with Airflow",
        "Expand to 3 more markets (Japan, India, Brazil)",
    ],
}

for category, items in project_retrospective.items():
    print(f"\n{category.replace('_', ' ').title()}:")
    for item in items:
        print(f"  - {item}")

What Went Well:
  - MLflow experiment tracking prevented confusion over 50+ runs
  - TimeSeriesSplit caught future leakage before production
What Could Improve:
  - Data pipeline should have been built before model training
Key Learnings:
  - Data quality > model complexity
Next Steps:
  - Add LSTM model for temporal pattern capture