Data Scientist Journey: User Churn Prediction for a B2B SaaS Product

  Current State                         Target State
  ┌────────────────────────┐            ┌────────────────────────┐
  │ CSM learns about       │            │ Model flags account    │
  │ churn at cancellation   │            │ 60 days before renewal │
  │                        │            │                        │
  │ Reactive: 0 days lead  │  ────▶     │ Proactive: 60 day lead │
  │ time, 5% save rate     │            │ time, 35% save rate    │
  │                        │            │                        │
  │ Revenue churn: 25%     │            │ Revenue churn target:  │
  │                        │            │ < 15%                  │
  └────────────────────────┘            └────────────────────────┘

{
  "source": "salesforce",
  "connector": "airbyte/source-salesforce",
  "config": {
    "streams": ["Account", "Opportunity", "Task", "Note"],
    "sync_mode": "incremental",
    "cursor_field": "SystemModstamp",
    "frequency": "every_30_minutes"
  },
  "schema_mapping": {
    "Account": {
      "destination_table": "crm_accounts",
      "primary_key": "Id",
      "field_mapping": {
        "Id": "account_id",
        "CloudFlow_Workspace_Id__c": "workspace_id",
        "OwnerId": "csm_owner",
        "Health_Score__c": "health_score",
        "Contract_End_Date__c": "renewal_date",
        "ARR__c": "arr"
      }
    }
  }
}

{
  "source": "zendesk",
  "connector": "airbyte/source-zendesk-support",
  "config": {
    "streams": ["tickets", "ticket_comments", "satisfaction_ratings"],
    "sync_mode": "incremental",
    "cursor_field": "updated_at",
    "frequency": "every_15_minutes"
  }
}

-- Data Workbench: Catalog profiling query
SELECT
    event_type,
    COUNT(*)                                    AS event_count,
    COUNT(DISTINCT user_id)                     AS unique_users,
    COUNT(DISTINCT workspace_id)                AS unique_workspaces,
    MIN(timestamp)                              AS first_seen,
    MAX(timestamp)                              AS last_seen
FROM events
WHERE timestamp >= CURRENT_DATE - INTERVAL '90' DAY
GROUP BY event_type
ORDER BY event_count DESC
LIMIT 20;

-- Profile monthly active days per user
SELECT
    CASE
        WHEN active_days >= 20 THEN 'Power User (20+ days)'
        WHEN active_days >= 10 THEN 'Regular (10-19 days)'
        WHEN active_days >= 5  THEN 'Light (5-9 days)'
        ELSE 'At Risk (< 5 days)'
    END                          AS user_segment,
    COUNT(*)                     AS user_count,
    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 1) AS pct
FROM (
    SELECT user_id, COUNT(DISTINCT DATE(timestamp)) AS active_days
    FROM events
    WHERE timestamp >= CURRENT_DATE - INTERVAL '30' DAY
    GROUP BY user_id
) user_activity
GROUP BY 1
ORDER BY 2 DESC;

event (raw)
  └──▶ session_aggregate (Pipeline Service, hourly)
         └──▶ user_activity_daily (Pipeline Service, daily)
                └──▶ workspace_health_score (Pipeline Service, daily)
                       └──▶ crm_accounts.health_score (Salesforce sync, bidirectional)

-- Feature: Usage frequency trends with rolling windows
-- Federated query: PostgreSQL (users) + Snowflake (events)
WITH daily_usage AS (
    SELECT
        u.workspace_id,
        DATE(e.timestamp)                       AS activity_date,
        COUNT(DISTINCT e.user_id)               AS active_users,
        COUNT(*)                                AS total_events,
        COUNT(DISTINCT e.event_type)            AS distinct_actions,
        COUNT(CASE WHEN e.event_type IN ('task_created', 'task_completed',
              'comment_added', 'file_uploaded') THEN 1 END) AS core_actions
    FROM snowflake.events e
    JOIN postgresql.users u ON e.user_id = u.user_id
    WHERE e.timestamp >= CURRENT_DATE - INTERVAL '90' DAY
    GROUP BY u.workspace_id, DATE(e.timestamp)
),
rolling_features AS (
    SELECT
        workspace_id,
        activity_date,
        -- 7-day rolling averages
        AVG(active_users)   OVER (PARTITION BY workspace_id
            ORDER BY activity_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW)
            AS avg_dau_7d,
        AVG(core_actions)   OVER (PARTITION BY workspace_id
            ORDER BY activity_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW)
            AS avg_core_actions_7d,
        -- 30-day rolling averages
        AVG(active_users)   OVER (PARTITION BY workspace_id
            ORDER BY activity_date ROWS BETWEEN 29 PRECEDING AND CURRENT ROW)
            AS avg_dau_30d,
        AVG(core_actions)   OVER (PARTITION BY workspace_id
            ORDER BY activity_date ROWS BETWEEN 29 PRECEDING AND CURRENT ROW)
            AS avg_core_actions_30d
    FROM daily_usage
)
SELECT
    workspace_id,
    activity_date,
    avg_dau_7d,
    avg_dau_30d,
    -- Usage trend: ratio of recent to longer-term average
    -- < 1.0 means declining usage (churn signal)
    ROUND(avg_dau_7d / NULLIF(avg_dau_30d, 0), 3)        AS usage_trend_ratio,
    ROUND(avg_core_actions_7d / NULLIF(avg_core_actions_30d, 0), 3)
                                                           AS action_trend_ratio
FROM rolling_features
WHERE activity_date = CURRENT_DATE - INTERVAL '1' DAY;

-- Feature: Feature adoption depth per workspace
WITH feature_usage AS (
    SELECT
        u.workspace_id,
        CASE
            WHEN e.event_type LIKE 'board_%'       THEN 'kanban_boards'
            WHEN e.event_type LIKE 'gantt_%'        THEN 'gantt_charts'
            WHEN e.event_type LIKE 'timeline_%'     THEN 'timelines'
            WHEN e.event_type LIKE 'automation_%'   THEN 'automations'
            WHEN e.event_type LIKE 'form_%'         THEN 'forms'
            WHEN e.event_type LIKE 'dashboard_%'    THEN 'dashboards'
            WHEN e.event_type LIKE 'integration_%'  THEN 'integrations'
            WHEN e.event_type LIKE 'report_%'       THEN 'reports'
            WHEN e.event_type LIKE 'api_%'          THEN 'api_access'
            WHEN e.event_type = 'invite_sent'       THEN 'team_invites'
        END AS feature_category
    FROM snowflake.events e
    JOIN postgresql.users u ON e.user_id = u.user_id
    WHERE e.timestamp >= CURRENT_DATE - INTERVAL '30' DAY
      AND e.event_type NOT IN ('page_viewed', 'session_start', 'session_end')
)
SELECT
    workspace_id,
    COUNT(DISTINCT feature_category)          AS features_adopted,
    ARRAY_AGG(DISTINCT feature_category)      AS feature_list
FROM feature_usage
WHERE feature_category IS NOT NULL
GROUP BY workspace_id;

-- Feature: Team collaboration patterns + support ticket sentiment
SELECT
    w.workspace_id,
    -- Collaboration signals
    COUNT(DISTINCT CASE WHEN e.event_type = 'comment_added'
          THEN e.user_id END)                         AS users_commenting,
    COUNT(CASE WHEN e.event_type = 'invite_sent'
          THEN 1 END)                                 AS invites_sent_30d,
    w.seat_count,
    ROUND(COUNT(DISTINCT e.user_id) * 1.0
          / NULLIF(w.seat_count, 0), 2)               AS seat_utilization,
    -- Support signals (from Zendesk)
    COUNT(DISTINCT t.ticket_id)                       AS support_tickets_30d,
    AVG(t.satisfaction_rating)                         AS avg_csat,
    COUNT(CASE WHEN t.priority IN ('high', 'urgent')
          THEN 1 END)                                 AS urgent_tickets_30d,
    -- Billing signals (from Stripe)
    s.plan                                            AS current_plan,
    CASE WHEN s.mrr < LAG(s.mrr) OVER (PARTITION BY s.workspace_id
         ORDER BY s.period_start) THEN 1 ELSE 0 END  AS downgraded_last_period
FROM postgresql.workspaces w
JOIN snowflake.events e ON w.workspace_id = e.workspace_id
    AND e.timestamp >= CURRENT_DATE - INTERVAL '30' DAY
LEFT JOIN zendesk.support_tickets t ON w.workspace_id = t.workspace_id
    AND t.created_at >= CURRENT_DATE - INTERVAL '30' DAY
LEFT JOIN stripe.subscriptions s ON w.workspace_id = s.workspace_id
    AND s.status = 'active'
GROUP BY w.workspace_id, w.seat_count, s.plan, s.mrr, s.period_start;

{
  "pipeline": {
    "name": "churn-prediction-daily",
    "schedule": "0 6 * * *",
    "description": "Daily churn risk scoring for all active workspaces",
    "stages": [
      {
        "name": "extract_features",
        "type": "sql_transform",
        "query_ref": "churn_features_v3.sql",
        "output_table": "ml.churn_features_daily",
        "timeout_minutes": 30
      },
      {
        "name": "quality_gate",
        "type": "data_quality",
        "depends_on": ["extract_features"],
        "checks": [
          {
            "name": "event_data_freshness",
            "type": "freshness",
            "table": "events",
            "column": "timestamp",
            "max_stale_hours": 2,
            "severity": "critical",
            "on_failure": "abort_pipeline"
          },
          {
            "name": "feature_completeness",
            "type": "column_not_null",
            "table": "ml.churn_features_daily",
            "columns": ["workspace_id", "avg_dau_7d", "features_adopted"],
            "max_null_pct": 5,
            "severity": "warning"
          },
          {
            "name": "workspace_count",
            "type": "row_count",
            "table": "ml.churn_features_daily",
            "min_rows": 3500,
            "max_rows": 5000,
            "severity": "critical"
          }
        ]
      },
      {
        "name": "score_accounts",
        "type": "model_inference",
        "depends_on": ["quality_gate"],
        "model": "churn-predictor-v3",
        "model_registry": "ml_workbench",
        "input_table": "ml.churn_features_daily",
        "output_table": "ml.churn_scores_daily",
        "batch_size": 500
      },
      {
        "name": "flag_at_risk",
        "type": "sql_transform",
        "depends_on": ["score_accounts"],
        "query": "INSERT INTO ml.at_risk_accounts SELECT workspace_id, churn_probability, risk_tier, scored_at FROM ml.churn_scores_daily WHERE churn_probability >= 0.65",
        "output_table": "ml.at_risk_accounts"
      },
      {
        "name": "sync_to_crm",
        "type": "reverse_sync",
        "depends_on": ["flag_at_risk"],
        "destination": "salesforce",
        "object": "Account",
        "mapping": {
          "workspace_id": "CloudFlow_Workspace_Id__c",
          "churn_probability": "Churn_Risk_Score__c",
          "risk_tier": "Risk_Tier__c"
        }
      }
    ]
  }
}

Pipeline Run: 2026-02-28 06:00 UTC
  ├── extract_features .............. PASSED (12m 34s)
  ├── quality_gate
  │   ├── event_data_freshness ..... PASSED (latest event: 05:58 UTC, lag: 2 min)
  │   ├── feature_completeness ..... PASSED (null rate: 1.2%)
  │   └── workspace_count .......... PASSED (4,187 workspaces scored)
  ├── score_accounts ............... PASSED (3m 12s)
  ├── flag_at_risk ................. PASSED (847 accounts flagged)
  └── sync_to_crm .................. PASSED (847 records synced)

-- Analyze churn rates by customer segment
SELECT
    CASE
        WHEN s.mrr >= 5000  THEN 'Enterprise ($5K+)'
        WHEN s.mrr >= 1000  THEN 'Mid-Market ($1K-$5K)'
        WHEN s.mrr >= 200   THEN 'SMB ($200-$1K)'
        ELSE 'Self-Serve (< $200)'
    END                                             AS segment,
    COUNT(DISTINCT w.workspace_id)                  AS total_accounts,
    COUNT(DISTINCT CASE WHEN w.status = 'cancelled'
          THEN w.workspace_id END)                  AS churned,
    ROUND(COUNT(DISTINCT CASE WHEN w.status = 'cancelled'
          THEN w.workspace_id END) * 100.0
          / COUNT(DISTINCT w.workspace_id), 1)      AS logo_churn_pct,
    SUM(CASE WHEN w.status = 'cancelled'
        THEN s.mrr ELSE 0 END)                      AS churned_mrr,
    ROUND(SUM(CASE WHEN w.status = 'cancelled' THEN s.mrr ELSE 0 END)
          * 100.0 / SUM(s.mrr), 1)                  AS revenue_churn_pct
FROM workspaces w
JOIN subscriptions s ON w.workspace_id = s.workspace_id
WHERE s.period_start >= CURRENT_DATE - INTERVAL '365' DAY
GROUP BY 1
ORDER BY revenue_churn_pct DESC;

-- Churn rate by number of core features adopted
SELECT
    features_adopted,
    COUNT(*)                                        AS account_count,
    COUNT(CASE WHEN churned = 1 THEN 1 END)         AS churned_count,
    ROUND(COUNT(CASE WHEN churned = 1 THEN 1 END)
          * 100.0 / COUNT(*), 1)                    AS churn_rate
FROM ml.churn_features_daily
GROUP BY features_adopted
ORDER BY features_adopted;

-- Ensure training data includes churned accounts from the past 12 months
SELECT
    CASE WHEN status = 'cancelled' THEN 'Churned' ELSE 'Active' END AS label,
    COUNT(*)                                                          AS count,
    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 1)               AS pct
FROM workspaces
WHERE created_at <= CURRENT_DATE - INTERVAL '90' DAY  -- exclude very new accounts
GROUP BY 1;

# ML Workbench: Notebook - churn_model_training_v3.ipynb
import pandas as pd
from sklearn.model_selection import train_test_split, StratifiedKFold
from sklearn.metrics import roc_auc_score, precision_recall_curve, auc
import lightgbm as lgb
import mlflow
 
# Load feature set from Query Engine
features_df = matih.query("""
    SELECT * FROM ml.churn_training_features
    WHERE snapshot_date BETWEEN DATE '2025-03-01' AND DATE '2026-02-28'
""")
 
# Feature columns
feature_cols = [
    'avg_dau_7d', 'avg_dau_14d', 'avg_dau_30d',
    'usage_trend_ratio_7d_30d', 'action_trend_ratio',
    'features_adopted', 'seat_utilization',
    'invites_sent_30d', 'users_commenting',
    'support_tickets_30d', 'urgent_tickets_30d', 'avg_csat',
    'days_since_last_login', 'mrr', 'account_age_days',
    'downgraded_last_period', 'nps_score'
]
 
X = features_df[feature_cols]
y = features_df['churned_within_60d']
 
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, stratify=y, random_state=42
)
 
# Train gradient boosting model
with mlflow.start_run(run_name="churn-gbm-v3"):
    model = lgb.LGBMClassifier(
        n_estimators=500,
        learning_rate=0.05,
        max_depth=6,
        num_leaves=31,
        min_child_samples=20,
        subsample=0.8,
        colsample_bytree=0.8,
        class_weight='balanced',
        random_state=42
    )
    model.fit(
        X_train, y_train,
        eval_set=[(X_test, y_test)],
        callbacks=[lgb.early_stopping(50)]
    )
 
    # Evaluate
    y_pred = model.predict_proba(X_test)[:, 1]
    auc_roc = roc_auc_score(y_test, y_pred)
    precision, recall, _ = precision_recall_curve(y_test, y_pred)
    auc_pr = auc(recall, precision)
 
    mlflow.log_metric("auc_roc", auc_roc)       # 0.83
    mlflow.log_metric("auc_pr", auc_pr)          # 0.61
    mlflow.log_metric("precision_at_80_recall", 0.52)
    mlflow.sklearn.log_model(model, "churn-predictor-v3")

{
  "deployment": {
    "name": "churn-predictor-v3",
    "model_uri": "models:/churn-predictor-v3/production",
    "serving": {
      "realtime": {
        "endpoint": "/api/v1/ml/churn-score",
        "runtime": "ray_serve",
        "num_replicas": 2,
        "max_batch_size": 32,
        "latency_sla_ms": 200,
        "trigger": "user_login_event"
      },
      "batch": {
        "schedule": "0 6 * * *",
        "pipeline": "churn-prediction-daily",
        "output_table": "ml.churn_scores_daily"
      }
    },
    "crm_sync": {
      "destination": "salesforce",
      "object": "Account",
      "field": "Churn_Risk_Score__c",
      "update_frequency": "daily"
    }
  }
}

{
  "monitoring": {
    "model": "churn-predictor-v3",
    "metrics": [
      {
        "name": "weekly_auc_roc",
        "type": "model_performance",
        "metric": "auc_roc",
        "window": "7d",
        "threshold": 0.78,
        "alert": "slack://ml-alerts"
      },
      {
        "name": "monthly_lift_chart",
        "type": "model_performance",
        "metric": "lift_at_decile",
        "window": "30d",
        "report": "bi_workbench://churn-model-performance"
      },
      {
        "name": "csm_intervention_tracking",
        "type": "business_outcome",
        "metric": "save_rate_for_flagged_accounts",
        "window": "90d",
        "target": 0.35
      }
    ],
    "drift_detection": [
      {
        "feature": "usage_trend_ratio_7d_30d",
        "type": "distribution_shift",
        "method": "psi",
        "threshold": 0.15,
        "note": "Seasonal patterns expected in Dec/Jan -- adjust threshold"
      },
      {
        "feature": "avg_dau_7d",
        "type": "distribution_shift",
        "method": "psi",
        "threshold": 0.10
      }
    ],
    "alerts": [
      {
        "condition": "auc_roc < 0.78 for 2 consecutive weeks",
        "action": "trigger_retraining_pipeline",
        "notify": ["zara@cloudflow.io", "slack://ml-alerts"]
      },
      {
        "condition": "prediction_volume drops > 20%",
        "action": "page_on_call",
        "note": "Likely ingestion pipeline failure, not model issue"
      }
    ]
  }
}

-- Monthly intervention outcome tracking
SELECT
    DATE_TRUNC('month', s.scored_at)        AS score_month,
    s.risk_tier,
    COUNT(DISTINCT s.workspace_id)          AS accounts_flagged,
    COUNT(DISTINCT CASE WHEN i.intervention_date IS NOT NULL
          THEN s.workspace_id END)          AS accounts_contacted,
    COUNT(DISTINCT CASE WHEN w.status = 'cancelled'
          AND w.cancelled_at <= s.scored_at + INTERVAL '90' DAY
          THEN s.workspace_id END)          AS actually_churned,
    ROUND(COUNT(DISTINCT CASE WHEN w.status = 'cancelled'
          AND w.cancelled_at <= s.scored_at + INTERVAL '90' DAY
          THEN s.workspace_id END) * 100.0
          / COUNT(DISTINCT s.workspace_id), 1) AS churn_rate
FROM ml.churn_scores_daily s
LEFT JOIN csm_interventions i ON s.workspace_id = i.workspace_id
    AND i.intervention_date BETWEEN s.scored_at AND s.scored_at + INTERVAL '14' DAY
LEFT JOIN workspaces w ON s.workspace_id = w.workspace_id
WHERE s.risk_tier = 'high'
GROUP BY 1, 2
ORDER BY 1;

{
  "experiment": {
    "name": "csm-outreach-strategy-q1-2026",
    "hypothesis": "Personalized outreach based on specific churn risk factors will have higher save rates than generic check-in calls",
    "variants": [
      {
        "name": "control",
        "description": "Standard quarterly check-in call",
        "allocation": 0.33
      },
      {
        "name": "risk_personalized",
        "description": "Outreach with specific risk factors: 'We noticed your team has not used [feature] recently -- can we schedule a training session?'",
        "allocation": 0.34
      },
      {
        "name": "value_demo",
        "description": "Proactive ROI report showing workspace productivity gains and cost savings",
        "allocation": 0.33
      }
    ],
    "primary_metric": "90_day_retention_rate",
    "secondary_metrics": ["expansion_mrr", "nps_change", "csm_time_per_account"],
    "duration_weeks": 12,
    "min_sample_per_variant": 100
  }
}