Computing period-over-period change with lag and lead

Use func.lag(value, 1).over(partition_by=..., order_by=...) to bring the previous period's value onto each row, so growth is computed, filtered and sorted in SQL — and decide explicitly what the first row of each partition compares against, because lag returns NULL there. This guide belongs to window functions and analytical queries.

Quick Answer

Comparing a row with the one before it does not need a Python loop, and doing it in SQL means the result can be filtered and sorted.

Before — the comparison done after loading every row:

from sqlalchemy import func, select

from shop.models import Order

rows = (await session.execute(
    select(
        func.date_trunc("month", Order.placed_at).label("month"),
        func.sum(Order.total_cents).label("revenue"),
    )
    .group_by(func.date_trunc("month", Order.placed_at))
    .order_by(func.date_trunc("month", Order.placed_at))
)).all()

growth = []
previous = None
for month, revenue in rows:                 # every month must be fetched
    growth.append((month, revenue, None if previous is None else revenue - previous))
    previous = revenue

After — lag brings the previous value onto the row:

from sqlalchemy import func, select

from shop.models import Order

monthly = (
    select(
        func.date_trunc("month", Order.placed_at).label("month"),
        func.sum(Order.total_cents).label("revenue_cents"),
    )
    .where(Order.status != "cancelled")
    .group_by(func.date_trunc("month", Order.placed_at))
    .cte("monthly")
)

previous = func.lag(monthly.c.revenue_cents, 1).over(order_by=monthly.c.month)

stmt = (
    select(
        monthly.c.month,
        monthly.c.revenue_cents,
        previous.label("previous_cents"),
        (monthly.c.revenue_cents - previous).label("change_cents"),
        (
            100.0 * (monthly.c.revenue_cents - previous)
            / func.nullif(previous, 0)
        ).label("change_pct"),
    )
    .order_by(monthly.c.month)
)

func.nullif(previous, 0) is the division guard: a previous value of zero yields NULL rather than a division error, and the first month — where lag is NULL — produces NULL throughout rather than a wrong number.

Execution Context & Async Workflow Integration

lag and lead are window functions that read another row of the same window. They do not aggregate; they navigate. lag(value, offset, default) reads the row offset positions earlier in the window's order, and lead reads forward. Both are computed after WHERE and GROUP BY, which is why a period-over-period query usually aggregates in one stage and navigates in the next — the CTE shape from using non-recursive CTEs to structure reporting queries.

partition_by restarts the sequence. Without it, one product's last month is followed by the next product's first, and the growth figure at that boundary is nonsense. With partition_by=monthly.c.product_id, each product gets its own ordered sequence and its own NULL at the start:

previous = func.lag(monthly.c.revenue_cents, 1).over(
    partition_by=monthly.c.product_id,
    order_by=monthly.c.month,
)

The third argument replaces the NULL for rows with no predecessor: func.lag(col, 1, 0) treats the period before the first as zero. Which is right depends on the question. For revenue growth, zero is often defensible; for a temperature reading, it is not, and NULL propagating through the calculation is the honest answer.

The subtle failure is that lag reads the previous row, not the previous month. If a product sold nothing in March, there is no March row, and April's lag reaches February — silently reporting two months of growth as one. Making the sequence dense before navigating it is the fix, and it needs a generated calendar joined to the aggregate, which is the subject of bucketing time series and filling gaps with generate_series.

Under async none of this is special: the whole query is built synchronously and executed with one await session.execute(stmt), returning rows rather than entities. That makes analytical queries some of the most comfortable code to write against an AsyncSession — there are no relationships to load and nothing that can lazy-load unexpectedly.

One performance note. Window functions require the rows to be sorted in the window's order, so an index matching (partition_by..., order_by...) lets PostgreSQL skip the sort entirely. On an aggregate stage the rows are usually few enough that it does not matter; over raw rows on a large table it matters a great deal, and reading EXPLAIN output for a SQLAlchemy query shows how to tell whether the sort was skipped.

Resolving Warnings, Errors & Common Mistakes

The last_value surprise is worth seeing in full, because it looks like a bug and is specified behaviour. With an ORDER BY and no frame clause, the window frame is "from the start of the partition to the current row", so last_value is the current row:

from sqlalchemy import func, select

from shop.models import Reading

# Wrong: returns each row's own value.
latest_wrong = func.last_value(Reading.value).over(
    partition_by=Reading.sensor_id, order_by=Reading.taken_at
)

# Right: the frame covers the whole partition.
latest = func.last_value(Reading.value).over(
    partition_by=Reading.sensor_id,
    order_by=Reading.taken_at,
    rows=(None, None),          # UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
)

SQLAlchemy's rows= and range= parameters take a two-tuple where None means unbounded and 0 means the current row, so rows=(None, 0) is the default frame and rows=(-2, 0) is a three-row trailing window. lag and lead ignore the frame entirely — they always navigate by position in the window order — which is why they are more predictable than the value functions.

A trailing average is the most common use of an explicit frame, and it reads naturally once the tuple convention is clear:

rolling_3 = func.avg(monthly.c.revenue_cents).over(
    order_by=monthly.c.month, rows=(-2, 0)      # this month and the two before
).label("rolling_3_month_avg")

Advanced: Year-Over-Year, Gaps and Sessionisation

Once lag is available, three common analytical questions become single queries.

Year-over-year is lag with an offset of twelve over a monthly series — which is correct only if the series is dense, because the offset counts rows:

from sqlalchemy import func, select

year_ago = func.lag(monthly.c.revenue_cents, 12).over(
    partition_by=monthly.c.product_id, order_by=monthly.c.month
)

stmt = select(
    monthly.c.month,
    monthly.c.revenue_cents,
    year_ago.label("year_ago_cents"),
    (
        100.0 * (monthly.c.revenue_cents - year_ago) / func.nullif(year_ago, 0)
    ).label("yoy_pct"),
)

If any month can be missing, the offset silently reaches the wrong row, and the only robust version joins the aggregate to a generated month series first.

Gap detection uses lag on a timestamp to find intervals longer than expected — missing readings, a stalled feed, a subscription lapse:

from sqlalchemy import func, select
from sqlalchemy.dialects.postgresql import INTERVAL

from shop.models import Reading

previous_at = func.lag(Reading.taken_at).over(
    partition_by=Reading.sensor_id, order_by=Reading.taken_at
)

readings = select(
    Reading.sensor_id,
    Reading.taken_at,
    previous_at.label("previous_at"),
    (Reading.taken_at - previous_at).label("interval"),
).cte("readings")

gaps = select(readings).where(
    readings.c.interval > func.cast("00:15:00", INTERVAL)
)

Sessionisation — grouping events into sessions separated by inactivity — is the same idea with a cumulative sum: mark each row where the gap exceeds the threshold, then count the marks to give every row a session number:

from sqlalchemy import Integer, case, cast, func, select
from sqlalchemy.dialects.postgresql import INTERVAL

from shop.models import Event

previous_at = func.lag(Event.occurred_at).over(
    partition_by=Event.customer_id, order_by=Event.occurred_at
)

marked = select(
    Event.customer_id,
    Event.occurred_at,
    cast(
        case((Event.occurred_at - previous_at > func.cast("00:30:00", INTERVAL), 1), else_=0),
        Integer,
    ).label("is_new_session"),
).cte("marked")

sessions = select(
    marked.c.customer_id,
    marked.c.occurred_at,
    func.sum(marked.c.is_new_session).over(
        partition_by=marked.c.customer_id, order_by=marked.c.occurred_at
    ).label("session_number"),
)

A running sum over a zero-or-one column is the standard way to turn "where does a group start" into "which group is this row in", and it is worth recognising because the same shape solves streak detection, version numbering and gap-and-island problems. The running-total mechanics are covered in writing window functions for running totals in Python.

Testing an Analytical Query

Analytical queries are worth testing precisely because their output is plausible when wrong: a growth figure is a number either way, and nothing errors. Three test shapes catch the mistakes that matter.

A fixture with hand-computed expectations. Small, explicit data, and the expected numbers written out rather than computed by the same logic under test:

import datetime as dt

import pytest

from shop.reports import monthly_growth


@pytest.fixture
async def three_months(session, order_factory):
    for day, cents in [
        (dt.date(2026, 7, 4), 1_000),
        (dt.date(2026, 7, 18), 500),     # July: 1,500
        (dt.date(2026, 8, 2), 3_000),    # August: 3,000
        (dt.date(2026, 9, 9), 1_500),    # September: 1,500
    ]:
        await order_factory(placed_on=day, total_cents=cents)
    await session.commit()


@pytest.mark.asyncio
async def test_monthly_growth(session, three_months):
    rows = (await session.execute(monthly_growth())).mappings().all()

    assert [row["revenue_cents"] for row in rows] == [1_500, 3_000, 1_500]
    assert rows[0]["previous_cents"] is None          # no month before July
    assert rows[0]["change_pct"] is None
    assert rows[1]["change_cents"] == 1_500
    assert rows[1]["change_pct"] == pytest.approx(100.0)
    assert rows[2]["change_cents"] == -1_500
    assert rows[2]["change_pct"] == pytest.approx(-50.0)

The first assertion about None is the valuable one: it pins the decision about what the first period compares against, so a later change to a default is a deliberate test change rather than a silent shift in every report.

A boundary case for partitions. Two products, each with two months, and an assertion that neither product's first month borrowed the other product's last. Without partition_by, that test fails and nothing else does.

A zero-baseline case. One month with zero revenue followed by a month with some, asserting the percentage is NULL rather than an error. That is the nullif guard, and it is the kind of row that appears in production data months after launch.

Run all of it against PostgreSQL. Window frames, date_trunc, interval arithmetic and NULL ordering differ or do not exist on SQLite, so a passing test there says very little — the container fixture in running tests against a Postgres testcontainer is what makes these assertions meaningful.

Frequently Asked Questions

Related

• Window Functions and Analytical Queries — The parent guide: partitions, frames and analytical patterns.
• Bucketing time series and filling gaps with generate_series — Making the sequence dense before navigating it.
• Writing window functions for running totals in Python — Frames and cumulative sums.
• Using non-recursive CTEs to structure reporting queries — Aggregate in one stage, navigate in the next.