03 - Versionare Dataset e Modelli: DVC vs LakeFS

# Installazione base
pip install dvc

# Con supporto remote storage (scegli in base al provider)
pip install dvc[s3]       # Amazon S3
pip install dvc[gs]       # Google Cloud Storage
pip install dvc[azure]    # Azure Blob Storage
pip install dvc[ssh]      # SSH/SFTP
pip install dvc[all]      # Tutti i backend

# Verifica installazione
dvc --version
# DVC 3.x.x

# Inizializza Git (se non già fatto)
git init
git add .
git commit -m "Initial commit"

# Inizializza DVC
dvc init

# Struttura creata da dvc init:
# .dvc/
# ├── config          # Configurazione DVC (tracciato da Git)
# ├── .gitignore      # Esclude cache e tmp
# ├── cache/          # Cache locale degli artefatti (NON tracciata da Git)
# └── tmp/

# Commit dei file di configurazione DVC
git add .dvc/ .gitignore
git commit -m "chore: initialize DVC"

# Aggiungere un dataset al tracking DVC
dvc add data/raw/training_data.csv

# DVC ha creato:
# - data/raw/training_data.csv.dvc  (puntatore, tracciato da Git)
# - data/raw/.gitignore             (esclude il file originale da Git)

# Contenuto del file .dvc generato:
# outs:
# - md5: a1b2c3d4e5f6...
#   size: 524288000
#   path: training_data.csv

# Aggiungere il puntatore a Git
git add data/raw/training_data.csv.dvc data/raw/.gitignore
git commit -m "feat: add training dataset v1.0"
git tag "dataset-v1.0"

# Aggiungere un modello addestrato
dvc add models/churn_model.pkl
git add models/churn_model.pkl.dvc models/.gitignore
git commit -m "feat: add trained churn model v1 (accuracy=0.94)"
git tag "model-v1.0"

# Workflow DVC quotidiano

# 1. Pull dati aggiornati dal remote
git pull
dvc pull

# 2. Modifica dataset o allena nuovo modello
python scripts/preprocess.py
python scripts/train.py

# 3. Traccia nuovi artefatti
dvc add data/processed/features.parquet
dvc add models/churn_model_v2.pkl

# 4. Commit codice e puntatori
git add .
git commit -m "feat: retrain model with new features (F1=0.92)"

# 5. Push dati al remote storage
dvc push

# 6. Push codice a Git
git push

# ---- Sul computer di un collega ----
git pull          # Scarica codice e .dvc files
dvc pull          # Scarica i dati dal remote storage

# ==================== AWS S3 ====================
# Prerequisiti: pip install dvc[s3], AWS credentials configurate

# Aggiungere S3 come remote di default
dvc remote add -d myremote s3://my-ml-bucket/dvc-storage

# Configurazione con access keys (per CI/CD, preferire IAM roles)
dvc remote modify myremote access_key_id $AWS_ACCESS_KEY_ID
dvc remote modify myremote secret_access_key $AWS_SECRET_ACCESS_KEY
dvc remote modify myremote region eu-west-1

# Per credenziali sensibili, usare config locale (non tracciata da Git)
dvc remote modify --local myremote access_key_id $AWS_ACCESS_KEY_ID
dvc remote modify --local myremote secret_access_key $AWS_SECRET_ACCESS_KEY


# ==================== Google Cloud Storage ====================
# Prerequisiti: pip install dvc[gs], gcloud auth configurato

dvc remote add -d gcsremote gs://my-ml-bucket/dvc-storage

# Con Service Account (per produzione)
dvc remote modify gcsremote credentialpath /path/to/service-account.json


# ==================== Azure Blob Storage ====================
# Prerequisiti: pip install dvc[azure]

dvc remote add -d azureremote azure://mycontainer/dvc-storage

# Configurazione connection string
dvc remote modify --local azureremote connection_string $AZURE_CONN_STRING


# ==================== Verifica e Operazioni ====================
# Mostra la configurazione remota
dvc remote list

# Push di tutti gli artefatti tracciati
dvc push

# Push solo di file specifici
dvc push models/churn_model.pkl.dvc

# Pull di tutti gli artefatti
dvc pull

# Verifica stato sincronizzazione
dvc status --cloud

# DagsHub come remote storage gratuito
# 1. Crea account su dagshub.com e un nuovo repository

# 2. Configura il remote DagsHub
dvc remote add origin https://dagshub.com/username/my-ml-project.dvc

# 3. Autenticazione con token personale
dvc remote modify origin --local auth basic
dvc remote modify origin --local user username
dvc remote modify origin --local password $DAGSHUB_TOKEN

# 4. Push e pull normali
dvc push
dvc pull

# dvc.yaml - Pipeline ML completa per churn prediction
stages:

  # Stage 1: Download e validazione dei dati raw
  fetch_data:
    cmd: python src/data/fetch_data.py
    deps:
      - src/data/fetch_data.py
    params:
      - data.source_url
      - data.date_range
    outs:
      - data/raw/transactions.parquet
      - data/raw/customers.csv

  # Stage 2: Preprocessing e feature engineering
  preprocess:
    cmd: python src/features/preprocess.py
    deps:
      - src/features/preprocess.py
      - data/raw/transactions.parquet
      - data/raw/customers.csv
    params:
      - features.window_days
      - features.aggregations
    outs:
      - data/processed/features.parquet
      - data/processed/feature_names.json

  # Stage 3: Split train/validation/test
  split:
    cmd: python src/data/split.py
    deps:
      - src/data/split.py
      - data/processed/features.parquet
    params:
      - split.train_ratio
      - split.val_ratio
      - split.random_seed
    outs:
      - data/splits/train.parquet
      - data/splits/val.parquet
      - data/splits/test.parquet

  # Stage 4: Training del modello
  train:
    cmd: python src/models/train.py
    deps:
      - src/models/train.py
      - data/splits/train.parquet
      - data/splits/val.parquet
    params:
      - model.type
      - model.n_estimators
      - model.max_depth
      - model.learning_rate
    outs:
      - models/churn_model.pkl
      - models/feature_importance.json
    metrics:
      - metrics/train_metrics.json:
          cache: false

  # Stage 5: Valutazione sul test set
  evaluate:
    cmd: python src/models/evaluate.py
    deps:
      - src/models/evaluate.py
      - models/churn_model.pkl
      - data/splits/test.parquet
    outs:
      - reports/confusion_matrix.png:
          cache: false
    metrics:
      - metrics/test_metrics.json:
          cache: false

# params.yaml - Configurazione centralizzata
data:
  source_url: "s3://my-data-bucket/raw/2025/"
  date_range: "2024-01-01:2025-01-01"

features:
  window_days: 30
  aggregations:
    - mean
    - std
    - max

split:
  train_ratio: 0.7
  val_ratio: 0.15
  random_seed: 42

model:
  type: "xgboost"
  n_estimators: 500
  max_depth: 6
  learning_rate: 0.05
  subsample: 0.8

# Eseguire l'intera pipeline (solo stage invalidati)
dvc repro

# Forzare ri-esecuzione di tutti gli stage
dvc repro --force

# Eseguire solo fino a uno stage specifico
dvc repro evaluate

# Visualizzare il DAG della pipeline
dvc dag

# Output:
#      +------------+
#      | fetch_data |
#      +------------+
#          *  *
#       ***    ***
#      *          *
# +-----------+  +----------+
# | preprocess|  |          |
# +-----------+  +----------+
#       *
#      ...

# Confrontare metriche tra commit
dvc metrics show
dvc metrics diff HEAD~1

# Confrontare parametri tra branch
dvc params diff main feature/new-model

# Visualizzare differenze nei dati
dvc diff

import dvc.api
import pandas as pd
import pickle
import json


# ==================== dvc.api.open() ====================
# Accede direttamente al remote storage senza download locale

# Leggere un dataset CSV da una versione specifica
with dvc.api.open(
    "data/raw/customers.csv",
    repo="https://github.com/myorg/ml-project",
    rev="dataset-v2.1",  # tag Git
    mode="r"
) as f:
    df = pd.read_csv(f)

print(f"Dataset caricato: {len(df)} righe")

# Leggere un file Parquet (file binario)
with dvc.api.open(
    "data/processed/features.parquet",
    rev="model-v3.0",
    mode="rb"
) as f:
    features_df = pd.read_parquet(f)


# ==================== dvc.api.get_url() ====================
# Ottiene l'URL diretto al remote storage per una versione specifica

url = dvc.api.get_url(
    "models/churn_model.pkl",
    repo="https://github.com/myorg/ml-project",
    rev="model-v2.5"
)
print(f"Model URL: {url}")
# Output: s3://my-ml-bucket/dvc-storage/ab/cd1234...


# ==================== Caricamento Modello Versioned ====================
def load_model(version: str):
    """Carica un modello da una specifica versione DVC."""
    with dvc.api.open(
        "models/churn_model.pkl",
        rev=version,
        mode="rb"
    ) as f:
        model = pickle.load(f)

    # Carica anche i metadata del modello
    with dvc.api.open(
        "models/feature_importance.json",
        rev=version,
        mode="r"
    ) as f:
        metadata = json.load(f)

    return model, metadata


# Uso in produzione
model, metadata = load_model("model-v3.0")
print(f"Modello caricato. Features: {metadata['n_features']}")

# Workflow Git+DVC per un ciclo di sviluppo ML

# ==================== Feature Branch Workflow ====================

# 1. Crea un branch per il nuovo esperimento
git checkout -b experiment/xgboost-v2

# 2. Aggiorna i parametri
# Modifica params.yaml: learning_rate: 0.01, n_estimators: 1000

# 3. Esegui la pipeline
dvc repro

# 4. Controlla le metriche
dvc metrics show
# Path                    accuracy    f1_score    auc_roc
# metrics/test_metrics.json   0.9423      0.8891      0.9234

# 5. Confronta con main
dvc metrics diff main
# Path                    Metric      Old     New     Change
# metrics/test_metrics.json accuracy   0.9301  0.9423  +0.0122

# 6. Commit e push
git add dvc.lock params.yaml metrics/
git commit -m "experiment: XGBoost v2 - accuracy +1.2% (0.9423)"
dvc push
git push origin experiment/xgboost-v2

# 7. Crea Pull Request e, dopo merge, tagga il modello
git checkout main
git merge experiment/xgboost-v2
git tag -a "model-v2.0" -m "XGBoost v2: accuracy=0.9423, F1=0.8891"
git push --tags


# ==================== Rollback a Versione Precedente ====================

# Situazione: il modello v2.0 ha un bug in produzione

# 1. Torna al commit del modello v1.0
git checkout model-v1.0

# 2. Ripristina i dati e i modelli a quella versione
dvc checkout

# 3. Verifica il modello ripristinato
python src/models/evaluate.py

# 4. Il sistema di serving può ora ricaricare model-v1.0

import mlflow
import mlflow.sklearn
import dvc.api
import subprocess
import pandas as pd
from sklearn.metrics import accuracy_score, f1_score, roc_auc_score
from xgboost import XGBClassifier


def get_dvc_metadata() -> dict:
    """Raccoglie metadata DVC per il run MLflow corrente."""
    # Hash del dataset corrente
    result = subprocess.run(
        ["dvc", "status", "--json"],
        capture_output=True, text=True
    )

    # Ottieni la revisione Git corrente (che punta ai dati)
    git_rev = subprocess.run(
        ["git", "rev-parse", "HEAD"],
        capture_output=True, text=True
    ).stdout.strip()

    # URL del remote storage per il dataset
    data_url = dvc.api.get_url("data/splits/train.parquet")

    return {
        "dvc.git_rev": git_rev,
        "dvc.data_url": data_url,
        "dvc.train_dataset": "data/splits/train.parquet",
        "dvc.data_version": "v2.1",
    }


def train_with_tracking(params: dict) -> None:
    """Training con tracking completo DVC + MLflow."""

    mlflow.set_experiment("churn-prediction")

    with mlflow.start_run(run_name="xgboost-dvc-integrated") as run:
        # Logga metadata DVC - collega run MLflow a versione dati
        dvc_metadata = get_dvc_metadata()
        mlflow.log_params(dvc_metadata)
        mlflow.log_params(params)

        # Carica dati dalla versione DVC corrente
        train_df = pd.read_parquet("data/splits/train.parquet")
        val_df = pd.read_parquet("data/splits/val.parquet")
        test_df = pd.read_parquet("data/splits/test.parquet")

        X_train = train_df.drop("churn", axis=1)
        y_train = train_df["churn"]
        X_test = test_df.drop("churn", axis=1)
        y_test = test_df["churn"]

        # Training
        model = XGBClassifier(**params)
        model.fit(X_train, y_train,
                  eval_set=[(val_df.drop("churn", axis=1), val_df["churn"])],
                  early_stopping_rounds=50,
                  verbose=False)

        # Metriche
        y_pred = model.predict(X_test)
        y_prob = model.predict_proba(X_test)[:, 1]

        metrics = {
            "accuracy": accuracy_score(y_test, y_pred),
            "f1_score": f1_score(y_test, y_pred),
            "auc_roc": roc_auc_score(y_test, y_prob),
            "best_iteration": model.best_iteration,
        }

        mlflow.log_metrics(metrics)
        mlflow.sklearn.log_model(model, "model")

        print(f"Run ID: {run.info.run_id}")
        print(f"Accuracy: {metrics['accuracy']:.4f}")
        print(f"Dataset version: {dvc_metadata['dvc.data_version']}")


if __name__ == "__main__":
    params = {
        "n_estimators": 500,
        "max_depth": 6,
        "learning_rate": 0.05,
        "subsample": 0.8,
        "use_label_encoder": False,
        "eval_metric": "logloss",
    }
    train_with_tracking(params)

# lakeFS: esempio di workflow con Python SDK
from lakefs_sdk import Client, Repository, Branch, Commit
import pandas as pd


# Connessione al server lakeFS
client = Client(
    host="https://lakefs.mycompany.com",
    username="access_key",
    password="secret_key"
)

# Crea un repository
repo = client.repositories.create_repository(
    "churn-data-lake",
    storage_namespace="s3://my-data-lake/churn/",
    default_branch="main"
)

# Workflow Git-like per dati
# 1. Crea branch per nuovo esperimento
experiment_branch = repo.branch("experiment/new-features").create(
    source_reference="main"
)

# 2. Upload nuovi dati sul branch sperimentale
# (non impatta il branch main - zero copy)
experiment_branch.object("data/features_v2.parquet").upload(
    content=new_features_df.to_parquet()
)

# 3. Commit sul branch
experiment_branch.commit(
    message="feat: add recency features for churn model",
    metadata={"model_version": "v3.0", "engineer": "alice"}
)

# 4. Merge solo se i risultati sono soddisfacenti
# (dopo validazione e A/B testing)
repo.branch("main").merge(source_ref="experiment/new-features")

# 5. Rollback immediato se qualcosa va storto
repo.branch("main").revert(
    reference="main~1"  # Torna al commit precedente
)

ml-project/
├── .dvc/
│   ├── config              # Remote storage config (no secrets)
│   └── .gitignore
├── data/
│   ├── raw/
│   │   ├── .gitignore      # Generato da DVC
│   │   ├── customers.csv.dvc
│   │   └── transactions.parquet.dvc
│   ├── processed/
│   │   └── features.parquet.dvc
│   └── splits/
│       ├── train.parquet.dvc
│       └── test.parquet.dvc
├── models/
│   ├── .gitignore
│   └── churn_model.pkl.dvc
├── metrics/
│   └── test_metrics.json   # Tracciato da Git (piccolo file JSON)
├── reports/
│   └── confusion_matrix.png # Tracciato da Git (con cache: false)
├── src/
│   ├── data/
│   │   ├── fetch_data.py
│   │   └── split.py
│   ├── features/
│   │   └── preprocess.py
│   └── models/
│       ├── train.py
│       └── evaluate.py
├── dvc.yaml                # Definizione pipeline
├── dvc.lock                # Snapshot stato pipeline (tracciato da Git)
├── params.yaml             # Iperparametri e configurazione
└── requirements.txt

# Schema di naming raccomandato per tag Git+DVC

# Dataset versioning
git tag "data/raw/v1.0"          # Prima raccolta dati
git tag "data/processed/v2.3"    # Dataset preprocessato v2.3
git tag "data/features/v1.0-30d" # Features con finestra 30 giorni

# Model versioning
git tag "model/baseline/v1.0"    # Modello baseline
git tag "model/xgboost/v2.1"     # XGBoost tuned v2.1
git tag "model/prod/v3.0"        # Versione in produzione attuale

# Pipeline snapshot
git tag "pipeline/2025-Q1"       # Snapshot pipeline Q1 2025

# Esempio di workflow release
git tag -a "model/prod/v3.0" \
  -m "Production model v3.0

  Metrics:
  - accuracy: 0.9423
  - f1_score: 0.8891
  - auc_roc: 0.9567

  Dataset: data/features/v2.1
  MLflow run: a1b2c3d4e5f6"

git push origin "model/prod/v3.0"

# .github/workflows/ml-pipeline.yml
name: ML Pipeline Validation

on:
  pull_request:
    branches: [main]
    paths:
      - 'src/**'
      - 'params.yaml'
      - 'dvc.yaml'

jobs:
  run-pipeline:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0  # Necessario per dvc metrics diff

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install dvc[s3] -r requirements.txt

      - name: Configure DVC remote
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
        run: |
          dvc remote modify --local myremote \
            access_key_id $AWS_ACCESS_KEY_ID
          dvc remote modify --local myremote \
            secret_access_key $AWS_SECRET_ACCESS_KEY

      - name: Pull data
        run: dvc pull data/splits/

      - name: Run pipeline
        run: dvc repro

      - name: Check metrics regression
        run: |
          # Confronta metriche con main branch
          dvc metrics diff main --md >> $GITHUB_STEP_SUMMARY

          # Fallisce se accuracy scende sotto soglia
          python scripts/check_metrics.py \
            --min-accuracy 0.90 \
            --metrics metrics/test_metrics.json

      - name: Push updated artifacts
        run: dvc push