SQLite Basics: Compute a Grouped Weighted Mean with a Python Aggregate

A weighted mean needs two totals: the sum of value times weight, and the sum of weights. Python can supply an aggregate class that SQLite instantiates separately for each group. The example also defines what to do with missing measurements and groups that have no positive weight.

Prepare a scratch directory

Tested on Linux with CPython 3.12.14 and SQLite 3.53.1. You need Python 3.11 or newer with SQLite support, Bash, a text editor, and standard mktemp, rm, and rmdir utilities. No third-party packages are required. Run Python without -O so the checks remain enabled.

Keep the same terminal open. Stop if setup fails, before saving the source:

start_dir="$PWD"
work_dir="$(mktemp -d)" || exit 1
cd -- "$work_dir" || exit 1

Run the example

Save this as weighted-sql-report.py:

import sqlite3


class WeightedMean:
    def __init__(self):
        self.weighted_total = 0.0
        self.weight_total = 0.0

    def step(self, value, weight):
        if value is None or weight is None:
            return
        if weight < 0:
            raise ValueError("weight must be nonnegative")
        self.weighted_total += value * weight
        self.weight_total += weight

    def finalize(self):
        if self.weight_total == 0:
            return None
        return self.weighted_total / self.weight_total


connection = sqlite3.connect(":memory:")
try:
    connection.create_aggregate("weighted_mean", 2, WeightedMean)
    connection.execute("CREATE TABLE readings (batch TEXT, value REAL, weight REAL)")
    connection.executemany("INSERT INTO readings VALUES (?, ?, ?)", [
        ("A", 10, 1), ("A", 20, 3), ("A", None, 9),
        ("B", 8, 2), ("B", 12, 2), ("C", 99, 0),
    ])
    rows = connection.execute(
        "SELECT batch, weighted_mean(value, weight) FROM readings "
        "GROUP BY batch ORDER BY batch"
    ).fetchall()
    assert rows == [("A", 17.5), ("B", 10.0), ("C", None)]
    for batch, mean in rows:
        print(f"{batch}: {'NULL' if mean is None else f'{mean:.2f}'}")
    empty = connection.execute(
        "SELECT weighted_mean(value, weight) FROM readings WHERE 0"
    ).fetchone()[0]
    assert empty is None
    print("Empty input:", "NULL")
finally:
    connection.close()

Run it:

python3 -B weighted-sql-report.py

Observed output:

A: 17.50
B: 10.00
C: NULL
Empty input: NULL

Apply the result

step() receives the two arguments for each input row. finalize() returns one result for the group. Batch A ignores the missing measurement and computes (10 × 1 + 20 × 3) / 4 = 17.5. Batch B starts with fresh state. Batch C has zero total weight, so it returns NULL. With no input rows at all, the query also returns NULL.

Make missing-data policy explicit: this implementation skips a row if either argument is missing and rejects negative weights. The fixture uses small finite numeric values. Production code should additionally validate types and finite ranges before aggregating. Floating-point accumulation is approximate and can depend on row order; it is not suitable for exact financial arithmetic.

For this particular calculation, ordinary SQL with SUM can also express the result. A callback is useful when a shared application rule is awkward to maintain in SQL, but it adds a Python call per input row and must be registered on every connection that uses it.

See the official reference for the API and its constraints. The output above comes from this controlled fixture.

Clean up

Remove the saved source, return to the starting directory, then remove the empty practice directory:

rm -- weighted-sql-report.py
cd -- "$start_dir" || exit 1
rmdir -- "$work_dir"
unset start_dir work_dir

If the directory is not empty, inspect its contents before removing anything else.

Acknowledgements

This post was created with assistance from AI.