app.py

import streamlit as st
import seaborn as sns
import pandas as pd
import numpy as np

from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.linear_model import LinearRegression, LogisticRegression
from sklearn.feature_selection import mutual_info_regression, mutual_info_classif
from sklearn.metrics import r2_score, accuracy_score
from sklearn.model_selection import train_test_split
#from scipy.stats import pearsonr
from scipy.stats import spearmanr

# ------------------------------------------------------------
# Configuration Globale
# ------------------------------------------------------------
TEST_SIZE = 0.3
RANDOM_STATE = 42

st.set_page_config(page_title="Analyse d'importance", layout="wide")

st.title("🔍 Analyse de l'importance des caractéristiques")
st.markdown(
    """
Cette application illustre la différence entre la pertinence marginale et la pertinence conditionnelle d'une caractéristique.

- Pertinence marginale : corrélation ou information mutuelle avec la cible.
- Pertinence conditionnelle : valeur ajoutée d'une variable excluant les redondances après contrôle.
"""
)

# ------------------------------------------------------------
# Sidebar: Dataset et Importation
# ------------------------------------------------------------
with st.sidebar:
    st.header("⚙️ Configuration")
    
    # Choix de la source de données
    data_source = st.radio(
        "Source des données",
        ["Jeu de données Seaborn", "Importer un fichier"],
        label_visibility="visible"
    )
    
    df = None
    
    if data_source == "Importer un fichier":
        uploaded_file = st.file_uploader("Importer un fichier CSV", type=["csv"])
        
        if uploaded_file is not None:
            try:
                df = pd.read_csv(uploaded_file, sep=None, engine='python')
                
                # Seuil de valeurs manquantes (configurable)
                missing_threshold = st.slider(
                    "Seuil max de valeurs manquantes (%)", 
                    min_value=0, 
                    max_value=100, 
                    value=50,
                    help="Les colonnes avec plus de X% de valeurs manquantes seront supprimées"
                )
                
                # Calcul du pourcentage de valeurs manquantes par colonne
                missing_pct = (df.isnull().sum() / len(df)) * 100
                cols_to_drop = missing_pct[missing_pct > missing_threshold].index.tolist()
                
                if cols_to_drop:
                    st.info(f"ℹ️ {len(cols_to_drop)} colonne(s) supprimée(s) (>{missing_threshold}% manquantes) : {', '.join(cols_to_drop)}")
                    df = df.drop(columns=cols_to_drop)
                
                # Suppression des lignes avec valeurs manquantes restantes
                df = df.dropna()
                
                if len(df) == 0:
                    st.error("❌ Aucune donnée après nettoyage. Essayez d'augmenter le seuil de valeurs manquantes.")
                    df = None
                else:
                    st.success(f"✅ Fichier CSV chargé ! ({len(df)} lignes, {len(df.columns)} colonnes)")
            except Exception as e:
                st.error(f"Erreur : {e}")
                df = None
    else:
        excluded_datasets = ['anagrams', 'anscombe', 'attention', 'brain_networks',
                             'car_crashes', 'dowjones','diamonds','flights','geyser',
                             'planets','seaice']
        available_datasets = [d for d in sorted(sns.get_dataset_names()) if d not in excluded_datasets]
        default_dataset = "iris"
        default_index = available_datasets.index(default_dataset) if default_dataset in available_datasets else 0
        dataset_name = st.selectbox(
            "Dataset d'exemple",
            available_datasets,
            index=default_index
        )
        #dataset_name = st.selectbox("Dataset d'exemple", available_datasets)
        try:
            df = sns.load_dataset(dataset_name)
            df = df.dropna()
            st.success(f"✅ Jeu '{dataset_name}' chargé")
        except Exception as e:
            st.error(f"Erreur : {e}")
            df = None

    if df is not None:
        target = st.selectbox("Sélection cible (Y)", df.columns)
        y = df[target]
        X = df.drop(columns=[target])
        
        # Vérification que X n'est pas vide après suppression de la cible
        if len(X.columns) == 0:
            st.warning("⚠️ Aucune variable disponible après sélection de la cible.")
            X = None
            y = None
            task = None
        else:
            task = "Regression" if (y.dtype.kind in "ifu" and y.nunique() > 10) else "Classification"
            excluded_features = st.multiselect("Variables à exclure :", X.columns.tolist(), default=[])
            
            if excluded_features:
                X = X.drop(columns=excluded_features)
                
            # Vérification après exclusion
            if len(X.columns) == 0:
                st.error("❌ Vous avez exclu toutes les variables ! Veuillez en garder au moins une.")
                X = None
                y = None
                task = None
    else:
        st.info("👈 Veuillez sélectionner ou importer un jeu de données.")
        X = None
        y = None
        task = None

# ------------------------------------------------------------
# Onglets
# ------------------------------------------------------------
if df is not None and X is not None and len(X.columns) > 0:
    tab1, tab2, tab3 = st.tabs(["📊 Analyse d'Importance", "📋 Données Brutes", "🔧 Types"])

    with tab2:
        st.dataframe(df.head(20), use_container_width=True)

    with tab3:
        st.header("Types des variables")
        
        num_cols = X.select_dtypes(include=[np.number]).columns.tolist()
        cat_cols = X.select_dtypes(exclude=[np.number]).columns.tolist()
        
        col1, col2 = st.columns(2)
        
        with col1:
            st.subheader("Numériques")
            for col in num_cols or ["None"]:
                st.write(f"- {col}")
        
        with col2:
            st.subheader("Catégorielles")
            for col in cat_cols or ["None"]:
                st.write(f"- {col}")
    
    # ------------------------------------------------------------
    # Analyse Principale (Tab 1)
    # ------------------------------------------------------------
    with tab1:
        if len(X.columns) > 0:
            try:
                num_cols = X.select_dtypes(include=[np.number]).columns.tolist()
                cat_cols = X.select_dtypes(exclude=[np.number]).columns.tolist()
                
                # Vérification qu'il y a au moins une variable
                if len(num_cols) == 0 and len(cat_cols) == 0:
                    st.warning("⚠️ Aucune variable disponible pour l'analyse. Veuillez ne pas tout exclure.")
                    st.stop()
                
                # Construction du préprocesseur seulement avec les colonnes qui existent
                transformers = []
                if num_cols:
                    transformers.append(("num", StandardScaler(), num_cols))
                if cat_cols:
                    transformers.append(("cat", OneHotEncoder(drop="first", handle_unknown="ignore", sparse_output=False), cat_cols))
                
                if not transformers:
                    st.warning("⚠️ Aucune colonne à traiter.")
                    st.stop()
                
                preprocess = ColumnTransformer(transformers=transformers)
                
                X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=TEST_SIZE, random_state=RANDOM_STATE)
                
                # Vérification qu'il y a assez de données pour le split
                if len(X_train) == 0 or len(X_test) == 0:
                    st.error("❌ Pas assez de données pour créer les ensembles d'entraînement et de test.")
                    st.info(f"Données disponibles : {len(X)} lignes. Minimum requis : 2 lignes.")
                    st.stop()
                
                X_train_proc = preprocess.fit_transform(X_train)
                
                # Vérification que les données transformées ne sont pas vides
                if X_train_proc.shape[0] == 0 or X_train_proc.shape[1] == 0:
                    st.error("❌ Erreur : Les données transformées sont vides.")
                    st.info(f"Shape après transformation : {X_train_proc.shape}")
                    st.info(f"Variables numériques : {num_cols}")
                    st.info(f"Variables catégorielles : {cat_cols}")
                    st.stop()
                
                feature_names = preprocess.get_feature_names_out()
                
                model = LinearRegression() if task == "Regression" else LogisticRegression(max_iter=1000)
                model.fit(X_train_proc, y_train)
                
                y_pred = model.predict(preprocess.transform(X_test))
                perf = r2_score(y_test, y_pred) if task == "Regression" else accuracy_score(y_test, y_pred)
                
                st.subheader("📊 Pertinence marginale vs conditionnelle")
                st.markdown(f"**🎯 Performance globale : {perf:.2f} ({'R²' if task == 'Regression' else 'Précision'})**")

                # Métriques
                mi = mutual_info_regression(X_train_proc, y_train, random_state=0) if task == "Regression" else mutual_info_classif(X_train_proc, y_train, random_state=0)
                coefs = model.coef_.ravel() if task == "Regression" else model.coef_[0]
                
                res = pd.DataFrame({
                    "Variable": feature_names,
                    "Importance seule (MI)": mi,
                    "Poids dans le modèle": np.abs(coefs),
                    "Sens": np.where(coefs > 0, "+", "-")
                })
                
                #if task == "Regression":
                #    res["Lien direct (Corr)"] = [pearsonr(X_train_proc[:, i], y_train)[0] for i in range(len(feature_names))]

                if task == "Regression":
                    res["Lien direct (Corr)"] = [spearmanr(X_train_proc[:, i], y_train)[0] for i in range(len(feature_names))]
                
                # Normalisation pour Score Synthétique
                def normalize(s): return (s - s.min()) / (s.max() - s.min() + 1e-10)
                mi_n = normalize(res["Importance seule (MI)"])
                poids_n = normalize(res["Poids dans le modèle"])



                if task == "Regression":
                    corr_n = normalize(res["Lien direct (Corr)"].abs())
                    res["Score synthétique"] = ((mi_n + corr_n) / 2 + poids_n) / 2
                else:
                    res["Score synthétique"] = (mi_n + poids_n) / 2

                res = res.sort_values("Score synthétique", ascending=False)

                # Réorganisation des colonnes
                cols = ["Variable", "Score synthétique", "Importance seule (MI)", "Poids dans le modèle", "Sens"]
                if task == "Regression":
                    cols = ["Variable", "Score synthétique", "Importance seule (MI)", "Lien direct (Corr)", "Poids dans le modèle", "Sens"]
                
                final_df = res[cols].copy()

                # --- STYLISATION ET AFFICHAGE ---
                # 1. Préparation du style pour la colonne Sens (couleurs)
                def style_sign(val):
                    color = 'color: #2ecc71;' if val == '+' else 'color: #e74c3c;'
                    return f'{color} font-weight: bold; font-size: 20px;'

                # 2. Application du formatage (2 décimales) et des gradients
                num_cols_to_style = [c for c in cols if c not in ["Variable", "Sens", "Score synthétique"]]
                
                styled_res = (final_df.style
                    .format({c: "{:.2f}" for c in cols if c not in ["Variable", "Sens"]})
                    .background_gradient(subset=num_cols_to_style, cmap="RdYlGn")
                    .map(style_sign, subset=['Sens'])
                )

                # 3. Affichage avec st.data_editor pour fixer la hauteur (6 lignes env = 250px)
                st.data_editor(
                    styled_res,
                    use_container_width=True,
                    height=250, # Limite la hauteur avec scrollbar
                    hide_index=True,
                    disabled=True, # Empêche l'édition, agit comme un dataframe
                    column_config={
                        "Sens": st.column_config.Column(
                            "Sens",
                            help="Direction de l'influence",
                            width="small"
                        )
                    }
                )

                st.subheader("📖 Guide de lecture")
                st.markdown(
                            """
                - **Score synthétique** : Note globale d'importance.
                - **Importance seule (MI)** : Mesure la dépendance globale entre la variable et la cible. Contrairement à la corrélation qui ne voit que les lignes droites, l'Information Mutuelle détecte toutes les formes de relations (courbes, motifs complexes, etc.). Elle indique quelle quantité d'information "pure" cette variable partage avec la cible, sans tenir compte des autres variables.
                - **Poids dans le modèle** : Contribution finale au modèle.
                - **Sens (+) / (-)** : Direction de l'impact sur la cible.
                            """
                        )
            
            except ValueError as e:
                if "Found array with 0 sample(s)" in str(e) or "shape=(0," in str(e):
                    st.error("❌ Erreur d'analyse : données insuffisantes ou incompatibles")
                    st.warning("⚠️ Vérifiez que :")
                    st.markdown("""
                    - Vous n'avez pas exclu toutes les variables
                    - La variable cible choisie est appropriée (elle ne doit pas être identique à une variable prédictive)
                    - Il reste suffisamment de données après nettoyage
                    - Les variables ont suffisamment de variance
                    """)
                else:
                    st.error(f"❌ Erreur : {str(e)}")
            
            except Exception as e:
                st.error(f"❌ Une erreur s'est produite lors de l'analyse")
                st.warning(f"Détails : {str(e)}")
                st.info("💡 Essayez de changer de variable cible ou de variables prédictives.")
                
        else:
            st.info("ℹ️ Veuillez sélectionner au moins une variable.")
else:
    st.info("👈 Veuillez sélectionner ou importer un jeu de données pour commencer l'analyse.")