Tutorial 20: Proyek MLOps End-to-End

Daftar Isi

Pendahuluan

MLOps adalah disiplin untuk men-deploy dan memelihara model machine learning di produksi secara andal dan efisien. Meskipun membangun model yang akurat itu penting, tantangan sesungguhnya terletak pada segala hal di sekitarnya: melakukan versioning data dan kode secara bersamaan, melacak eksperimen secara reproducible, mengotomatisasi pipeline pelatihan dan deployment, memantau performa model di produksi, dan merespons data drift.

Tutorial ini memandu Anda melalui proyek MLOps yang lengkap dari data mentah hingga monitoring produksi. Kita akan membangun sistem prediksi churn pelanggan menggunakan alat standar industri: DVC untuk versioning data, MLflow untuk pelacakan eksperimen dan registry model, GitHub Actions untuk CI/CD, Docker untuk kontainerisasi, dan Evidently untuk monitoring produksi.

Prasyarat

• Python 3.9+
• Git dan akun GitHub
• Docker dan Docker Compose
• AWS CLI atau CLI cloud sejenis (untuk deployment)
• Pemahaman dasar pelatihan model ML

# Instal semua paket yang diperlukan
pip install dvc[s3] mlflow scikit-learn pandas evidently
pip install fastapi uvicorn docker boto3

import os
print("Tutorial MLOps E2E - Pengaturan Lingkungan")

Gambaran Proyek

Struktur proyek kita mengikuti praktik terbaik MLOps dengan pemisahan tanggung jawab yang jelas.

prediksi-churn/
├── .github/
│   └── workflows/
│       ├── train.yml
│       ├── test.yml
│       └── deploy.yml
├── data/
│   ├── raw/
│   │   └── pelanggan.csv.dvc
│   └── processed/
│       └── fitur.csv.dvc
├── src/
│   ├── data/
│   │   ├── init.py
│   │   ├── persiapan.py
│   │   └── validasi.py
│   ├── fitur/
│   │   ├── init.py
│   │   └── bangunfitur.py

│   ├── model/
│   │   ├── init.py
│   │   ├── latih.py
│   │   └── prediksi.py
│   └── monitoring/
│       ├── init.py
│       └── deteksidrift.py
├── serving/
│   ├── app.py
│   ├── Dockerfile
│   └── requirements.txt
├── tests/
│   ├── testdata.py

│   ├── testmodel.py
│   └── testapi.py

├── configs/
│   └── config.yaml
├── dvc.yaml
├── dvc.lock
├── params.yaml
├── docker-compose.yml
└── requirements.txt

# Inisialisasi DVC di repositori Git Anda
cd prediksi-churn
dvc init

Konfigurasi penyimpanan remote (S3 dalam contoh ini)
dvc remote add -d myremote s3://my-ml-bucket/dvc-store
dvc remote modify myremote region us-east-1

Lacak file data
dvc add data/raw/pelanggan.csv
git add data/raw/pelanggan.csv.dvc data/raw/.gitignore
git commit -m "Lacak data pelanggan mentah dengan DVC"

Push data ke penyimpanan remote
dvc push

# dvc.yaml - Mendefinisikan pipeline ML yang reproducible
stages:
  persiapan:
    cmd: python src/data/persiapan.py
    deps:
src/data/persiapan.py
data/raw/pelanggan.csv
    params:

persiapan.testsize
persiapan.randomseed

    outs:

data/processed/train.csv
data/processed/test.csv

  fiturisasi:
    cmd: python src/fitur/bangunfitur.py
    deps:
src/fitur/bangunfitur.py

data/processed/train.csv
data/processed/test.csv
    params:

fitur
    outs:

data/processed/trainfitur.csv
data/processed/testfitur.csv


  pelatihan:
    cmd: python src/model/latih.py
    deps:
src/model/latih.py
data/processed/trainfitur.csv
    params:

pelatihan
    outs:

models/model.pkl
    metrics:

metrics/metrikpelatihan.json:

          cache: false

    plots:
metrics/kurvaroc.json:
          cache: false

          x: fpr
          y: tpr

  evaluasi:
    cmd: python src/model/evaluasi.py
    deps:
src/model/evaluasi.py
models/model.pkl
data/processed/testfitur.csv

    metrics:

metrics/metrikevaluasi.json:
          cache: false

# params.yaml - Konfigurasi parameter terpusat
persiapan:
  testsize: 0.2

  randomseed: 42

fitur:
  numerik:
masaberlangganan

biayabulanan
totalbiaya

  kategorikal:

tipekontrak
metodepembayaran

layananinternet
  rekayasa:

biayaperbulan
kelompokmasalangganan

pelatihan:
  model: xgboost
  nestimators: 500

  maxdepth: 6
  learningrate: 0.01

  subsample: 0.8
  colsamplebytree: 0.8
  earlystoppingrounds: 50

Skrip Persiapan Data

# src/data/persiapan.py
import pandas as pd
from sklearn.modelselection import traintestsplit

import yaml
import os

def persiapkandata():
    """Pisahkan data mentah menjadi set pelatihan dan pengujian."""
    with open("params.yaml", "r") as f:
        params = yaml.safeload(f)["persiapan"]


    # Muat data mentah
    df = pd.readcsv("data/raw/pelanggan.csv")

    # Pembersihan dasar
    df = df.dropna(subset=["churn"])
    df["totalbiaya"] = pd.tonumeric(
        df["totalbiaya"], errors="coerce"

    )
    df = df.fillna(df.median(numericonly=True))

    # Pemisahan
    traindf, testdf = traintestsplit(
        df,
        testsize=params["testsize"],
        randomstate=params["randomseed"],
        stratify=df["churn"]
    )

    os.makedirs("data/processed", existok=True)

    traindf.tocsv("data/processed/train.csv", index=False)

    testdf.tocsv("data/processed/test.csv", index=False)


    print(f"Train: {len(traindf)} baris, Test: {len(testdf)} baris")

    print(f"Tingkat churn - Train: {traindf['churn'].mean():.3f}, "
          f"Test: {testdf['churn'].mean():.3f}")


if name == "main":
    persiapkandata()

Pelacakan Eksperimen dengan MLflow

MLflow menyediakan pelacakan eksperimen, pencatatan model, dan registry model — semuanya penting untuk ML yang reproducible.

Menyiapkan MLflow

# Mulai server pelacakan MLflow
mlflow server --backend-store-uri sqlite:///mlflow.db \
--default-artifact-root ./mlruns \
--host 0.0.0.0 --port 5000

import mlflow
import mlflow.sklearn
import mlflow.xgboost

Konfigurasi MLflow
mlflow.settrackinguri("http://localhost:5000")
mlflow.setexperiment("prediksi-churn")

# src/model/latih.py
import pandas as pd
import numpy as np
import xgboost as xgb
from sklearn.metrics import (
    accuracyscore, precisionscore, recallscore,
    f1score, rocaucscore, roccurve
)
import mlflow
import mlflow.xgboost
import yaml
import json
import joblib
import os

def latihmodel():

    """Latih model dengan pelacakan MLflow penuh."""
    with open("params.yaml", "r") as f:
        params = yaml.safeload(f)

    parampelatihan = params["pelatihan"]


    # Muat fitur
    traindf = pd.readcsv("data/processed/trainfitur.csv")
    Xtrain = traindf.drop(columns=["churn"])
    ytrain = traindf["churn"]

    # Konfigurasi MLflow
    mlflow.settrackinguri("http://localhost:5000")
    mlflow.setexperiment("prediksi-churn")


    with mlflow.startrun(runname="pelatihanxgboost") as run:
        # Catat parameter
        mlflow.logparams(parampelatihan)
        mlflow.logparam("jumlahfitur", Xtrain.shape[1])

        mlflow.logparam("sampelpelatihan", Xtrain.shape[0])

        # Catat informasi dataset
        mlflow.settag("versidata",
                       os.popen("dvc version").read().strip())
        mlflow.settag("gitcommit",
                       os.popen("git rev-parse HEAD").read().strip())

        # Latih model
        model = xgb.XGBClassifier(
            nestimators=parampelatihan["nestimators"],

            maxdepth=parampelatihan["maxdepth"],
            learningrate=parampelatihan["learningrate"],

            subsample=parampelatihan["subsample"],
            colsamplebytree=parampelatihan["colsamplebytree"],

            evalmetric="logloss",
            uselabelencoder=False,
            randomstate=42

        )

        # Latih dengan split validasi untuk early stopping
        from sklearn.modelselection import traintestsplit
        Xtr, Xval, ytr, yval = traintestsplit(
            Xtrain, ytrain, testsize=0.15, randomstate=42
        )

        model.fit(
            Xtr, ytr,
            evalset=[(Xval, yval)],

            verbose=False
        )

        # Prediksi dan metrik
        ypred = model.predict(Xval)

        yprob = model.predictproba(Xval)[:, 1]

        metrik = {
            "accuracy": accuracyscore(yval, ypred),

            "precision": precisionscore(yval, ypred),
            "recall": recallscore(yval, ypred),

            "f1": f1score(yval, ypred),
            "aucroc": rocaucscore(yval, yprob),

        }

        # Catat metrik
        mlflow.logmetrics(metrik)

        # Catat pentingnya fitur
        pentingnyafitur = dict(zip(

            Xtrain.columns,
            model.featureimportances.tolist()
        ))
        mlflow.logdict(pentingnyafitur, "pentingnyafitur.json")


        # Catat data kurva ROC
        fpr, tpr,  = roccurve(yval, yprob)

        dataroc = [{"fpr": f, "tpr": t}
                    for f, t in zip(fpr.tolist(), tpr.tolist())]

        # Catat model
        mlflow.xgboost.logmodel(

            model,
            artifactpath="model",
            registeredmodelname="prediktor-churn",
            inputexample=Xval.iloc[:3],
        )

        # Simpan lokal untuk pelacakan DVC
        os.makedirs("models", existok=True)

        joblib.dump(model, "models/model.pkl")

        # Simpan metrik untuk DVC
        os.makedirs("metrics", existok=True)
        with open("metrics/metrikpelatihan.json", "w") as f:

            json.dump(metrik, f, indent=2)

        with open("metrics/kurvaroc.json", "w") as f:
            json.dump(dataroc, f)


        print(f"Run ID: {run.info.runid}")
        for nama, nilai in metrik.items():
            print(f"  {nama}: {nilai:.4f}")

        return run.info.runid


if name == "main":
    latihmodel()

Model Registry

MLflow Model Registry menyediakan pusat sentral untuk mengelola tahapan siklus hidup model.

# src/model/registry.py
import mlflow
from mlflow.tracking import MlflowClient

client = MlflowClient("http://localhost:5000")

def promosikanmodel(namamodel: str, versi: int, tahap: str):
    """Promosikan versi model ke tahap baru."""
    tahapvalid = ["Staging", "Production", "Archived"]

    if tahap not in tahapvalid:
        raise ValueError(f"Tahap harus salah satu dari {tahapvalid}")


    client.transitionmodelversionstage(
        name=namamodel,

        version=versi,
        stage=tahap,
        archiveexistingversions=(tahap == "Production")

    )
    print(f"Model {namamodel} v{versi} dipromosikan ke {tahap}")


def ambilmodelproduksi(namamodel: str):

    """Muat model produksi saat ini."""
    modeluri = f"models:/{namamodel}/Production"

    model = mlflow.pyfunc.loadmodel(modeluri)

    return model


def bandingkanmodel(namamodel: str, metrik: str = "aucroc"):
    """Bandingkan semua versi model berdasarkan metrik tertentu."""
    versilist = client.searchmodelversions(f"name='{namamodel}'")

    hasil = []
    for v in versilist:

        run = client.getrun(v.runid)

        nilaimetrik = run.data.metrics.get(metrik, 0)
        hasil.append({
            "versi": v.version,
            "tahap": v.currentstage,

            "metrik": nilaimetrik,
            "runid": v.runid,
        })

    hasil.sort(key=lambda x: x["metrik"], reverse=True)

    print(f"\nModel: {namamodel} | Metrik: {metrik}")

    print("-"  60)

    for r in hasil:
        print(f"  v{r['versi']} ({r['tahap']}): {r['metrik']:.4f}")

    return hasil


Penggunaan
if name == "main":
    bandingkanmodel("prediktor-churn", metrik="aucroc")

    # promosikanmodel("prediktor-churn", versi=3, tahap="Production")

# .github/workflows/train.yml
name: Pipeline Pelatihan ML

on:
  push:
    paths:
'src/'

'params.yaml'
'dvc.yaml'
  workflowdispatch:

    inputs:
      latihulang:
        description: 'Paksa pelatihan ulang'
        required: false
        default: 'false'

jobs:
  latih:
    runs-on: ubuntu-latest
    steps:
uses: actions/checkout@v4
name: Siapkan Python
        uses: actions/setup-python@v5

        with:
          python-version: '3.11'
name: Instal dependensi
        run: |

          pip install -r requirements.txt
          pip install dvc[s3]
name: Konfigurasi kredensial AWS
        uses: aws-actions/configure-aws-credentials@v4

        with:
          aws-access-key-id: ${{ secrets.AWSACCESSKEYID }}

          aws-secret-access-key: ${{ secrets.AWSSECRETACCESSKEY }}
          aws-region: us-east-1
name: Tarik data dari DVC
        run: dvc pull

name: Jalankan pipeline DVC
        env:

          MLFLOWTRACKINGURI: ${{ secrets.MLFLOWTRACKINGURI }}
        run: dvc repro
name: Jalankan pengujian
        run: pytest tests/ -v --tb=short

name: Periksa gerbang kualitas model
        run: |

          python -c "
          import json
          with open('metrics/metrikevaluasi.json') as f:

              metrik = json.load(f)
          assert metrik['aucroc'] > 0.85, \
              f'AUC {metrik[\"aucroc\"]:.4f} di bawah ambang 0.85'

          assert metrik['f1'] > 0.75, \
              f'F1 {metrik[\"f1\"]:.4f} di bawah ambang 0.75'
          print('Gerbang kualitas LOLOS')
          "
name: Push metrik
        run: |

          git config user.name github-actions
          git config user.email github-actions@github.com
          git add metrics/
          git diff --cached --quiet || git commit -m "Perbarui metrik [skip ci]"
          git push

  deploy:
    needs: latih
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
uses: actions/checkout@v4
name: Build dan push image Docker
        run: |

          docker build -t churn-api:${{ github.sha }} serving/
          docker tag churn-api:${{ github.sha }} \
            ${{ secrets.ECRREGISTRY }}/churn-api:latest
          docker push ${{ secrets.ECRREGISTRY }}/churn-api:latest

name: Deploy ke ECS
        run: |

          aws ecs update-service \
            --cluster ml-production \
            --service churn-api \
            --force-new-deployment

# .github/workflows/test.yml
name: Pengujian ML

on:
  pullrequest:
    branches: [main]

jobs:
  pengujian:
    runs-on: ubuntu-latest
    steps:
uses: actions/checkout@v4
name: Siapkan Python
        uses: actions/setup-python@v5

        with:
          python-version: '3.11'
name: Instal dependensi
        run: pip install -r requirements.txt

name: Jalankan unit test
        run: pytest tests/testdata.py tests/testmodel.py -v

name: Jalankan tes API
        run: pytest tests/testapi.py -v

name: Lint kode
        run: |

          pip install ruff
          ruff check src/

# serving/Dockerfile
FROM python:3.11-slim

WORKDIR /app

Instal dependensi
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

Salin kode aplikasi
COPY app.py .
COPY models/ models/

Health check
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000

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

# serving/app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import numpy as np
import pandas as pd
import mlflow
import logging
import time
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

app = FastAPI(title="API Prediksi Churn", version="1.0.0")

Muat model saat startup
model = None
versimodel = None

@app.onevent("startup")

async def muatmodel():
    global model, versimodel

    try:
        # Coba muat dari registry MLflow terlebih dahulu
        mlflowuri = os.environ.get("MLFLOWTRACKINGURI")
        if mlflowuri:

            mlflow.settrackinguri(mlflowuri)
            model = mlflow.pyfunc.loadmodel(

                "models:/prediktor-churn/Production"
            )
            versimodel = "mlflow-production"
        else:
            model = joblib.load("models/model.pkl")
            versimodel = "lokal"

        logger.info(f"Model dimuat: {versimodel}")
    except Exception as e:
        logger.error(f"Gagal memuat model: {e}")
        raise


class PermintaanPrediksi(BaseModel):
    masaberlangganan: float

    biayabulanan: float
    totalbiaya: float

    tipekontrak: str
    metodepembayaran: str

    layananinternet: str


class ResponPrediksi(BaseModel):
    probabilitaschurn: float

    prediksichurn: bool
    versimodel: str

    waktuprediksims: float



@app.get("/health")
async def kesehatan():
    return {
        "status": "sehat",
        "modeldimuat": model is not None,
        "versimodel": versimodel,
        "timestamp": datetime.utcnow().isoformat()
    }


@app.post("/predict", responsemodel=ResponPrediksi)

async def prediksi(permintaan: PermintaanPrediksi):
    if model is None:
        raise HTTPException(statuscode=503, detail="Model belum dimuat")

    waktumulai = time.time()


    try:
        # Buat DataFrame fitur
        fitur = pd.DataFrame([permintaan.dict()])

        # Dapatkan prediksi
        if hasattr(model, "predictproba"):
            probabilitas = float(model.predictproba(fitur)[0, 1])

        else:
            probabilitas = float(model.predict(fitur)[0])

        prediksi = probabilitas >= 0.5
        berlalums = (time.time() - waktumulai)  1000

        # Catat prediksi untuk monitoring
        logger.info(
            f"Prediksi: prob={probabilitas:.4f}, "
            f"churn={prediksi}, waktu={berlalums:.1f}ms"
        )

        return ResponPrediksi(
            probabilitaschurn=round(probabilitas, 4),

            prediksichurn=prediksi,
            versimodel=versimodel,
            waktuprediksims=round(berlalums, 2)

        )

    except Exception as e:
        logger.error(f"Error prediksi: {e}")
        raise HTTPException(statuscode=500, detail=str(e))


@app.post("/predict/batch")
async def prediksibatch(daftarpermintaan: list[PermintaanPrediksi]):
    """Endpoint prediksi batch untuk beberapa pelanggan."""
    if model is None:
        raise HTTPException(statuscode=503, detail="Model belum dimuat")


    fitur = pd.DataFrame([p.dict() for p in daftarpermintaan])

    if hasattr(model, "predictproba"):

        probabilitas = model.predictproba(fitur)[:, 1]
    else:
        probabilitas = model.predict(fitur)

    return {
        "prediksi": [
            {
                "probabilitaschurn": round(float(p), 4),

                "prediksichurn": bool(p >= 0.5)
            }
            for p in probabilitas
        ],
        "jumlah": len(daftarpermintaan),

        "versimodel": versimodel

    }

# docker-compose.yml
version: '3.8'

services:
  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.10.0
    ports:
"5000:5000"
    command: >

      mlflow server
        --backend-store-uri sqlite:///mlflow.db
        --default-artifact-root /mlruns
        --host 0.0.0.0
    volumes:
mlflow-data:/mlruns

  api:
    build: ./serving
    ports:
"8000:8000"
    environment:

MLFLOWTRACKINGURI=http://mlflow:5000

    dependson:

mlflow

  monitoring:
    build: ./monitoring
    ports:
"8501:8501"
    environment:

APIURL=http://api:8000

    dependson:

api

volumes:
  mlflow-data:

Deployment

Skrip Deploy Produksi

# scripts/deploy.py
import boto3
import json
import subprocess
import sys

def deploykeproduksi(tagimage: str, cluster: str = "ml-production",

                       service: str = "churn-api"):
    """Deploy versi model baru ke produksi."""

    # Langkah 1: Jalankan tes pra-deployment
    print("Menjalankan tes pra-deployment...")
    result = subprocess.run(
        ["pytest", "tests/", "-v", "--tb=short"],
        captureoutput=True, text=True
    )
    if result.returncode != 0:
        print(f"Tes gagal:\n{result.stdout}")
        sys.exit(1)

    # Langkah 2: Validasi kualitas model
    print("Memvalidasi kualitas model...")
    with open("metrics/metrikevaluasi.json") as f:

        metrik = json.load(f)

    ambangbatas = {"aucroc": 0.85, "f1": 0.75, "precision": 0.70}

    for namametrik, ambang in ambangbatas.items():

        if metrik.get(namametrik, 0) < ambang:
            print(f"GAGAL: {namametrik}={metrik[namametrik]:.4f} < {ambang}")
            sys.exit(1)

    print("Semua gerbang kualitas lolos")

    # Langkah 3: Perbarui layanan ECS
    print(f"Men-deploy {tagimage} ke {cluster}/{service}...")

    ecs = boto3.client("ecs")
    ecs.updateservice(
        cluster=cluster,
        service=service,
        forceNewDeployment=True
    )

    print("Deployment dimulai. Pantau di AWS Console.")

if name == "main":
    deploykeproduksi(tagimage=sys.argv[1] if len(sys.argv) > 1

                       else "latest")

# src/monitoring/deteksidrift.py
import pandas as pd
import numpy as np
from evidently.report import Report
from evidently.metricpreset import (

    DataDriftPreset,
    TargetDriftPreset,
    ClassificationPreset
)
from evidently.metrics import (
    DataDriftTable,
    DatasetDriftMetric,
    ColumnDriftMetric
)
from evidently.testsuite import TestSuite
from evidently.tests import (
    TestShareOfDriftedColumns,
    TestColumnDrift,
    TestShareOfMissingValues
)
import json
from datetime import datetime
import logging

logger = logging.getLogger(name)

class PemantauModel:
    """Pantau performa model dan data drift di produksi."""

    def init(self, datareferensi: pd.DataFrame,

                 kolomfitur: list, kolomtarget: str = "churn"):

        self.datareferensi = datareferensi

        self.kolomfitur = kolomfitur

        self.kolomtarget = kolomtarget


    def periksadatadrift(self, datasaatini: pd.DataFrame,

                            ambangdrift: float = 0.3) -> dict:
        """Periksa data drift antara data referensi dan data saat ini."""
        report = Report(metrics=[
            DatasetDriftMetric(),
            DataDriftTable(),
        ])

        report.run(
            referencedata=self.datareferensi[self.kolomfitur],

            currentdata=datasaatini[self.kolomfitur]

        )

        hasil = report.asdict()
        driftdataset = hasil["metrics"][0]["result"]


        laporandrift = {
            "timestamp": datetime.utcnow().isoformat(),
            "driftdatasetterdeteksi": driftdataset["datasetdrift"],
            "proporsidrift": driftdataset["shareofdriftedcolumns"],

            "jumlahkolom": driftdataset["numberofcolumns"],

            "jumlahkolomdrift":

                driftdataset["numberofdriftedcolumns"],

            "kolomdrift": [],
        }

        # Identifikasi kolom mana yang mengalami drift
        hasilkolom = hasil["metrics"][1]["result"]["driftbycolumns"]

        for namakolom, datakolom in hasilkolom.items():
            if datakolom["driftdetected"]:
                laporandrift["kolomdrift"].append({
                    "kolom": namakolom,

                    "skordrift": datakolom["driftscore"],
                    "namastattest": datakolom["stattestname"],

                })

        kritis = (
            laporandrift["proporsidrift"] > ambangdrift
        )
        laporandrift["tindakandiperlukan"] = kritis

        return laporandrift


    def jalankansuitetes(self, datasaatini: pd.DataFrame) -> dict:

        """Jalankan tes kualitas otomatis pada data saat ini."""
        suite = TestSuite(tests=[
            TestShareOfDriftedColumns(lt=0.3),
            TestShareOfMissingValues(lt=0.05),
        ])

        # Tambahkan tes drift per kolom untuk fitur kritis
        for kolom in ["biayabulanan", "masaberlangganan", "totalbiaya"]:
            if kolom in self.kolomfitur:

                suite.tests.append(TestColumnDrift(columnname=kolom))

        suite.run(
            referencedata=self.datareferensi,
            currentdata=datasaatini

        )

        hasil = suite.asdict()
        return {
            "timestamp": datetime.utcnow().isoformat(),
            "ringkasan": hasil["summary"],
            "semualolos": hasil["summary"]["allpassed"],
            "tes": [
                {
                    "nama": t["name"],
                    "status": t["status"],
                    "deskripsi": t.get("description", ""),
                }
                for t in hasil["tests"]
            ]
        }

    def buatlaporanmonitoring(self, datasaatini: pd.DataFrame,
                                 pathoutput: str = "laporanmonitoring.html"):
        """Buat laporan HTML monitoring yang komprehensif."""
        report = Report(metrics=[
            DataDriftPreset(),
            TargetDriftPreset(),
        ])

        report.run(
            referencedata=self.datareferensi,
            currentdata=datasaatini

        )

        report.savehtml(pathoutput)

        logger.info(f"Laporan monitoring disimpan ke {pathoutput}")


Penggunaan
if name == "main":
    # Muat data referensi (data pelatihan)
    referensi = pd.readcsv("data/processed/trainfitur.csv")

    # Simulasikan data produksi
    saatini = pd.readcsv("data/produksi/batchterbaru.csv")


    kolomfitur = [
        "masaberlangganan", "biayabulanan", "totalbiaya",

        "tipekontrak", "metodepembayaran", "layananinternet"
    ]

    pemantau = PemantauModel(referensi, kolomfitur)


    # Periksa drift
    drift = pemantau.periksadatadrift(saatini)
    print(f"Drift terdeteksi: {drift['driftdatasetterdeteksi']}")
    print(f"Kolom drift: {len(drift['kolomdrift'])}")


    if drift["tindakandiperlukan"]:
        print("TINDAKAN DIPERLUKAN: Data drift signifikan terdeteksi!")

    # Jalankan tes kualitas
    hasiltes = pemantau.jalankansuitetes(saatini)
    print(f"Tes lolos: {hasiltes['semualolos']}")

Alerting (Sistem Peringatan)

# src/monitoring/alerting.py
import requests
import json
import logging
from datetime import datetime

logger = logging.getLogger(name)

class ManajerPeringatan:
    """Kirim peringatan ketika monitoring mendeteksi masalah."""

    def init(self, urlwebhookslack: str = None,
                 kuncipagerduty: str = None):

        self.slackwebhook = urlwebhookslack
        self.kuncipagerduty = kuncipagerduty

    def kirimperingatanslack(self, judul: str, pesan: str,
                                tingkatkeparahan: str = "warning"):

        """Kirim peringatan ke channel Slack."""
        if not self.slackwebhook:
            logger.warning("Webhook Slack belum dikonfigurasi")
            return

        petawarna = {

            "info": "#36a64f",
            "warning": "#ff9900",
            "critical": "#ff0000"
        }

        payload = {
            "attachments": [{
                "color": petawarna.get(tingkatkeparahan, "#ff9900"),

                "title": f"[Peringatan ML] {judul}",
                "text": pesan,
                "fields": [
                    {"title": "Tingkat Keparahan",
                     "value": tingkatkeparahan, "short": True},
                    {"title": "Waktu",
                     "value": datetime.utcnow().isoformat(),
                     "short": True},
                ],
                "footer": "Sistem Monitoring MLOps"
            }]
        }

        response = requests.post(self.slackwebhook, json=payload)

        if response.statuscode == 200:
            logger.info(f"Peringatan Slack terkirim: {judul}")
        else:
            logger.error(f"Peringatan Slack gagal: {response.text}")

    def periksadanperingatkan(self, laporandrift: dict,

                                 hasiltes: dict):
        """Evaluasi hasil monitoring dan kirim peringatan."""
        # Peringatan data drift
        if laporandrift.get("tindakandiperlukan"):
            kolomdrift = ", ".join(

                [d["kolom"] for d in laporandrift["kolomdrift"]]

            )
            self.kirimperingatanslack(

                judul="Data Drift Terdeteksi",
                pesan=(
                    f"Data drift signifikan terdeteksi di produksi.\n"
                    f"Kolom drift ({laporandrift['proporsidrift']:.0%}): "

                    f"{kolomdrift}\n"
                    f"Pertimbangkan untuk melatih ulang model."
                ),
                tingkatkeparahan="critical"

            )

        # Peringatan kegagalan tes
        if not hasiltes.get("semualolos", True):

            gagal = [t["nama"] for t in hasiltes.get("tes", [])
                     if t["status"] != "SUCCESS"]
            self.kirimperingatanslack(
                judul="Tes Kualitas Gagal",
                pesan=(
                    f"Tes kualitas data produksi gagal.\n"
                    f"Tes yang gagal: {', '.join(gagal)}"
                ),
                tingkatkeparahan="warning"

            )


Job monitoring terjadwal
def jalankanjobmonitoring():

    """Jalankan pipeline monitoring lengkap (panggil dari cron atau scheduler)."""
    from deteksidrift import PemantauModel
    import pandas as pd

    referensi = pd.readcsv("data/processed/trainfitur.csv")
    saatini = pd.readcsv("data/produksi/batchterbaru.csv")


    kolomfitur = [