SaaS Churn Prediction: Early Warning Signals, ML Models, and Retention Playbooks

Churn is the silent killer of SaaS businesses. A 5% monthly churn means you replace your entire customer base every 20 months — a treadmill that no acquisition strategy can outrun. The antidote isn't reactive support tickets; it's predicting who will churn before they decide to leave, then intervening with targeted retention actions.

This post covers the engineering side of churn prediction: the behavioral signals that predict churn, building a health score, training a churn prediction model with scikit-learn, and automating retention playbooks triggered by risk signals.

---

The Churn Prediction Stack

Product Events (Kafka / PostHog)
    │
Feature Engineering (Prefect daily pipeline)
    │
Health Score (rule-based, interpretable)
    │
Churn Probability Model (XGBoost, weekly inference)
    │
Risk Segmentation (High / Medium / Low)
    │
Retention Playbooks (automated + CS-assisted)
    │
Feedback Loop (outcome tracking → model retraining)

---

Churn Signals by Category

Not all signals are equal. Some predict churn weeks in advance; others are lagging indicators.

---

Health Score Implementation

A health score (0–100) gives Customer Success a single number to prioritize attention. Keep it simple and interpretable — black-box scores that nobody trusts don't get used.

# src/health_score.py
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional
import math

@dataclass
class CustomerMetrics:
    customer_id: str
    # Usage
    dau_last_30: float          # Daily active users, last 30 days avg
    feature_adoption_pct: float # % of core features used in last 30 days
    api_calls_last_7d: int
    api_calls_prev_7d: int      # For trend calculation
    # Engagement
    logins_last_30d: int
    logins_prev_30d: int
    support_tickets_open: int
    last_support_sentiment: Optional[float]  # -1 to 1
    # Commercial
    seats_used: int
    seats_contracted: int
    days_until_renewal: int
    payment_overdue_days: int
    # Feedback
    nps_score: Optional[int]    # 0–10
    last_qbr_days_ago: Optional[int]

def compute_health_score(m: CustomerMetrics) -> dict:
    """
    Compute 0–100 health score with component breakdown.
    Higher = healthier.
    """
    scores = {}

    # Usage component (30 points)
    usage_score = 0
    # DAU trend: compare last 7 to prior 7 (normalized to 0–10)
    api_trend = (m.api_calls_last_7d - m.api_calls_prev_7d) / max(m.api_calls_prev_7d, 1)
    usage_score += max(0, min(10, 5 + api_trend * 10))  # 5 is neutral, trend ±5
    # Feature adoption
    usage_score += m.feature_adoption_pct * 10
    # Login frequency
    login_trend = (m.logins_last_30d - m.logins_prev_30d) / max(m.logins_prev_30d, 1)
    usage_score += max(0, min(10, 5 + login_trend * 10))
    scores['usage'] = round(usage_score)

    # Support component (20 points)
    support_score = 20
    support_score -= min(m.support_tickets_open * 4, 12)  # -4 per open ticket, max -12
    if m.last_support_sentiment is not None:
        # Sentiment -1 to 1 → 0 to 8 points
        support_score += (m.last_support_sentiment + 1) / 2 * 8 - 4
    scores['support'] = round(max(0, support_score))

    # Seat utilization component (20 points)
    utilization = m.seats_used / max(m.seats_contracted, 1)
    if utilization >= 0.8:
        seat_score = 20        # High utilization = happy customer
    elif utilization >= 0.5:
        seat_score = 14
    elif utilization >= 0.3:
        seat_score = 8
    else:
        seat_score = 2         # Low utilization = at-risk
    scores['seat_utilization'] = seat_score

    # Commercial health component (20 points)
    commercial_score = 20
    if m.payment_overdue_days > 0:
        commercial_score -= min(m.payment_overdue_days * 2, 15)
    if m.days_until_renewal <= 30:
        commercial_score -= 5   # Renewal pressure
    scores['commercial'] = round(max(0, commercial_score))

    # Feedback component (10 points)
    feedback_score = 5  # Neutral if no data
    if m.nps_score is not None:
        if m.nps_score >= 9:
            feedback_score = 10   # Promoter
        elif m.nps_score >= 7:
            feedback_score = 7    # Passive
        else:
            feedback_score = 2    # Detractor
    if m.last_qbr_days_ago is not None and m.last_qbr_days_ago > 90:
        feedback_score -= 2       # No QBR in 90+ days
    scores['feedback'] = round(max(0, feedback_score))

    total = sum(scores.values())
    risk_level = (
        'critical' if total < 30
        else 'high' if total < 50
        else 'medium' if total < 70
        else 'low'
    )

    return {
        'customer_id': m.customer_id,
        'health_score': total,
        'risk_level': risk_level,
        'components': scores,
        'computed_at': datetime.utcnow().isoformat(),
    }

---

Churn Prediction Model

The health score is interpretable but rule-based. A trained model captures non-linear interactions between signals.

# src/churn_model.py
import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split, StratifiedKFold, cross_val_score
from sklearn.metrics import roc_auc_score, classification_report
import mlflow
import mlflow.sklearn
import shap

FEATURE_COLS = [
    # Usage
    'avg_dau_30d', 'feature_adoption_pct', 'api_call_trend_7d',
    'login_trend_30d', 'days_since_last_login',
    # Support
    'open_tickets', 'avg_ticket_sentiment', 'ticket_volume_trend',
    # Commercial
    'seat_utilization', 'days_until_renewal', 'payment_overdue_days',
    'months_as_customer', 'mrr_usd',
    # Feedback
    'nps_score', 'last_nps_days_ago',
    # Health
    'health_score',  # Include as a feature — captures domain expertise
]

TARGET_COL = 'churned_90d'  # 1 if customer churned within 90 days

def prepare_features(df: pd.DataFrame) -> tuple[np.ndarray, np.ndarray]:
    X = df[FEATURE_COLS].copy()
    y = df[TARGET_COL].values
    return X.values, y

def train_churn_model(df: pd.DataFrame) -> str:
    X, y = prepare_features(df)
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, stratify=y, random_state=42
    )

    mlflow.set_experiment("churn-prediction")

    with mlflow.start_run() as run:
        pipeline = Pipeline([
            ('imputer', SimpleImputer(strategy='median')),
            ('scaler', StandardScaler()),
            ('model', GradientBoostingClassifier(
                n_estimators=300,
                max_depth=4,
                learning_rate=0.05,
                subsample=0.8,
                min_samples_leaf=15,
                random_state=42,
            )),
        ])

        pipeline.fit(X_train, y_train)

        # Evaluation
        y_proba = pipeline.predict_proba(X_test)[:, 1]
        y_pred = (y_proba >= 0.5).astype(int)

        auc = roc_auc_score(y_test, y_proba)
        mlflow.log_metrics({
            "auc_roc": auc,
            "churn_rate_actual": y_test.mean(),
        })

        # Cross-validation
        cv_scores = cross_val_score(
            pipeline, X, y, cv=StratifiedKFold(5), scoring='roc_auc'
        )
        mlflow.log_metrics({
            "cv_auc_mean": cv_scores.mean(),
            "cv_auc_std": cv_scores.std(),
        })

        # SHAP values for feature importance (model explainability)
        explainer = shap.TreeExplainer(pipeline.named_steps['model'])
        shap_values = explainer.shap_values(
            pipeline.named_steps['scaler'].transform(
                pipeline.named_steps['imputer'].transform(X_test)
            )
        )

        feature_importance = pd.DataFrame({
            'feature': FEATURE_COLS,
            'mean_abs_shap': np.abs(shap_values).mean(axis=0),
        }).sort_values('mean_abs_shap', ascending=False)

        mlflow.log_text(
            feature_importance.to_string(), "feature_importance.txt"
        )
        print(f"\nTop 5 churn drivers:\n{feature_importance.head(5).to_string()}")
        print(f"\nAUC-ROC: {auc:.4f} | CV: {cv_scores.mean():.4f} ± {cv_scores.std():.4f}")

        # Log model
        mlflow.sklearn.log_model(
            pipeline, "model",
            registered_model_name="churn-predictor",
            input_example=pd.DataFrame([X_test[0]], columns=FEATURE_COLS),
        )

        return run.info.run_id

---

Retention Playbooks

Risk segmentation triggers specific playbooks for Customer Success:

// src/retention/playbooks.ts
interface ChurnRisk {
  customerId: string;
  riskLevel: 'critical' | 'high' | 'medium' | 'low';
  churnProbability: number;
  topSignals: string[];
  csOwner: string;
  renewalDate: Date;
}

const PLAYBOOKS: Record<string, Playbook> = {
  // Critical risk (<30 days to churn, >80% probability)
  critical_imminent: {
    name: 'Critical: Executive Escalation',
    trigger: (r) => r.riskLevel === 'critical' && daysUntil(r.renewalDate) < 30,
    actions: [
      { type: 'create_task', owner: 'cs_manager', title: 'Executive call within 48h', priority: 'urgent' },
      { type: 'slack_alert', channel: '#cs-escalations', message: (r) =>
        `🚨 CRITICAL CHURN RISK: ${r.customerId} | ${Math.round(r.churnProbability * 100)}% probability | ${daysUntil(r.renewalDate)} days to renewal` },
      { type: 'send_email', template: 'executive_outreach', from: 'ceo@myapp.com' },
    ],
  },

  // High risk, usage drop is primary signal
  high_usage_drop: {
    name: 'High Risk: Usage Recovery',
    trigger: (r) => r.riskLevel === 'high' && r.topSignals.includes('login_trend_30d'),
    actions: [
      { type: 'create_task', owner: r => r.csOwner, title: 'Proactive health check call', dueInDays: 3 },
      { type: 'send_email', template: 'usage_recovery', includeUsageReport: true },
      { type: 'in_app_message', template: 'feature_tips', targetUserId: 'admin' },
    ],
  },

  // Medium risk, approaching renewal
  medium_renewal: {
    name: 'Medium Risk: Pre-Renewal Check-In',
    trigger: (r) => r.riskLevel === 'medium' && daysUntil(r.renewalDate) < 60,
    actions: [
      { type: 'create_task', owner: r => r.csOwner, title: 'Renewal conversation', dueInDays: 7 },
      { type: 'send_email', template: 'renewal_value_summary' },
    ],
  },
};

export async function executeRetentionPlaybooks(risks: ChurnRisk[]): Promise<void> {
  for (const risk of risks) {
    for (const [key, playbook] of Object.entries(PLAYBOOKS)) {
      if (playbook.trigger(risk)) {
        // Check if this playbook already ran for this customer recently
        const alreadyRan = await db.query(
          `SELECT 1 FROM retention_actions
           WHERE customer_id = $1 AND playbook = $2
             AND created_at > now() - interval '14 days'`,
          [risk.customerId, key],
        );

        if (!alreadyRan.rows.length) {
          await executePlaybook(playbook, risk);
          await db.query(
            `INSERT INTO retention_actions (customer_id, playbook, risk_level, churn_probability)
             VALUES ($1, $2, $3, $4)`,
            [risk.customerId, key, risk.riskLevel, risk.churnProbability],
          );
        }
        break; // Only execute the highest-priority matching playbook
      }
    }
  }
}

---

Churn Cohort Analysis SQL

-- Monthly churn rate by customer segment
WITH cohorts AS (
  SELECT
    customer_id,
    date_trunc('month', subscription_start) AS cohort_month,
    mrr_usd,
    CASE
      WHEN mrr_usd >= 1000 THEN 'enterprise'
      WHEN mrr_usd >= 200  THEN 'mid_market'
      ELSE                      'smb'
    END AS segment
  FROM customers
  WHERE subscription_start >= now() - interval '24 months'
),
churn_events AS (
  SELECT customer_id, date_trunc('month', churned_at) AS churn_month
  FROM customers
  WHERE churned_at IS NOT NULL
)
SELECT
  c.cohort_month,
  c.segment,
  count(DISTINCT c.customer_id)                                           AS cohort_size,
  count(DISTINCT ce.customer_id)                                          AS churned,
  round(count(DISTINCT ce.customer_id)::numeric / count(DISTINCT c.customer_id) * 100, 1) AS churn_rate_pct,
  sum(CASE WHEN ce.customer_id IS NOT NULL THEN c.mrr_usd ELSE 0 END)    AS mrr_lost
FROM cohorts c
LEFT JOIN churn_events ce
  ON ce.customer_id = c.customer_id
  AND ce.churn_month = c.cohort_month + interval '3 months'  -- 3-month churn window
GROUP BY 1, 2
ORDER BY 1 DESC, 2;

---

Churn Prediction ROI

A well-tuned churn prediction system typically prevents 10–25% of predicted churn through targeted interventions, paying back the engineering investment within 2–4 months at $1M+ ARR.

---

Working With Viprasol

Our data and AI team builds churn prediction systems for SaaS products — from behavioral health scores through ML churn models and automated CS playbooks.

What we deliver:

• Behavioral health score design and implementation
• XGBoost/GBM churn prediction model with SHAP explainability
• Prefect pipeline for weekly risk scoring
• CS dashboard with risk segmentation and top signals
• Automated playbook triggers (Slack alerts, CRM tasks, email sequences)

→ Discuss your retention strategy → AI and machine learning services

---

See Also

• SaaS Onboarding Engineering
• MLOps in Production
• Startup Growth Metrics
• Real-Time Analytics with ClickHouse
• SaaS Pricing Strategy