Aller au contenu

Pipeline Excellence Promotionnelle Lactalis

Vue d'ensemble

Cette pipeline automatisée analyse l'efficacité des promotions commerciales pour les différentes business units de Lactalis. Elle transforme les données brutes de ventes (sell-in/sell-out) en insights actionnables sur la performance promotionnelle, incluant le calcul du ROI, l'analyse de la cannibalisation et la classification des événements promotionnels.

Objectifs principaux

  • Analyser l'impact réel des promotions sur les volumes et marges
  • Calculer les effets de volume (uplift, cannibalisation, forward buying, halo)
  • Évaluer le ROI et classifier les promotions (value/volume generator/destroyer)
  • Fournir des données structurées pour l'outil de simulation promotionnelle
  • Automatiser le processus end-to-end avec gestion d'erreurs robuste

Business Units supportées

  • Italy Galbani : Produits fromagers avec données sell-out
  • Italy Parmalat : Produits laitiers UHT avec spécificités cannibalisation
  • Autres pays : Framework extensible pour nouveaux marchés

Architecture technique

Vue d'ensemble de la pipeline

┌─────────────────────────────────────────────────────────────────────┐
│                        PIPELINE COMPLÈTE                              │
├─────────────────────────────────────────────────────────────────────┤
│                                                                       │
│  Step 00: Download Inputs          Step 01: Pre-Processing Italy     │
│  ├─ Connexion SFTP                 ├─ Transformation Excel → TXT     │
│  ├─ Retry intelligent              ├─ Normalisation caractères       │
│  └─ → /app/inputs/                └─ Tables référence MariaDB       │
│                                                                       │
│  Step 02: Data Integration         Step 03: Data Cleaning            │
│  ├─ Mapping colonnes/types         ├─ Re-classification produits    │
│  ├─ Parsing dates                  ├─ Calcul coûts promo             │
│  └─ → Tables MariaDB               └─ 8 Quality Checks               │
│                                                                       │
│  Step 04: Model Database           Step 05: Building Baseline        │
│  ├─ Scoping (filtrage volumes)     ├─ Calcul uplift/cannib          │
│  ├─ Méthodologies baseline         ├─ Forward buying (Galbani)      │
│  └─ Standard vs Statistical        └─ Quality check stock-up         │
│                                                                       │
│  Step 06: Margins & Cannib         Step 07: Aggregation Promo        │
│  ├─ Calcul marges ajustées         ├─ Agrégation niveau promo       │
│  ├─ Valorisation baseline          ├─ Classification événements      │
│  └─ Cannibalisation famille        └─ Performance YTD/LYTD          │
│                                                                       │
│  Step 08: Simulation Inputs        Step 09: Upload Outputs           │
│  ├─ Export Excel formaté           ├─ Upload SFTP avec retry         │
│  ├─ Année max uniquement           ├─ Logs et résultats             │
│  └─ → /app/outputs/                └─ → Serveur central              │
│                                                                       │
└─────────────────────────────────────────────────────────────────────┘

Technologies utilisées

  • Langage : Python 3.x
  • Base de données : MariaDB
  • Librairies principales : pandas, SQLAlchemy, paramiko
  • Orchestration : Bash avec gestion mémoire et heartbeat
  • Containerisation : Docker avec volumes persistants

Installation et configuration

Prérequis système

  • Docker et Docker Compose
  • Accès SFTP aux serveurs Lactalis
  • 16GB RAM minimum (32GB recommandé pour Italy)
  • 50GB espace disque

Variables d'environnement requises

# Base de données
DATABASE_HOST=mariadb
DATABASE_PORT=3306
DATABASE_NAME=promo_analysis
DATABASE_USERNAME=promo_user
DATABASE_PASSWORD=****

# SFTP
SFTP_HOST=sftp.lactalis.com
SFTP_PORT=22
SFTP_USERNAME=****
SFTP_PASSWORD=****

# Business Unit
BUSINESS_UNIT=italy_galbani  # ou italy_parmalat, france_president, etc.

Structure des répertoires

/app/
├── src/
│   ├── scripts/          # Scripts Python par étape
│   ├── db_config.py      # Configuration DB
│   ├── logger.py         # Système de logging
│   └── utils.py          # Fonctions utilitaires
├── inputs/
│   └── {business_unit}/  # Données téléchargées
├── outputs/
│   └── {business_unit}/  # Résultats et QC
├── logs/
│   └── {business_unit}/  # Logs d'exécution
└── MariaDB/
    └── Tool_architecture/  # Mappings et schémas

Guide d'utilisation

Lancement de la pipeline

# Lancement complet
docker-compose up

# Ou directement le script
./run_pipeline.sh

Monitoring de l'exécution

# Suivre les logs en temps réel
tail -f /app/logs/{business_unit}/pipeline_*.log

# Vérifier le heartbeat
cat /tmp/heartbeat

# Vérifier l'utilisation mémoire
docker stats

Résultats produits

Fichiers de sortie principaux

  1. Promotion_Level_Table.xlsx : Performance détaillée par promotion
  2. Baseline_Table.xlsx : Volumes baseline hebdomadaires
  3. Quality checks/ : 8 rapports de contrôle qualité

Tables MariaDB clés

  • step_6_03_final_with_scope_promo_table : Données promotions enrichies
  • step_6_04_perf_evolution_* : Analyses comparatives YTD/LYTD
  • Tables de référence spécifiques par business unit

Description détaillée des étapes

Step 00: Download Inputs

Objectif : Télécharger les fichiers sources depuis SFTP

Fonctionnalités clés : - Retry automatique avec backoff exponentiel - Gestion des variantes de nommage (underscore simple/double) - Support des connexions instables

Voir : Documentation Step 00

Step 01: Specific Pre-Processing (Italie uniquement)

Objectif : Préparer les données spécifiques italiennes

Fonctionnalités clés : - Transformation Excel → format standard TXT - Normalisation des caractères italiens (à → a) - Chargement des tables de référence (13 pour Galbani, 10 pour Parmalat)

Voir : Documentation Step 01

Step 02: Data Integration

Objectif : Intégrer les données dans MariaDB

Fonctionnalités clés : - Application des mappings colonnes/types par pays - Parsing intelligent des dates - Gestion des encodages multiples

Voir : Documentation Step 02

Step 03: Data Cleaning & Quality Checks

Objectif : Nettoyer les données et valider la qualité

Fonctionnalités clés : - Re-classification produits et catégories - Calcul des coûts promotionnels - 8 contrôles qualité automatisés

Voir : Documentation Step 03

Step 04: Building Model Database

Objectif : Construire la base du modèle avec baselines

Fonctionnalités clés : - Filtrage volumes (low-volume retailers) - Calcul baselines (Standard vs Statistical) - Gestion In&Out et redistribution volumes

Voir : Documentation Step 04

Step 05: Building Baseline Effects

Objectif : Calculer les effets de volume

Fonctionnalités clés : - Uplift et cannibalisation - Forward buying (Galbani avec sell-out) - Halo et pantry loading

Voir : Documentation Step 05

Step 06: Margins & Cannibalization

Objectif : Valoriser les effets et calculer la cannibalisation

Fonctionnalités clés : - Marges ajustées par business unit - Cannibalisation intra-famille - Valorisation des effets baseline

Voir : Documentation Step 06

Step 07: Promo Level Aggregation

Objectif : Agréger au niveau promotion et classifier

Fonctionnalités clés : - Forward buying et cannibalisation finale - Calcul ROI et classification - Analyse performance YTD/LYTD

Voir : Documentation Step 07

Step 08: Simulation Inputs

Objectif : Préparer les données pour l'outil de simulation

Fonctionnalités clés : - Export année courante uniquement - Format Excel standardisé - Gestion spécifique Parmalat (sans EAN_desc)

Voir : Documentation Step 08

Step 09: Upload Outputs

Objectif : Transférer les résultats vers SFTP

Fonctionnalités clés : - Upload robuste avec retry - Transfer des logs pour traçabilité - Gestion des chemins distants

Voir : Documentation Step 09

Spécificités par business unit

Italy Galbani

  • Données sell-out : 3 sources (Galbani, competitors, private label)
  • Forward buying : Calcul avec delta sell-in/sell-out
  • Catégories : Focus produits fromagers
  • Tables spécifiques : 13 tables de configuration

Italy Parmalat

  • Sans sell-out : Uniquement données sell-in
  • Cannibalisation : Limitée aux produits UHT
  • Redistribution volumes : Avant calcul baseline
  • Marges pré-calculées : Dans Step 3

Framework générique

  • Extensible : Support facile de nouveaux pays
  • Configurable : Via promo_config.json
  • Fallback : Fonctions default si spécifique absent

Gestion des erreurs et robustesse

Stratégies implémentées

  1. Retry intelligent : Backoff exponentiel pour opérations réseau
  2. Checkpoints : Sauvegarde après chaque étape majeure
  3. Validation données : Contrôles qualité automatiques
  4. Logs détaillés : Traçabilité complète des opérations
  5. Gestion mémoire : Chunking et garbage collection

Points de contrôle critiques

  • Cohérence row count entre étapes
  • Validation types de données
  • Détection valeurs aberrantes
  • Vérification complétude

Maintenance et évolution

Ajout d'un nouveau pays

  1. Créer la configuration dans promo_config.json : 

    {
    "Country": "spain_president",
    "Dayfirst_datetime": true,
    "Years_in_scope_for_tool_refresh": "2023,2024,2025"
}

  2. Ajouter les mappings dans Tool_architecture/{country}/

  3. Implémenter les fonctions spécifiques si nécessaire : 

    def compute_margin_spain_president(engine):
    # Logique spécifique

  4. Tester avec données exemple

Modification des calculs

Les points d'extension principaux : - Marges : Step_06 compute_margin_{country}() - Baseline : Step_04 data_model_{data_type}_{country}() - Classification : Step_07 classify_events_{country}()

Monitoring performance

-- Vérifier volumes de données
SELECT 
    TABLE_NAME,
    TABLE_ROWS,
    ROUND(DATA_LENGTH/1024/1024, 2) as DATA_MB
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'promo_analysis'
ORDER BY TABLE_ROWS DESC;

-- Analyser durées d'exécution
SELECT 
    step_name,
    start_time,
    end_time,
    TIMESTAMPDIFF(MINUTE, start_time, end_time) as duration_minutes
FROM execution_logs
ORDER BY start_time DESC;

Troubleshooting

Problèmes fréquents

  1. Out of Memory
  2. Symptôme : Container killed

  3. Solution : Augmenter RAM ou réduire chunk size

  4. Timeout SFTP

  5. Symptôme : EOFError pendant download/upload

  6. Solution : Augmenter keepalive et timeouts

  7. Baseline manquants

  8. Symptôme : Beaucoup de NaN

  9. Solution : Vérifier seuils et historique données

  10. ROI aberrants

  11. Symptôme : Valeurs > 1000%
  12. Solution : Vérifier calcul coûts promo

Logs de diagnostic

# Erreurs critiques
grep -i error /app/logs/*/pipeline_*.log

# Warnings
grep -i warning /app/logs/*/pipeline_*.log

# Progression étapes
grep "Starting\|Completed" /app/logs/*/pipeline_*.log

Support et contact

Pour toute question ou problème : 1. Consulter la documentation détaillée de l'étape concernée 2. Vérifier les logs pour identifier l'erreur 3. Contacter l'équipe Data Lactalis

Versions et changelog

Version 1.0 (Current)

  • Pipeline complète pour Italy (Galbani/Parmalat)
  • 9 étapes automatisées
  • Quality checks intégrés
  • Documentation complète

Pipeline développée pour Lactalis - Excellence Promotionnelle