Aller au contenu

Step_06_Building_Promo_Analysis - Analyse des marges et cannibalisation

Vue d'ensemble

Cette étape calcule les marges promotionnelles et les effets de cannibalisation en trois phases distinctes : calcul des marges (main_margin_computation), valorisation des effets baseline (main_valuation_computation), et calcul de la cannibalisation (main_cannibalization_computation). C'est l'étape qui transforme les volumes en valeur financière.

Objectif principal

  • Calculer les marges promotionnelles ajustées par produit
  • Valoriser les effets de volume (baseline, uplift, halo, pantry loading)
  • Calculer la cannibalisation intra-catégorie
  • Préparer les données pour l'analyse finale des promotions

Position dans la pipeline

  • Étape précédente : Step_05_Building_Baseline
  • Étape suivante : Step_07_Building_Products_Classification et Substep_601
  • Appels distincts :
  • main_margin_computation : Calcul des marges unitaires
  • main_valuation_computation : Multiplication volumes × marges
  • main_cannibalization_computation : Répartition cannibalisation

Architecture technique

Flux de données

Phase 1 : Calcul des marges
───────────────────────────
Entrée
    ├── step_2_03_sell_in_base (données transactionnelles)
    └── step_4_01_model_with_effects_model_data_baseline

Traitement
    ├── Galbani
    │   ├── Adjusted_gross_margin = GM + (Supply_chain_gap × Volumes)
    │   └── Promo_margin = Moyenne pondérée par EAN/Year
    │
    └── Parmalat
        └── Récupération marges pré-calculées (Step 3)

Sortie → step_5_01_model_with_margin_kpi

Phase 2 : Valorisation baseline
──────────────────────────────
Entrée
    └── step_5_01_model_with_margin_kpi

Traitement
    ├── Baseline_GM_excl_promo_adjustment = Baseline_Volume × Promo_margin
    └── Colonnes conservées pour compatibilité

Sortie → step_5_02_model_margins

Phase 3 : Cannibalisation
─────────────────────────
Entrée
    └── step_5_02_model_margins

Traitement
    ├── Agrégation famille (New_category × Retailer × Year × Week)
    ├── Ratio_Cannibalizing = Uplift_i / ΣUplift_famille
    ├── Cannibalizing_Volume = Ratio × Family_Cannib_Volume
    └── Cannibalizing_GM = Volume × Ponderated_Family_GM

Sortie → step_5_03_model_cannibalization_effects

Concepts clés

1. Marges ajustées (Galbani)

  • Adjusted_gross_margin : Marge brute + écart supply chain
  • Promo_margin : Moyenne pondérée par volumes sur l'année
  • Agrégation : EAN × EAN_desc × Year

2. Marges pré-calculées (Parmalat)

  • Utilisation Promo_margin_per_unit de Step 3
  • Méthodologie : réallocation contribution marginale
  • Agrégation : EAN × Year (sans EAN_desc)

3. Valorisation des effets

Multiplication simple : Volume × Marge unitaire - Baseline_GM_excl_promo_adjustment - Plus tard : Uplift_GM, Halo_GM, Pantry_Loading_GM, Cannib_GM

4. Cannibalisation

  • Famille : produits de même catégorie
  • Ratio : part de l'uplift du produit dans l'uplift famille
  • Répartition : proportionnelle à la contribution uplift

Implémentation détaillée

1. Phase 1 : Calcul des marges

Galbani : compute_margin_italy_galbani()

# Calcul marge ajustée
df_sell_in['Adjusted_gross_margin'] = (
    df_sell_in['Gross_margin'] + 
    (df_sell_in['Supply_chain_gap'] * df_sell_in['Volumes'])
)

# Moyenne pondérée par produit/année
def compute_weighted_average(data):
    weighted_margin_sum = np.sum(data['Adjusted_gross_margin'])
    total_volumes = np.sum(data['Volumes'])
    if total_volumes > 0:
        return weighted_margin_sum / total_volumes
    else:
        return np.nan

weighted_promo_margin = df_sell_in.groupby(['EAN', 'EAN_desc', 'Year']).apply(
    compute_weighted_average
).reset_index(name='Promo_margin')

Parmalat : compute_margin_italy_parmalat()

# Récupération marges existantes
if 'Promo_margin_per_unit' not in df_sell_in.columns:
    log_message("WARNING: Promo_margin_per_unit not found")
    # Fallback calculation
    df_sell_in['Promo_margin_total'] = (
        df_sell_in['Product_margin'] + 
        df_sell_in['Supply_chain_gap'] + 
        df_sell_in['Fixed_industrial_costs']
    )
    df_sell_in['Promo_margin_per_unit'] = (
        df_sell_in['Promo_margin_total'] / df_sell_in['Volumes']
    )

# Agrégation sans EAN_desc
weighted_promo_margin = df_sell_in.groupby(['EAN', 'Year']).apply(
    aggregate_weighted_margin
).reset_index(name='Promo_margin')

Default : compute_margin_default()

Pour autres business units : 

# Somme tous les coûts promo
promo_columns = [
    'On_invoice_promo_costs_*',
    'Promo_funding_off_invoice_*',
    'Other_off_invoice_promo_costs',
    'Other_promo_costs'
]
df['Total_Promo_Costs'] = df[promo_columns].sum(axis=1)

# Marge hors promo
df['Gross_Margin_excluding_all_promo_costs'] = (
    df['Gross_margin'] - df['Total_Promo_Costs']
)
df['Gross_Margin_excluding_all_promo_costs_per_unit'] = (
    df['Gross_Margin_excluding_all_promo_costs'] / df['Volumes']
)

2. Phase 2 : Valorisation baseline

Identique pour tous : `compute_valuation_baseline_effects_*()``

# Simple multiplication
df['Baseline_GM_excl_promo_adjustment'] = (
    df['Baseline_Volume'] * df['Promo_margin']
)

Note : Les autres valorisations (Uplift_GM, etc.) sont calculées plus tard ou dans d'autres étapes.

3. Phase 3 : Cannibalisation

Galbani : compute_cannibalization_italy_galbani()

Spécificités : - Famille = New_category - Ratio basé sur Uplift_Volume uniquement - Pas de Halo dans le calcul

# Marge cannibalisée originale
df['Cannib_GM'] = df['Cannib_Volume'] * df['Promo_margin']

# Agrégation famille
family_aggregates = df.groupby(['New_category', 'Retailer_name', 'Year', 'Week']).agg({
    'Uplift_Volume': 'sum',  # → Family_Uplift_Volume
    'Cannib_Volume': 'sum',  # → Family_Cannib_Volume
    'Cannib_GM': 'sum'       # → Total_Family_Cannib_GM
})

# Marge pondérée famille
family['Ponderated_Family_Cannib_GM'] = (
    family['Total_Family_Cannib_GM'] / family['Family_Cannib_Volume']
)

# Répartition
df['Ratio_Cannibalizing'] = safe_divide(
    df['Uplift_Volume'], 
    df['Family_Uplift_Volume']
)
df['Cannibalizing_Volume'] = df['Ratio_Cannibalizing'] * df['Family_Cannib_Volume']
df['Cannibalizing_GM'] = (
    df['Ratio_Cannibalizing'] * 
    df['Family_Cannib_Volume'] * 
    df['Ponderated_Family_Cannib_GM']
)

Parmalat : compute_cannibalization_italy_parmalat()

Restriction importante : 

# Cannibalisation uniquement pour le lait UHT
CANNIBALIZATION_CATEGORIES = ['UHT normal milk', 'UHT lactose free milk']

Traitement : - Catégories UHT : calcul normal - Autres catégories : tous metrics = 0

Default : compute_cannibalization_default()

  • Famille = Category
  • Ratio inclut Halo_Volume : (Uplift + Halo) / (Family_Uplift + Family_Halo)
  • Applicable à tous produits

Gestion des erreurs

Division par zéro

Fonction safe_divide() systématique : 

def safe_divide(numerator, denominator, default_value=0):
    safe_denominator = denominator.replace({0: np.nan})
    safe_numerator = numerator.replace({np.inf: np.nan, -np.inf: np.nan})
    result = safe_numerator.div(safe_denominator).fillna(default_value)
    return result

Colonnes manquantes

# Parmalat margin fallback
if 'Promo_margin_per_unit' not in df_sell_in.columns:
    log_message("WARNING: Promo_margin_per_unit not found")
    # Calcul alternatif

Volumes nuls

# Protection moyenne pondérée
if total_volumes > 0:
    return weighted_margin_sum / total_volumes
else:
    return np.nan

Performance et optimisation

1. Agrégations efficaces

  • GroupBy avec fonctions natives (sum, mean)
  • Apply uniquement pour calculs complexes
  • Éviter les boucles sur DataFrame

2. Gestion mémoire

# Remplacement explicite pour éviter warning
pd.set_option('future.no_silent_downcasting', True)
family_aggregates['Family_Cannib_Volume'] = (
    family_aggregates['Family_Cannib_Volume'].replace(0, np.nan)
)

3. Réduction requêtes

  • Une seule lecture par table source
  • Agrégations pandas plutôt que SQL
  • Sauvegarde unique en fin de traitement

Points d'attention maintenance

1. Cohérence agrégations

  • Galbani : toujours EAN × EAN_desc
  • Parmalat : toujours EAN seul
  • Default : adapter selon structure

2. Colonnes marges

Noms différents selon business unit : - Galbani : Gross_margin, Supply_chain_gap - Parmalat : Product_margin, Fixed_industrial_costs - Default : liste exhaustive coûts promo

3. Catégories cannibalisation

  • Galbani : toutes New_category
  • Parmalat : seulement lait UHT
  • Default : toutes Category

4. Ordre d'exécution

  1. main_margin_computation
  2. main_valuation_computation
  3. main_cannibalization_computation

Important : Respecter l'ordre, chaque phase dépend de la précédente

Troubleshooting

Problème : Marges négatives ou extrêmes

Diagnostic : 

SELECT EAN, Year, Promo_margin, 
       COUNT(*) as occurrences
FROM step_5_01_model_with_margin_kpi
WHERE Promo_margin < -10 OR Promo_margin > 100
GROUP BY EAN, Year, Promo_margin;

Causes : - Volumes très faibles → division instable - Données financières incohérentes - Écarts supply chain anormaux

Problème : Cannibalisation = 0 partout

Vérifications : 1. Cannib_Volume dans données source ? 2. Catégories correctes (Parmalat : UHT uniquement) ? 3. Family aggregates calculés ?

-- Vérifier volumes cannibalisation source
SELECT New_category, 
       SUM(CASE WHEN Cannib_Volume < 0 THEN 1 ELSE 0 END) as negative_cannib,
       COUNT(*) as total_rows
FROM step_5_02_model_margins
GROUP BY New_category;

Problème : Ratio_Cannibalizing > 1

Symptôme : Un produit cannibalisé plus que l'uplift famille

Cause : Incohérence agrégation famille

Solution : 

# Ajouter contrôle
df['Ratio_Cannibalizing'] = df['Ratio_Cannibalizing'].clip(upper=1.0)

Exemples d'utilisation

Ajout nouveau business unit

def compute_margin_new_country(engine):
    # Adapter colonnes selon structure
    df_sell_in = pd.read_sql("SELECT * FROM sell_in_base", engine)

    # Calcul marge spécifique
    df_sell_in['Specific_margin'] = ...

    # Agrégation produit
    margin_by_product = df_sell_in.groupby(['EAN', 'Year']).agg({
        'Specific_margin': 'sum',
        'Volumes': 'sum'
    })

    # Merge et sauvegarde
    ...

Modification catégories cannibalisation

# Dans compute_cannibalization_*
CANNIBALIZATION_CATEGORIES = [
    'Category_A', 
    'Category_B',
    'New_Category_C'  # Ajout
]

Debug marges par retailer

# Ajouter avant agrégation finale
debug_margins = df_sell_in.groupby(['Retailer_name', 'EAN', 'Year']).agg({
    'Adjusted_gross_margin': 'sum',
    'Volumes': 'sum',
    'Promo_margin': lambda x: (x * df_sell_in.loc[x.index, 'Volumes']).sum() / df_sell_in.loc[x.index, 'Volumes'].sum()
})
debug_margins.to_csv('debug_margins_by_retailer.csv')

Forcer recalcul marges Parmalat

# Dans compute_margin_italy_parmalat()
# Forcer recalcul même si colonnes existent
FORCE_RECALCULATE = True

if FORCE_RECALCULATE or 'Promo_margin_per_unit' not in df_sell_in.columns:
    # Recalcul complet
    df_sell_in['Promo_margin_total'] = ...