classical_ml_utils.py

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)