Projet 02 : Analyseur de données de ventes et reporting financier

Dans les départements Data, Finance et Opérations, l'automatisation du traitement des fichiers d'exportation quotidienne est un cas d'usage fondamental. Bien que des bibliothèques externes spécialisées telles que Pandas existent, maîtriser la manipulation de données en pur Python standard sans dépendance externe est indispensable : cela permet de construire des microservices ultralégers, des fonctions Serverless (AWS Lambda, Google Cloud Functions) à démarrage instantané et des scripts système hautement portables.

L'objectif de ce deuxième projet est de construire un pipeline ETL (Extract, Transform, Load) complet capable de lire un fichier CSV de transactions commerciales, de nettoyer les données, d'agréger des indicateurs clés de performance (KPI) et de produire un rapport financier double : un affichage console élégant et un export structuré au standard JSON.

---

1. Cahier des charges fonctionnel

Notre analyseur doit traiter un fichier de transactions de vente contenant les colonnes suivantes : id_commande, date, client, categorie, produit, quantite, prix_unitaire.

Exigences du traitement :

1. Extraction & Nettoyage robuste :
2. Parcourir le fichier CSV avec le module standard csv.DictReader.
3. Ignorer les lignes corrompues ou incomplètes sans faire planter le script (valeurs manquantes, quantités ou prix négatifs ou non convertibles).
4. Convertir les données textuelles en types natifs appropriés (int pour les quantités, float pour les prix).
5. Calculs analytiques & Métriques financières :
6. Chiffre d'affaires brut total : somme des montants de chaque ligne (\(\text{quantite} \times \text{prix\_unitaire}\)).
7. Nombre total de commandes uniques et nombre total d'articles vendus.
8. Panier moyen par commande : \(\frac{\text{CA Total}}{\text{Nombre de commandes}}\).
9. Ventilation par catégorie : montant des ventes, part en pourcentage du CA global et volume d'articles.
10. Palmarès des clients (Top 3) : les 3 clients ayant généré le plus de valeur d'achat.
11. Palmarès des produits (Top 3) : les 3 articles les plus vendus en volume unitaire.
12. Double restitution des livrables :
13. Restitution textuelle formatée avec des tableaux ASCII soignés en console.
14. Export d'un résumé analytique prêt à l'emploi au format JSON (resume_ventes.json), encodé en UTF-8 et indenté.

---

2. Architecture du pipeline ETL

flowchart TD
    A[Fichier ventes.csv] --> B[Module csv.DictReader]
    B --> C{Validation ligne par ligne}
    C -- Erreur de conversion ou donnee negative --> D[Journalisation avertissement & Ligne ignoree]
    C -- Ligne valide --> E[Conversion de types : int, float]
    E --> F[Agrégation en mémoire : dictionnaires de cumul]
    F --> G[Calcul des indicateurs agreges et pourcentages]
    G --> H[Tri et extraction des Top 3 Clients et Produits]
    H --> I[Rendu visuel formatee en console f-strings]
    H --> J[Export vers fichier resume_ventes.json]

---

3. Implémentation complète en Python 3.10+

Voici le script complet, modulaire, typé et prêt à l'emploi. Il intègre une fonction d'auto-génération de données de test au premier lancement si le fichier source n'existe pas encore sur votre machine :

"""
Analyseur de données de ventes et générateur de reporting financier en pur Python.
Auteur : Joseph Konka (AlgoJungle)
"""

import csv
import json
from collections import defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple


@dataclass
class Transaction:
    """Structure de données représentant une ligne de vente valide."""
    id_commande: str
    date: str
    client: str
    categorie: str
    produit: str
    quantite: int
    prix_unitaire: float

    @property
    def total_ligne(self) -> float:
        """Montant total généré par la ligne."""
        return self.quantite * self.prix_unitaire


def creer_jeu_de_donnees_demo(chemin: Path) -> None:
    """Génère un fichier CSV d'exemple pour tester immédiatement le pipeline."""
    donnees_demo = [
        ["id_commande", "date", "client", "categorie", "produit", "quantite", "prix_unitaire"],
        ["CMD-101", "2026-03-01", "Alice Dupont", "Informatique", "Clavier Mécanique", "2", "89.90"],
        ["CMD-102", "2026-03-01", "Bob Martin", "Mobilier", "Chaise Ergonomique", "1", "249.00"],
        ["CMD-103", "2026-03-02", "Claire Moreau", "Informatique", "Écran 27 pouces", "1", "319.50"],
        ["CMD-104", "2026-03-02", "Alice Dupont", "Fournitures", "Cahier Réutilisable", "5", "12.00"],
        ["CMD-105", "2026-03-03", "David Cohen", "Informatique", "Souris Sans Fil", "3", "45.00"],
        ["CMD-106", "2026-03-03", "Bob Martin", "Informatique", "Clavier Mécanique", "1", "89.90"],
        ["CMD-107", "2026-03-04", "Claire Moreau", "Mobilier", "Bureau Assis-Debout", "1", "480.00"],
        ["CMD-108", "2026-03-04", "Emma Laurent", "Fournitures", "Stylo Ergonomique", "10", "4.50"],
        ["CMD-109", "2026-03-05", "David Cohen", "Mobilier", "Lampe de Bureau LED", "2", "35.00"],
        ["CMD-110", "2026-03-05", "Alice Dupont", "Informatique", "Souris Sans Fil", "2", "45.00"],
        # Lignes volontairement corrompues pour tester la robustesse du filtre :
        ["CMD-111", "2026-03-05", "Erreur Client", "Fournitures", "Produit Invalide", "-2", "15.00"],
        ["CMD-112", "2026-03-05", "Erreur Prix", "Informatique", "Câble HDMI", "1", "Gratuit"],
    ]
    with open(chemin, mode="w", encoding="utf-8", newline="") as f:
        writer = csv.writer(f)
        writer.writerows(donnees_demo)
    print(f"📁 Fichier d'exemple généré : {chemin.resolve()}")


def charger_et_nettoyer_ventes(chemin_csv: Path) -> Tuple[List[Transaction], int]:
    """
    Lit le fichier CSV, filtre les lignes invalides et retourne les objets Transaction.
    Retourne également le nombre de lignes ignorées.
    """
    if not chemin_csv.exists():
        raise FileNotFoundError(f"Le fichier source '{chemin_csv}' est introuvable.")

    transactions_valides: List[Transaction] = []
    lignes_ignorees: int = 0

    with open(chemin_csv, mode="r", encoding="utf-8") as f:
        lecteur = csv.DictReader(f)
        for num_ligne, ligne in enumerate(lecteur, start=2):
            try:
                quantite = int(ligne["quantite"])
                prix = float(ligne["prix_unitaire"])

                if quantite <= 0 or prix <= 0:
                    lignes_ignorees += 1
                    continue

                tx = Transaction(
                    id_commande=ligne["id_commande"].strip(),
                    date=ligne["date"].strip(),
                    client=ligne["client"].strip(),
                    categorie=ligne["categorie"].strip(),
                    produit=ligne["produit"].strip(),
                    quantite=quantite,
                    prix_unitaire=prix,
                )
                transactions_valides.append(tx)
            except (ValueError, KeyError):
                lignes_ignorees += 1

    return transactions_valides, lignes_ignorees


def analyser_donnees(transactions: List[Transaction]) -> Dict[str, Any]:
    """Agrège et calcule tous les indicateurs clés de performance (KPI)."""
    if not transactions:
        raise ValueError("Aucune transaction valide à analyser.")

    ca_total: float = sum(tx.total_ligne for tx in transactions)
    commandes_uniques: set[str] = {tx.id_commande for tx in transactions}
    total_articles: int = sum(tx.quantite for tx in transactions)
    panier_moyen: float = ca_total / len(commandes_uniques)

    # 1. Cumul par Catégorie
    ca_par_categorie: Dict[str, float] = defaultdict(float)
    articles_par_categorie: Dict[str, int] = defaultdict(int)

    # 2. Cumul par Client
    ca_par_client: Dict[str, float] = defaultdict(float)

    # 3. Cumul par Produit (volume)
    volume_par_produit: Dict[str, int] = defaultdict(int)

    for tx in transactions:
        ca_par_categorie[tx.categorie] += tx.total_ligne
        articles_par_categorie[tx.categorie] += tx.quantite
        ca_par_client[tx.client] += tx.total_ligne
        volume_par_produit[tx.produit] += tx.quantite

    # Construction du détail des catégories avec pourcentages
    synthese_categories: List[Dict[str, Any]] = []
    for cat, montant in sorted(ca_par_categorie.items(), key=lambda item: item[1], reverse=True):
        part_ca = (montant / ca_total) * 100 if ca_total > 0 else 0.0
        synthese_categories.append({
            "categorie": cat,
            "chiffre_affaires": round(montant, 2),
            "pourcentage_ca": round(part_ca, 2),
            "articles_vendus": articles_par_categorie[cat],
        })

    # Classements Top 3
    top_clients = sorted(ca_par_client.items(), key=lambda item: item[1], reverse=True)[:3]
    top_produits = sorted(volume_par_produit.items(), key=lambda item: item[1], reverse=True)[:3]

    return {
        "chiffre_affaires_total": round(ca_total, 2),
        "nombre_commandes": len(commandes_uniques),
        "total_articles_vendus": total_articles,
        "panier_moyen": round(panier_moyen, 2),
        "categories": synthese_categories,
        "top_3_clients": [
            {"client": c, "depenses": round(m, 2)} for c, m in top_clients
        ],
        "top_3_produits": [
            {"produit": p, "quantite_vendue": q} for p, q in top_produits
        ],
    }


def formater_rapport_console(rapport: Dict[str, Any], lignes_ignorees: int) -> str:
    """Génère un affichage textuel élégant et tabulé pour la console."""
    lignes: List[str] = [
        "=" * 68,
        "          📊 RAPPORT D'ACTIVITÉ COMMERCIALE & FINANCIÈRE",
        "=" * 68,
        f"💰 Chiffre d'affaires brut : {rapport['chiffre_affaires_total']:>12,.2f} €",
        f"📦 Nombre de commandes     : {rapport['nombre_commandes']:>12}",
        f"🏷️  Articles vendus au total : {rapport['total_articles_vendus']:>12}",
        f"🛒 Panier moyen par commande: {rapport['panier_moyen']:>12,.2f} €",
        f"⚠️  Lignes erronées écartées : {lignes_ignorees:>12}",
        "-" * 68,
        "📁 VENTES PAR CATÉGORIE :",
        f"{'Catégorie':<18} | {'CA Réalisé (€)':>15} | {'Part (%)':>10} | {'Articles':>8}",
        "-" * 68,
    ]

    for cat in rapport["categories"]:
        lignes.append(
            f"{cat['categorie']:<18} | {cat['chiffre_affaires']:>15,.2f} | {cat['pourcentage_ca']:>9.1f}% | {cat['articles_vendus']:>8}"
        )

    lignes.append("-" * 68)
    lignes.append("🏆 TOP 3 CLIENTS (Valeur d'achat) :")
    for rang, c in enumerate(rapport["top_3_clients"], start=1):
        lignes.append(f"  {rang}. {c['client']:<25} : {c['depenses']:>10,.2f} €")

    lignes.append("-" * 68)
    lignes.append("⭐ TOP 3 PRODUITS LES PLUS VENDUS (Volume) :")
    for rang, p in enumerate(rapport["top_3_produits"], start=1):
        lignes.append(f"  {rang}. {p['produit']:<25} : {p['quantite_vendue']:>5} unités")

    lignes.append("=" * 68)
    return "\n".join(lignes)


def exporter_vers_json(rapport: Dict[str, Any], chemin_sortie: Path) -> None:
    """Exporte les indicateurs sous forme de JSON structuré et indenté."""
    with open(chemin_sortie, mode="w", encoding="utf-8") as f:
        json.dump(rapport, f, ensure_ascii=False, indent=4)
    print(f"💾 Rapport financier JSON exporté avec succès dans : {chemin_sortie.resolve()}")


def executer_pipeline(chemin_source: str = "ventes.csv", chemin_json: str = "resume_ventes.json") -> None:
    """Coordonne l'ensemble des étapes du pipeline d'analyse."""
    path_csv = Path(chemin_source)
    path_json = Path(chemin_json)

    # Génération automatique du fichier d'exemple si absent
    if not path_csv.exists():
        print(f"ℹ️ Aucun fichier '{path_csv}' détecté. Création d'un jeu de test...")
        creer_jeu_de_donnees_demo(path_csv)

    print(f"\n⏳ Analyse en cours du fichier : {path_csv.name}...")
    transactions, lignes_rejet = charger_et_nettoyer_ventes(path_csv)
    metriques = analyser_donnees(transactions)

    # 1. Rendu console
    rapport_texte = formater_rapport_console(metriques, lignes_rejet)
    print("\n" + rapport_texte + "\n")

    # 2. Rendu fichier JSON
    exporter_vers_json(metriques, path_json)


if __name__ == "__main__":
    executer_pipeline()

---

4. Exemple de rendu et d'exportation

Sortie en console texte

À l'exécution du script, l'ingénieur reçoit immédiatement un résumé synthétique :

====================================================================
          📊 RAPPORT D'ACTIVITÉ COMMERCIALE & FINANCIÈRE
====================================================================
💰 Chiffre d'affaires brut :     1,570.70 €
📦 Nombre de commandes     :           10
🏷️  Articles vendus au total :           29
🛒 Panier moyen par commande:       157.07 €
⚠️  Lignes erronées écartées :            2
--------------------------------------------------------------------
📁 VENTES PAR CATÉGORIE :
Catégorie          |  CA Réalisé (€) |   Part (%) | Articles
--------------------------------------------------------------------
Mobilier           |          799.00 |      50.9% |        4
Informatique       |          666.70 |      42.4% |       10
Fournitures        |          105.00 |       6.7% |       15
--------------------------------------------------------------------
🏆 TOP 3 CLIENTS (Valeur d'achat) :
  1. Claire Moreau             :     799.50 €
  2. Alice Dupont              :     364.80 €
  3. Bob Martin                :     338.90 €
--------------------------------------------------------------------
⭐ TOP 3 PRODUITS LES PLUS VENDUS (Volume) :
  1. Stylo Ergonomique         :    10 unités
  2. Souris Sans Fil           :     5 unités
  3. Cahier Réutilisable       :     5 unités
====================================================================

Fichier JSON généré (resume_ventes.json)

Le fichier JSON résultant peut être consommé directement par une API Web, une base de données NoSQL ou un tableau de bord Power BI :

{
    "chiffre_affaires_total": 1570.7,
    "nombre_commandes": 10,
    "total_articles_vendus": 29,
    "panier_moyen": 157.07,
    "categories": [
        {
            "categorie": "Mobilier",
            "chiffre_affaires": 799.0,
            "pourcentage_ca": 50.87,
            "articles_vendus": 4
        },
        {
            "categorie": "Informatique",
            "chiffre_affaires": 666.7,
            "pourcentage_ca": 42.45,
            "articles_vendus": 10
        },
        {
            "categorie": "Fournitures",
            "chiffre_affaires": 105.0,
            "pourcentage_ca": 6.68,
            "articles_vendus": 15
        }
    ],
    "top_3_clients": [
        { "client": "Claire Moreau", "depenses": 799.5 },
        { "client": "Alice Dupont", "depenses": 364.8 },
        { "client": "Bob Martin", "depenses": 338.9 }
    ],
    "top_3_produits": [
        { "produit": "Stylo Ergonomique", "quantite_vendue": 10 },
        { "produit": "Souris Sans Fil", "quantite_vendue": 5 },
        { "produit": "Cahier Réutilisable", "quantite_vendue": 5 }
    ]
}

---

5. Mise en pratique & Défis d'extension

Pour aller plus loin dans la maîtrise des données :

Défi 1 : Filtrage temporel avec le module datetime

• Objectif : Ajoutez deux arguments facultatifs date_debut et date_fin à la fonction charger_et_nettoyer_ventes. À l'aide de datetime.strptime(tx.date, "%Y-%m-%d"), filtrez le reporting pour n'analyser que les transactions comprises dans la période spécifiée.

Défi 2 : Tracé d'un diagramme à barres horizontal en ASCII

• Objectif : Implémentez une fonction qui restitue la répartition des ventes par catégorie sous forme de barres graphiques dans la console terminal (ex : chaque caractère # représente 2 % de part de marché) :

Mobilier     [#########################] 50.9%
Informatique [#####################    ] 42.4%
Fournitures  [###                      ]  6.7%

Défi 3 : Détection statistique des commandes hors-normes

• Objectif : À l'aide du module statistics, calculez l'écart-type des montants des commandes et signalez dans le rapport les transactions dont le montant est supérieur à \(\mu + 2\sigma\) (valeurs aberrantes ou commandes exceptionnelles nécessitant une validation managériale).

Partagez sur les réseaux sociaux

Commentaires