Summary  
This chapter covers advanced window functions for quantile and percentile analysis—specifically NTILE for dividing rows into equal groups, PERCENT_RANK for computing relative ranks as a fraction, and CUME_DIST for calculating cumulative distribution within a partition.

General domain of usage  
Customer segmentation and targeting in marketing analytics.

In advanced analytics, window functions allow you to perform complex calculations across sets of rows related to your current row. Beyond ranking and running totals, SQL provides powerful window functions specifically designed for quantile and percentile analysis: **NTILE**, **PERCENT_RANK**, and **CUME_DIST**.

The **NTILE** function divides your result set into a specified number of roughly equal groups, assigning a group number to each row. For example, using `NTILE(4)` on a sorted list will assign each row to a quartile (1 to 4), which is useful for segmenting customers, employees, or products into performance buckets.

**PERCENT_RANK** computes the relative rank of a row within a partition as a percentage between 0 and 1, showing how a value compares to others in its group. **CUME_DIST** calculates the cumulative distribution of a value—essentially, the proportion of rows with values less than or equal to the current row.

A **quantile** divides a dataset into equal-sized, ordered groups. Common quantiles include quartiles (4 groups), quintiles (5), and deciles (10). A **percentile** indicates the percentage of data points below a particular value. In business analytics, quantiles and percentiles help identify top and bottom performers, target segments, or outliers for focused strategies.

Definition

SELECT
  c.customer_id,
  c.first_name,
  c.last_name,
  SUM(o.total) AS lifetime_value,
  NTILE(4) OVER (ORDER BY SUM(o.total) DESC) AS revenue_quartile
FROM
  customers c
  JOIN orders o ON c.customer_id = o.customer_id
GROUP BY
  c.customer_id, c.first_name, c.last_name
ORDER BY
  revenue_quartile, lifetime_value DESC;

Imagine a scenario where the marketing team wants to target the top 10% of customers by lifetime value. You can use **PERCENT_RANK** to identify which customers fall into this elite decile.

SELECT
  customer_id,
  first_name,
  last_name,
  lifetime_value,
  pct_rank
FROM (
  SELECT
    c.customer_id,
    c.first_name,
    c.last_name,
    SUM(o.total) AS lifetime_value,
    PERCENT_RANK() OVER (ORDER BY SUM(o.total) DESC) AS pct_rank
  FROM customers c
  JOIN orders o ON c.customer_id = o.customer_id
  GROUP BY c.customer_id, c.first_name, c.last_name
) ranked
WHERE pct_rank <= 0.1
ORDER BY pct_rank;

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 _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_columns_present(self):
        """Check that all required columns are present in the output."""
        df = self._execute_student_query()
        expected_columns = [
            "employee_id", "first_name", "last_name",
            "department_id", "department_name", "score", "performance_quartile"
        ]
        missing = [col for col in expected_columns if col not in df.columns]
        _dynamic_test(
            self,
            len(missing) == 0,
            "All required columns are present.",
            f"Missing columns: {missing}"
        )

    def test_quartiles_range(self):
        """Check that performance_quartile values are between 1 and 4."""
        df = self._execute_student_query()
        if "performance_quartile" not in df.columns:
            self.skipTest("performance_quartile column missing.")
        valid = df["performance_quartile"].apply(lambda x: x in [1, 2, 3, 4]).all()
        _dynamic_test(
            self,
            valid,
            "All performance_quartile values are between 1 and 4.",
            "Some performance_quartile values are not between 1 and 4."
        )

    def test_quartile_partition_by_department(self):
        """Check that quartile assignment starts at 1 within each department."""
        df = self._execute_student_query()
        if "department_id" not in df.columns or "performance_quartile" not in df.columns:
            self.skipTest("department_id or performance_quartile column missing.")
        issues = []
        for dept_id, group in df.groupby("department_id"):
            quartiles = set(group["performance_quartile"])
            if not all(1 <= q <= 4 for q in quartiles):
                issues.append((dept_id, sorted(quartiles)))
            if min(quartiles) != 1:
                issues.append((dept_id, "min_quartile", min(quartiles)))
        _dynamic_test(
            self,
            len(issues) == 0,
            "Quartile assignment is partitioned by department and starts at 1 for each department.",
            f"Quartile assignment issue(s) found: {issues}"
        )

    def test_quartile_ordering_desc(self):
        """Check that higher scores get lower quartile numbers within each department."""
        df = self._execute_student_query()
        if "department_id" not in df.columns or "performance_quartile" not in df.columns or "score" not in df.columns:
            self.skipTest("department_id, performance_quartile, or score column missing.")
        issues = []
        for dept_id, group in df.groupby("department_id"):
            # Compare average score per quartile — higher quartile number = lower avg score
            avg_score_by_quartile = group.groupby("performance_quartile")["score"].mean().sort_index()
            scores = avg_score_by_quartile.tolist()
            if scores != sorted(scores, reverse=True):
                issues.append((dept_id, dict(avg_score_by_quartile)))
        _dynamic_test(
            self,
            len(issues) == 0,
            "Quartile assignment uses descending order of score in each department.",
            f"Quartile ordering issue(s) found: {issues}"
        )

    def test_ntile_distribution(self):
        """Check that NTILE distributes employees as evenly as possible across 4 quartiles."""
        df = self._execute_student_query()
        if "department_id" not in df.columns or "performance_quartile" not in df.columns:
            self.skipTest("department_id or performance_quartile column missing.")
        issues = []
        for dept_id, group in df.groupby("department_id"):
            n = len(group)
            counts = group["performance_quartile"].value_counts().sort_index()
            min_count = n // 4
            max_count = min_count + (1 if n % 4 != 0 else 0)
            for q, cnt in counts.items():
                if not (min_count <= cnt <= max_count + 1):
                    issues.append((dept_id, f"quartile {q} has {cnt} employees, expected {min_count}-{max_count+1}"))
        _dynamic_test(
            self,
            len(issues) == 0,
            "NTILE distributes employees evenly across quartiles.",
            f"Uneven NTILE distribution: {issues}"
        )


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.

Advanced Window Functions: NTILE, PERCENT_RANK, CUME_DIST

Soluzione