Summary  
This chapter covers using conditional expressions with CASE WHEN inside aggregate functions like SUM to classify and count records by stage.  

General domain of usage  
Marketing analytics (funnel analysis)

Funnel analysis is a core technique in analytics for understanding how users or leads progress through a series of defined business stages, such as from initial awareness to final purchase. In SQL, you can build funnel reports by using the `CASE WHEN` statement to classify records into funnel stages, and then aggregate these classifications to count the number of users or leads at each stage. This approach allows you to visualize where users drop off and where they convert, providing actionable insights for optimizing business processes.

A **funnel stage** represents a key step in a user's journey, such as "Lead," "Trial," or "Paid."  
A **conversion rate** is the percentage of users who move from one stage to the next. For example, if 100 users start a trial and 30 become paid customers, the conversion rate from "Trial" to "Paid" is 30%.

Definition

SELECT
  COUNT(*) AS total_leads,
  SUM(CASE WHEN status = 'Contacted' THEN 1 ELSE 0 END) AS contacted_leads,
  SUM(CASE WHEN status = 'Converted' THEN 1 ELSE 0 END) AS converted_leads
FROM leads
WHERE campaign_id = 8;

The marketing team often needs to understand the effectiveness of their campaigns. For example, they may request a funnel report showing the number of leads generated, how many of those leads started a trial, and how many ultimately converted to paid customers for the last campaign. This type of report helps them identify which steps in the funnel are performing well and which need improvement.

SELECT
  COUNT(*) AS total_leads,
  SUM(CASE WHEN status = 'Contacted' THEN 1 ELSE 0 END) AS trials_started,
  SUM(CASE WHEN l.status = 'Converted' AND c.conversion_id IS NOT NULL THEN 1 ELSE 0 END) AS paid_conversions
FROM leads l
LEFT JOIN conversions c ON l.lead_id = c.lead_id
WHERE campaign_id = (
  SELECT MAX(campaign_id) FROM campaigns
);

Now, you will apply these concepts to measure drop-off rates between each stage in the sales funnel. Understanding drop-off at each step is crucial for optimizing the funnel and improving conversion rates.

import unittest
import psycopg2
import pandas as pd

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 get_query_result(self):
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        return pd.DataFrame(rows, columns=columns)

    def test_total_leads_count(self):
        """Check that total_leads is correct for the most recent campaign."""
        df = self.get_query_result()
        # Find MAX campaign_id
        self.cursor.execute("SELECT MAX(campaign_id) FROM campaigns;")
        max_campaign_id = self.cursor.fetchone()[0]
        self.cursor.execute("SELECT COUNT(*) FROM leads WHERE campaign_id = %s;", (max_campaign_id,))
        expected = self.cursor.fetchone()[0]
        actual = int(df['total_leads'][0])
        _dynamic_test(self, actual == expected, f"Total leads count is correct: {actual}", f"Total leads should be {expected}, got {actual}")

    def test_contacted_leads_count(self):
        """Check that contacted_leads is correct for the most recent campaign."""
        df = self.get_query_result()
        self.cursor.execute("SELECT MAX(campaign_id) FROM campaigns;")
        max_campaign_id = self.cursor.fetchone()[0]
        self.cursor.execute("SELECT COUNT(*) FROM leads WHERE campaign_id = %s AND status = 'Contacted';", (max_campaign_id,))
        expected = self.cursor.fetchone()[0]
        actual = int(df['contacted_leads'][0])
        _dynamic_test(self, actual == expected, f"Contacted leads count is correct: {actual}", f"Contacted leads should be {expected}, got {actual}")

    def test_converted_leads_count(self):
        """Check that converted_leads is correct for the most recent campaign."""
        df = self.get_query_result()
        self.cursor.execute("SELECT MAX(campaign_id) FROM campaigns;")
        max_campaign_id = self.cursor.fetchone()[0]
        self.cursor.execute("SELECT COUNT(*) FROM leads WHERE campaign_id = %s AND status = 'Converted';", (max_campaign_id,))
        expected = self.cursor.fetchone()[0]
        actual = int(df['converted_leads'][0])
        _dynamic_test(self, actual == expected, f"Converted leads count is correct: {actual}", f"Converted leads should be {expected}, got {actual}")

    def test_dropoff_lead_to_contacted_pct(self):
        """Check drop-off percentage from Lead to Contacted."""
        df = self.get_query_result()
        total = int(df['total_leads'][0])
        contacted = int(df['contacted_leads'][0])
        expected = round(100.0 * (total - contacted) / total, 2) if total != 0 else None
        actual = float(df['dropoff_lead_to_contacted_pct'][0]) if 'dropoff_lead_to_contacted_pct' in df.columns else None
        _dynamic_test(
            self,
            actual == expected,
            f"Drop-off from Lead to Contacted is correct: {actual}%",
            f"Drop-off from Lead to Contacted should be {expected}%, got {actual}%"
        )

    def test_dropoff_contacted_to_converted_pct(self):
        """Check drop-off percentage from Contacted to Converted."""
        df = self.get_query_result()
        contacted = int(df['contacted_leads'][0])
        converted = int(df['converted_leads'][0])
        expected = round(100.0 * (contacted - converted) / contacted, 2) if contacted != 0 else None
        actual = float(df['dropoff_contacted_to_converted_pct'][0]) if 'dropoff_contacted_to_converted_pct' in df.columns else None
        _dynamic_test(
            self,
            actual == expected,
            f"Drop-off from Contacted to Converted is correct: {actual}%",
            f"Drop-off from Contacted to Converted should be {expected}%, got {actual}%"
        )

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.

Funnel Analysis with CASE WHEN + SUM

Solução