Summary  
The chapter demonstrates how to use SQL joins, GROUP BY, aggregate functions, and date-interval filtering to identify entities that lack recent activity or events.

General domain of usage  
Customer churn prediction

Churn prediction is a crucial analytical process for identifying customers or users who are likely to stop using a product or service. In the context of SaaS or subscription businesses, **churn prediction** typically relies on analyzing user activity and transaction patterns to flag those at high risk of leaving. By understanding and detecting these behavioral signals early, companies can proactively engage with at-risk users, aiming to retain more customers and reduce revenue loss.

To predict churn, you often analyze user activity logs, transaction histories, and subscription records. A common SQL approach is to flag users who have not performed any activity—such as logins, purchases, or payments—within a recent time window (for example, the last 90 days). This is especially relevant for SaaS platforms, where regular engagement is expected. You can adapt this logic to your available data—such as using order dates, payment dates, or subscription changes as proxies for engagement.

Churn refers to the loss of customers or users who stop using a product or service. In SQL analytics, common churn indicators include **extended periods of inactivity** (e.g., no logins or transactions for 30, 60, or 90 days), **subscription downgrades**, **cancellations**, or **payment failures**.

Definition

SELECT
    u.user_id,
    u.username,
    u.full_name,
    MAX(p.payment_date) AS last_payment
FROM users u
LEFT JOIN subscriptions s ON u.user_id = s.user_id
LEFT JOIN payments p ON s.subscription_id = p.subscription_id
GROUP BY u.user_id, u.username, u.full_name
HAVING
    MAX(p.payment_date) IS NULL
    OR MAX(p.payment_date) < '2025-01-01'::date - INTERVAL '90 days'
ORDER BY last_payment ASC NULLS FIRST;

Suppose the SaaS customer success team needs a current list of users who are likely to churn, based on recent inactivity. They want to prioritize outreach to these users to prevent churn and improve retention.

SELECT
    u.user_id,
    u.username,
    u.full_name,
    MAX(p.payment_date) AS last_payment
FROM
    users u
LEFT JOIN
    subscriptions s ON u.user_id = s.user_id
LEFT JOIN
    payments p ON s.subscription_id = p.subscription_id AND p.status = 'Completed'
GROUP BY
    u.user_id, u.username, u.full_name
HAVING
    MAX(p.payment_date) IS NULL
    OR MAX(p.payment_date) < CURRENT_DATE - INTERVAL '90 days';

import unittest
import psycopg2
import pandas as pd
import re


REFERENCE_DATE = '2023-01-01'


class TestPostgresQuery(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.connection = psycopg2.connect(
            host="localhost",
            user="agent",
            password="qwerty",
            dbname="testdb",
            sslmode="disable"
        )
        cls.cursor = cls.connection.cursor()

    def _execute_student_query(self):
        with open("query.sql", "r") as f:
            query = f.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0].lower() for desc in self.cursor.description]
        return pd.DataFrame(rows, columns=columns)

    def test_expired_subscriptions_included(self):
        """Check that users with Expired subscriptions after reference date are included."""
        df = self._execute_student_query()
        self.cursor.execute("""
            SELECT u.user_id, u.username, u.full_name, s.plan, s.status, s.end_date
            FROM users u
            JOIN subscriptions s ON u.user_id = s.user_id
            WHERE s.status = 'Expired'
              AND s.end_date >= %s
        """, (REFERENCE_DATE,))
        expected = pd.DataFrame(
            self.cursor.fetchall(),
            columns=["user_id", "username", "full_name", "plan", "status", "end_date"]
        )
        missing = []
        for _, row in expected.iterrows():
            mask = (df["user_id"] == row["user_id"]) & (df["status"] == row["status"])
            if not mask.any():
                missing.append(row.to_dict())
        _dynamic_test(
            self,
            len(missing) == 0,
            "All users with Expired subscriptions after reference date are included.",
            f"Missing Expired users: {missing}"
        )

    def test_plan_changed_included(self):
        """Check that users whose plan differs from plan_at_signup are included."""
        df = self._execute_student_query()
        self.cursor.execute("""
            SELECT u.user_id, u.username, u.full_name, s.plan, s.status, s.end_date
            FROM users u
            JOIN subscriptions s ON u.user_id = s.user_id
            WHERE s.plan <> u.plan_at_signup
              AND s.end_date >= %s
        """, (REFERENCE_DATE,))
        expected = pd.DataFrame(
            self.cursor.fetchall(),
            columns=["user_id", "username", "full_name", "plan", "status", "end_date"]
        )
        missing = []
        for _, row in expected.iterrows():
            mask = (df["user_id"] == row["user_id"]) & (df["plan"] == row["plan"])
            if not mask.any():
                missing.append(row.to_dict())
        _dynamic_test(
            self,
            len(missing) == 0,
            "All users with plan change after reference date are included.",
            f"Missing plan-changed users: {missing}"
        )

    def test_no_subscriptions_before_reference_date(self):
        """Check that no rows with end_date before reference date are included."""
        df = self._execute_student_query()
        if df.empty:
            _dynamic_test(self, True, "No rows returned, nothing to check.", "")
            return
        df["end_date"] = pd.to_datetime(df["end_date"])
        too_old = df[df["end_date"] < pd.to_datetime(REFERENCE_DATE)]
        _dynamic_test(
            self,
            too_old.empty,
            "No subscriptions with end_date before reference date included.",
            f"Rows with end_date before {REFERENCE_DATE}: {too_old.to_dict('records')}"
        )

    def test_no_false_positives(self):
        """Check that no rows are included that don't meet Expired or plan-change condition."""
        df = self._execute_student_query()
        self.cursor.execute("""
            SELECT u.user_id, s.plan, u.plan_at_signup, s.status, s.end_date
            FROM users u
            JOIN subscriptions s ON u.user_id = s.user_id
            WHERE s.end_date >= %s
        """, (REFERENCE_DATE,))
        all_recent = pd.DataFrame(
            self.cursor.fetchall(),
            columns=["user_id", "plan", "plan_at_signup", "status", "end_date"]
        )
        false_positives = []
        for _, row in df.iterrows():
            match = all_recent[
                (all_recent["user_id"] == row["user_id"]) &
                (all_recent["plan"] == row["plan"]) &
                (all_recent["status"] == row["status"])
            ]
            if not match.empty:
                orig = match.iloc[0]
                if not (orig["status"] == "Expired" or orig["plan"] != orig["plan_at_signup"]):
                    false_positives.append(row.to_dict())
        _dynamic_test(
            self,
            len(false_positives) == 0,
            "No false positives — all rows meet at least one condition.",
            f"False positives found: {false_positives}"
        )


def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)


def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()


if __name__ == "__main__":
    unittest.main()

test_solution.py

Master advanced SQL techniques for building analytical reports. Learn to solve real-world business problems using multi-level aggregations, window functions, cohort and funnel analysis, KPI calculations, and more. Each chapter features theory, a realistic business use case, and a hands-on coding task.

Explore advanced SQL concepts and apply them to real-world analytical reporting scenarios across multiple business domains.

Churn Prediction with SQL

解答