import json import os from collections import Counter from datetime import datetime from typing import List import joblib from sklearn.ensemble import RandomForestClassifier from sklearn.linear_model import LogisticRegression from sklearn.model_selection import GridSearchCV, StratifiedKFold from sklearn.neighbors import KNeighborsClassifier from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler from sklearn.svm import SVC from config import CV_FOLDS, session_model_dir, session_meta_dir from metrics_utils import compute_classification_metrics, save_confusion_matrix_figure CLF_TYPE_MAP = { "SVM": "svm", "Régression logistique": "logreg", "k-NN": "knn", "Forêt aléatoire": "rf", } # Grille de recherche par algorithme : le SEUL hyperparamètre exposé dans # l'UI est aussi celui balayé par la validation croisée — c'est ce balayage # (et non le score de stabilité d'une valeur choisie à la main) qui constitue # le véritable rôle pédagogique de la CV : sélectionner le meilleur # hyperparamètre plutôt que de simplement le deviner. PARAM_GRIDS = { "svm": ("C", [0.01, 0.1, 1.0, 10.0, 100.0]), "logreg": ("C", [0.01, 0.1, 1.0, 10.0, 100.0]), "knn": ("n_neighbors", [1, 3, 5, 7, 9, 15]), "rf": ("n_estimators", [50, 100, 200, 300]), } def classifier_path(model_name: str, session_id: str) -> str: return os.path.join(session_model_dir(session_id), f"{model_name}.joblib") def meta_path(model_name: str, session_id: str) -> str: return os.path.join(session_meta_dir(session_id), f"{model_name}.json") def build_pipeline(clf_type: str, **params) -> Pipeline: key = CLF_TYPE_MAP.get(clf_type, clf_type) if key == "svm": clf = SVC( C=params.get("C", 1.0), kernel=params.get("kernel", "rbf"), gamma=params.get("gamma", "scale"), probability=True, random_state=42, ) elif key == "logreg": clf = LogisticRegression( C=params.get("C", 1.0), max_iter=params.get("max_iter", 1000), random_state=42, ) elif key == "knn": clf = KNeighborsClassifier( n_neighbors=params.get("n_neighbors", 5), metric=params.get("metric", "euclidean"), ) elif key == "rf": clf = RandomForestClassifier( n_estimators=params.get("n_estimators", 100), max_depth=None, random_state=42, n_jobs=-1, ) else: raise ValueError(f"Classifieur inconnu : {clf_type}") return Pipeline([("scaler", StandardScaler()), ("clf", clf)]) def _grid_search_cv(clf_type: str, X_train, y_train, manual_value, **other_params): """Recherche par grille en validation croisée : le véritable rôle de la CV n'est pas de noter la valeur choisie à la main, mais de comparer plusieurs valeurs candidates et de sélectionner celle qui généralise le mieux. La valeur choisie manuellement par l'étudiant·e est incluse dans la grille pour qu'il/elle puisse comparer son choix au choix retenu par CV. """ key = CLF_TYPE_MAP.get(clf_type, clf_type) hp_name, default_grid = PARAM_GRIDS[key] candidates = sorted(set(default_grid) | ({manual_value} if manual_value is not None else set())) min_class_count = min(Counter(y_train.tolist()).values()) folds = max(2, min(CV_FOLDS, min_class_count)) skf = StratifiedKFold(n_splits=folds, shuffle=True, random_state=42) pipeline = build_pipeline(clf_type, **{hp_name: candidates[0], **other_params}) param_grid = {f"clf__{hp_name}": candidates} search = GridSearchCV( pipeline, param_grid=param_grid, cv=skf, scoring="f1_macro", refit=True, n_jobs=-1 ) search.fit(X_train, y_train) grid_results = [] for candidate_params, mean_score, std_score in zip( search.cv_results_["params"], search.cv_results_["mean_test_score"], search.cv_results_["std_test_score"], ): grid_results.append({ hp_name: candidate_params[f"clf__{hp_name}"], "cv_f1_macro_mean": round(float(mean_score), 4), "cv_f1_macro_std": round(float(std_score), 4), }) grid_results.sort(key=lambda r: r[hp_name]) cv_metrics = { "cv_folds": folds, "hyperparameter_name": hp_name, "manual_value": manual_value, "cv_best_value": search.best_params_[f"clf__{hp_name}"], "cv_best_f1_macro": round(float(search.best_score_), 4), "cv_grid_results": grid_results, } return cv_metrics, search.best_estimator_ def train_classical_model( clf_type: str, features_cache: dict, class_names: List[str], session_id: str, model_tag: str = "", use_cv: bool = False, **params, ): X_train = features_cache["train"]["X"] y_train = features_cache["train"]["y"] X_test = features_cache["test"]["X"] y_test = features_cache["test"]["y"] key = CLF_TYPE_MAP.get(clf_type, clf_type) if use_cv: hp_name, _ = PARAM_GRIDS[key] manual_value = params.get(hp_name) other_params = {k: v for k, v in params.items() if k != hp_name} cv_metrics, pipeline = _grid_search_cv(clf_type, X_train, y_train, manual_value, **other_params) # Le modèle sauvegardé utilise l'hyperparamètre sélectionné par CV, # pas nécessairement celui choisi à la main dans l'UI. params = {**params, hp_name: cv_metrics["cv_best_value"]} else: cv_metrics = None pipeline = build_pipeline(clf_type, **params) pipeline.fit(X_train, y_train) y_pred = pipeline.predict(X_test) metrics = compute_classification_metrics(y_test.tolist(), y_pred.tolist(), class_names) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") safe_tag = model_tag.strip().replace(" ", "_") if model_tag.strip() else CLF_TYPE_MAP.get(clf_type, "clf") model_name = f"{safe_tag}_{timestamp}" joblib.dump(pipeline, classifier_path(model_name, session_id)) cm_path = save_confusion_matrix_figure(metrics["confusion_matrix"], model_name) config_dict = { "model_type": CLF_TYPE_MAP.get(clf_type, clf_type), "clf_type_label": clf_type, "class_names": class_names, "num_classes": len(class_names), **{k: v for k, v in params.items() if v is not None}, } training_summary = { "test_accuracy": metrics["accuracy"], "test_f1_macro": metrics["f1_macro"], "test_f1_weighted": metrics["f1_weighted"], "train_samples": int(len(X_train)), "test_samples": int(len(X_test)), **(cv_metrics or {}), } with open(meta_path(model_name, session_id), "w", encoding="utf-8") as f: json.dump( { "model_name": model_name, "config": config_dict, "training_summary": training_summary, "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }, f, indent=2, ensure_ascii=False, ) return { "model_name": model_name, "summary": training_summary, "classification_report": metrics["classification_report"], "confusion_matrix": metrics["confusion_matrix"], "confusion_matrix_path": cm_path, } def load_classical_pipeline(model_name: str, session_id: str) -> Pipeline: path = classifier_path(model_name, session_id) if not os.path.exists(path): raise FileNotFoundError(f"Classifieur introuvable : {model_name}") return joblib.load(path)