ML Adoption Playbook — LeadFlow CRM

# Data contract: validación de schema en runtime
from pydantic import BaseModel, Field
from typing import Optional, Literal

class LeadFeatures(BaseModel):
    source: Literal['organic', 'paid', 'referral', 'event', 'unknown']
    company_size: Literal['S', 'M', 'L', 'XL']
    industry: Optional[str] = 'other'
    pages_visited: int = Field(..., ge=0, le=1000)
    email_opens: int = Field(..., ge=0, le=500)
    demo_requested: bool = False

# Nótese: days_to_close EXCLUIDO (leakage)

import joblib
from django.conf import settings
from .data_contract import LeadFeatures

class LeadScoringService:
    """Servicio desacoplado: única interfaz entre Django y el modelo."""

    def __init__(self):
        self._model = None

    def _load(self):
        if self._model is None:
            self._model = joblib.load(settings.LEAD_SCORING_MODEL_PATH)

    def score(self, features: LeadFeatures) -> dict:
        # Feature flag: desactivado → fallback a regla
        if not settings.FEATURE_FLAGS.get('ml_lead_scoring'):
            return self._fallback_rule(features)
        try:
            self._load()
            prob = self._model.predict_proba([features.dict()])[0][1]
            return {'score': round(prob, 3), 'source': 'ml'}
        except Exception as e:
            # Fallback automático si el modelo falla (<200ms SLA)
            return self._fallback_rule(features)

    def _fallback_rule(self, f: LeadFeatures) -> dict:
        score = 0.1
        if f.demo_requested: score += 0.4
        if f.email_opens > 3:  score += 0.2
        if f.pages_visited > 5: score += 0.15
        return {'score': round(score, 3), 'source': 'rule'}

import numpy as np
import joblib, json
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_auc_score, classification_report

SEED = 42
np.random.seed(SEED)

# --- Paso 1: Split cronológico (evitar leakage temporal) ---
df_train = df[df['created_at'] < '2025-01-01']
df_test  = df[df['created_at'] >= '2025-01-01']

FEATURES_NUM = ['pages_visited', 'email_opens', 'demo_requested']
FEATURES_CAT = ['source', 'company_size', 'industry']
TARGET = 'converted'

# --- Paso 2: Pipeline (baseline: LogReg) ---
preprocessor = ColumnTransformer([
    ('num', StandardScaler(), FEATURES_NUM),
    ('cat', OneHotEncoder(handle_unknown='ignore'), FEATURES_CAT),
])
model = Pipeline([
    ('prep', preprocessor),
    ('clf', LogisticRegression(
        class_weight='balanced', random_state=SEED, max_iter=500
    ))
])

# --- Paso 3: Entrenar y evaluar ---
model.fit(df_train[FEATURES_NUM + FEATURES_CAT], df_train[TARGET])
y_prob = model.predict_proba(df_test[FEATURES_NUM + FEATURES_CAT])[:,1]
auc = roc_auc_score(df_test[TARGET], y_prob)
print(f"AUC-ROC: {auc:.4f}")  # objetivo: > 0.78

# --- Paso 4: Guardar artefacto ---
joblib.dump(model, 'models/lead_scoring_v1.pkl')
json.dump({'auc': auc, 'seed': SEED, 'version': '1.0'},
         open('models/eval_report.json', 'w'))

- name: Evaluate lead scoring model
  run: |
    python leadflow/ml/train.py --eval-only
    AUC=$(jq '.auc' models/eval_report.json)
    if (( $(echo "$AUC < 0.75" | bc -l) )); then
      echo "::error::AUC $AUC below threshold 0.75"
      exit 1
    fi