macro_ingest v2 → The Warehouse

Time	Block
0:00 – 0:10	30.1 The final architecture
0:10 – 0:40	30.2 Resilient BoC and StatCan vector sources
0:40 – 0:48	Break
0:48 – 1:10	30.3 Full StatCan tables: change detection + chunked loading
1:10 – 1:30	30.4 ingest.py v2
1:30 – 1:38	Break
1:38 – 2:10	30.5 Switching the warehouse to raw_dlt, with a regression test
2:10 – 2:30	30.6 Runbook, pipeline script · Module 3 recap

---

30.1 The Final Architecture (10 min)

      Bank of Canada Valet   FRED   StatCan WDS (vectors)   StatCan full tables
               │               │            │                        │
               ▼               ▼            ▼                        ▼
         ┌───────────────────────────────────────────────────────────────┐
         │ macro_ingest (dlt)                                            │
         │  RESTClient · retries · throttle · batching · circuit breaker │
         │  payload validation · per-series cursors · merge · contracts  │
         │  release-time change detection · ErrorLog                     │
         └───────────────────────────────────────────────────────────────┘
               │  python ingest.py
               ▼
  macro.duckdb ─ raw_dlt.*   boc/fred/statcan_observations, *_metadata,
               │             statcan_cpi_full, statcan_gdp_full,
               │             statcan_releases, extraction_errors, ingest_runs
               │  python load_warehouse.py   (seeds → staging → intermediate → marts)
               ▼
               seeds.* · staging.* · intermediate.* · marts.*
               │  python run_tests.py
               ▼
         data/marts/*.parquet  →  dashboards (course notebook)

The Day 10 path (run_extract.py, macro_extract/, raw.observations, raw.statcan_*_full) is retired today. Keep the code in Git history for reference, but nothing reads from it anymore.

---

30.2 Resilient BoC and StatCan Vector Sources (30 min)

macro_ingest/clients.py

"""Configured REST clients, one per provider."""
from dlt.sources.helpers.rest_client import RESTClient
from dlt.sources.helpers.rest_client.paginators import SinglePagePaginator

USER_AGENT = {"User-Agent": "macro-data-course/1.0 (educational use)"}


def valet_client():
    return RESTClient(base_url="https://www.bankofcanada.ca/valet/",
                      headers=USER_AGENT, paginator=SinglePagePaginator())


def wds_client():
    return RESTClient(base_url="https://www150.statcan.gc.ca/t1/wds/rest/",
                      headers=USER_AGENT, paginator=SinglePagePaginator())

macro_ingest/boc.py (v2)

"""Bank of Canada Valet — resilient, incremental, isolated per series."""
from datetime import date, timedelta

import dlt

from .clients import valet_client
from .parsers import fetch_valet_series
from .resilience import CircuitBreaker, CircuitOpen, ErrorLog, Throttle

OBS_CONTRACT = {"columns": "freeze", "data_type": "freeze"}


@dlt.source(name="boc", section="boc")
def boc_source(series_ids: list = dlt.config.value,
               start_date: str = dlt.config.value,
               lookback_days: int = 14,
               min_interval: float = 0.3,
               max_consecutive_failures: int = 3,
               error_log: ErrorLog = None):
    client = valet_client()
    errors = error_log if error_log is not None else ErrorLog()

    @dlt.resource(name="boc_observations", write_disposition="merge",
                  primary_key=("series_id", "date"),
                  columns={"value": {"data_type": "double"}},
                  schema_contract=OBS_CONTRACT)
    def observations():
        cursors = dlt.current.resource_state().setdefault("last_date", {})
        throttle = Throttle(min_interval)
        breaker = CircuitBreaker(max_consecutive_failures)

        for sid in series_ids:
            start = (date.fromisoformat(cursors[sid]) - timedelta(days=lookback_days)
                     if sid in cursors else date.fromisoformat(start_date))
            throttle.wait()
            try:
                rows = fetch_valet_series(client, sid, start)
            except CircuitOpen:
                raise
            except Exception as e:
                errors.add("boc", sid, e)
                breaker.failure()
                continue
            breaker.success()
            if rows:
                cursors[sid] = max(r["date"] for r in rows).isoformat()
            yield rows

    @dlt.resource(name="boc_series_metadata", write_disposition="replace",
                  primary_key="series_id",
                  schema_contract={"columns": "evolve", "data_type": "freeze"})
    def series_metadata():
        for sid in series_ids:
            try:
                resp = client.get(f"series/{sid}/json")
                resp.raise_for_status()
                details = resp.json().get("seriesDetails", {})
                yield {"series_id": sid, "label": details.get("label"),
                       "description": details.get("description")}
            except Exception as e:
                errors.add("boc", sid, e, stage="metadata")

    return observations, series_metadata

Notes:

• The error_log argument lets the runner pass in its own ErrorLog and read it afterwards. Explicit arguments bypass configuration injection (Day 23).
• Metadata now uses Valet's series/{id}/json endpoint (Day 8.4). Check its response keys once with print(resp.json().keys()) and adjust seriesDetails if your response differs.

macro_ingest/statcan.py (v2): batched, isolated, incremental

"""Statistics Canada WDS vectors — batched POSTs, per-item isolation, merge."""
import dlt

from .clients import wds_client
from .parsers import parse_wds_items
from .resilience import CircuitBreaker, CircuitOpen, ErrorLog, Throttle


def _chunked(items, size):
    for i in range(0, len(items), size):
        yield items[i:i + size]


@dlt.source(name="statcan", section="statcan")
def statcan_source(vectors: list = dlt.config.value,
                   initial_latest_n: int = 250,
                   latest_n: int = 36,
                   batch_size: int = 25,
                   min_interval: float = 0.5,
                   max_consecutive_failures: int = 3,
                   error_log: ErrorLog = None):
    client = wds_client()
    errors = error_log if error_log is not None else ErrorLog()

    @dlt.resource(name="statcan_observations", write_disposition="merge",
                  primary_key=("series_id", "date"),
                  columns={"value": {"data_type": "double"}},
                  schema_contract={"columns": "freeze", "data_type": "freeze"})
    def observations():
        loaded = dlt.current.resource_state().setdefault("loaded_vectors", [])
        throttle = Throttle(min_interval)
        breaker = CircuitBreaker(max_consecutive_failures)

        for batch in _chunked(list(vectors), batch_size):
            body = [{"vectorId": v,
                     "latestN": latest_n if f"v{v}" in loaded else initial_latest_n}
                    for v in batch]
            throttle.wait()
            try:
                resp = client.post("getDataFromVectorsAndLatestNPeriods", json=body)
                resp.raise_for_status()
                rows = parse_wds_items(resp.json(), errors)     # FAILED items → ErrorLog
            except CircuitOpen:
                raise
            except Exception as e:
                errors.add("statcan", ",".join(f"v{v}" for v in batch), e, stage="batch")
                breaker.failure()
                continue
            breaker.success()
            for sid in {r["series_id"] for r in rows}:
                if sid not in loaded:
                    loaded.append(sid)
            yield rows

    return observations

The first successful load of a vector uses initial_latest_n (a long history), and later runs use latest_n (a revision window, merged). Vector IDs are recorded as loaded only when data arrived.

Keep the Day 27 fred.py, adding the same error_log, throttle and circuit-breaker pattern around each series (an exercise; it mirrors boc.py).

---

30.3 Full StatCan Tables: Change Detection + Chunked Loading (23 min)

The staging models (Days 18–20) need the full CPI and GDP tables. Loading them through dlt (instead of Day 13's read_csv in SQL) gives them the same run tracking as everything else. Two techniques keep this efficient:

1. Change detection: download only tables whose releaseTime changed since the last successful load (Day 27.5).
2. Chunked loading: read the 1M-row CSV in pieces and yield pandas DataFrames, which dlt loads through its fast columnar path.

macro_ingest/statcan_tables.py

"""Full StatCan tables: release-time change detection, streamed download, chunked load."""
import io
import os
import zipfile

import dlt
import pandas as pd
from dlt.sources.helpers import requests

from .clients import wds_client

RAW_DIR = "data/raw/statcan"


def current_release_times(product_ids):
    """{product_id (str): releaseTime} from getCubeMetadata."""
    resp = wds_client().post("getCubeMetadata",
                             json=[{"productId": int(pid)} for pid in product_ids])
    resp.raise_for_status()
    out = {}
    for item in resp.json():
        if item.get("status") == "SUCCESS":
            obj = item["object"]
            out[str(obj.get("productId"))] = obj.get("releaseTime")
    return out


def _download_zip(product_id):
    meta = wds_client().get(f"getFullTableDownloadCSV/{product_id}/en")
    meta.raise_for_status()
    info = meta.json()
    if info.get("status") != "SUCCESS":
        raise RuntimeError(f"Table {product_id}: {info.get('status')}")
    os.makedirs(RAW_DIR, exist_ok=True)
    path = os.path.join(RAW_DIR, f"{product_id}.zip")
    with requests.get(info["object"], stream=True, timeout=300) as r:
        r.raise_for_status()
        with open(path, "wb") as f:
            for chunk in r.iter_content(chunk_size=1024 * 1024):
                f.write(chunk)
    return path


@dlt.source(name="statcan_tables", section="statcan_tables")
def statcan_tables_source(tables: dict = dlt.config.value, chunk_rows: int = 200_000):
    """`tables` maps product id → destination table name."""

    def make_resource(product_id, table_name):
        @dlt.resource(name=table_name, write_disposition="replace")
        def table():
            path = _download_zip(product_id)
            with zipfile.ZipFile(path) as z:
                name = [f for f in z.namelist()
                        if f.endswith(".csv") and not f.endswith("_MetaData.csv")][0]
                with z.open(name) as fh:
                    reader = pd.read_csv(io.TextIOWrapper(fh, encoding="utf-8-sig"),
                                         dtype=str, chunksize=chunk_rows)
                    for chunk in reader:
                        chunk.columns = chunk.columns.str.strip()
                        yield chunk                       # all text: raw layer (Day 18)
        return table

    return [make_resource(pid, name) for pid, name in tables.items()]

Configuration:

[sources.statcan_tables]
tables = { "18100004" = "statcan_cpi_full", "36100104" = "statcan_gdp_full" }

Points worth noticing:

• dtype=str keeps every value as text, as a raw layer should (Day 18). Staging casts.
• make_resource(pid, name) is a small factory function: it creates one resource per table and avoids the loop-variable gotcha from Day 22.3.
• Column names will be normalised by dlt: REF_DATE → ref_date, Products and product groups → products_and_product_groups, and so on. Staging must be updated accordingly (30.5).

The order of operations matters

Record a release marker only after its data has loaded successfully.

If you recorded "CPI release of 2026-09-16 processed" and then the download failed, the next run would think the table is up to date and never retry. ingest.py therefore:

1. asks the API for current release times;
2. compares them with the last loaded release times stored in raw_dlt.statcan_releases;
3. loads only the changed tables;
4. then appends their release times to raw_dlt.statcan_releases.

---

30.4 ingest.py v2 (20 min)

"""ingest.py — resilient ingestion of all macro sources into DuckDB (dataset raw_dlt).

Usage:
  python ingest.py                     # incremental
  python ingest.py --full-refresh      # drop and reload all resources
  python ingest.py --only boc,statcan  # a subset of sources

Exit codes: 0 all ok · 2 partial failure · 1 nothing loaded
"""
import sys
import time
from datetime import datetime, timezone

import dlt

from macro_ingest.boc import boc_source
from macro_ingest.fred import fred_source
from macro_ingest.resilience import ErrorLog, notify
from macro_ingest.statcan import statcan_source
from macro_ingest.statcan_tables import current_release_times, statcan_tables_source

FULL_REFRESH = "--full-refresh" in sys.argv
ONLY = next((a.split("=", 1)[1] if "=" in a else sys.argv[sys.argv.index(a) + 1]
             for a in sys.argv if a.startswith("--only")), None)

SOURCES = {"boc": boc_source, "fred": fred_source, "statcan": statcan_source}


def loaded_release_times(pipeline):
    """Last successfully loaded releaseTime per product, from the warehouse itself."""
    try:
        with pipeline.sql_client() as client:
            table = client.make_qualified_table_name("statcan_releases")
            rows = client.execute_sql(
                f"SELECT product_id, MAX(release_time) FROM {table} GROUP BY 1")
        return {str(pid): rt for pid, rt in rows}
    except Exception:
        return {}                                       # first run: table doesn't exist yet


def run_source(pipeline, name, make_source, errors):
    t0 = time.perf_counter()
    status, message = "ok", None
    try:
        info = pipeline.run(make_source(error_log=errors),
                            refresh="drop_resources" if FULL_REFRESH else None)
        info.raise_on_failed_jobs()
    except Exception as e:
        status, message = "failed", f"{type(e).__name__}: {str(e)[:300]}"
    return {"source": name, "status": status, "message": message,
            "duration_s": round(time.perf_counter() - t0, 1)}


def main():
    started = datetime.now(timezone.utc)
    pipeline = dlt.pipeline(pipeline_name="macro_ingest", destination="duckdb",
                            dataset_name="raw_dlt", export_schema_path="schemas/export")
    wanted = set(ONLY.split(",")) if ONLY else set(SOURCES) | {"statcan_tables"}
    summaries, errors = [], ErrorLog()

    # 1. Observation sources, each isolated
    for name, make_source in SOURCES.items():
        if name in wanted:
            s = run_source(pipeline, name, make_source, errors)
            summaries.append(s)
            print(f"{'✓' if s['status'] == 'ok' else '✗'} {name} ({s['duration_s']}s)")

    # 2. Full tables, only when republished
    if "statcan_tables" in wanted:
        t0 = time.perf_counter()
        try:
            tables = dlt.config["sources.statcan_tables.tables"]
            current = current_release_times(tables.keys())
            loaded = loaded_release_times(pipeline)
            changed = {pid: tbl for pid, tbl in tables.items()
                       if FULL_REFRESH or current.get(pid) != loaded.get(pid)}
            if changed:
                info = pipeline.run(statcan_tables_source(tables=changed))
                info.raise_on_failed_jobs()
                pipeline.run([{"product_id": pid, "release_time": current.get(pid),
                               "loaded_at": datetime.now(timezone.utc)} for pid in changed],
                             table_name="statcan_releases", write_disposition="append")
            print(f"✓ statcan_tables: {len(changed)} table(s) refreshed")
            summaries.append({"source": "statcan_tables", "status": "ok",
                              "message": f"refreshed {sorted(changed)}",
                              "duration_s": round(time.perf_counter() - t0, 1)})
        except Exception as e:
            print(f"✗ statcan_tables: {e}")
            summaries.append({"source": "statcan_tables", "status": "failed",
                              "message": str(e)[:300],
                              "duration_s": round(time.perf_counter() - t0, 1)})

    # 3. Record errors and the run summary (always)
    if errors.records:
        pipeline.run(errors.records, table_name="extraction_errors", write_disposition="append")
    for s in summaries:
        s.update({"run_started_at": started, "series_errors": len(errors)})
    pipeline.run(summaries, table_name="ingest_runs", write_disposition="append")

    # 4. Alert and exit
    failed = [s["source"] for s in summaries if s["status"] != "ok"]
    if failed or errors.records:
        notify(f"macro_ingest: failed sources {failed or 'none'}; "
               f"{len(errors)} series-level error(s). See raw_dlt.extraction_errors.")
    if len(failed) == len(summaries):
        sys.exit(1)
    if failed or errors.records:
        sys.exit(2)


if __name__ == "__main__":
    main()

Run it:

python ingest.py --full-refresh
python ingest.py                     # incremental: fast, and tables are skipped if unchanged

SELECT run_started_at, source, status, duration_s, series_errors, message
FROM raw_dlt.ingest_runs ORDER BY run_started_at DESC, source LIMIT 10;

SELECT * FROM raw_dlt.extraction_errors ORDER BY occurred_at DESC LIMIT 10;
SELECT * FROM raw_dlt.statcan_releases ORDER BY loaded_at DESC;

Try it 30.4 — prove the isolation works

Temporarily add a bogus code ("NOT_A_SERIES") to [sources.boc] series_ids. Run python ingest.py --only boc and verify that: (a) the other BoC series still load; (b) an error row appears in extraction_errors; (c) the exit code is 2 (echo $?); (d) an alert is printed or sent. Then remove the bogus code.

---

30.5 Switching the Warehouse to raw_dlt, with a Regression Test (33 min)

Step 1 — Snapshot the marts before the switch

mkdir -p data/migration
duckdb macro.duckdb -readonly -c "
COPY marts.macro_monthly   TO 'data/migration/before_macro_monthly.parquet'   (FORMAT parquet);
COPY marts.macro_quarterly TO 'data/migration/before_macro_quarterly.parquet' (FORMAT parquet);
COPY marts.policy_cycles   TO 'data/migration/before_policy_cycles.parquet'   (FORMAT parquet);"

Step 2 — A series seed replaces raw.dim_series

Add to sql/03_seeds.sql. This turns the Day 20 stretch goal into reality: the aggregation rule lives in data, not in a CASE.

-- Grain: one row per ingested series
CREATE OR REPLACE TABLE seeds.seed_series (
    series_id           VARCHAR PRIMARY KEY,
    series_name         VARCHAR NOT NULL UNIQUE,
    source              VARCHAR NOT NULL,
    frequency           VARCHAR NOT NULL,
    monthly_agg         VARCHAR NOT NULL CHECK (monthly_agg IN ('last', 'mean')),
    max_staleness_days  INTEGER NOT NULL
);
INSERT INTO seeds.seed_series VALUES
    ('V39079',             'overnight_rate',    'boc',     'daily',   'last', 10),
    ('FXCADUSD',           'cad_usd',           'boc',     'daily',   'mean', 10),
    ('BD.CDN.2YR.DQ.YLD',  'yield_2yr',         'boc',     'daily',   'mean', 10),
    ('BD.CDN.10YR.DQ.YLD', 'yield_10yr',        'boc',     'daily',   'mean', 10),
    ('FEDFUNDS',           'fed_funds',         'fred',    'monthly', 'last', 75),
    ('DCOILWTICO',         'wti',               'fred',    'daily',   'mean', 15),
    ('CANLOLITOAASTSAM',   'oecd_cli',          'fred',    'monthly', 'mean', 120),
    ('v41690973',          'cpi_all_items',     'statcan', 'monthly', 'last', 75),
    ('v2062815',           'unemployment_rate', 'statcan', 'monthly', 'last', 60);

max_staleness_days reflects each series' publication rhythm (e.g. monthly CPI appears several weeks after the month ends). Adjust the numbers to what you observe.

Step 3 — Point staging at raw_dlt

sql/12_stg_macro__observations.sql

-- staging.stg_macro__observations
-- Source: raw_dlt.{boc,fred,statcan}_observations (dlt, incremental, merged)
-- Grain:  one row per series × observation date
CREATE OR REPLACE VIEW staging.stg_macro__observations AS
WITH unioned AS (
    SELECT date, series_id, value, source, _dlt_load_id FROM raw_dlt.boc_observations
    UNION ALL
    SELECT date, series_id, value, source, _dlt_load_id FROM raw_dlt.fred_observations
    UNION ALL
    SELECT date, series_id, value, source, _dlt_load_id FROM raw_dlt.statcan_observations
)
SELECT u.date          AS obs_date,
       u.series_id,
       s.series_name,
       u.source,
       u.value,
       s.frequency,
       s.monthly_agg,
       l.inserted_at   AS loaded_at
FROM unioned AS u
JOIN seeds.seed_series AS s USING (series_id)
LEFT JOIN raw_dlt._dlt_loads AS l ON l.load_id = u._dlt_load_id;

sql/23_int_rates__monthly.sql: replace the hard-coded CASE WHEN series_name IN (...) with the seed's rule:

CASE WHEN monthly_agg = 'last' THEN arg_max(value, obs_date) ELSE AVG(value) END AS value

(and add monthly_agg to that CTE's GROUP BY ALL, which happens automatically).

sql/10_stg_statcan__cpi.sql: the source is now raw_dlt.statcan_cpi_full, with normalised names:

WITH source AS (
    SELECT * FROM raw_dlt.statcan_cpi_full
),
renamed AS (
    SELECT
        CAST(ref_date || '-01' AS DATE)          AS month,
        geo,
        products_and_product_groups              AS component,
        vector                                   AS vector_id,
        uom                                      AS index_base,
        TRY_CAST(value AS DOUBLE)                AS cpi,
        NULLIF(TRIM(status), '')                 AS status_flag,
        COALESCE(terminated = 't', false)        AS is_terminated
    FROM source
)
-- final SELECT unchanged from Day 18

sql/11_stg_statcan__gdp.sql: same idea: ref_date, estimates, prices, seasonal_adjustment, uom, scalar_factor, vector, value, status, geo from raw_dlt.statcan_gdp_full.

Always confirm the real column names with DESCRIBE raw_dlt.statcan_cpi_full; before editing. The staging layer's job is to hide such names from everything downstream, and that's why only these few files change.

Step 4 — Find everything else that still reads the old tables

grep -rn "raw\.observations\|raw\.statcan\|raw\.dim_series" sql/ load_warehouse.py

Typical hits and fixes:

• sql/32_policy_cycles.sql: yields from staging.stg_macro__observations WHERE series_name = 'yield_2yr' (column obs_date); CPI from staging.stg_statcan__cpi WHERE geo_code = 'CA' AND component = 'All-items'.
• sql/02_raw_tables.sql: delete the loads of raw.observations, raw.statcan_cpi, raw.statcan_cpi_full, raw.statcan_gdp_full. Keep raw.boc_decisions (it's hand-maintained, more like a seed).
• load_warehouse.py: delete the raw.dim_series creation.
• Old tests referencing raw.observations: point them at staging.stg_macro__observations or raw_dlt.*.

Step 5 — New tests

-- sql/tests/series_without_seed.sql
-- severity: error
-- Every ingested series must be described in seeds.seed_series
SELECT DISTINCT series_id FROM (
    SELECT series_id FROM raw_dlt.boc_observations
    UNION SELECT series_id FROM raw_dlt.fred_observations
    UNION SELECT series_id FROM raw_dlt.statcan_observations)
ANTI JOIN seeds.seed_series USING (series_id)

-- sql/tests/series_freshness.sql
-- severity: warn
SELECT s.series_name, MAX(o.obs_date) AS latest, s.max_staleness_days
FROM seeds.seed_series AS s
LEFT JOIN staging.stg_macro__observations AS o USING (series_id)
GROUP BY ALL
HAVING MAX(o.obs_date) IS NULL
    OR MAX(o.obs_date) < current_date - s.max_staleness_days

-- sql/tests/last_ingest_run_ok.sql
-- severity: warn
SELECT source, status, message
FROM raw_dlt.ingest_runs
WHERE run_started_at = (SELECT MAX(run_started_at) FROM raw_dlt.ingest_runs)
  AND status <> 'ok'

Step 6 — Rebuild and compare with the snapshot

python load_warehouse.py && python run_tests.py

-- Months whose key values changed after the migration
WITH before AS (
    SELECT month, ROUND(overnight_rate, 6) AS overnight_rate,
           ROUND(headline_yoy, 6) AS headline_yoy, ROUND(real_rate, 6) AS real_rate
    FROM 'data/migration/before_macro_monthly.parquet'
),
after AS (
    SELECT month, ROUND(overnight_rate, 6), ROUND(headline_yoy, 6), ROUND(real_rate, 6)
    FROM marts.macro_monthly
)
SELECT 'only_before' AS side, * FROM (SELECT * FROM before EXCEPT SELECT * FROM after)
UNION ALL
SELECT 'only_after', * FROM (SELECT * FROM after EXCEPT SELECT * FROM before)
ORDER BY month, side;

How to read the result:

• Differences only in the latest month(s): new data published between snapshots. Expected.
• Differences in old months: revisions picked up by the lookback window, or a bug in the migration. Investigate each one.
• Nothing at all: a clean migration.

This technique, snapshot → change → compare, is called a regression test. Use it whenever you refactor a pipeline: it proves you changed the plumbing, not the numbers.

---

30.6 Runbook, Pipeline Script · Module 3 Recap (20 min)

pipeline.sh (updated)

#!/usr/bin/env bash
# pipeline.sh — ingest → build → test. Tolerates partial ingestion failures (exit 2).
set -e
source .venv/bin/activate

python ingest.py || [ $? -eq 2 ]
python load_warehouse.py
python run_tests.py
echo "✓ Pipeline complete: data/marts/ is up to date"

cmd || [ $? -eq 2 ] means "accept exit code 2 (partial success) and continue". A complete ingestion failure (exit 1) still stops the pipeline, and the warehouse keeps its last good data.

RUNBOOK.md: what to do when things break

A runbook is written for someone else (or you, at 7 a.m., half-awake). Create it at the repository root:

# Macro pipeline runbook

## Quick health check
    duckdb macro.duckdb -readonly -c "SELECT * FROM raw_dlt.ingest_runs ORDER BY run_started_at DESC LIMIT 5"
    python run_tests.py

## Symptoms → actions
| Symptom | Likely cause | Action |
|---|---|---|
| Alert: series-level errors | typo in config, discontinued series | check raw_dlt.extraction_errors; fix .dlt/config.toml and seeds.seed_series |
| Alert: CircuitOpen for a source | provider outage | wait; rerun `python ingest.py --only <source>` later |
| PipelineStepFailed at normalize | API changed shape (contract) | inspect error + `git diff schemas/`; adapt parser; evolve deliberately |
| "database is locked" | an open DuckDB session | close CLI/notebook sessions; rerun |
| Freshness warning, no errors | provider hasn't published yet | check provider release calendar; no action |
| Marts changed for old months | revisions via lookback | expected; note in release notes |

## Full rebuild
    python ingest.py --full-refresh && python load_warehouse.py && python run_tests.py

## Adding a series
1. Add the code to `.dlt/config.toml` (sources.<provider>.series_ids / vectors)
2. Add a row to `seeds.seed_series` in `sql/03_seeds.sql`
3. `./pipeline.sh`; check the `series_without_seed` and freshness tests pass

Adding a series is now a two-line change with a checklist. That's the payoff of separating configuration, seeds and code.

Module 3 recap (Days 21–30)

Day	Capability
21	pipelines, resources, generators, write dispositions
22	sources vs resources, transformers, metadata as dimensions
23	configuration and secrets, dev/prod destinations, the macro_ingest package
24	schema inference, evolution, variant columns, schema files in Git
25	schema contracts, Pydantic validation, contract tests
26	nested JSON, child tables, JSON columns, the CPI product hierarchy
27	incremental cursors, revisions with lookback + merge, change detection, vintages
28	RESTClient, auth, pagination, declarative REST sources
29	failure taxonomy, retries, throttling, isolation, circuit breakers, mocked tests, alerts
30	resilient v2 extractors, full tables via dlt, warehouse migration with a regression test, runbook

What still hurts, and what fixes it next

Pain point	Fixed by
Numbered SQL files and a hand-written loader decide build order	dbt: dependencies inferred from ref()
Staging/intermediate/marts conventions enforced only by discipline	dbt project structure, materialisations, macros
SQL tests are hand-rolled files with a custom runner	dbt tests (Days 31–40), then Great Expectations (Days 41–50)
pipeline.sh has no schedule, retries at the step level, or UI	Prefect (Days 51–60)

Day 30 checklist

• My BoC and StatCan extractors use RESTClient, validation, throttling, isolation and circuit breakers
• Full StatCan tables load through dlt, only when republished, in chunks, with markers recorded after success
• ingest.py records errors and run summaries, alerts on problems, and returns meaningful exit codes
• Staging reads raw_dlt, series metadata lives in seeds.seed_series, and the old path is retired
• A snapshot comparison shows my marts are unchanged apart from explainable differences
• My runbook explains how to diagnose failures and add a series

./pipeline.sh
git add .
git commit -m "Day 30: resilient macro_ingest v2, warehouse on raw_dlt, runbook — Module 3 complete"
git push