CI deploy Fri Nov 21 11:21:49 UTC 2025

Dockerfile

@@ -2,9 +2,9 @@ FROM python:3.13-slim
 
 WORKDIR /app
 
-# Installer dépendances système
+# Installer dépendances système + cron
 RUN apt-get update && apt-get install -y \
-    gcc build-essential libpq-dev libssl-dev libffi-dev \
+    gcc build-essential libpq-dev libssl-dev libffi-dev cron \
     && rm -rf /var/lib/apt/lists/*
 
 # Installer Poetry
@@ -23,6 +23,21 @@ COPY ./ ./
 # Installer le projet
 RUN poetry install --only-root --no-interaction --no-ansi
 
+# Créer le fichier cron : exécution le 1er de chaque mois à 02:00
+RUN echo "0 2 1 * * cd /app && /usr/local/bin/poetry run python -m src.drift.monitoring >> /var/log/drift_cron.log 2>&1" > /etc/cron.d/drift_cron
+
+# Donner les bons droits au fichier cron
+RUN chmod 0644 /etc/cron.d/drift_cron && crontab /etc/cron.d/drift_cron
+
+# Créer le fichier de log
+RUN touch /var/log/drift_cron.log
+
+# Script de démarrage qui lance à la fois l'API et cron
+RUN echo '#!/bin/bash\n\
+cron\n\
+exec uvicorn src.api.main:app --host 0.0.0.0 --port 7860' > /start.sh && \
+chmod +x /start.sh
+
 EXPOSE 7860
 
-CMD ["uvicorn", "src.api.main:app", "--host", "0.0.0.0", "--port", "7860"]
+CMD ["/start.sh"]

README.md

@@ -124,4 +124,25 @@ Il est nécessaire de préconfigurer cet environnement avec les variables suivan
 * `HF_PIPELINE` : pré-rempli dans `.env.dist`
 * `XDG_CACHE_HOME` : `/tmp/.cache`
 
+## Détection automatique du Datadrift
+
+Le code permettant de gérer le rapport HTML du datadrift est inclus dans `src/drift/monitoring.py`
+Voici son fonctionnement : 
+
+* Récupère les données de logs de prédictions depuis la base de données (`predict_logs`)
+* Compare ces données de production au jeu d'entrainement local avec **Evidently**
+* Génère le rapport au format HTML dans `.data/drift/report.html`
+
+Ce script de monitoring s'exécute en production comme en local une fois par mois
+La configuration est faite dans le `Dockerfile`, par l'utilisation de `crontab`
+
+## Profiling automatique du code de prédiction du modèle
+
+
+Le profiling du code s'effectue automatiquement, il lancé lors de la création de la stack locale avec `make start` (il n'est pas utile de l'exporter en production), il permet : 
+* de simuler un appel à `/predict` et analyse la stack d'appels avec `cProfile`
+* de générer un rapport d'analyse dans `.data/profiling/`, en html et en `.prof`
+* de lancer le container docker `p8-snakeviz` pour visualiser les résultats du rapports à l'adresse suivante : 
+  * http://localhost:8082/snakeviz/%2Fapp%2Fprofiling%2Fpredict_inference.prof (**port à adapter à votre configuration**)
+
 

coverage.xml

@@ -1,5 +1,5 @@
 <?xml version="1.0" ?>
-<coverage version="7.12.0" timestamp="1763654250338" lines-valid="290" lines-covered="242" line-rate="0.8345" branches-valid="16" branches-covered="7" branch-rate="0.4375" complexity="0">
+<coverage version="7.12.0" timestamp="1763724058110" lines-valid="290" lines-covered="242" line-rate="0.8345" branches-valid="16" branches-covered="7" branch-rate="0.4375" complexity="0">
 	<!-- Generated by coverage.py: https://coverage.readthedocs.io/en/7.12.0 -->
 	<!-- Based on https://raw.githubusercontent.com/cobertura/web/master/htdocs/xml/coverage-04.dtd -->
 	<sources>

src/data/models/__init__.py

@@ -3,5 +3,5 @@ from .predict_logs import PredictLogs
 
 __all__ = [
     "Base",
-    "PredictLogs",
+    "PredictLogs"
 ]

src/drift/monitoring.py

@@ -0,0 +1,115 @@
+import sys
+from pathlib import Path
+
+import pandas as pd
+from sqlalchemy import text
+from evidently import Report
+from evidently.presets import DataDriftPreset
+
+# Ajuste ce chemin à ton projet si besoin
+ROOT_DIR = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT_DIR))
+
+from src.data.database import get_db
+
+# Config
+DATA_DIR = ROOT_DIR / ".data"
+TRAIN_PATH = DATA_DIR / "application_train.csv"
+WINDOW_DAYS = 100
+REPORT_OUTPUT = DATA_DIR / "drift" / "report.html"
+
+def extract_prod_data() -> pd.DataFrame:
+    """Extrait les données prod depuis la table predict_logs."""
+    db = next(get_db())
+
+    query = f"""
+        SELECT input_payload->'application_data' AS data
+        FROM public.predict_logs
+        WHERE status = 'success'
+          AND date >= NOW() - INTERVAL '{WINDOW_DAYS} days'
+    """
+
+    rows = db.execute(text(query)).fetchall()
+    db.close()
+
+    if not rows:
+        return pd.DataFrame()
+
+    data = [row[0] for row in rows]
+    return pd.DataFrame(data)
+
+
+def load_reference_data() -> pd.DataFrame:
+    """Charge les données d'entraînement comme référence."""
+    print(f"Chargement des données de référence depuis {TRAIN_PATH}")
+    return pd.read_csv(TRAIN_PATH)
+
+
+def generate_drift_report(reference_data: pd.DataFrame, current_data: pd.DataFrame) -> None:
+    """Génère un rapport HTML de drift avec Evidently."""
+
+    # Créer le dossier reports s'il n'existe pas
+    REPORT_OUTPUT.parent.mkdir(parents=True, exist_ok=True)
+
+    # Aligner les colonnes entre référence et production
+    common_cols = list(set(reference_data.columns) & set(current_data.columns))
+
+    if not common_cols:
+        print("Aucune colonne commune entre les données de référence et de production!")
+        return
+
+    print(f"Colonnes communes détectées: {len(common_cols)}")
+
+    reference_subset = reference_data[common_cols]
+    current_subset = current_data[common_cols]
+
+    print("Génération du rapport de drift...")
+
+    # Ici : API Evidently "nouvelle" avec Report + metric_preset
+    report = Report(
+        metrics=[
+            DataDriftPreset(),
+        ]
+    )
+
+    eval = report.run(
+        reference_data=reference_subset,
+        current_data=current_subset,
+    )
+
+    # Cette version‑là de Report a bien save_html
+    eval.save_html(str(REPORT_OUTPUT))
+
+    print(f"Rapport de drift généré: {REPORT_OUTPUT}")
+
+
+def main():
+    """Point d'entrée principal."""
+    print("=" * 60)
+    print("ANALYSE DE DRIFT DU MODÈLE")
+    print("=" * 60)
+
+    # 1. Charger les données de référence (train)
+    reference_data = load_reference_data()
+    print(f"Données de référence: {reference_data.shape}")
+
+    # 2. Extraire les données de production
+    print(f"\nExtraction des données de production ({WINDOW_DAYS} derniers jours)...")
+    current_data = extract_prod_data()
+
+    if current_data.empty:
+        print("Aucune donnée de production trouvée!")
+        return
+
+    print(f"Données de production: {current_data.shape}")
+
+    # 3. Générer le rapport de drift
+    generate_drift_report(reference_data, current_data)
+
+    print("\n" + "=" * 60)
+    print("ANALYSE TERMINÉE")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    main()

src/scripts/api_simulation.py

@@ -0,0 +1,161 @@
+import pandas as pd
+import requests
+from pathlib import Path
+from datetime import datetime
+import time
+import os
+
+# Configuration
+API_URL = os.getenv("API_URL")
+API_TOKEN = os.getenv("API_TOKEN")
+DATA_DIR = Path(".data")
+
+
+def load_data():
+    """Charge les données application_test et bureau"""
+    app_test = pd.read_csv(DATA_DIR / "application_test.csv")
+    bureau = pd.read_csv(DATA_DIR / "bureau.csv")
+    return app_test, bureau
+
+
+def get_bureau_records(sk_id_curr, bureau_df):
+    """Récupère tous les enregistrements bureau pour un SK_ID_CURR donné"""
+    bureau_records = bureau_df[bureau_df['SK_ID_CURR'] == sk_id_curr]
+
+    if bureau_records.empty:
+        return []
+
+    # Convertir en liste de dictionnaires
+    bureau_list = bureau_records.to_dict('records')
+
+    # Remplacer les NaN par None pour le JSON
+    for record in bureau_list:
+        for key, value in record.items():
+            if pd.isna(value):
+                record[key] = None
+
+    return bureau_list
+
+
+def create_api_payload(app_row, bureau_records):
+    """Crée le payload JSON pour l'API"""
+    # Convertir la ligne application en dict
+    app_dict = app_row.to_dict()
+
+    # Remplacer les NaN par None
+    for key, value in app_dict.items():
+        if pd.isna(value):
+            app_dict[key] = None
+
+    payload = {
+        "application_data": app_dict,
+        "bureau_data": bureau_records
+    }
+
+    return payload
+
+
+def call_api(payload):
+    """Appelle l'API et retourne la réponse"""
+    try:
+        start_time = time.time()
+        response = requests.post(
+            API_URL+'/predict',
+            json=payload,headers={
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {API_TOKEN}"
+            },
+            timeout=30
+        )
+        inference_time = time.time() - start_time
+
+        if response.status_code == 200:
+            result = response.json()
+            result['inference_time_ms'] = round(inference_time * 1000, 2)
+            result['status'] = 'success'
+            return result
+        else:
+            return {
+                'status': 'error',
+                'status_code': response.status_code,
+                'error_message': response.text,
+                'inference_time_ms': round(inference_time * 1000, 2)
+            }
+    except Exception as e:
+        return {
+            'status': 'error',
+            'error_message': str(e),
+            'inference_time_ms': None
+        }
+
+
+def generate_production_data(start_index=0, num_records=10):
+    """
+    Génère des données de production en appelant l'API
+
+    Args:
+        start_index: Index de départ dans application_test.csv
+        num_records: Nombre d'enregistrements à traiter
+        api_url: URL de l'API
+    """
+    print(f"🚀 Génération de {num_records} prédictions à partir de l'index {start_index}")
+
+    # Charger les données
+    print("Chargement des données...")
+    app_test, bureau = load_data()
+
+    # Sélectionner les lignes
+    end_index = start_index + num_records
+    selected_rows = app_test.iloc[start_index:end_index]
+
+    print(f"{len(selected_rows)} lignes sélectionnées (index {start_index} à {end_index - 1})")
+
+    # Préparer les résultats
+    results = []
+    payloads = []
+
+    # Traiter chaque ligne
+    for idx, (_, row) in enumerate(selected_rows.iterrows(), 1):
+        sk_id_curr = row['SK_ID_CURR']
+        print(f"\n[{idx}/{num_records}] Traitement de SK_ID_CURR: {sk_id_curr}")
+
+        # Récupérer les données bureau
+        bureau_records = get_bureau_records(sk_id_curr, bureau)
+        print(f"  📋 {len(bureau_records)} enregistrements bureau trouvés")
+
+        # Créer le payload
+        payload = create_api_payload(row, bureau_records)
+        payloads.append(payload)
+
+        # Appeler l'API
+        print(f"Appel de l'API...")
+        api_response = call_api(payload)
+
+        # Ajouter les métadonnées
+        result = {
+            'timestamp': datetime.now().isoformat(),
+            'sk_id_curr': int(sk_id_curr),
+            'start_index': start_index,
+            'record_index': start_index + idx - 1,
+            'api_response': api_response
+        }
+
+        results.append(result)
+
+        if api_response['status'] == 'success':
+            print(f"  ✅ Succès - Temps: {api_response.get('inference_time_ms', 'N/A')} ms")
+        else:
+            print(f"  ❌ Erreur: {api_response.get('error_message', 'Unknown')}")
+
+        # Pause pour ne pas surcharger l'API
+        #time.sleep(0.1)
+    return results
+
+
+if __name__ == "__main__":
+
+    # Génère 100 prédictions à partir de l'index 0
+    results = generate_production_data(
+        start_index=0,
+        num_records=500
+    )