Summary  
This chapter covers using window functions to calculate running totals (cumulative sums) and fixed-window moving averages over ordered data.  

General domain of usage  
Business analytics

Running totals and moving averages are essential analytical techniques for understanding trends over time in business data. A **running total** (also known as a cumulative sum) calculates the sum of a metric up to each point in a sequence, such as the total sales up to each day. This helps you see how a value accumulates, revealing growth patterns or seasonality. A **moving average** smooths out short-term fluctuations by averaging values over a fixed window (like the last 7 days), making it easier to spot underlying trends.

These calculations are widely used in business scenarios:

- A retail manager might use running totals to track cumulative sales during a promotion;
- A SaaS company could use a 7-day moving average of new signups to understand ongoing growth, filtering out daily spikes or dips;
- Finance teams often rely on moving averages for revenue forecasting;
- Marketing analysts use running totals to monitor campaign performance as it unfolds.


Window frame clauses, such as `ROWS BETWEEN`, define the subset of rows used for window function calculations in SQL. For running totals, the frame is typically specified as `ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW`, meaning the calculation includes all rows from the start of the partition up to the current row.

Definition

SELECT
  sale_date,
  total_amount,
  SUM(total_amount) OVER (ORDER BY sale_date
                         ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_total
FROM sales
ORDER BY sale_date;

A common use case for moving averages is when the finance team wants to analyze trends in daily revenue, but without the noise of daily fluctuations. Suppose the finance team requests a report showing the **7-day moving average of daily revenue** for the SaaS platform. This helps them understand the underlying revenue trend and make better forecasts.

SELECT
  order_date::date AS day,
  SUM(total) AS daily_revenue,
  AVG(SUM(total)) OVER (
    ORDER BY order_date::date
    ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
  ) AS moving_avg_7d
FROM orders
GROUP BY order_date::date
ORDER BY day;

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 test_month_grouping(self):
        """Check that users are grouped by correct month."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        # All values in 'month' should be in 'YYYY-MM' format
        pattern = r"^\\d{4}-\\d{2}$"
        condition = df['month'].str.match(pattern).all()
        _dynamic_test(self, condition,
            "All months are correctly formatted as 'YYYY-MM'",
            "Some months are not correctly formatted as 'YYYY-MM'."
        )

    def test_new_users_count(self):
        """Check that new_users counts are correct for each month."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Reference solution
        solution_query = '''
        SELECT TO_CHAR(signup_date, 'YYYY-MM') AS month, COUNT(*) AS new_users
        FROM users
        GROUP BY TO_CHAR(signup_date, 'YYYY-MM')
        ORDER BY month;
        '''
        self.cursor.execute(solution_query)
        expected = pd.DataFrame(self.cursor.fetchall(), columns=['month', 'new_users'])
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result = pd.DataFrame(rows, columns=columns)
        # Only compare the new_users columns for matching months
        merged = pd.merge(expected, result, on='month', suffixes=('_exp', '_user'))
        condition = (merged['new_users_exp'] == merged['new_users_user']).all()
        _dynamic_test(self, condition,
            "The number of new users per month is correct.",
            "The number of new users per month is incorrect."
        )

    def test_cumulative_users_increasing(self):
        """Check that cumulative_users is non-decreasing and correct."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Reference solution
        solution_query = '''
        SELECT
          TO_CHAR(signup_date, 'YYYY-MM') AS month,
          COUNT(*) AS new_users,
          SUM(COUNT(*)) OVER (ORDER BY TO_CHAR(signup_date, 'YYYY-MM')) AS cumulative_users
        FROM users
        GROUP BY TO_CHAR(signup_date, 'YYYY-MM')
        ORDER BY month;
        '''
        self.cursor.execute(solution_query)
        expected = pd.DataFrame(self.cursor.fetchall(), columns=['month', 'new_users', 'cumulative_users'])
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result = pd.DataFrame(rows, columns=columns)
        # Compare cumulative_users for each month
        merged = pd.merge(expected, result, on='month', suffixes=('_exp', '_user'))
        # Check cumulative_users match exactly
        correct_cumulative = (merged['cumulative_users_exp'] == merged['cumulative_users_user']).all()
        # Check non-decreasing
        non_decreasing = (merged['cumulative_users_user'].diff().fillna(0) >= 0).all()
        condition = correct_cumulative and non_decreasing
        _dynamic_test(self, condition,
            "Cumulative users column is correct and non-decreasing.",
            "Cumulative users column is incorrect or decreases at some point."
        )

    def test_column_names(self):
        """Check that the result columns are named 'month', 'new_users', and 'cumulative_users'."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected = {'month', 'new_users', 'cumulative_users'}
        condition = set(columns) == expected
        _dynamic_test(self, condition,
            "Result columns are correctly named.",
            f"Result columns are {columns}, expected: {expected}.")

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.

Running Totals and Moving Averages

Рішення