DuckDB and the Customer × Week Panel

3.1 DuckDB: SQL directly on files (50 min)

Concept. DuckDB is an analytical database that runs inside Python: no server, no installation beyond pip. It speaks SQL, the language of every company data warehouse, and it can query CSV and Parquet files directly. In this course, Polars and DuckDB are partners: Polars for Python-style transformations, DuckDB for SQL models (and, from Day 6, dbt).

Create day03/duck_intro.py:

# day03/duck_intro.py
import duckdb

# 1. SQL straight on a file: no loading step
print(duckdb.sql("SELECT COUNT(*) AS n_lines FROM 'data/clean/orders.parquet'"))

# 2. A familiar GROUP BY, this time on the raw CSV
print(duckdb.sql("""
    SELECT status, COUNT(*) AS n, ROUND(SUM(quantity * unit_price)) AS revenue
    FROM 'data/raw/orders.csv'
    GROUP BY status
"""))

# 3. Join two files
print(duckdb.sql("""
    SELECT c.region, COUNT(DISTINCT o.customer_id) AS buyers
    FROM 'data/clean/orders.parquet' AS o
    JOIN 'data/clean/customers.parquet' AS c USING (customer_id)
    GROUP BY c.region
    ORDER BY buyers DESC
"""))

# 4. Hand a result to Polars
df = duckdb.sql("SELECT * FROM 'data/clean/customers.parquet' WHERE is_loyalty").pl()
print(type(df), df.shape)

┌───────────┬───────┬───────────┐
│  status   │   n   │  revenue  │
├───────────┼───────┼───────────┤
│ completed │ 18020 │ 2838607.0 │
│ cancelled │   522 │   83637.0 │
└───────────┴───────┴───────────┘
...
<class 'polars.dataframe.frame.DataFrame'> (1194, 5)

Notes:

• FROM 'file.parquet' treats a file as a table. Example 2 runs on the raw CSV, which still contains cancellations and duplicates.
• .pl() hands a DuckDB result to Polars with no copying. You can move between SQL and Polars freely.

Try it. Write a DuckDB query on the clean orders that returns lines and revenue by month. (Hint: strftime(order_date, '%Y-%m').)

import duckdb
print(duckdb.sql("""
    SELECT strftime(order_date, '%Y-%m') AS month,
           COUNT(*) AS n_lines,
           ROUND(SUM(revenue)) AS revenue
    FROM 'data/clean/orders.parquet'
    GROUP BY month
    ORDER BY month
"""))

│ 2026-01 │    4399 │ 689407.0 │
│ 2026-02 │    4622 │ 712361.0 │
│ 2026-03 │    4800 │ 763991.0 │
│ 2026-04 │    4018 │ 644815.0 │

3.2 A persistent local warehouse (30 min)

Concept. Querying files is great for exploration. For a pipeline we want named, organised tables. We create one database file, data/northmart.duckdb, with schemas (folders for tables): clean for cleaned sources and, later, marts for analysis-ready tables.

Create day03/build_warehouse.py:

# day03/build_warehouse.py
"""Load the clean Parquet files into a persistent DuckDB database: data/northmart.duckdb"""
import duckdb

TABLES = ["customers", "orders", "products", "email_campaign"]

con = duckdb.connect("data/northmart.duckdb")
con.sql("CREATE SCHEMA IF NOT EXISTS clean")
for t in TABLES:
    con.sql(f"CREATE OR REPLACE TABLE clean.{t} AS SELECT * FROM 'data/clean/{t}.parquet'")
    n = con.sql(f"SELECT COUNT(*) FROM clean.{t}").fetchall()[0][0]
    print(f"clean.{t:<15} {n:>7,} rows")
con.close()

clean.customers         4,000 rows
clean.orders           17,839 rows
clean.products             40 rows
clean.email_campaign    1,279 rows

Getting a single value. con.sql(...).fetchall() returns a list of rows, each a tuple, so [0][0] is the first column of the first row. This form works in both DuckDB 1.x and 2.x.

Only one program can write to a DuckDB file at a time. If you get a "lock" error, close other scripts or notebooks that have the database open.

3.3 Weeks and the calendar spine (55 min)

Concept. Most causal methods in this course (DiD, event studies, synthetic control) need a panel: one row per unit per period. Raw data only has rows when something happened. A customer who bought nothing in week 5 has no row for week 5, but their true outcome that week is 0, not missing.

If we forget this, averages are computed only over active weeks and every estimate is biased upwards. The fix is a calendar spine: a table of all periods, crossed with all units, onto which we left-join the events.

Create day03/calendar_spine.py:

# day03/calendar_spine.py
import duckdb

con = duckdb.connect("data/northmart.duckdb")

# 1. Weekly orders: date_trunc('week', ...) gives the Monday of each week
print(con.sql("""
    SELECT date_trunc('week', order_date)::DATE AS week_start, COUNT(*) AS n_orders
    FROM clean.orders
    GROUP BY week_start
    ORDER BY week_start
    LIMIT 5
"""))

# 2. The problem: a customer with no orders in a week has NO ROW for that week
print(con.sql("""
    SELECT customer_id, COUNT(DISTINCT date_trunc('week', order_date)) AS weeks_with_orders
    FROM clean.orders
    GROUP BY customer_id
    ORDER BY weeks_with_orders
    LIMIT 3
"""))

# 3. The fix: a calendar spine of every week, crossed with every customer
con.sql("""
    CREATE OR REPLACE TABLE clean.weeks AS
    SELECT
        week_start::DATE AS week_start,
        ROW_NUMBER() OVER (ORDER BY week_start) AS week_no
    FROM generate_series(DATE '2026-01-05', DATE '2026-04-20', INTERVAL 7 DAY) AS g(week_start)
""")
print(con.sql("SELECT * FROM clean.weeks ORDER BY week_no"))

n = con.sql("SELECT COUNT(*) FROM clean.customers CROSS JOIN clean.weeks").fetchall()[0][0]
print(f"Customers x weeks = {n:,} rows (4,000 x 16)")
con.close()

┌────────────┬──────────┐
│ week_start │ n_orders │
├────────────┼──────────┤
│ 2026-01-05 │     1050 │
...
│ C00040      │                 1 │      <- ordered in only 1 of 16 weeks
...
Customers x weeks = 64,000 rows (4,000 x 16)

• date_trunc('week', d) returns the Monday of the week, so our weeks run Monday to Sunday. Week 9 starts on 2 March, the email date.
• generate_series creates every Monday from 5 January to 20 April: 16 weeks.
• CROSS JOIN pairs every customer with every week.

Try it. For customer C00004, list all 16 weeks with their number of orders, showing 0 for weeks with no orders. (Hint: left-join that customer's weekly orders onto clean.weeks and use COALESCE.)

import duckdb
con = duckdb.connect("data/northmart.duckdb")
print(con.sql("""
    WITH weekly AS (
        SELECT customer_id, date_trunc('week', order_date)::DATE AS week_start, COUNT(*) AS n_orders
        FROM clean.orders GROUP BY ALL
    )
    SELECT w.week_start, COALESCE(o.n_orders, 0) AS n_orders
    FROM clean.weeks AS w
    LEFT JOIN (SELECT * FROM weekly WHERE customer_id = 'C00004') AS o USING (week_start)
    ORDER BY w.week_start
"""))
con.close()

3.4 Window functions: lags, running totals, pre-period values (50 min)

Concept. A window function computes a value from other rows related to the current one, without collapsing the table. PARTITION BY customer_id means "look only at this customer's rows"; ORDER BY week_start sets the order within that customer. Causal work uses windows constantly: lagged outcomes, cumulative exposure, and pre-period averages repeated on every row.

Create day03/windows.py:

# day03/windows.py
import duckdb

con = duckdb.connect("data/northmart.duckdb")

weekly = """
    SELECT customer_id,
           date_trunc('week', order_date)::DATE AS week_start,
           COUNT(*) AS n_orders
    FROM clean.orders
    WHERE customer_id = 'C00004'
    GROUP BY ALL
"""

# 1. LAG: last week's value next to this week's
print(con.sql(f"""
    SELECT *, LAG(n_orders) OVER (PARTITION BY customer_id ORDER BY week_start) AS prev_orders
    FROM ({weekly})
    ORDER BY week_start
"""))

# 2. Running total
print(con.sql(f"""
    SELECT *, SUM(n_orders) OVER (PARTITION BY customer_id ORDER BY week_start) AS cumulative
    FROM ({weekly})
    ORDER BY week_start
"""))

# 3. A pre-period average repeated on every row of the customer (a filtered window)
print(con.sql(f"""
    SELECT *,
           AVG(n_orders) FILTER (WHERE week_start < DATE '2026-03-02')
               OVER (PARTITION BY customer_id) AS pre_avg_active_weeks
    FROM ({weekly})
    ORDER BY week_start
"""))
con.close()

│ C00004      │ 2026-01-19 │        3 │        NULL │
│ C00004      │ 2026-02-02 │        1 │           3 │
...
│ C00004      │ 2026-01-19 │        3 │                  1.4 │

Look carefully at example 3. The "pre-period average" of 1.4 is wrong: it averages only the 5 weeks with orders, not all 8 pre-period weeks (the true figure is 7 / 8 = 0.875). This is the missing-zeros problem from 3.3 again. Window functions are only correct when they run on the complete panel. In the mini-project we compute them after the spine.

• LAG(x) gives the previous row's value; the first row has no previous row, hence NULL.
• AGG(...) FILTER (WHERE ...) restricts which rows enter the calculation: here, pre-period weeks only.

Try it. Rank customers by total revenue and keep the top 5. (Hints: RANK() OVER (ORDER BY SUM(revenue) DESC); DuckDB's QUALIFY filters on a window result the way HAVING filters on an aggregate.)

import duckdb
con = duckdb.connect("data/northmart.duckdb")
print(con.sql("""
    SELECT customer_id,
           ROUND(SUM(revenue), 2) AS total_revenue,
           RANK() OVER (ORDER BY SUM(revenue) DESC) AS revenue_rank
    FROM clean.orders
    GROUP BY customer_id
    QUALIFY revenue_rank <= 5
    ORDER BY revenue_rank
"""))
con.close()

3.5 Day 3 mini-project: the customer × week panel (70 min)

Goal. Build the table every later module starts from: marts.panel_customer_week, with one row per customer per week and these columns:

The SQL lives in its own file so it can be read, reviewed and (from Day 6) moved into dbt. Create sql/panel_customer_week.sql:

-- sql/panel_customer_week.sql
-- One row per customer per week (4,000 x 16 = 64,000 rows).
-- Grain: customer_id x week_start. Treatment: SPRING26 email, sent 2026-03-02 (week 9).

CREATE OR REPLACE TABLE marts.panel_customer_week AS
WITH weekly_orders AS (          -- 1. outcomes, only where orders exist
    SELECT customer_id,
           date_trunc('week', order_date)::DATE AS week_start,
           COUNT(DISTINCT order_id)             AS n_orders,
           SUM(revenue)                         AS revenue
    FROM clean.orders
    GROUP BY ALL
),
spine AS (                       -- 2. every customer in every week
    SELECT c.customer_id, w.week_start, w.week_no
    FROM clean.customers AS c
    CROSS JOIN clean.weeks AS w
),
filled AS (                      -- 3. attach outcomes; no order = 0, not NULL
    SELECT s.*,
           COALESCE(o.n_orders, 0) AS n_orders,
           COALESCE(o.revenue, 0)  AS revenue
    FROM spine AS s
    LEFT JOIN weekly_orders AS o USING (customer_id, week_start)
)
SELECT
    f.customer_id,
    f.week_start,
    f.week_no,
    f.week_no - 9                                       AS event_week,   -- 0 = email week
    -- treatment
    (e.customer_id IS NOT NULL)::INTEGER                AS treated,
    (f.week_start >= DATE '2026-03-02')::INTEGER        AS post,
    ((e.customer_id IS NOT NULL)
       AND f.week_start >= e.sent_date)::INTEGER        AS treated_post,
    -- outcomes
    f.n_orders,
    ROUND(f.revenue, 2)                                 AS revenue,
    -- pre-treatment covariates (fixed per customer)
    c.region,
    c.age,
    c.is_loyalty::INTEGER                               AS is_loyalty,
    SUM(f.n_orders) FILTER (WHERE f.week_no <= 8)
        OVER (PARTITION BY f.customer_id)               AS pre_orders,
    ROUND(SUM(f.revenue) FILTER (WHERE f.week_no <= 8)
        OVER (PARTITION BY f.customer_id), 2)           AS pre_revenue
FROM filled AS f
JOIN clean.customers AS c USING (customer_id)
LEFT JOIN clean.email_campaign AS e USING (customer_id)
ORDER BY f.customer_id, f.week_start;

Read it top to bottom: the WITH clauses are steps, just like a Polars pipeline. Then create day03/build_panel.py, which runs the SQL, checks the result and exports it:

# day03/build_panel.py
"""Mini-project: build, check and export the customer x week panel."""
from pathlib import Path

import duckdb

con = duckdb.connect("data/northmart.duckdb")
con.sql("CREATE SCHEMA IF NOT EXISTS marts")
con.sql(Path("sql/panel_customer_week.sql").read_text())


def scalar(sql: str):
    return con.sql(sql).fetchall()[0][0]


# ---- Checks: each one is a claim about the data that must be true ----
checks = {
    "one row per customer-week":
        scalar("SELECT COUNT(*) FROM marts.panel_customer_week")
        == scalar("SELECT COUNT(*) FROM clean.customers") * 16,
    "no duplicate keys":
        scalar("SELECT COUNT(*) - COUNT(DISTINCT (customer_id, week_start)) FROM marts.panel_customer_week") == 0,
    "no missing outcomes":
        scalar("SELECT COUNT(*) FROM marts.panel_customer_week WHERE n_orders IS NULL OR revenue IS NULL") == 0,
    "revenue matches clean orders":
        abs(scalar("SELECT SUM(revenue) FROM marts.panel_customer_week")
            - scalar("SELECT SUM(revenue) FROM clean.orders")) < 1,
    "treated_post only after the email":
        scalar("SELECT COUNT(*) FROM marts.panel_customer_week WHERE treated_post = 1 AND post = 0") == 0,
    "pre_orders constant within customer":
        scalar("""SELECT COUNT(*) FROM (SELECT customer_id FROM marts.panel_customer_week
                  GROUP BY customer_id HAVING COUNT(DISTINCT pre_orders) > 1)""") == 0,
}
for name, ok in checks.items():
    print(f"[{'PASS' if ok else 'FAIL'}] {name}")
assert all(checks.values()), "Fix the failing checks before using the panel."

# ---- A first look at the panel ----
print(con.sql("""
    SELECT treated, post,
           COUNT(DISTINCT customer_id)  AS customers,
           ROUND(AVG(n_orders), 3)      AS orders_per_week,
           ROUND(AVG(revenue), 2)       AS revenue_per_week
    FROM marts.panel_customer_week
    GROUP BY treated, post
    ORDER BY treated, post
"""))

Path("data/marts").mkdir(parents=True, exist_ok=True)
con.sql("COPY marts.panel_customer_week TO 'data/marts/panel_customer_week.parquet' (FORMAT parquet)")
print("Exported data/marts/panel_customer_week.parquet")
con.close()

python day03/build_panel.py

[PASS] one row per customer-week
[PASS] no duplicate keys
[PASS] no missing outcomes
[PASS] revenue matches clean orders
[PASS] treated_post only after the email
[PASS] pre_orders constant within customer
┌─────────┬───────┬───────────┬─────────────────┬──────────────────┐
│ treated │ post  │ customers │ orders_per_week │ revenue_per_week │
├─────────┼───────┼───────────┼─────────────────┼──────────────────┤
│       0 │     0 │      2721 │           0.231 │            35.99 │
│       0 │     1 │      2721 │           0.203 │            32.64 │
│       1 │     0 │      1279 │           0.404 │            63.22 │
│       1 │     1 │      1279 │           0.416 │            65.44 │
└─────────┴───────┴───────────┴─────────────────┴──────────────────┘
Exported data/marts/panel_customer_week.parquet

Each check is a claim about the data written as code. The last two protect causal validity directly: no "treated" rows before the email existed, and pre-period covariates that cannot change after treatment. On Day 8 these checks become formal data contracts.

The 2 × 2 table above is the raw material of difference-in-differences: (0.416 − 0.404) − (0.203 − 0.231) ≈ +0.04 orders per customer-week, about +11% on what treated customers would otherwise have ordered (0.404 − 0.028 = 0.376). Day 2's growth-rate version gave +15.5%. Which scale the "parallel trends" assumption should hold on, levels or growth rates, is a Module 5 question, along with standard errors.

Try it. Load the exported Parquet file in Polars and build an event-study table: average weekly orders by event_week for treated and control customers, side by side. (Hint: group_by then pivot.) Describe what you see before and after week 0.

import polars as pl
panel = pl.read_parquet("data/marts/panel_customer_week.parquet")
print(panel.shape)
print(panel.group_by("event_week", "treated")
      .agg(orders=pl.col("n_orders").mean().round(3))
      .pivot(on="treated", index="event_week", values="orders")
      .sort("event_week"))

│ -8         ┆ 0.229 ┆ 0.334 │
│ -7         ┆ 0.238 ┆ 0.379 │
...
│ 6          ┆ 0.205 ┆ 0.431 │
│ 7          ┆ 0.218 ┆ 0.435 │

3.6 Save with Git and look ahead (15 min)

Prove the whole pipeline is reproducible: delete every generated file and rebuild from scratch.

rm -rf data/raw data/clean data/marts data/northmart.duckdb
python data/make_northmart.py
python day02/clean.py
python day03/build_warehouse.py
python day03/calendar_spine.py
python day03/build_panel.py

All six checks should pass again with identical numbers. Then commit:

git add .
git commit -m "Day 3: DuckDB warehouse and tested customer x week panel"
git push

Coming next (Days 4–10). Today the pipeline is five commands you must run in the right order. Next we make it professional: Days 4–5 ingest data from an API with dlt; Days 6–7 move the SQL into dbt models with tests; Day 8 adds data contracts that catch leakage and treatment-timing errors; Day 9 schedules everything with Prefect; Day 10 is the Module 1 capstone, the Causal Data Mart.

Day 3 checklist

• I can query CSV and Parquet files with DuckDB and pass results to Polars
• I can build a persistent DuckDB database with schemas
• I can explain why a panel needs a calendar spine and zero-filled outcomes
• I can use LAG, running sums and FILTER inside window functions
• I have a tested panel_customer_week table and can rebuild it from scratch