Step_08_Simulation_Inputs - Préparation des données pour l'outil de simulation¶

Vue d'ensemble¶

Cette étape constitue le pont entre l'analyse des promotions historiques et l'outil de simulation. Elle extrait et formate les données essentielles des étapes précédentes pour créer deux fichiers Excel qui serviront d'inputs à l'outil de simulation promotionnelle. C'est une étape de transformation pure qui prépare les données pour une utilisation externe.

Objectif principal¶

Extraire les données de performance promotionnelle de l'année la plus récente
Extraire les baselines hebdomadaires correspondantes
Formater les données selon les spécifications de l'outil de simulation
Exporter en fichiers Excel prêts à l'emploi

Position dans la pipeline¶

Étape précédente : Step_07_Aggregation_at_promo_level
Étape suivante : Step_09_Upload_Outputs
Appel unique : main() → extract_and_export_data()

Architecture technique¶

Flux de données¶

Extraction des données
─────────────────────
Entrée
    ├── step_6_03_final_with_scope_promo_table
    │   └── Données promotionnelles avec KPIs et classification
    │
    └── step_3_02_model_data_baseline
        └── Volumes baseline hebdomadaires

Traitement
    ├── Filtrage année max uniquement
    ├── Sélection colonnes pertinentes
    ├── Renommage pour interface utilisateur
    └── Formatage (arrondis, types)

Sortie
    ├── Promotion_Level_Table.xlsx
    │   └── Performance par promotion (année courante)
    │
    └── Baseline_Table.xlsx
        └── Baselines hebdomadaires (année courante)

Concepts clés¶

1. Année de référence¶

Sélection : MAX(Year) de chaque table
Objectif : Fournir les données les plus récentes pour simulation
Cohérence : Les deux exports utilisent la même année

2. Mapping des colonnes¶

Transformation des noms techniques en noms business : - Adj_Promo_Volume → Promo_Volume - Promo_margin_at_event_level → Promo_margin - NSV_at_event_level → Promo_NSV - In_Scope → Filter - Volume_Uplift_% → Volume Uplift - On_invoice_discount → Sell-in discount - Category → High-level classification - Event_Type → Segmentation - Event_duration_in_weeks → Event duration

3. Gestion Parmalat¶

Spécificité : Absence de colonne EAN_desc
Impact : Requêtes et sélections adaptées
Détection : Via business_unit == 'italy_parmalat'

Implémentation détaillée¶

Structure principale¶

def main():
    """Point d'entrée principal"""
    # Connexion DB
    engine = create_engine(
        f'mysql+pymysql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}'
    )

    # Extraction et export
    extract_and_export_data(engine)

    log_message("Script executed successfully")

Fonction d'extraction : `extract_and_export_data()`¶

1. Détection du business unit¶

def extract_and_export_data(engine):
    # Détection Parmalat pour adaptation
    is_parmalat = business_unit.lower() == 'italy_parmalat'

    log_message(
        f"Starting data extraction process"
        f"{' for Parmalat (without EAN_desc)' if is_parmalat else ''}..."
    )

2. Extraction table promotions¶

# Query universelle (toutes colonnes)
query_step_6_03 = """
SELECT *
FROM step_6_03_final_with_scope_promo_table
WHERE Year = (SELECT MAX(Year) FROM step_6_03_final_with_scope_promo_table)
"""

# Chargement
df_step_6_03 = pd.read_sql_query(query_step_6_03, engine)

# Sélection colonnes selon business unit
if is_parmalat:
    columns_to_select_6_03 = [
        'Retailer_name', 'EAN', 'New_category', 'Promotion_Code', 'Year',
        'Baseline_Volume', 'Adj_Promo_Volume', 'Promo_margin_at_event_level', 
        'NSV_at_event_level', 'Baseline_Margin', 'Promo_Costs',
        'In_Scope', 'ROI', 'Volume_Uplift_%', 'Incremental_Margin', 
        'On_invoice_discount', 'First_Week', 'Last_Week', 'Category', 
        'Event_Type', 'Event_duration_in_weeks'
    ]
else:
    # Inclut EAN_desc
    columns_to_select_6_03 = [
        'Retailer_name', 'EAN', 'EAN_desc', 'New_category', 'Promotion_Code', 
        # ... reste identique
    ]

df_step_6_03 = df_step_6_03[columns_to_select_6_03]

3. Extraction table baseline¶

# Query adaptée selon business unit
if is_parmalat:
    query_step_3_02 = """
    SELECT
        Retailer_name,
        EAN,
        Year,
        CAST(Week AS INTEGER) AS Week,
        Baseline_Volume
    FROM step_3_02_model_data_baseline
    WHERE Year = (SELECT MAX(Year) FROM step_3_02_model_data_baseline)
    """
else:
    query_step_3_02 = """
    SELECT
        Retailer_name,
        EAN,
        EAN_desc,
        Year,
        CAST(Week AS INTEGER) AS Week,
        Baseline_Volume
    FROM step_3_02_model_data_baseline
    WHERE Year = (SELECT MAX(Year) FROM step_3_02_model_data_baseline)
    """

df_step_3_02 = pd.read_sql_query(query_step_3_02, engine)

4. Renommage des colonnes¶

# Mapping différencié selon présence EAN_desc
if is_parmalat:
    df_step_6_03.columns = [
        'Retailer_name', 'EAN', 'New_category', 'Promotion_Code', 'Year',
        'Baseline_Volume', 'Promo_Volume', 'Promo_margin', 'Promo_NSV', 
        'Baseline_Margin', 'Promo_Costs', 'Filter', 'ROI', 'Volume Uplift', 
        'Incremental Margin', 'Sell-in discount', 'First Week', 'Last Week', 
        'High-level classification', 'Segmentation', 'Event duration'
    ]
else:
    df_step_6_03.columns = [
        'Retailer_name', 'EAN', 'EAN_desc', 'New_category', 'Promotion_Code', 
        # ... reste identique avec EAN_desc en position 3
    ]

5. Formatage et export¶

# Arrondi des discounts
df_step_6_03['Sell-in discount'] = df_step_6_03['Sell-in discount'].round(1)

# Export Excel
output_path = f"/app/outputs/{business_unit}/Promotion_Level_Table.xlsx"
df_step_6_03.to_excel(output_path, index=False)

output_path_baseline = f"/app/outputs/{business_unit}/Baseline_Table.xlsx"
df_step_3_02.to_excel(output_path_baseline, index=False)

Structure des fichiers de sortie¶

1. Promotion_Level_Table.xlsx¶

Colonne	Description	Format
Retailer_name	Nom du retailer	String
EAN	Code produit	String
EAN_desc	Description produit (sauf Parmalat)	String
New_category	Catégorie produit	String
Promotion_Code	Code unique promotion	String
Year	Année	Integer
Baseline_Volume	Volume baseline total	Float
Promo_Volume	Volume promotionnel ajusté	Float
Promo_margin	Marge unitaire promo	Float
Promo_NSV	NSV unitaire promo	Float
Baseline_Margin	Marge unitaire hors promo	Float
Promo_Costs	Coûts promotionnels totaux	Float
Filter	Flag In_Scope (0/1)	Integer
ROI	Return on Investment	Float
Volume Uplift	Uplift en %	Float
Incremental Margin	Marge incrémentale	Float
Sell-in discount	Discount en % (1 décimale)	Float
First Week	Première semaine promo	Integer
Last Week	Dernière semaine promo	Integer
High-level classification	Catégorie générale	String
Segmentation	Type d'événement	String
Event duration	Durée en semaines	Integer

2. Baseline_Table.xlsx¶

Colonne	Description	Format
Retailer_name	Nom du retailer	String
EAN	Code produit	String
EAN_desc	Description produit (sauf Parmalat)	String
Year	Année	Integer
Week	Numéro de semaine	Integer
Baseline_Volume	Volume baseline hebdo	Float

Gestion des erreurs¶

1. Tables sources manquantes¶

try:
    df_step_6_03 = pd.read_sql_query(query_step_6_03, engine)
except Exception as e:
    log_message(f"ERROR: Could not read step_6_03_final_with_scope_promo_table: {e}")
    raise

2. Colonnes manquantes¶

Protection implicite via sélection explicite :

# Si colonne manque, pandas lèvera KeyError
df_step_6_03 = df_step_6_03[columns_to_select_6_03]

3. Données vides¶

if df_step_6_03.empty:
    log_message("WARNING: No data found for max year in promotions table")
    # Continue quand même pour créer fichier vide

4. Erreurs d'export¶

try:
    df_step_6_03.to_excel(output_path, index=False)
except PermissionError:
    log_message(f"ERROR: Cannot write to {output_path} - file may be open")
    raise
except Exception as e:
    log_message(f"ERROR: Failed to export Excel: {e}")
    raise

Performance et optimisation¶

1. Requêtes optimisées¶

Utilisation de sous-requêtes pour MAX(Year)
Sélection uniquement des colonnes nécessaires
CAST explicite pour types (Week AS INTEGER)

2. Mémoire¶

Pas de copie inutile des DataFrames
Sélection colonnes avant renommage
Export direct sans transformation intermédiaire

3. Logs structurés¶

log_message("Loading data for step_6_03...")
# Action
log_message("Data loaded for step_6_03.")

Points d'attention maintenance¶

1. Synchronisation colonnes¶

Si ajout/modification colonnes dans étapes précédentes : 1. Mettre à jour columns_to_select_6_03 2. Adapter le mapping de renommage 3. Vérifier ordre des colonnes

2. Nouveaux business units¶

Pour ajouter un business unit sans EAN_desc :

# Ajouter à la condition
is_special_case = business_unit.lower() in ['italy_parmalat', 'new_unit']

3. Format Excel¶

Utilise openpyxl par défaut
Index=False pour éviter colonne index
Encoding UTF-8 automatique

4. Chemins de sortie¶

Structure fixe : /app/outputs/{business_unit}/ - Créer le répertoire si inexistant - Vérifier permissions écriture

Troubleshooting¶

Problème : Année max différente entre tables¶

Symptôme : Incohérence temporelle entre les deux exports

Diagnostic :

SELECT 
    'promotions' as source,
    MAX(Year) as max_year
FROM step_6_03_final_with_scope_promo_table
UNION ALL
SELECT 
    'baseline' as source,
    MAX(Year) as max_year
FROM step_3_02_model_data_baseline;

Solution : Utiliser année commune

# Année de référence commune
max_year_query = """
SELECT MIN(max_year) as reference_year
FROM (
    SELECT MAX(Year) as max_year FROM step_6_03_final_with_scope_promo_table
    UNION ALL
    SELECT MAX(Year) as max_year FROM step_3_02_model_data_baseline
) t
"""
reference_year = pd.read_sql_query(max_year_query, engine).iloc[0, 0]

Problème : Fichier Excel corrompu¶

Symptôme : Excel ne peut pas ouvrir le fichier

Causes possibles : - Caractères spéciaux dans les données - Valeurs infinies ou NaN - Taille excessive

Solution :

# Nettoyage avant export
df_step_6_03 = df_step_6_03.replace([np.inf, -np.inf], np.nan)
df_step_6_03 = df_step_6_03.fillna('')

# Limite taille si nécessaire
if len(df_step_6_03) > 1048576:  # Limite Excel
    log_message("WARNING: Truncating to Excel row limit")
    df_step_6_03 = df_step_6_03.head(1048576)

Problème : Colonnes dans le désordre¶

Symptôme : Ordre des colonnes incorrect dans Excel

Solution : Forcer l'ordre

# Définir ordre explicite
column_order = ['Retailer_name', 'EAN', 'EAN_desc', ...]
df_step_6_03 = df_step_6_03[column_order]

Exemples d'utilisation¶

Export avec filtrage additionnel¶

def extract_and_export_data_filtered(engine, retailer_filter=None):
    # Query modifiée
    where_clause = "WHERE Year = (SELECT MAX(Year) FROM step_6_03_final_with_scope_promo_table)"
    if retailer_filter:
        where_clause += f" AND Retailer_name = '{retailer_filter}'"

    query_step_6_03 = f"""
    SELECT *
    FROM step_6_03_final_with_scope_promo_table
    {where_clause}
    """

Export multi-format¶

def export_multiple_formats(df, base_path):
    # Excel standard
    df.to_excel(f"{base_path}.xlsx", index=False)

    # CSV pour compatibilité
    df.to_csv(f"{base_path}.csv", index=False, encoding='utf-8-sig')

    # Parquet pour performance
    df.to_parquet(f"{base_path}.parquet", index=False)

Validation pré-export¶

def validate_data_before_export(df):
    issues = []

    # Vérifier colonnes requises
    required_cols = ['Retailer_name', 'EAN', 'Promotion_Code']
    missing = [col for col in required_cols if col not in df.columns]
    if missing:
        issues.append(f"Missing columns: {missing}")

    # Vérifier données manquantes
    null_counts = df[required_cols].isnull().sum()
    if null_counts.any():
        issues.append(f"Null values found: {null_counts[null_counts > 0].to_dict()}")

    # Vérifier types
    if 'Year' in df.columns and not pd.api.types.is_integer_dtype(df['Year']):
        issues.append("Year column is not integer type")

    if issues:
        log_message(f"WARNING: Data validation issues: {issues}")

    return len(issues) == 0

Export avec métadonnées¶

def add_metadata_sheet(excel_path):
    """Ajoute une feuille de métadonnées au fichier Excel"""
    from openpyxl import load_workbook

    wb = load_workbook(excel_path)
    ws = wb.create_sheet("Metadata", 0)  # Première position

    metadata = [
        ["Generated", datetime.now().strftime("%Y-%m-%d %H:%M:%S")],
        ["Business Unit", business_unit],
        ["Source Tables", "step_6_03_final_with_scope_promo_table, step_3_02_model_data_baseline"],
        ["Pipeline Version", "1.0"],
        ["Row Count", len(df_step_6_03)]
    ]

    for row in metadata:
        ws.append(row)

    wb.save(excel_path)

Configuration par JSON¶

def load_export_config():
    """Charge configuration d'export depuis JSON"""
    config_path = f"/app/inputs/{business_unit}/export_config.json"

    if os.path.exists(config_path):
        with open(config_path, 'r') as f:
            config = json.load(f)

        return {
            'columns_to_export': config.get('simulation_columns', []),
            'rename_mapping': config.get('column_names', {}),
            'filters': config.get('data_filters', {}),
            'format_options': config.get('excel_format', {})
        }
    else:
        return None  # Utiliser config par défaut