Tests et optimisation

Section 14.8.3 : Tests et optimisation

🎯 Objectif pédagogique

Tester, monitorer et optimiser le RevOps stack en conditions réelles : data quality, model monitoring, et itération continue.

Data quality : la fondation

## Tests de qualité des données (automatisés)

### Freshness checks
- Les données CRM sont-elles à jour ? (< 1h de retard)
- Les scores ML sont-ils recalculés ? (< 24h)
- Les dashboards sont-ils rafraîchis ? (< 15 min)

### Completeness checks
- Quel % de contacts ont un email valide ?
- Quel % de deals ont un montant renseigné ?
- Quel % de comptes ont un health score ?

### Consistency checks
- Le pipeline total dans le CRM = pipeline dans le dashboard ?
- Le nombre de MQLs dans HubSpot = MQLs dans le rapport ?
- Les scores ML dans le CRM = scores dans le modèle ?

### Anomaly detection
- Le nombre de MQLs a-t-il changé de > 50% vs semaine dernière ?
- Un commercial a-t-il 0 activité depuis 3 jours ?
- Un health score a-t-il chuté de > 30 points en 1 semaine ?

ML Model monitoring

# Model monitoring — détecter le drift

class ModelMonitor:
    def __init__(self, model, baseline_metrics):
        self.model = model
        self.baseline = baseline_metrics

    def check_prediction_drift(self, recent_predictions):
        """Vérifier si la distribution des scores a changé"""
        baseline_mean = self.baseline['score_mean']
        baseline_std = self.baseline['score_std']

        recent_mean = recent_predictions.mean()
        recent_std = recent_predictions.std()

        drift_score = abs(recent_mean - baseline_mean) / baseline_std

        if drift_score > 2:
            alert("CRITICAL: Score distribution drift detected! "
                  f"Baseline mean: {baseline_mean:.1f}, "
                  f"Current mean: {recent_mean:.1f}")
            return 'critical'
        elif drift_score > 1:
            alert("WARNING: Moderate score drift detected.")
            return 'warning'
        return 'ok'

    def check_feature_drift(self, recent_features):
        """Vérifier si les features d entrée ont changé"""
        for feature in recent_features.columns:
            baseline_stats = self.baseline['feature_stats'][feature]
            current_mean = recent_features[feature].mean()
            drift = abs(current_mean - baseline_stats['mean']) / baseline_stats['std']

            if drift > 3:
                alert(f"Feature drift: {feature} "
                      f"(baseline: {baseline_stats['mean']:.2f}, "
                      f"current: {current_mean:.2f})")

    def check_performance(self, predictions, actuals):
        """Vérifier que la performance n a pas dégradé"""
        from sklearn.metrics import roc_auc_score
        current_auc = roc_auc_score(actuals, predictions)
        baseline_auc = self.baseline['auc']

        degradation = (baseline_auc - current_auc) / baseline_auc

        if degradation > 0.1:  # > 10% de dégradation
            alert(f"Model performance degraded! "
                  f"Baseline AUC: {baseline_auc:.3f}, "
                  f"Current AUC: {current_auc:.3f}")
            return False
        return True

Boucle d'optimisation

Checklist d'optimisation

## Optimisations du stack par ordre de priorité

### Quick Wins (Semaine 1)
- [ ] Ajouter des alertes Slack pour les anomalies
- [ ] Créer un dashboard de monitoring des modèles
- [ ] Documenter les runbooks (que faire si X casse)

### Moyen terme (Mois 1)
- [ ] A/B test des seuils de scoring (Hot > 70 vs Hot > 60)
- [ ] Ajouter des features d enrichissement (Clearbit)
- [ ] Optimiser les séquences de nurturing (timing, contenu)

### Long terme (Trimestre)
- [ ] Re-training automatique mensuel des modèles
- [ ] Feedback loop : les commerciaux valident les scores
- [ ] Expansion des agents IA (plus de tools, plus de data sources)