Machine Learning: Implantação do Projeto

# Stack de implantação de nível de produção
services:
  api: FastAPI + Uvicorn (serving de modelo)
  redis: Camada de cache (previsões quentes)
  nginx: Balanceador de carga + rate limiting
  prometheus: Coleta de métricas
  grafana: Dashboard de monitoramento
  airflow: Scheduler de re-treinamento

# Arquivo: 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")

# Modelo global (carregado na inicialização)
model = None

@app.on_event("startup")
async def load_model():
    global model
    model = joblib.load("model/salespredict_lgbm.joblib")
    logger.info("Modelo carregado com sucesso")

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

        # Intervalo de confiança simples (±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"Erro de previsão: {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"Tamanho do lote excede {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

# Estágio 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

# Estágio 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;
    }
}

# Adicionar ao app/main.py
from prometheus_client import Counter, Histogram, generate_latest
from fastapi import Response

PREDICTIONS_COUNT = Counter("predictions_total", "Total de previsões feitas")
PREDICTION_LATENCY = Histogram("prediction_latency_seconds", "Latência de previsão",
                                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")

# Instrumentar o endpoint de previsão
@app.post("/predict", response_model=PredictionOutput)
def predict(input_data: PredictionInput):
    start = time.time()
    PREDICTIONS_COUNT.inc()
    # ... (lógica de previsão) ...
    PREDICTION_LATENCY.observe(time.time() - start)
    # ... (retornar resultado) ...

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

# Arquivo: 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):
        """Executa verificação semanal de drift em todas as 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"),
                }

        # Registrar resultados
        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"ALERTA DE DRIFT em {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)
        # Em produção: enviar para Slack/PagerDuty

# Job de monitoramento semanal
monitor = DriftMonitor("model/reference_stats.npy")
# features = load_current_week_features()
# results, drift = monitor.check_weekly_drift(features)

DriftMonitor initialized with reference stats
Verificação semanal agendada: PSI threshold=0.2

# Arquivo: dags/retrain_pipeline.py (conceito de DAG do Airflow)
# from airflow import DAG
# from airflow.operators.python import PythonOperator

def retrain_pipeline():
    """Pipeline de re-treinamento automatizado disparado por detecção de drift."""
    # Passo 1: Verificar drift
    # drift_detected = check_weekly_drift()

    # Passo 2: Buscar dados recentes
    # data = fetch_recent_data(months=3)

    # Passo 3: Re-treinar modelo
    # new_model, metrics = train_lightgbm(data)

    # Passo 4: Comparar com produção
    # if metrics["mape"] < production_mape:
    #     register_model(new_model, stage="Staging")

    # Passo 5: Teste A/B (2 semanas)
    # run_ab_test(new_model, duration_weeks=2)

    # Passo 6: Promover se o teste A/B mostrar melhora
    # if ab_test_significant:
    #     promote_to_production(new_model)

    # Passo 7: Atualizar estatísticas de referência
    # update_reference_stats(new_model)
    pass

# Definição do DAG
# 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' registrado
Schedule: toda segunda-feira 06:00 UTC
Tarefas: check_drift >> retrain >> ab_test >> promote

project_retrospective = {
    "what_went_well": [
        "Rastreamento de experimentos com MLflow evitou confusão entre 50+ runs",
        "TimeSeriesSplit capturou vazamento de futuro antes da produção",
        "Optuna encontrou melhores parâmetros que a busca manual em 1/10 do tempo",
        "Implantação Docker habilitou atualizações de modelo sem tempo de inatividade",
    ],
    "what_could_improve": [
        "O pipeline de dados deveria ter sido construído antes do treinamento do modelo",
        "Um feature store reduziria a duplicação entre treinamento e serving",
        "O teste A/B deveria ter rodado por mais tempo (3 semanas em vez de 2)",
        "O dashboard de monitoramento deveria ter sido configurado desde o dia 1",
    ],
    "key_learnings": [
        "Qualidade dos dados > complexidade do modelo (dados limpos + modelo simples supera dados sujos + modelo complexo)",
        "Design primeiro, código depois (1 semana de design economizou 4 semanas de retrabalho)",
        "Implante cedo, itere rápido (feedback de produção > experimentos de laboratório)",
        "Monitore tudo (detecção de drift capturou o problema do Double-11 em 3 dias)",
    ],
    "next_steps": [
        "Adicionar modelo LSTM para captura de padrões temporais",
        "Implementar serving de features em tempo real com Feast",
        "Construir pipeline de re-treinamento automatizado com Airflow",
        "Expandir para 3 novos mercados (Japão, Índia, Brasil)",
    ],
}

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

O que Deu Certo:
  - Rastreamento de experimentos com MLflow evitou confusão entre 50+ runs
  - TimeSeriesSplit capturou vazamento de futuro antes da produção
O Que Pode Melhorar:
  - O pipeline de dados deveria ter sido construído antes do treinamento do modelo
Aprendizados-Chave:
  - Qualidade dos dados > complexidade do modelo
Próximos Passos:
  - Adicionar modelo LSTM para captura de padrões temporais