# src/scripts/profile_predict.py """ Script de profiling pour identifier les goulots d'étranglement lors de l'inférence (prédiction), simulant le comportement de l'API FastAPI. Usage: poetry run python src/scripts/profile_predict.py Prérequis: - Fichiers de test dans ./.data/extract/: * test_sample_application.csv * test_sample_bureau.csv - Variables d'environnement HF_REPOSITORY, HF_MODEL, HF_PIPELINE """ import cProfile import pstats import io from pathlib import Path from datetime import datetime import os import joblib import pandas as pd import numpy as np from huggingface_hub import hf_hub_download # Variables globales (comme dans l'API) model = None pipeline = None _model_loaded = False def load_model(): """ Charge le modèle et le pipeline depuis Hugging Face (lazy loading). Simule exactement le comportement de l'API. """ global model, pipeline, _model_loaded if _model_loaded: print("/////////////////////////////// Modèle déjà chargé !! //////////") return # déjà chargé print("#######################################################") print("Chargement du modèle et du pipeline depuis Hugging Face (lazy)...") hf_repository = os.getenv("HF_REPOSITORY") hf_model = os.getenv("HF_MODEL") hf_pipeline = os.getenv("HF_PIPELINE") if not all([hf_repository, hf_model, hf_pipeline]): raise RuntimeError( "Variables d'environnement HF manquantes (HF_REPOSITORY, HF_MODEL, HF_PIPELINE)" ) # Téléchargement et chargement du modèle model_path = hf_hub_download( repo_id=hf_repository, filename=hf_model ) model = joblib.load(model_path) print(f"Modèle chargé depuis: {model_path}") # Téléchargement et chargement de la pipeline pipeline_path = hf_hub_download( repo_id=hf_repository, filename=hf_pipeline ) pipeline = joblib.load(pipeline_path) print(f"Pipeline chargée depuis: {pipeline_path}") _model_loaded = True def load_test_data(): """Charge les données de test pour le profiling.""" data_dir = Path('./.data/extract') app_path = data_dir / 'test_sample_application.csv' bureau_path = data_dir / 'test_sample_bureau.csv' if not app_path.exists(): raise FileNotFoundError( f"Fichier {app_path} introuvable. " f"Placez vos données de test dans {data_dir}/" ) df_application = pd.read_csv(app_path) print(f"Données application chargées: {len(df_application)} lignes") # Bureau peut être vide if bureau_path.exists(): df_bureau = pd.read_csv(bureau_path) print(f"Données bureau chargées: {len(df_bureau)} lignes") else: df_bureau = pd.DataFrame() print("⚠ Pas de données bureau (fichier absent)") return df_application, df_bureau def simulate_api_prediction(df_application, df_bureau): """ Simule une requête API : lazy loading + prédiction. Args: df_application: DataFrame avec les données application (1 ligne) df_bureau: DataFrame avec les données bureau Returns: dict: Résultat de la prédiction avec temps d'inférence """ start = datetime.now() # Lazy loading (comme dans l'API) if model is None or pipeline is None: load_model() else: print("############################ modèle déjà chargé !!!! ##########") # Transformation via la pipeline (comme dans l'API) X_processed = pipeline.transform(df_application, df_bureau) # Prédiction pred = int(model.predict(X_processed)[0]) # Calcul de la probabilité if hasattr(model, "predict_proba"): proba = float(model.predict_proba(X_processed)[0][1]) elif hasattr(model, "decision_function"): decision = model.decision_function(X_processed)[0] proba = float(1 / (1 + np.exp(-decision))) else: proba = float(pred) elapsed_ms = (datetime.now() - start).total_seconds() * 1000 return { 'prediction': pred, 'probability': round(proba, 4), 'inference_time_ms': round(elapsed_ms, 2) } def run_profiling(df_application, df_bureau, n_iterations=100): """ Exécute n prédictions pour profiler les performances. Args: df_application: DataFrame avec les données application df_bureau: DataFrame avec les données bureau n_iterations: Nombre d'itérations pour le profiling """ print(f"\n🔄 Exécution de {n_iterations} prédictions pour profiling...") results = [] for i in range(n_iterations): result = simulate_api_prediction(df_application, df_bureau) result['iteration'] = i + 1 results.append(result) # Statistiques (en excluant la première itération qui inclut le chargement) times = [r['inference_time_ms'] for r in results[1:]] # Skip première itération print(f"\n Statistiques sur {len(times)} prédictions (hors chargement initial):") print(f" - Temps moyen: {np.mean(times):.2f} ms") print(f" - Temps médian: {np.median(times):.2f} ms") print(f" - Temps min: {np.min(times):.2f} ms") print(f" - Temps max: {np.max(times):.2f} ms") print(f" - Écart-type: {np.std(times):.2f} ms") # Temps de la première prédiction (avec chargement) first_time = results[0]['inference_time_ms'] print(f"\n⏱ Temps première prédiction (avec lazy loading): {first_time:.2f} ms") return results def main(): """Fonction principale encapsulant tout le profiling.""" print("=" * 80) print("PROFILING DE L'INFÉRENCE - Simulation du comportement de l'API") print("=" * 80) # 1. Chargement des données de test df_application, df_bureau = load_test_data() # 2. Exécution des prédictions (à profiler) # Note: Le lazy loading se fera à la première itération results = run_profiling( df_application=df_application, df_bureau=df_bureau, n_iterations=100 ) print("\n Profiling terminé avec succès") if __name__ == "__main__": # Créer le dossier de sortie pour les résultats de profiling Path('.data/profiling').mkdir(parents=True, exist_ok=True) # Activer le profiler profiler = cProfile.Profile() profiler.enable() # Exécuter le profiling main() # Désactiver le profiler profiler.disable() # Afficher les résultats dans la console s = io.StringIO() ps = pstats.Stats(profiler, stream=s).sort_stats("cumulative") ps.print_stats(50) # Top 50 fonctions les plus coûteuses print("\n" + "=" * 80) print("PROFILING RESULTS - Top 50 fonctions par temps cumulé") print("=" * 80) print(s.getvalue()) # Sauvegarder le fichier .prof prof_file = '.data/profiling/predict_inference.prof' profiler.dump_stats(prof_file) print(f"\n Fichier de profiling sauvegardé : {prof_file}")