Day 10: Capstone: The Causal Data Mart · Applied Causal Inference & Decision Science

Written lesson

Time	Block
0:00 – 0:30	10.1 The brief and how it is assessed
0:30 – 1:15	10.2 Two fixes for a pipeline that runs all year
1:15 – 1:30	Break
1:30 – 2:30	10.3 Replay the 16 weeks
2:30 – 3:15	10.4 Acceptance tests
3:15 – 3:30	Break
3:30 – 4:15	10.5 The data card
4:15 – 4:45	10.6 The handover memo
4:45 – 5:00	10.7 Final commit and Module 1 retrospective

10.1 The brief and how it is assessed (30 min)

The brief. NorthMart's analytics lead writes:

"Next month the team will estimate the effect of the SPRING26 email, first with a DAG, then with difference-in-differences and matching. Before anyone estimates anything, I need a data mart I can trust: rebuilt automatically every week, checked for the problems that ruin causal studies, and documented well enough that a new analyst can use it without asking you. Show me it would have worked all spring, week by week."

Deliverables.

#	Deliverable	Where
D1	A pipeline that runs every week of the study without manual steps	`pipeline/northmart_flow.py`
D2	A week-by-week replay from an empty warehouse	`day10/reset_warehouse.py`, `day10/replay_weeks.py`
D3	Acceptance tests on the published release	`day10/acceptance.py`
D4	A generated data card	`day10/data_card.py`, `docs/DATA_CARD.md`
D5	A contract register with fire-drill results and runbook	`docs/CONTRACTS.md`
D6	A one-page handover memo	`docs/HANDOVER_MEMO.md`

Assessment rubric (used again, in extended form, for the final course capstone):

Criterion	Weight	Excellent
Reproducibility	20%	Replay runs end to end from an empty warehouse with one command
Causal validity	25%	Leakage, timing, fixed-treatment and completeness contracts enforced; fire drill documented
Correctness	20%	Acceptance checks pass, including agreement with the independent Day 3 pipeline
Documentation	15%	Data card generated from data, contract register, runbook
Communication	20%	Memo states what the data can and cannot support, in plain language

10.2 Two fixes for a pipeline that runs all year (45 min)

Concept. So far we always ran the pipeline with all 16 weeks available. A real pipeline starts on its first Sunday with an empty warehouse. Simulate that. Create day10/reset_warehouse.py:

# day10/reset_warehouse.py
"""Empty the warehouse and the loader's memory, as on the first day of a new pipeline."""
import shutil
from pathlib import Path

import duckdb

ROOT = Path(__file__).resolve().parent.parent


def reset_warehouse() -> None:
    con = duckdb.connect(str(ROOT / "data" / "northmart.duckdb"))
    for schema in ["raw", "staging", "intermediate", "marts"]:
        con.sql(f"DROP SCHEMA IF EXISTS {schema} CASCADE")
    con.close()
    shutil.rmtree(ROOT / ".dlt_pipelines" / "northmart", ignore_errors=True)
    shutil.rmtree(ROOT / "data" / "published", ignore_errors=True)
    print("Warehouse emptied: raw, staging, intermediate and marts dropped; loader state cleared.")


if __name__ == "__main__":
    reset_warehouse()

Then empty the warehouse and run the flow as of Sunday 18 January:

python day10/reset_warehouse.py
python pipeline/northmart_flow.py --as-of 2026-01-18

build_models - 2 of 32 ERROR creating sql view model staging.stg_campaign_recipients .......... [ERROR in 0.29s]
build_models - Task run failed with exception: RuntimeError('dbt build failed, data NOT published.
               Done. PASS=16 WARN=0 ERROR=1 SKIP=15 NO-OP=0 REUSED=0 TOTAL=32')

Problem 1: no recipient table. Before 2 March nobody has been emailed, and dlt only creates tables that receive rows (Day 5). stg_campaign_recipients points at a table that doesn't exist yet.

Fix that (Fix 1 below) and run again, and a second problem appears:

build_models - 27 of 32 FAIL 14 assert_no_empty_weeks ......................................... [FAIL 14 in 0.08s]
build_models - Done. PASS=30 WARN=1 ERROR=1 SKIP=0 NO-OP=0 REUSED=0 TOTAL=32

Problem 2: empty future weeks. int_weeks always generates 16 weeks. On 18 January, weeks 3–16 have no orders yet, and the completeness contract (rightly) rejects all 14 of them.

Both are design errors in our code, not in the data, and the contracts caught both before anything was published.

Fix 1 — a treatment log that may not exist yet. Replace transform/models/staging/stg_campaign_recipients.sql:

-- Customers who received a campaign email, with the send date.
-- Before the first email is sent, dlt has not created raw.campaign_recipients yet
-- (it only creates tables that received rows). In that case: an empty table, nobody treated.
{% set recipients = adapter.get_relation(
       database=source('raw', 'campaign_recipients').database,
       schema=source('raw', 'campaign_recipients').schema,
       identifier=source('raw', 'campaign_recipients').identifier) %}

{% if recipients is not none %}
select
    customer_id,
    campaign_id,
    cast(sent_date as date) as sent_date
from {{ source('raw', 'campaign_recipients') }}
{% else %}
select
    cast(null as varchar) as customer_id,
    cast(null as varchar) as campaign_id,
    cast(null as date)    as sent_date
where false
{% endif %}

adapter.get_relation asks the database whether the table exists when the model runs. If it doesn't, the model returns an empty table with the right columns, which means exactly what the business situation means: nobody is treated yet.

Fix 2 — a calendar that ends with the data. Replace transform/models/intermediate/int_weeks.sql:

-- The calendar spine: one row per Monday, numbered 1, 2, 3...
-- It stops at the last week that has data, so a weekly pipeline run in, say, February
-- doesn't create empty "future" weeks (which the completeness contract would reject).
with bounds as (
    select least(
        date '{{ var("last_week") }}',
        cast(date_trunc('week', max(order_date)) as date)
    ) as last_monday
    from {{ ref('stg_orders') }}
)
select
    cast(g.week_start as date)                        as week_start,
    row_number() over (order by g.week_start)         as week_no
from bounds,
     generate_series(date '{{ var("first_week") }}', bounds.last_monday, interval 7 day) as g(week_start)

The spine now ends at the earlier of the study's last week and the last week with orders. Our clock always stops on a Sunday, so every week included is complete. (A pipeline that ran mid-week would also need to drop the current, partial week; otherwise it would look like a sales drop.)

Run the flow for 18 January again:

python pipeline/northmart_flow.py --as-of 2026-01-18

build_models - Done. PASS=31 WARN=1 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=32
Flow run 'origami-caracara' - Data as of 2026-01-18: panel_customer_week 8,000 rows, customer_features 4,000 rows

Two weeks × 4,000 customers, nobody treated yet, every contract passing. The overlap warning now fires for every region (no emailed customers exist yet), which is correct and, being a warning, doesn't block the release.

10.3 Replay the 16 weeks (60 min)

Goal. Start from an empty warehouse (the script calls reset_warehouse() itself) and run the full flow once per week, each Sunday from 11 January to 26 April, exactly as the scheduled pipeline would have. After each run, record what was published.

Create day10/replay_weeks.py:

# day10/replay_weeks.py
"""
Capstone step 1: live through the 16 weeks as NorthMart's data team did.

Starts from an empty warehouse, then runs the full pipeline once per week,
every Sunday from 11 January to 26 April 2026. Each run loads only that week's
new orders, rebuilds the models, checks every contract, and publishes.
"""
import sys
from datetime import date, timedelta
from pathlib import Path

import polars as pl

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "pipeline"))
sys.path.insert(0, str(ROOT / "day10"))
from northmart_flow import northmart_pipeline  # noqa: E402
from reset_warehouse import reset_warehouse  # noqa: E402

FIRST_SUNDAY = date(2026, 1, 11)
N_WEEKS = 16


def snapshot(as_of: date) -> dict:
    panel = pl.read_parquet(ROOT / "data" / "published" / "panel_customer_week.parquet")
    return {
        "as_of": as_of,
        "weeks": panel["week_start"].n_unique(),
        "panel_rows": panel.height,
        "orders_so_far": int(panel["n_orders"].sum()),
        "treated_customers": int(panel.filter(pl.col("week_no") == 1)["treated"].sum()),
    }


def main() -> None:
    reset_warehouse()
    history = []
    for k in range(N_WEEKS):
        as_of = FIRST_SUNDAY + timedelta(weeks=k)
        northmart_pipeline(as_of.isoformat())
        history.append(snapshot(as_of))

    table = pl.DataFrame(history)
    with pl.Config(tbl_rows=-1):
        print(table)
    table.write_csv(ROOT / "data" / "published" / "replay_history.csv")


if __name__ == "__main__":
    main()

With the API running (and prefect config unset PREFECT_API_URL if you set it yesterday), run it. It takes about three minutes:

python day10/replay_weeks.py

┌────────────┬───────┬────────────┬───────────────┬───────────────────┐
│ as_of      ┆ weeks ┆ panel_rows ┆ orders_so_far ┆ treated_customers │
╞════════════╪═══════╪════════════╪═══════════════╪═══════════════════╡
│ 2026-01-11 ┆ 1     ┆ 4000       ┆ 1050          ┆ 0                 │
│ 2026-01-18 ┆ 2     ┆ 8000       ┆ 2183          ┆ 0                 │
│ 2026-01-25 ┆ 3     ┆ 12000      ┆ 3350          ┆ 0                 │
│ 2026-02-01 ┆ 4     ┆ 16000      ┆ 4567          ┆ 0                 │
│ 2026-02-08 ┆ 5     ┆ 20000      ┆ 5770          ┆ 0                 │
│ 2026-02-15 ┆ 6     ┆ 24000      ┆ 6920          ┆ 0                 │
│ 2026-02-22 ┆ 7     ┆ 28000      ┆ 8098          ┆ 0                 │
│ 2026-03-01 ┆ 8     ┆ 32000      ┆ 9168          ┆ 0                 │
│ 2026-03-08 ┆ 9     ┆ 36000      ┆ 10298         ┆ 1279              │
│ 2026-03-15 ┆ 10    ┆ 40000      ┆ 11384         ┆ 1279              │
│ 2026-03-22 ┆ 11    ┆ 44000      ┆ 12417         ┆ 1279              │
│ 2026-03-29 ┆ 12    ┆ 48000      ┆ 13511         ┆ 1279              │
│ 2026-04-05 ┆ 13    ┆ 52000      ┆ 14509         ┆ 1279              │
│ 2026-04-12 ┆ 14    ┆ 56000      ┆ 15582         ┆ 1279              │
│ 2026-04-19 ┆ 15    ┆ 60000      ┆ 16690         ┆ 1279              │
│ 2026-04-26 ┆ 16    ┆ 64000      ┆ 17839         ┆ 1279              │
└────────────┴───────┴────────────┴───────────────┴───────────────────┘

Read it row by row:

All 16 runs passed every contract and published.
The panel grows by exactly 4,000 rows (one week of customers) per run.
The 1,279 treated customers appear in the week the email was sent, not before.
The final run has 17,839 orders: the same number as Day 2's cleaning of the CSV, reached here through 16 incremental API loads.

10.4 Acceptance tests (45 min)

Concept. dbt contracts check the data inside the warehouse. Acceptance tests check the published release from the user's side: is it complete, does it agree with an independent source, and does it support the analysis it was built for?

Create day10/acceptance.py:

# day10/acceptance.py
"""Capstone step 2: acceptance tests for the published Causal Data Mart."""
import json
from pathlib import Path

import polars as pl

PUB = Path("data/published")
KEY = ["customer_id", "week_start"]
results: list[tuple[str, bool, str]] = []


def check(name: str, ok: bool, detail: str = "") -> None:
    results.append((name, bool(ok), detail))


manifest = json.loads((PUB / "manifest.json").read_text())
panel = pl.read_parquet(PUB / "panel_customer_week.parquet").sort(KEY)
features = pl.read_parquet(PUB / "customer_features.parquet")

# A. The release is complete and clean
check("A1 data covers all 16 weeks", panel["week_start"].n_unique() == 16, str(manifest["data_as_of"]))
check("A2 dbt build had no errors", "ERROR=0" in manifest["dbt"], manifest["dbt"])
check("A3 one feature row per customer", features.height == features["customer_id"].n_unique() == 4000)

# B. It agrees with the independent Day 3 pipeline
day3_file = Path("data/marts/panel_customer_week.parquet")
if day3_file.exists():
    day3 = pl.read_parquet(day3_file).sort(KEY)
    same = all(
        ((panel[c].cast(pl.Float64) - day3[c].cast(pl.Float64)).abs() < 1e-6).all()
        if panel[c].dtype.is_numeric() else (panel[c] == day3[c]).all()
        for c in day3.columns
    )
    check("B1 identical to the Day 3 panel", same and panel.height == day3.height)
else:
    check("B1 identical to the Day 3 panel", False, "run Days 2-3 scripts to create the comparison file")

# C. It supports the causal analysis of Modules 2-5
cust = panel.group_by("customer_id", "treated").agg(
    pre=pl.col("n_orders").filter(pl.col("post") == 0).sum(),
    post=pl.col("n_orders").filter(pl.col("post") == 1).sum(),
)
m = cust.group_by("treated").agg(pl.col("pre").mean(), pl.col("post").mean()).sort("treated")
c, t = m.row(0, named=True), m.row(1, named=True)
naive = t["post"] / c["post"] - 1
growth_gap = (t["post"] / t["pre"] - 1) - (c["post"] / c["pre"] - 1)
check("C1 naive post-period gap reproduces Day 2", abs(naive - 1.06) < 0.01, f"{naive:+.1%}")
check("C2 difference in growth reproduces Day 2", abs(growth_gap - 0.155) < 0.005, f"{growth_gap:+.1%}")
check("C3 features balance: treated spent more pre-email",
      features.filter(pl.col("treated") == 1)["pre_revenue"].mean()
      > features.filter(pl.col("treated") == 0)["pre_revenue"].mean())

for name, ok, detail in results:
    print(f"[{'PASS' if ok else 'FAIL'}] {name}" + (f"  ({detail})" if detail else ""))
passed = sum(ok for _, ok, _ in results)
print(f"\n{passed}/{len(results)} acceptance checks passed")
raise SystemExit(0 if passed == len(results) else 1)

python day10/acceptance.py

[PASS] A1 data covers all 16 weeks  (2026-04-26)
[PASS] A2 dbt build had no errors  (Done. PASS=31 WARN=1 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=32)
[PASS] A3 one feature row per customer
[PASS] B1 identical to the Day 3 panel
[PASS] C1 naive post-period gap reproduces Day 2  (+105.6%)
[PASS] C2 difference in growth reproduces Day 2  (+15.5%)
[PASS] C3 features balance: treated spent more pre-email

7/7 acceptance checks passed

Group C is the bridge to the rest of the course. The published mart reproduces both Day 2 numbers: the misleading naive gap (+106%) and the growth comparison close to the true +15%. Module 2 explains, with a DAG, why the first is biased; Module 5 turns the second into a proper difference-in-differences estimate with standard errors.

10.5 The data card (45 min)

Concept. A data card tells a new analyst what a dataset is for, how it was built, what each variable means, and what its known weaknesses are. Written by hand, it goes out of date with the next load. So we generate it from the published data and manifest.

Create day10/data_card.py:

# day10/data_card.py
"""Capstone step 3: write docs/DATA_CARD.md from the published data, so it is never out of date."""
import json
from pathlib import Path

import polars as pl

PUB = Path("data/published")
manifest = json.loads((PUB / "manifest.json").read_text())
panel = pl.read_parquet(PUB / "panel_customer_week.parquet")
features = pl.read_parquet(PUB / "customer_features.parquet")

balance = (features.group_by("treated")
           .agg(customers=pl.len(),
                pre_orders=pl.col("pre_orders").mean().round(2),
                pre_revenue=pl.col("pre_revenue").mean().round(0),
                loyalty_share=pl.col("is_loyalty").mean().round(3),
                age=pl.col("age").mean().round(1),
                tenure_days=pl.col("tenure_days").mean().round(0))
           .sort("treated"))


def md_table(df: pl.DataFrame) -> str:
    head = "| " + " | ".join(df.columns) + " |\n|" + "---|" * len(df.columns) + "\n"
    return head + "\n".join("| " + " | ".join(str(v) for v in row) + " |" for row in df.iter_rows())


card = f"""# Data card: NorthMart Causal Data Mart

*Generated by `day10/data_card.py` from the release published {manifest['published_at']}.
Do not edit by hand: rerun the script.*

## What question this data supports

Did the SPRING26 email (sent Monday 2 March 2026) change how much customers buy?

## Release

| Item | Value |
|---|---|
| Data as of | {manifest['data_as_of']} |
| dbt build | {manifest['dbt']} |
| Weeks | {panel['week_start'].min()} to {panel['week_start'].max()} ({panel['week_start'].n_unique()} weeks) |
| Customers | {panel['customer_id'].n_unique():,} |
| Emailed customers | {int(features['treated'].sum()):,} ({features['treated'].mean():.1%}) |

## Tables

| Table | Grain | Rows | Use it for |
|---|---|---|---|
| `panel_customer_week` | customer × week | {manifest['tables']['panel_customer_week']['rows']:,} | difference-in-differences, event studies (Module 5) |
| `customer_features` | customer | {manifest['tables']['customer_features']['rows']:,} | matching, weighting, causal ML (Modules 4 and 6) |

## Definitions

- **Treatment:** `treated = 1` if the customer appears in the SPRING26 recipient log. Fixed per customer.
- **Timing:** `post = 1` from week 9 (week of 2 March 2026); `event_week = 0` is the email week.
- **Outcomes:** `n_orders` (completed orders) and `revenue` (quantity × unit price after discount), 0 in weeks without orders. Cancelled orders are excluded.
- **Covariates:** region, age, loyalty, tenure, and pre-period behaviour, all measured **before** 2 March 2026 (contract C9, tested by time travel).

## Pre-treatment balance

{md_table(balance)}

Emailed customers were heavier shoppers and more often loyalty members **before** the email: the
assignment was targeted, not random. A raw comparison of outcomes is biased; see Modules 4-5.

## Known limitations

- Region is missing for {int((features['region'] == 'UNKNOWN').sum())} customers (`UNKNOWN`); this group has fewer than 30 emailed customers (warning C11).
- The pre-period covers 8 weeks only; longer histories would strengthen pre-trend checks.
- Data are simulated: the true effect is a 15% increase in order frequency, so methods can be graded.

## Rebuild

```bash
python api/northmart_api.py          # in a separate terminal
python pipeline/northmart_flow.py    # load, build, test, publish
python day10/acceptance.py
python day10/data_card.py

Contracts are listed in docs/CONTRACTS.md. """ Path("docs").mkdir(exist_ok=True) Path("docs/DATA_CARD.md").write_text(card) print(card[:1200]) print("...\nSaved docs/DATA_CARD.md")


```bash
python day10/data_card.py

Open docs/DATA_CARD.md in VS Code (right-click → Open Preview). The pre-treatment balance section reads:

| treated | customers | pre_orders | pre_revenue | loyalty_share | age | tenure_days |
|---|---|---|---|---|---|---|
| 0 | 2721 | 1.85 | 288.0 | 0.208 | 46.7 | 607.0 |
| 1 | 1279 | 3.23 | 506.0 | 0.491 | 47.3 | 594.0 |

Age and tenure are balanced; shopping intensity and loyalty are not. That single table tells a reader the email was targeted, and that any estimate must deal with it.

Make it run every week. Add the acceptance tests and the data card to the flow so each release documents itself. In pipeline/northmart_flow.py, add a task:

@task
def document() -> None:
    run([sys.executable, "day10/acceptance.py"])
    run([sys.executable, "day10/data_card.py"])

and call document() at the end of northmart_pipeline, after publish. (Acceptance check A1 expects all 16 weeks, so add it only once the replay is done, or relax A1 to "weeks published = weeks loaded".)

10.6 The handover memo (30 min)

Write docs/HANDOVER_MEMO.md, one page, for the analytics lead. Use these headings and answer in plain language; a non-technical manager should follow it.

The question and the data. What decision does this data inform? What are the unit, the treatment, the outcome and the period?
What the mart guarantees. Which contracts run on every release, and what happens when one fails? (Refer to CONTRACTS.md; don't repeat it.)
What the raw numbers say, and why not to trust them. The naive gap is +106%. Explain in two sentences why that can't be the effect, using the balance table.
What the data can support. Name one comparison that is more credible than the naive one and the assumption it relies on. (Hint: Day 2's growth comparison. What must be true of the two groups' trends?)
Known limitations and next steps. At least three, including the UNKNOWN region and the 8-week pre-period.

Model answer for section 3

Emailed customers placed twice as many orders as other customers after 2 March, but they were already placing 75% more before the email was sent: marketing targeted its best customers. Most of the post-email gap is therefore a difference between the kinds of customers, not an effect of the email, and the raw comparison overstates the effect several times over.

Model answer for section 4

Comparing how each group's orders changed from before to after the email removes the stable difference between the groups. Emailed customers' orders grew by 3% while other customers' fell by 12%, a gap of about 15 percentage points. This is credible only if, without the email, both groups' orders would have followed the same trend (parallel trends). Eight weeks of pre-period data let us check that partially; Module 5 does so formally.

10.7 Final commit and Module 1 retrospective (15 min)

git add .
git status        # data/, .dlt/secrets.toml, .dlt_pipelines/ and transform/target/ must NOT appear
git commit -m "Day 10: Causal Data Mart capstone: replay, acceptance, data card, memo"
git push

What you built in Module 1.

Day	Skill	Causal reason
1	Environment, causal vocabulary	Reproducible estimates; roles of variables
2	Polars cleaning, naive comparison	Selection bias, seen in the data
3	DuckDB panel with a calendar spine	Zeros are outcomes, not missing data
4–5	APIs and incremental dlt loads	Treatment logs live in company systems
6–7	dbt models and tests	Every transformation reviewed and tested
8	Causal data contracts	Leakage and timing errors stopped automatically
9	Prefect orchestration with a gate	Analysts never see data that failed a contract
10	Replay, acceptance, data card	A mart a team can trust and understand

Coming next: Module 2, Causal Thinking, Estimands & DAGs. You'll state precisely what "the effect of the email" means (the ATT, not the ATE, and why), draw the DAG behind the targeting you documented today, and use it to decide which variables in customer_features to adjust for, and which to leave alone.

Day 10 checklist

My pipeline runs correctly on any Sunday of the study, before and after the email
I have replayed all 16 weeks from an empty warehouse with every contract passing
My published release passes all acceptance checks, including agreement with Day 3
My data card is generated from the data and explains the treatment imbalance
My handover memo explains, in plain language, what the data can and cannot support

Capstone: The Causal Data Mart

What you'll learn today

Today's plan