Statistics Canada WDS and Your Own Extraction Package

10.1 How Statistics Canada Organises Data (20 min)

StatCan's Web Data Service (WDS) exposes three levels:

The notebook downloads whole tables (getFullTableDownloadCSV): over a million rows for CPI, just to keep a handful of series. That's simple but slow. When you know which series you need, vectors are far more efficient.

Finding a vector ID

1. Open the table page on statcan.gc.ca (search "18-10-0004-01").
2. Click Add/Remove data and select the series you want.
3. Under Customize layout, tick Display Vector identifier and Coordinate.
4. The table now shows a v... ID beside each series.

Always verify a vector ID by fetching its metadata (10.2, Example 1) before building on it. A wrong ID returns a valid but different series, and nothing crashes.

10.2 The Vector Endpoints (30 min)

These WDS endpoints use POST with a JSON body: a list of requests. Create day10/statcan_vectors.py.

Example 1 — Verify vectors via their metadata

import requests

WDS = "https://www150.statcan.gc.ca/t1/wds/rest"

vectors = [41690973, 2062815]      # CPI All-items; unemployment rate (LFS)
body = [{"vectorId": v} for v in vectors]

r = requests.post(f"{WDS}/getSeriesInfoFromVector", json=body, timeout=30)
r.raise_for_status()

for item in r.json():
    if item.get("status") != "SUCCESS":
        print("✗", item)
        continue
    obj = item["object"]
    print(f"v{obj['vectorId']}: table {obj.get('productId')} · "
          f"{obj.get('SeriesTitleEn', '(no title)')}")

json=body tells requests to convert the Python list to JSON and set the right headers. Check that the printed titles match what you expect. If they don't, look up the correct vector (10.1) before continuing.

Example 2 — Fetch the latest n periods

body = [{"vectorId": v, "latestN": 24} for v in vectors]
r = requests.post(f"{WDS}/getDataFromVectorsAndLatestNPeriods", json=body, timeout=30)
r.raise_for_status()
payload = r.json()

first = payload[0]["object"]
print(first.keys())
print(first["vectorDataPoint"][-1])     # refPer (reference period), value, ...

Example 3 — Parse into our standard long format

import pandas as pd


def parse_vectors(payload):
    """StatCan vector payload → long DataFrame (date, series_id, value)."""
    rows = []
    for item in payload:
        if item.get("status") != "SUCCESS":
            continue
        obj = item["object"]
        sid = f"v{obj['vectorId']}"
        for dp in obj.get("vectorDataPoint", []):
            rows.append({"date": dp.get("refPer"), "series_id": sid,
                         "value": dp.get("value")})
    df = pd.DataFrame(rows, columns=["date", "series_id", "value"])
    df["date"] = pd.to_datetime(df["date"], errors="coerce")
    df["value"] = pd.to_numeric(df["value"], errors="coerce")
    return df.dropna(subset=["date", "value"]).reset_index(drop=True)


stat = parse_vectors(payload)
print(stat.pivot(index="date", columns="series_id", values="value").tail())

Compare with parse_valet and parse_fred_csv: three different source formats, one output schema.

Try it 10.2

Fetch the last 240 months of CPI All-items (latestN: 240) and compute the latest YoY inflation with pct_change(12) (Day 5). Does it match the value from your Day 7 panel?

r = requests.post(f"{WDS}/getDataFromVectorsAndLatestNPeriods",
                  json=[{"vectorId": 41690973, "latestN": 240}], timeout=30)
r.raise_for_status()
cpi = parse_vectors(r.json()).set_index("date")["value"].sort_index()
yoy = cpi.pct_change(12) * 100
print(f"{yoy.iloc[-1]:.1f}% ({yoy.index[-1]:%b %Y})")

For long histories between specific dates, WDS also offers getDataFromVectorByReferencePeriodRange. For entire tables, use the full download (next block).

10.3 Full-Table Downloads, Done Efficiently (23 min)

Sometimes you need the whole table, e.g. all CPI components for all provinces. Let's improve the notebook's statcan_fetch in three ways:

1. Stream the ZIP to disk in chunks instead of holding it all in memory.
2. Cache it: skip the download if a recent copy exists.
3. Read only the columns you need and filter immediately.

Create day10/statcan_tables.py:

import os
import time
import zipfile

import pandas as pd
import requests

WDS = "https://www150.statcan.gc.ca/t1/wds/rest"


def download_table(table_id, session, folder="data/raw/statcan", max_age_hours=24):
    """Download a full StatCan table ZIP, reusing a recent cached copy."""
    os.makedirs(folder, exist_ok=True)
    zip_path = os.path.join(folder, f"{table_id}.zip")

    if os.path.exists(zip_path):
        age_hours = (time.time() - os.path.getmtime(zip_path)) / 3600
        if age_hours < max_age_hours:
            print(f"  using cached {zip_path} ({age_hours:.1f}h old)")
            return zip_path

    meta = session.get(f"{WDS}/getFullTableDownloadCSV/{table_id}/en", timeout=30)
    meta.raise_for_status()
    info = meta.json()
    if info.get("status") != "SUCCESS":
        raise ValueError(f"Table {table_id}: {info.get('status')}")

    with session.get(info["object"], stream=True, timeout=300) as r:
        r.raise_for_status()
        with open(zip_path, "wb") as f:                    # "wb" = write bytes
            for chunk in r.iter_content(chunk_size=1024 * 1024):
                f.write(chunk)
    print(f"  downloaded {zip_path} ({os.path.getsize(zip_path) / 1e6:.1f} MB)")
    return zip_path


def read_table(zip_path, columns, filters):
    """Read selected columns from a StatCan ZIP and apply {column: [values]} filters."""
    with zipfile.ZipFile(zip_path) as z:
        name = [f for f in z.namelist()
                if f.endswith(".csv") and not f.endswith("_MetaData.csv")][0]
        df = pd.read_csv(z.open(name), encoding="utf-8-sig", low_memory=False,
                         usecols=lambda c: c.strip() in columns)
    df.columns = df.columns.str.strip()
    for col, allowed in filters.items():
        df = df[df[col].isin(allowed)]
    return df

Usage, reproducing the notebook's CPI extraction:

session = requests.Session()
path = download_table("18100004", session)
cpi = read_table(
    path,
    columns=["REF_DATE", "GEO", "Products and product groups", "VALUE"],
    filters={"GEO": ["Canada"],
             "Products and product groups": ["All-items", "Shelter", "Energy"]},
)
print(cpi.shape)
print(cpi.tail())

Run it twice: the second run uses the cache and finishes in seconds.

usecols=lambda c: ... passes a small anonymous function (a lambda) that pandas calls for each column name. We strip spaces first, remembering the notebook's hidden-BOM problem.

10.4 Building the macro_extract Package (35 min)

Now we assemble everything into a package: a folder of modules with an __init__.py file. It's the Day 3 macrotools.py idea, grown up.

mkdir -p macro_extract tests data/staging
touch macro_extract/__init__.py

macro_extract/net.py — one shared, polite, retrying session

"""Shared HTTP session with retries (Day 9)."""
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def make_session(retries=4, backoff=1.0):
    retry = Retry(total=retries, backoff_factor=backoff,
                  status_forcelist=[429, 500, 502, 503, 504],
                  allowed_methods=["GET", "POST"])
    session = requests.Session()
    session.mount("https://", HTTPAdapter(max_retries=retry))
    session.headers.update({"User-Agent": "macro-data-course/1.0 (educational use)"})
    return session

We call this module net.py, not http.py, because Python already has a built-in module named http. Avoid naming your files after standard modules (http, json, csv, io...).

macro_extract/schema.py — one output format for every source

"""Standard long schema shared by all sources, plus validation."""
import pandas as pd

COLUMNS = ["date", "series_id", "value", "source"]


def standardize(df, source):
    out = df.copy()
    out["date"] = pd.to_datetime(out["date"], errors="coerce")
    out["value"] = pd.to_numeric(out["value"], errors="coerce")
    out["source"] = source
    out = out.dropna(subset=["date", "value"])
    return out[COLUMNS].sort_values(["series_id", "date"]).reset_index(drop=True)


def validate(df, label, min_rows=1):
    missing = [c for c in COLUMNS if c not in df.columns]
    if missing:
        raise ValueError(f"{label}: missing columns {missing}")
    if len(df) < min_rows:
        raise ValueError(f"{label}: only {len(df)} rows")
    dupes = int(df.duplicated(["series_id", "date"]).sum())
    if dupes:
        raise ValueError(f"{label}: {dupes} duplicate (series_id, date) rows")
    return df

macro_extract/rawstore.py — the landing zone

"""Save untouched API responses under data/raw/ (Day 9.6)."""
import os
from datetime import datetime, timezone


def save_raw(content, source, series_id, extension, root="data/raw"):
    stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H-%M-%S")
    folder = os.path.join(root, source, series_id.replace("/", "_"))
    os.makedirs(folder, exist_ok=True)
    path = os.path.join(folder, f"{stamp}.{extension}")
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return path

macro_extract/boc.py

"""Bank of Canada Valet API."""
import pandas as pd

from .rawstore import save_raw
from .schema import standardize

BASE = "https://www.bankofcanada.ca/valet"


def parse_valet(payload):
    series_ids = list(payload.get("seriesDetail", {}).keys())
    rows = []
    for obs in payload.get("observations", []):
        for sid in series_ids:
            cell = obs.get(sid)
            if isinstance(cell, dict):
                rows.append({"date": obs.get("d"), "series_id": sid,
                             "value": cell.get("v")})
    return standardize(pd.DataFrame(rows, columns=["date", "series_id", "value"]), "boc")


def fetch(series_id, session, start="2006-01-01"):
    r = session.get(f"{BASE}/observations/{series_id}/json",
                    params={"start_date": start}, timeout=30)
    r.raise_for_status()
    save_raw(r.text, "boc", series_id, "json")
    return parse_valet(r.json())

The dot in from .schema import ... means "from this same package". That's a relative import.

macro_extract/fred.py

"""FRED graph CSV endpoint (no key required)."""
import io

import pandas as pd

from .rawstore import save_raw
from .schema import standardize

URL = "https://fred.stlouisfed.org/graph/fredgraph.csv"


def parse_fred_csv(text, series_id):
    df = pd.read_csv(io.StringIO(text)).iloc[:, :2]
    df.columns = ["date", "value"]
    df["series_id"] = series_id
    return standardize(df, "fred")


def fetch(series_id, session, start="2006-01-01"):
    r = session.get(URL, params={"id": series_id}, timeout=30)
    r.raise_for_status()
    save_raw(r.text, "fred", series_id, "csv")
    df = parse_fred_csv(r.text, series_id)
    return df[df["date"] >= start].reset_index(drop=True)

macro_extract/statcan.py

"""Statistics Canada WDS: vector endpoint."""
import json

import pandas as pd

from .rawstore import save_raw
from .schema import standardize

WDS = "https://www150.statcan.gc.ca/t1/wds/rest"


def parse_vectors(payload):
    rows = []
    for item in payload:
        if item.get("status") != "SUCCESS":
            continue
        obj = item["object"]
        for dp in obj.get("vectorDataPoint", []):
            rows.append({"date": dp.get("refPer"),
                         "series_id": f"v{obj['vectorId']}",
                         "value": dp.get("value")})
    return standardize(pd.DataFrame(rows, columns=["date", "series_id", "value"]),
                       "statcan")


def fetch(series_id, session, latest_n=250):
    """series_id like 'v41690973'."""
    vector = int(series_id.lstrip("v"))
    r = session.post(f"{WDS}/getDataFromVectorsAndLatestNPeriods",
                     json=[{"vectorId": vector, "latestN": latest_n}], timeout=30)
    r.raise_for_status()
    payload = r.json()
    save_raw(json.dumps(payload), "statcan", series_id, "json")
    if not payload or payload[0].get("status") != "SUCCESS":
        raise ValueError(f"StatCan {series_id}: request not successful")
    return parse_vectors(payload)

macro_extract/catalog.py — what to fetch, separate from how

"""Series catalogue: add a row here to add a series to the pipeline."""

CATALOG = [
    # source     series_id               friendly name
    ("boc",     "V39079",               "overnight_rate"),
    ("boc",     "FXCADUSD",             "cad_usd"),
    ("boc",     "BD.CDN.2YR.DQ.YLD",    "yield_2yr"),
    ("boc",     "BD.CDN.10YR.DQ.YLD",   "yield_10yr"),
    ("fred",    "FEDFUNDS",             "fed_funds"),
    ("fred",    "DCOILWTICO",           "wti"),
    ("fred",    "CANLOLITOAASTSAM",     "oecd_cli"),
    ("statcan", "v41690973",            "cpi_all_items"),   # verify (10.1)
    ("statcan", "v2062815",             "unemployment_rate"),  # verify (10.1)
]

Separating configuration (which series) from code (how to fetch) is a core data-engineering idea. dlt sources (Days 21–30) and dbt sources.yml (Days 31+) are built on it.

run_extract.py (repository root) — one command for everything

"""Run all extractions in the catalogue and write one tidy staging file.

Usage:  python run_extract.py
"""
import logging
import os
import sys
import time

import pandas as pd

from macro_extract import boc, fred, statcan
from macro_extract.catalog import CATALOG
from macro_extract.net import make_session
from macro_extract.schema import validate

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s | %(levelname)-7s | %(message)s")
log = logging.getLogger("extract")

FETCHERS = {"boc": boc.fetch, "fred": fred.fetch, "statcan": statcan.fetch}


def main():
    session = make_session()
    frames, failures = [], []

    for source, series_id, name in CATALOG:
        try:
            df = FETCHERS[source](series_id, session)
            validate(df, f"{source}:{series_id}")
            df["name"] = name
            frames.append(df)
            log.info(f"✓ {name:<18} {len(df):>6} rows, latest "
                     f"{df['value'].iloc[-1]:.3f} ({df['date'].iloc[-1]:%Y-%m-%d})")
        except Exception as e:
            log.error(f"✗ {name:<18} {source}:{series_id} → {e}")
            failures.append(name)
        time.sleep(0.3)          # be polite

    if not frames:
        log.error("Every extraction failed — nothing written.")
        sys.exit(1)

    out = pd.concat(frames, ignore_index=True)
    out["fetched_at"] = pd.Timestamp.now(tz="UTC")
    os.makedirs("data/staging", exist_ok=True)
    out.to_parquet("data/staging/observations.parquet", index=False)

    log.info(f"Wrote {len(out):,} rows for {out['name'].nunique()} series "
             f"→ data/staging/observations.parquet")
    if failures:
        log.warning(f"{len(failures)} series failed: {failures}")
        sys.exit(2)              # partial success: a scheduler can detect this


if __name__ == "__main__":
    main()

sys.exit(1) and sys.exit(2) return exit codes: 0 means success, anything else means a problem. Schedulers such as Prefect (Days 51–60) use them to decide whether a run succeeded.

10.5 Testing Parsers with pytest (18 min)

Because parsing is separate from fetching, we can test parsers with small hand-made inputs, with no internet needed. That makes tests fast and reliable.

pip install pytest
pip freeze > requirements.txt

Create tests/test_parsers.py:

import pandas as pd
import pytest

from macro_extract.boc import parse_valet
from macro_extract.fred import parse_fred_csv
from macro_extract.schema import COLUMNS, validate
from macro_extract.statcan import parse_vectors


def test_valet_skips_bad_values():
    payload = {
        "seriesDetail": {"V39079": {"label": "Overnight"}},
        "observations": [
            {"d": "2026-01-05", "V39079": {"v": "2.25"}},
            {"d": "2026-01-06", "V39079": {"v": "Bank holiday"}},
            {"d": "2026-01-07"},                              # key missing
        ],
    }
    df = parse_valet(payload)
    assert list(df.columns) == COLUMNS
    assert len(df) == 1
    assert df["value"].iloc[0] == 2.25


def test_fred_handles_either_header_and_blanks():
    for header in ["DATE", "observation_date"]:
        text = f"{header},FEDFUNDS\n2026-01-01,3.64\n2026-02-01,.\n2026-03-01,3.63\n"
        df = parse_fred_csv(text, "FEDFUNDS")
        assert len(df) == 2
        assert (df["series_id"] == "FEDFUNDS").all()


def test_statcan_ignores_failed_items():
    payload = [
        {"status": "SUCCESS", "object": {"vectorId": 41690973, "vectorDataPoint": [
            {"refPer": "2026-06-01", "value": 168.2},
            {"refPer": "2026-07-01", "value": 169.0}]}},
        {"status": "FAILED", "object": "Vector not found"},
    ]
    df = parse_vectors(payload)
    assert set(df["series_id"]) == {"v41690973"}
    assert len(df) == 2


def test_validate_rejects_duplicates():
    df = pd.DataFrame({"date": pd.to_datetime(["2026-01-01"] * 2),
                       "series_id": ["X", "X"], "value": [1.0, 1.0],
                       "source": ["test", "test"]})
    with pytest.raises(ValueError, match="duplicate"):
        validate(df, "test")

Run the tests from the repository root:

python -m pytest -v

You should see four green PASSED lines. Using python -m pytest rather than plain pytest makes sure Python can find your macro_extract package.

Break it on purpose: change parse_valet so it no longer skips "Bank holiday", rerun the tests, and read the failure message. Tests are your safety net whenever you change code. Great Expectations (Days 41–50) does the same job for the data itself.

Try it 10.5

Write a test proving that parse_valet handles a payload with two series in one observation (as in 8.4, Example 2) and returns two rows per date.

def test_valet_multiple_series():
    payload = {
        "seriesDetail": {"A": {}, "B": {}},
        "observations": [
            {"d": "2026-01-05", "A": {"v": "2.27"}, "B": {"v": "3.75"}},
            {"d": "2026-01-06", "A": {"v": "2.30"}, "B": {"v": "3.80"}},
        ],
    }
    df = parse_valet(payload)
    assert len(df) == 4
    assert df.groupby("date").size().eq(2).all()

10.6 Run It End-to-End, Recap, and Look Ahead (10 min)

python -m pytest -q
python run_extract.py

Then inspect the result:

import pandas as pd

obs = pd.read_parquet("data/staging/observations.parquet")
print(obs.groupby(["source", "name"])["date"].agg(["min", "max", "count"]))

wide = (obs.assign(date=obs["date"].dt.to_period("M").dt.to_timestamp())
           .pivot_table(index="date", columns="name", values="value", aggfunc="last"))
print(wide.tail())

Stretch goals

• Connect to Day 7: rewrite panel.py to read data/staging/observations.parquet instead of the get_course_data.py files, then delete get_course_data.py.
• Add a source type: support ("statcan_table", "18100004", ...) in the catalogue using download_table and read_table from 10.3.
• Incremental loading: read the existing staging file first and only request dates after its latest date. This is the core idea of dlt incremental loading (Days 24–27).

What you built, mapped to the course notebook

Looking ahead: Days 11–20

Your pipeline now writes one tidy, long Parquet file. On Day 11 you'll query it directly with DuckDB, an analytical SQL database that reads Parquet without any import step:

SELECT name, COUNT(*) AS n, MAX(date) AS latest
FROM 'data/staging/observations.parquet'
GROUP BY name;

The long (date, series_id, value, source) schema you designed today is exactly what makes that SQL simple.

Day 10 checklist

• I understand StatCan tables, vectors and coordinates, and I verify vector IDs
• I can POST JSON requests to the WDS vector endpoints
• I can stream, cache and selectively read large table downloads
• My package separates network, parsing, schema, storage and configuration
• My parsers are covered by offline pytest tests
• One command (python run_extract.py) extracts every series and reports failures

git add .
git status            # confirm data/ and .env are not staged
git commit -m "Day 10: StatCan WDS, macro_extract package, tests"
git push