Maîtriser les Opérations Avancées de Pandas pour le Traitement de Données en Production

1. Gestion Avancée des Données Manquantes et Imputations

Définition

La gestion des données manquantes est l'art de traiter les valeurs NaN, NULL ou vides dans un DataFrame de manière stratégique. Ce n'est pas simplement supprimer les lignes défectueuses, mais choisir la meilleure stratégie d'imputation selon le contexte métier et les caractéristiques des données.

Explication Détaillée

En production, les données manquantes sont inévitables. Elles proviennent de sources défaillantes, d'erreurs de collecte, ou de processus métier incomplets. Ignorer ce problème compromet la qualité des modèles de machine learning. Les stratégies d'imputation varient selon le type de données (numériques vs catégoriques), la proportion de valeurs manquantes, et l'impact business attendu. Une mauvaise imputation introduit des biais systématiques qui faussent les analyses ultérieures.

Bloc de Code

import pandas as pd
import numpy as np
from sklearn.impute import SimpleImputer, KNNImputer

# Création d'un dataset avec valeurs manquantes
df = pd.DataFrame({
    'age': [25, np.nan, 35, 28, np.nan, 42],
    'salaire': [50000, 60000, np.nan, 55000, 65000, np.nan],
    'departement': ['IT', 'HR', 'IT', np.nan, 'Finance', 'IT'],
    'experience': [2, 5, np.nan, 3, 7, 4]
})

print("Dataset original:")
print(df)
print("\nPourcentage de données manquantes par colonne:")
print((df.isnull().sum() / len(df) * 100).round(2))

# Stratégie 1: Suppression conditionnelle
df_drop = df.dropna(subset=['age', 'salaire'], how='any')
print("\nAprès suppression des lignes avec NaN dans age/salaire:")
print(df_drop)

# Stratégie 2: Imputation par la médiane (données numériques)
imputer_numeric = SimpleImputer(strategy='median')
df['age'] = imputer_numeric.fit_transform(df[['age']])
print("\nAprès imputation médiane pour l'âge:")
print(df['age'])

# Stratégie 3: Imputation par la mode (données catégoriques)
df['departement'].fillna(df['departement'].mode()[0], inplace=True)
print("\nAprès imputation mode pour département:")
print(df['departement'])

# Stratégie 4: Imputation KNN (utilise la similarité avec voisins)
knn_imputer = KNNImputer(n_neighbors=3)
numeric_cols = ['age', 'salaire', 'experience']
df[numeric_cols] = knn_imputer.fit_transform(df[numeric_cols])
print("\nAprès imputation KNN pour variables numériques:")
print(df[numeric_cols])

# Stratégie 5: Créer un indicateur de valeur imputée
df['salaire_impute'] = df['salaire'].isnull()
df['salaire'].fillna(df['salaire'].mean(), inplace=True)
print("\nDataFrame final avec indicateur d'imputation:")
print(df)

Tableau Comparatif des Stratégies

Astuce Professionnelle

Toujours créer un indicateur binaire (_is_imputed) avant l'imputation pour tracer quelles données étaient manquantes. Cela permet au modèle de capturer si la missingness elle-même est informative. En production, cette colonne devient précieuse pour auditer et déboguer.

⚠️ Attention Critique

Ne jamais imputer les données manquantes AVANT de splitter train/test. L'imputation doit apprendre des paramètres (médiane, mode, voisins) uniquement sur l'ensemble d'entraînement, puis appliquer ces paramètres au test. Sinon, vous causez une fuite de données (data leakage) qui invalide vos métriques.

2. Groupby, Agrégations et Transformations Complexes

Définition

L'opération groupby segmente un DataFrame en groupes selon une ou plusieurs colonnes, puis applique des fonctions d'agrégation ou de transformation. C'est le cœur des analyses exploratoires et des feature engineering en data science.

Explication Détaillée

groupby est essentiellement un split-apply-combine: vous divisez les données en groupes logiques, appliquez une opération à chaque groupe, puis consolidez les résultats. Contrairement aux simples filtres, groupby permet des analyses par segment, des calculs de statistiques par groupe, et la création de features dépendantes du contexte. Maîtriser les différences entre agg, transform, et apply est crucial pour écrire du code Pandas efficace et lisible.

Bloc de Code

import pandas as pd
import numpy as np

# Dataset de ventes réalistes
df_ventes = pd.DataFrame({
    'date': pd.date_range('2023-01-01', periods=100, freq='D'),
    'vendeur': np.random.choice(['Alice', 'Bob', 'Charlie', 'David'], 100),
    'produit': np.random.choice(['Laptop', 'Phone', 'Tablet'], 100),
    'ventes': np.random.randint(1000, 10000, 100),
    'region': np.random.choice(['Nord', 'Sud', 'Est', 'Ouest'], 100)
})

print("Dataset original (5 premières lignes):")
print(df_ventes.head())

# Agrégation simple: total des ventes par vendeur
print("\n=== AGREGATION SIMPLE ===")
ventes_par_vendeur = df_ventes.groupby('vendeur')['ventes'].sum()
print("Total des ventes par vendeur:")
print(ventes_par_vendeur)

# Agrégations multiples
print("\n=== AGREGATIONS MULTIPLES ===")
statistiques = df_ventes.groupby('vendeur')['ventes'].agg([
    ('total', 'sum'),
    ('moyenne', 'mean'),
    ('min', 'min'),
    ('max', 'max'),
    ('count', 'count'),
    ('std', 'std')
])
print("Statistiques complètes par vendeur:")
print(statistiques)

# Groupby multi-colonnes
print("\n=== GROUPBY MULTI-COLONNES ===")
ventes_produit_region = df_ventes.groupby(['produit', 'region'])['ventes'].agg({
    'total': 'sum',
    'nombre_transactions': 'count',
    'moyenne': 'mean'
}).reset_index()
print("Ventes par produit et région:")
print(ventes_produit_region)

# TRANSFORM vs AGG (différence cruciale)
print("\n=== TRANSFORM vs AGG ===")
# AGG: retourne une valeur par groupe
agg_result = df_ventes.groupby('vendeur')['ventes'].agg('mean')
print("AGG - Moyenne par vendeur (1 ligne par groupe):")
print(agg_result)
print(f"Shape: {agg_result.shape}")

# TRANSFORM: retourne une valeur pour chaque ligne originale
df_ventes['moyenne_vendeur'] = df_ventes.groupby('vendeur')['ventes'].transform('mean')
print("\nTRANSFORM - Moyenne répliquée (même nombre de lignes):")
print(df_ventes[['vendeur', 'ventes', 'moyenne_vendeur']].head(10))
print(f"Shape: {df_ventes[['moyenne_vendeur']].shape}")

# Custom aggregation avec lambda
print("\n=== CUSTOM AGGREGATION ===")
custom_agg = df_ventes.groupby('vendeur').agg({
    'ventes': [
        ('coeff_var', lambda x: x.std() / x.mean()),  # Coefficient de variation
        ('percentile_75', lambda x: x.quantile(0.75)),
        ('range', lambda x: x.max() - x.min())
    ]
})
print("Agrégations personnalisées:")
print(custom_agg)

# Groupby avec filtrage: sales > 5000
print("\n=== GROUPBY AVEC FILTRAGE ===")
vendeurs_hauts_revenus = df_ventes.groupby('vendeur')['ventes'].sum()
vendeurs_filtres = vendeurs_hauts_revenus[vendeurs_hauts_revenus > 200000]
print("Vendeurs avec ventes > 200000:")
print(vendeurs_filtres)

# Ranking au sein des groupes
print("\n=== RANKING DANS LES GROUPES ===")
df_ventes['rang_ventes'] = df_ventes.groupby('vendeur')['ventes'].rank(ascending=False)
print("Ranking des ventes par vendeur:")
print(df_ventes[['vendeur', 'ventes', 'rang_ventes']].head(15))

# Apply pour logique complexe
print("\n=== APPLY POUR LOGIQUE COMPLEXE ===")
def categorie_performance(groupe):
    moyenne = groupe['ventes'].mean()
    if moyenne > 5000:
        return 'Top Performer'
    elif moyenne > 3000:
        return 'Standard'
    else:
        return 'À Développer'

df_ventes['performance'] = df_ventes.groupby('vendeur').apply(
    lambda x: pd.Series([categorie_performance(x)] * len(x))
).values
print("Catégorisation de performance:")
print(df_ventes[['vendeur', 'ventes', 'performance']].drop_duplicates(subset=['vendeur']))

Tableau des Opérations Groupby

Astuce Professionnelle

Pour déboguer un groupby complexe, utilisez groupby().head(2) ou groupby().get_group('nom_groupe') pour inspecter rapidement un groupe. En production, nommez explicitement vos agrégations avec un dictionnaire au lieu de tuples: {'ventes': ['sum', 'mean']} plutôt que simples listes.

⚠️ Attention Critique

L'ordre des lignes n'est pas garanti après groupby() + agg() en versions anciennes de Pandas. Toujours utiliser reset_index() ou sort_values() si l'ordre importe. Aussi, attention à la NaN-semantics: par défaut, les NaN ne participent pas aux groupes (utilisez dropna=False pour les inclure).

3. Fusion et Jointures Multi-Tables Optimisées

Définition

Les fusions (merge, join) combinent deux ou plusieurs DataFrames selon une clé commune. C'est l'opération fondamentale pour enrichir des données provenant de sources multiples, un scénario très courant en data engineering.

Explication Détaillée

En pratique, vos données proviennent rarement d'une seule source. Vous devez fusionner client_data, transaction_data, product_catalog, etc. Les jointures SQL dans Pandas offrent quatre stratégies (inner, outer, left, right) avec des implications importantes sur la taille du résultat et les valeurs NaN introduites. Comprendre quand utiliser quelle jointure, gérer les clés dupliquées, et optimiser les performances avec merge_ordered ou merge_asof sont des compétences critiques.

Bloc de Code

import pandas as pd
import numpy as np

# Création de datasets à fusionner
clients = pd.DataFrame({
    'client_id': [1, 2, 3, 4, 5],
    'nom': ['Alice', 'Bob', 'Charlie', 'David', 'Eve'],
    'segment': ['Premium', 'Standard', 'Premium', 'Standard', 'Premium']
})

commandes = pd.DataFrame({
    'commande_id': [101, 102, 103, 104, 105, 106],
    'client_id': [1, 1, 2, 3, 3, 6],  # Note: 6 n'existe pas dans clients
    'montant': [500, 300, 800, 1200, 450, 600],
    'date': pd.date_range('2023-01-01', periods=6, freq='M')
})

produits = pd.DataFrame({
    'commande_id': [101, 102, 103, 104, 105],
    'produit': ['Laptop', 'Mouse', 'Monitor', 'Keyboard', 'Monitor'],
    'prix_unitaire': [800, 20, 300, 50, 300]
})

print("=== DATASETS ORIGINAUX ===")
print("Clients:\n", clients)
print("\nCommandes:\n", commandes)
print("\nProduits:\n", produits)

# MERGE INNER (intersection)
print("\n=== MERGE INNER ===")
result_inner = pd.merge(clients, commandes, on='client_id', how='inner')
print("Clients avec au moins 1 commande:")
print(result_inner)
print(f"Lignes: {len(result_inner)} (client_id 6 éliminé)")

# MERGE LEFT (tous les clients, même sans commande)
print("\n=== MERGE LEFT ===")
result_left = pd.merge(clients, commandes, on='client_id', how='left')
print("Tous les clients avec leurs commandes (NaN si aucune):")
print(result_left)
print(f"Lignes: {len(result_left)}")

# MERGE OUTER (union)
print("\n=== MERGE OUTER ===")
result_outer = pd.merge(clients, commandes, on='client_id', how='outer')
print("Tous les enregistrements (clients ET commandes):")
print(result_outer)
print(f"Lignes: {len(result_outer)}")

# Merge sur plusieurs clés
print("\n=== MERGE SUR PLUSIEURS CLÉS ===")
result_multi = pd.merge(
    commandes, 
    produits, 
    on='commande_id', 
    how='left'
)
print("Commandes avec détails produits:")
print(result_multi)

# Fusion multi-tables (chaînée)
print("\n=== FUSION MULTI-TABLES ===")
# D'abord clients + commandes, puis ajouter produits
merged = pd.merge(clients, commandes, on='client_id', how='left')
merged = pd.merge(merged, produits, on='commande_id', how='left')
print("Vue consolidée (clients + commandes + produits):")
print(merged[['nom', 'segment', 'montant', 'produit']].head(10))

# Suffixes pour colonnes dupliquées
print("\n=== GESTION DES COLONNES DUPLIQUÉES ===")
meta1 = pd.DataFrame({
    'id': [1, 2, 3],
    'prix': [100, 200, 300],
    'qualite': ['A', 'B', 'A']
})

meta2 = pd.DataFrame({
    'id': [1, 2, 3],
    'prix': [95, 210, 290],  # Colonne prix dans les deux
    'origine': ['China', 'USA', 'Germany']
})

result_suffix = pd.merge(meta1, meta2, on='id', suffixes=('_fournisseur1', '_fournisseur2'))
print("Fusion avec suffixes (colonnes prix dupliquées):")
print(result_suffix)

# MERGE_ASOF (fusion sur condition de proximité - timeseries!)
print("\n=== MERGE_ASOF (pour séries temporelles) ===")
trades = pd.DataFrame({
    'timestamp': pd.to_datetime(['2023-01-01 09:00', '2023-01-01 10:30', '2023-01-01 11:45']),
    'prix_achat': [100, 102, 101]
}).sort_values('timestamp')

quotes = pd.DataFrame({
    'timestamp': pd.to_datetime(['2023-01-01 09:30', '2023-01-01 10:00', '2023-01-01 11:00', '2023-01-01 12:00']),
    'prix_marche': [99, 101, 102, 100]
}).sort_values('timestamp')

result_asof = pd.merge_asof(trades, quotes, on='timestamp', direction='backward')
print("Achats appairés avec prix de marché le plus proche (antérieur):")
print(result_asof)

# Index-based join
print("\n=== JOIN PAR INDEX ===")
serie1 = pd.Series([10, 20, 30], index=['A', 'B', 'C'], name='valeur1')
serie2 = pd.Series([100, 200, 300], index=['A', 'B', 'C'], name='valeur2')
df_join = pd.concat([serie1, serie2], axis=1)
print("Concaténation par index (join implicite):")
print(df_join)

# Performance: merge_ordered
print("\n=== MERGE_ORDERED (fusion et tri) ===")
result_ordered = pd.merge_ordered(
    clients[['client_id', 'nom']], 
    commandes[['client_id', 'montant']], 
    on='client_id'
)
print("Fusion ordonnée (résultat trié):")
print(result_ordered)

Tableau des Types de Jointures

Astuce Professionnelle

Avant une fusion, nettoyez toujours les clés (strip whitespace, lowercase, types cohérents). Utilisez .merge(..., validate='m:1') ou '1:1' pour vérifier que la cardinalité est celle attendue—cela capture les bugs silencieux comme des clés dupliquées non-intentionnelles. Pour les grosses fusions, considérez pd.merge(..., indicator=True) pour tracer quelles lignes proviennent de quelle source.

⚠️ Attention Critique

Les fusions peuvent exploser en taille! Un inner join sur deux tables de 1M lignes avec clés mal définis peut créer un Cartésien accidentel (milliards de lignes). Vérifiez les doublons dans les clés avec .duplicated() avant toute fusion. Aussi, attention aux types: un client_id en int64 vs string ne matchera jamais—convertissez explicitement.

4. Optimisation des Performances et Gestion de la Mémoire

Définition

L'optimisation en Pandas concerne réduire l'empreinte mémoire et accélérer les opérations sur gros volumes. C'est essentiel quand vous travaillez avec des fichiers multi-gigaoctets ou des pipelines production à latence critique.

Explication Détaillée

Par défaut, Pandas charge tout en RAM. Sur un dataset de 10 GB, c'est irréaliste. Les stratégies incluent: réduire les types de données (int64 → int32, float64 → float32), utiliser les catégories pour colonnes répétitives, charger incrementalement avec chunksize, utiliser des formats efficaces (parquet vs CSV), et paralléliser avec Dask. Connaître le profil mémoire de votre DataFrame et identifier les goulots d'étranglement est une compétence data science/engineering critique.

Bloc de Code

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# Création d'un dataset "gros" pour la démo
print("=== DIAGNOSTIC MÉMOIRE ===")
df_large = pd.DataFrame({
    'id': range(100000),
    'timestamp': [datetime.now() - timedelta(seconds=i) for i in range(100000)],
    'valeur': np.random.randn(100000),
    'categorie': np.random.choice(['A', 'B', 'C', 'D', 'E'], 100000),
    'description': ['texte_' + str(i) for i in range(100000)]
})

# Inspection mémoire avant optimisation
print("Mémoire utilisée AVANT optimisation:")
print(df_large.memory_usage(deep=True))
print(f"\nTotal: {df_large.memory_usage(deep=True).sum() / 1024**2:.2f} MB")

# Analyse par colonne
print("\n=== ANALYSE PAR COLONNE ===")
for col in df_large.columns:
    dtype = df_large[col].dtype
    mem = df_large[col].memory_usage(deep=True) / 1024
    print(f"{col:15} | Dtype: {str(dtype):15} | Mémoire: {mem:8.2f} KB")

# OPTIMISATION 1: Réduire les types numériques
print("\n=== OPTIMISATION 1: RÉDUCTION TYPES NUMÉRIQUES ===")
df_opt = df_large.copy()

# int64 → int32 (réduit de 50% si possible)
if df_opt['id'].min() >= 0 and df_opt['id'].max() < 2**31 - 1:
    df_opt['id'] = df_opt['id'].astype('int32')
    
# float64 → float32 (réduit de 50%)
df_opt['valeur'] = df_opt['valeur'].astype('float32')

print("Avant: id=int64, valeur=float64")
print("Après: id=int32, valeur=float32")
print(f"Mémoire économisée: {(df_large.memory_usage(deep=True).sum() - df_opt.memory_usage(deep=True).sum()) / 1024:.2f} KB")

# OPTIMISATION 2: Catégories pour colonnes répétitives
print("\n=== OPTIMISATION 2: CATÉGORIES ===")
df_opt['categorie'] = df_opt['categorie'].astype('category')

mem_avant = df_large['categorie'].memory_usage(deep=True) / 1024
mem_apres = df_opt['categorie'].memory_usage(deep=True) / 1024
print(f"Colonne 'categorie' - Avant: {mem_avant:.2f} KB, Après: {mem_apres:.2f} KB")
print(f"Réduction: {((mem_avant - mem_apres) / mem_avant * 100):.1f}%")

# OPTIMISATION 3: Datetime parsé correctement
print("\n=== OPTIMISATION 3: DATETIME ===")
df_opt['timestamp'] = pd.to_datetime(df_opt['timestamp'])
print(f"Type timestamp: {df_opt['timestamp'].dtype}")
print(f"Mémoire timestamp: {df_opt['timestamp'].memory_usage(deep=True) / 1024:.2f} KB")

# Mémoire totale après optimisations
print("\n=== RÉSUMÉ OPTIMISATION ===")
print(f"Avant: {df_large.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
print(f"Après: {df_opt.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
taux_reduction = ((df_large.memory_usage(deep=True).sum() - df_opt.memory_usage(deep=True).sum()) / df_large.memory_usage(deep=True).sum() * 100)
print(f"Réduction totale: {taux_reduction:.1f}%")

# LECTURE CHUNKED (pour fichiers énormes)
print("\n=== LECTURE CHUNKED (simulation) ===")
# Créer un CSV temporaire
df_large.to_csv('/tmp/gros_dataset.csv', index=False)

# Lire par chunks
chunk_size = 10000
chunks = []
for chunk in pd.read_csv('/tmp/gros_dataset.csv', chunksize=chunk_size, dtype={
    'id': 'int32',
    'valeur': 'float32',
    'categorie': 'category'
}):
    # Traitement par chunk
    chunk['valeur_carre'] = chunk['valeur'] ** 2
    chunks.append(chunk)

df_processed = pd.concat(chunks, ignore_index=True)
print(f"Traité {len(df_processed)} lignes par chunks de {chunk_size}")
print(f"Mémoire finale: {df_processed.memory_usage(deep=True).sum() / 1024**2:.2f} MB")

# PARQUET (format compressé efficace)
print("\n=== PARQUET VS CSV ===")
# Écrire en parquet
df_opt.to_parquet('/tmp/dataset.parquet', compression='snappy')

import os
csv_size = os.path.getsize('/tmp/gros_dataset.csv') / 1024**2
parquet_size = os.path.getsize('/tmp/dataset.parquet') / 1024**2
print(f"CSV size: {csv_size:.2f} MB")
print(f"Parquet size: {parquet_size:.2f} MB")
print(f"Compression: {((csv_size - parquet_size) / csv_size * 100):.1f}%")

# Temps de lecture
print("\nTemps de lecture (rough benchmark):")
import time

start = time.time()
df_csv = pd.read_csv('/tmp/gros_dataset.csv')
time_csv = time.time() - start

start = time.time()
df_parquet = pd.read_parquet('/tmp/dataset.parquet')
time_parquet = time.time() - start

print(f"CSV: {time_csv*1000:.2f}ms")
print(f"Parquet: {time_parquet*1000:.2f}ms")

# Sélection de colonnes en parquet (lazy evaluation)
print("\n=== LAZY LOADING PARQUET ===")
df_subset = pd.read_parquet('/tmp/dataset.parquet', columns=['id', 'valeur'])
print(f"Chargement seulement 2 colonnes: {df_subset.memory_usage(deep=True).sum() / 1024**2:.2f} MB")

# Copy() vs View()
print("\n=== COPY vs VIEW (piège courant) ===")
df1 = df_opt.copy()  # Copie vraie (nouveau memory)
df2 = df_opt[['id', 'valeur']]  # View (même memory)
print(f"Après copy(): {df1.memory_usage(deep=True).sum() / 1024**2:.2f} MB (indépendant)")
print(f"Après slicing: {df2.memory_usage(deep=True).sum() / 1024**2:.2f} MB (plus léger)")

Tableau d'Optimisation

Astuce Professionnelle

Toujours profiler votre code avec %timeit ou line_profiler. 90% du temps, le goulot est une boucle implicite ou un copie DataFrame non-nécessaire. Préférez les opérations vectorisées Numpy/Pandas aux boucles Python. Utilisez .loc (label-based) plutôt que .iloc (integer-based) quand vous connaissez les labels—c'est souvent plus rapide.

⚠️ Attention Critique

Les catégories créent une dépendance: si vous mergez deux DataFrames avec catégories, les catégories non-matchées deviennent NaN. Évitez les .astype('category') après filtrage si les anciennes catégories manquent—nettoyez avec .cat.remove_unused_categories(). Aussi, attention au .copy() implicite: assigner un DataFrame à une variable crée une vue, pas une copie.

5. Feature Engineering et Pipelines de Transformation Robustes

Définition

Le feature engineering est l'art de créer ou modifier des colonnes (features) pour améliorer la performance des modèles ML. Un pipeline robuste automatise et reproduit ces transformations de manière testable et versionnable.

Explication Détaillée