Cliente si iscrive → ASPETTA 10 minuti → INVIA welcome email

Cliente aggiunge prodotto al carrello → ENTRO 15 minuti senza completare → INVIA email urgente "Il tuo articolo e in esaurimento"

Il modello di machine learning predice che questo cliente ha 75% di probabilita di churnare entro 30 giorni → INVIA offerta di ritenzione personalizzata immediatamente

import requests
import psycopg2
from datetime import datetime, timedelta
import math
import numpy as np

DB_PARAMS = {
    "dbname": "analytics",
    "user": "dbuser",
    "host": "localhost",
    "password": "secretpassword"
}
SLACK_WEBHOOK = "https://hooks.slack.com/services/T00000/B00000/XXXXXXXX"

# --- Soglie Configurabili ---
THRESHOLD_CR = 0.02  # 2% Conversion Rate minimo accettabile
THRESHOLD_TRAFFIC_SPIKE = 1.5  # +50% rispetto alla baseline
CHURN_SCORE_THRESHOLD = 0.75  # 75% probabilita di churn

def check_conversion_rate():
    """Calcola il CR dell'ultima ora e confronta con baseline a 7 giorni."""
    conn = psycopg2.connect(**DB_PARAMS)
    cur = conn.cursor()

    # CR dell'ultima ora
    cur.execute("""
        SELECT
            COUNT(CASE WHEN event = 'purchase' THEN 1 END) * 1.0
                / NULLIF(COUNT(DISTINCT user_id), 0) AS cr_latest_hour,
            COUNT(DISTINCT user_id) AS utenti_ultima_ora,
            COUNT(CASE WHEN event = 'purchase' THEN 1 END) AS acquisti_ultima_ora
        FROM events
        WHERE timestamp > NOW() - INTERVAL '1 hour'
    """)
    cr_latest, utenti_attivi, acquisti = cur.fetchone()

    # Baseline: media dell'ultima settimana, stessa ora del giorno
    cur.execute("""
        SELECT AVG(hourly_cr) AS baseline_cr
        FROM (
            SELECT
                COUNT(CASE WHEN event = 'purchase' THEN 1 END) * 1.0 /
                NULLIF(COUNT(DISTINCT user_id), 0) AS hourly_cr
            FROM events
            WHERE timestamp BETWEEN NOW() - INTERVAL '8 days' AND NOW() - INTERVAL '1 day'
              AND EXTRACT(HOUR FROM timestamp) = EXTRACT(HOUR FROM NOW())
            GROUP BY DATE_TRUNC('hour', timestamp)
        ) sub
    """)
    baseline_cr = cur.fetchone()[0]
    conn.close()

    # Controllo: volume minimo per significativita
    if utenti_attivi < 50:
        return

    # Trigger se CR scende sotto soglia OR scende >30% vs baseline
    if baseline_cr:
        drop_pct = (baseline_cr - cr_latest) / baseline_cr * 100 if cr_latest else 100
    else:
        drop_pct = 0

    if cr_latest and cr_latest < THRESHOLD_CR:
        severity = "CRITICA"
    elif drop_pct > 30:
        severity = "ALTA"
    else:
        return

    message = (
        f"*{severity}: CALO CONVERSION RATE*\n"
        f"CR ultima ora: *{cr_latest:.2%}*\n"
        f"Baseline 7gg stessa ora: {baseline_cr:.2%}\n"
        f"Calo: *{drop_pct:.0f}%*\n"
        f"Volume: {utenti_attivi} visitatori, {acquisti} acquisti\n"
        f"Timestamp: {datetime.now().strftime('%H:%M del %d/%m/%Y')}\n"
        f"\n_Azioni rapide:_\n"
        f"• Verifica: bug nel checkout? problemi con i gateway di pagamento?\n"
        f"• Controlla: c'e un'ondata di bot visitatori?\n"
        f"• Slack: #data-critical per escalation"
    )
    send_slack_alert(message, channel="#critical-alerts", severity=severity)

def check_traffic_spike():
    """Rileva picchi di traffico anomalo basato su deviazione standard."""
    conn = psycopg2.connect(**DB_PARAMS)
    cur = conn.cursor()

    # Statistiche: media e stddev delle ultime 2 settimane alla stessa ora
    cur.execute("""
        WITH baseline AS (
            SELECT
                COUNT(DISTINCT session_id) AS hourly_sessions,
                STDDEV_POP(COUNT(DISTINCT session_id)) OVER () AS sessions_stddev
            FROM events
            WHERE timestamp BETWEEN NOW() - INTERVAL '14 days' AND NOW() - INTERVAL '1 hour'
              AND EXTRACT(HOUR FROM timestamp) = EXTRACT(HOUR FROM NOW())
            GROUP BY DATE_TRUNC('hour', timestamp)
        ),
        current_hour AS (
            SELECT COUNT(DISTINCT session_id) AS sessions_now FROM events
            WHERE timestamp > NOW() - INTERVAL '1 hour'
        ),
        stats AS (
            SELECT AVG(hourly_sessions) AS avg_sessions, AVG(sessions_stddev) AS stddev_sessions
            FROM baseline
        )
        SELECT c.sessions_now, s.avg_sessions, s.stddev_sessions
        FROM current_hour c, stats s
    """)

    result = cur.fetchone()
    conn.close()

    if not result or not result[1]:
        return

    sessions_now, avg_sessions, stddev = result

    # Z-score: quanto le sessioni di adesso differiscono dalla media in termine di deviazioni standard
    z_score = (sessions_now - avg_sessions) / stddev if stddev > 0 else 0

    # Trigger se z-score > 2.5 (98.7° percentile di anormalita) o > 50% spike
    ratio = sessions_now / avg_sessions if avg_sessions > 0 else 0

    if z_score > 2.5 or ratio > THRESHOLD_TRAFFIC_SPIKE:
        message = (
            f"*PICCO DI TRAFFICO ANOMALO*\n"
            f"Sessioni ultima ora: *{int(sessions_now)}*\n"
            f"Media (2 settimane): {int(avg_sessions)} ± {int(stddev)}\n"
            f"Z-Score: *{z_score:.1f}σ* | Ratio: *{ratio:.1f}x*\n"
            f"\n_Probabili cause:_\n"
            f"• Traffic virale (check Twitter, news, PR)\n"
            f"• Attacco bot (check CloudFlare / WAF logs)\n"
            f"• Press coverage (check press releases tempo)\n"
            f"• Campagna accidentalmente lanciata"
        )
        send_slack_alert(message, channel="#data-alerts")

def check_churn_predictions():
    """Carica i punteggi di churn calcolati dal modello ML e identifica clienti ad alto rischio."""
    conn = psycopg2.connect(**DB_PARAMS)
    cur = conn.cursor()

    # Clienti ad altissimo rischio churn che non sono stati contattati ieri
    cur.execute("""
        SELECT user_id, email, churn_score, monetary_value, last_purchase_date
        FROM customer_churn_scores
        WHERE churn_score > %s
          AND date_calculated = CURRENT_DATE()
          AND (
            SELECT COUNT(*) FROM retention_campaigns
            WHERE user_id = customer_churn_scores.user_id
              AND sent_date > NOW() - INTERVAL '1 day'
          ) = 0
        ORDER BY churn_score DESC
        LIMIT 50
    """, (CHURN_SCORE_THRESHOLD,))

    high_risk_customers = cur.fetchall()
    conn.close()

    if high_risk_customers:
        # Prepara le azioni: email, SMS, account manager assignment
        customer_list = "\n".join([
            f"  • {user_id} ({email}): churn={score:.1%}, LTV={monetary:.0f}€"
            for user_id, email, score, monetary, last_purchase in high_risk_customers
        ])

        message = (
            f"*ALERT: {len(high_risk_customers)} CLIENTI AD ALTISSIMO RISCHIO CHURN*\n"
            f"Churn Score > 75% identificati oggi\n\n"
            f"{customer_list}\n\n"
            f"_Azioni automatiche:_\n"
            f"✅ Email win-back personalizzata schedulata\n"
            f"⏳ Task per account manager assegnati\n"
            f"📱 SMS offer code inviato"
        )
        send_slack_alert(message, channel="#retention-team")

        # Trigger automazione in Braze / Klaviyo
        trigger_retention_campaign(high_risk_customers)

def send_slack_alert(message: str, channel: str = "#alerts", severity: str = "INFO"):
    """Invia notifica su Slack via webhook con colore indicativo della severity."""
    color_map = {
        "CRITICA": "#ff0000",
        "ALTA": "#ff8800",
        "MEDIA": "#ffff00",
        "INFO": "#0099ff"
    }

    payload = {
        "text": message,
        "channel": channel,
        "username": "Analytics Trigger Bot",
        "attachments": [
            {
                "color": color_map.get(severity, "#0099ff"),
                "footer": f"Severity: {severity}"
            }
        ]
    }
    response = requests.post(SLACK_WEBHOOK, json=payload, timeout=10)
    response.raise_for_status()

def trigger_retention_campaign(customers):
    """Invia API request a Braze / Klaviyo per attivare campagna di retention."""
    for user_id, email, churn_score, monetary, last_purchase in customers:
        payload = {
            "user_id": user_id,
            "email": email,
            "churn_probability": churn_score,
            "ltv": monetary,
            "trigger_name": "high_churn_score_trigger"
        }
        # Usa l'API di Braze / Klaviyo per mandar e il trigger
        # requests.post("https://api.braze.com/users/trigger/send", json=payload)
        pass

if __name__ == "__main__":
    # Questo script gira ogni 15 minuti via cron job
    # */15 * * * * /usr/bin/python3 /path/to/marketing_triggers.py

    print(f"[{datetime.now()}] Running trigger checks...")
    check_conversion_rate()
    check_traffic_spike()
    check_churn_predictions()
    print(f"[{datetime.now()}] Trigger checks completed.")

🚨 CONVERSION RATE CALO 35%
CR ultima ora: 1.4% (baseline: 2.1%)
Sessioni: 450 | Acquisti: 6 (baseline: 9)

VERIFICA SUBITO:
1. Bug checkout? (Sentry logs)
2. Problema payment gateway? (Stripe status)
3. Traffico bot inflazionato? (CloudFlare)

Day 0 — 00:00: Signup
  ↓ Trigger: account_created
    ✉️ Welcome email (brand story, prodotti bestseller)

Day 0 — 24:00:
  ↓ Se NO email aperta
    ✉️ Reminder email (incentivo con discount "welcome only")

Day 2:
  ↓ Trigger: vedi il comportamento del nuovo utente
    ✉️ Se ha visitato Product XYZ ma non acquistato
      → "Questa feature ti interessava: costo solo €"

Day 7:
  ↓ Trigger: se non ancora convertito
    🎁 Email: "Sconto 15% solo per te, ultimi 3 giorni"

Purchase event → Trigger immediato
  ↓
  ✉️ Order confirmation (con tracking link)
  📱 SMS "Il tuo ordine e in spedizione"

Day 1 post-delivery:
  ↓ Trigger: delivery_confirmed
    ✉️ Email di post-acquisto "Come ti e andato?"
    📞 Review solicitation (soprattutto se ordine > €100)

Routine: ogni 7 giorni se inattivo
  ↓ Trigger: 7_day_inactivity
    ✉️ "Clienti come te adorano [related products]"

Ogni 14 giorni: upsell
  ↓ Trigger: 14_day_engagement
    ✉️ "Upgrade a [premium plan] e ottieni 2x la capacita"

Trigger: customer_clv > €500 + 3 mesi attivo
  ↓
  💎 Exclusive: accesso early-access a nuove features
  📞 Dedicated account manager offer
  💰 VIP discount (15% off annual plan)

Monthly check:
  ↓ Se usage crolla del 50%
    ✉️ "Ti manca qualcosa? Ecco le feature che non conosci"
    📞 Quick call con support team

Quartic warning:
  ↓ Se intent segnali (checking competitors) rilevati
    💎 Retention offer (15% + extra credits)

Trigger: 60 giorni inattivita (clienti ex-paying)
  ↓
  ✉️ "Torniamo insieme: offerta speciale 50% off per 3 mesi"

Trigger: 120 giorni inattivita (clienti free)
  ↓ Cancella account email
    💬 "Se torni, riattivo il tuo account con i tuoi dati"

-- Query per monitorare l'efficacia dei trigger ogni giorno
SELECT
  trigger_name,
  DATE(trigger_fired_at) AS data,
  COUNT(*) AS trigger_count,
  COUNT(DISTINCT CASE WHEN user_opened_email THEN user_id END) AS email_opens,
  COUNT(DISTINCT CASE WHEN user_clicked_email THEN user_id END) AS email_clicks,
  COUNT(DISTINCT CASE WHEN user_converted THEN user_id END) AS conversions,

  ROUND(
    COUNT(DISTINCT CASE WHEN user_opened_email THEN user_id END) * 100.0
    / NULLIF(COUNT(*), 0), 2
  ) AS open_rate_pct,

  ROUND(
    COUNT(DISTINCT CASE WHEN user_converted THEN user_id END) * 100.0
    / NULLIF(COUNT(*), 0), 2
  ) AS trigger_conversion_rate,

  ROUND(
    (COUNT(DISTINCT CASE WHEN user_converted THEN user_id END) * avg_order_value) -
    (email_send_cost * COUNT(*)),
    2
  ) AS roi_totale

FROM trigger_events
WHERE trigger_fired_at >= CURRENT_DATE() - INTERVAL 30 DAY
GROUP BY 1, 2
ORDER BY data DESC, conversion_rate DESC;