"""Build a tidy, point-in-time table from FRED-MD monthly vintages (McCracken-Ng).
FRED-MD ships one CSV per monthly **vintage**: row 1 is the header (``sasdate`` + ~128 series),
row 2 is the ``Transform:`` row of stationarity codes (tcodes), and the body is monthly levels with
``sasdate`` (the reference period) as ``M/1/YYYY``.
This builder stacks vintages into one wide table ``[ref_date, vintage, <series...>]`` — the
**vintage axis becomes a column** (Croushore-Stark's term; the same word as numeraire's
``data_vintage``) so revisions are first-class. Naming: ``ref_date`` = the reference period
(FRED-MD's ``sasdate``, anchored to month-end); ``vintage`` = the snapshot identity (the file
month, as a month-end date).
Two rules that keep downstream point-in-time use correct:
* **tcode transforms are applied per vintage, here at build time** — never across vintages and never
in numeraire's main pipeline. ``transform=False`` keeps raw levels.
* **The availability lag ``L`` is NOT encoded here.** It stays a read-time parameter in numeraire
(``asof`` uses ``vintage + L``), so it can be swept. This table stores only what was observed.
Different series carry different publication lags, so each vintage's latest reference periods are
missing for late-released series — the **ragged edge** (Wallis 1986). It is preserved as NaN.
Sources. Recent single vintages download from the St. Louis Fed site as ``YYYY-MM-md.csv``; the full
history lives in two zip archives (1999-2015 and 2015-01..2024-12). Filenames across these use three
conventions — ``YYYY-MM.csv``, ``FRED-MD_YYYYmMM.csv``, ``YYYY-MM-md.csv`` — all handled here.
Downloads go through ``curl`` (the CDN's Akamai TLS fingerprinting rejects Python's urllib).
"""
from __future__ import annotations
import contextlib
import datetime as dt
import hashlib
import json
import re
import subprocess
import tempfile
import zipfile
from collections.abc import Sequence
from importlib.resources import files
from pathlib import Path
import numpy as np
import pandas as pd
from numeraire_dataset.paths import data_home
# Recent single vintages: new site (``-md.csv``, 2025-04+) with the legacy S3 path as a fallback.
_FREDMD_NEW = (
"https://www.stlouisfed.org/-/media/project/frbstl/stlouisfed/research/fred-md/monthly"
)
_FREDMD_S3 = "https://files.stlouisfed.org/files/htdocs/fred-md/monthly"
# Full-history zip archives. The Sitecore media path resolves without the ``?sc_lang=&hash=``
# query string (the hash is a rotating cache-buster), so we use the clean URL.
_ARCHIVE_BASE = "https://www.stlouisfed.org/-/media/project/frbstl/stlouisfed/research/fred-md"
HIST_ARCHIVES = {
"1999-2015": f"{_ARCHIVE_BASE}/historical_fred-md.zip",
"2015-2024": f"{_ARCHIVE_BASE}/historical-vintages-of-fred-md-2015-01-to-2024-12.zip",
}
REF_COL = "ref_date"
VINTAGE_COL = "vintage"
_DATE_COL = "sasdate"
_TRANSFORM_LABEL = "Transform:"
_PAT_DASH = re.compile(r"(\d{4})-(\d{2})") # YYYY-MM.csv / YYYY-MM-md.csv
_PAT_M = re.compile(r"(\d{4})m(\d{1,2})") # FRED-MD_YYYYmMM.csv
# A tiny slice that exercises the whole point-in-time machinery rather than covering the economy:
# INDPRO (output, tcode 5 Δlog, heavily revised), UNRATE (labor, tcode 2 Δ, lightly revised),
# T10YFFM (10y-FedFunds spread, tcode 1 level, never revised). Three tcodes, three revision
# regimes (heavy/light/none), all US-gov public sources with names stable across every vintage.
REPRESENTATIVE_SERIES = ["INDPRO", "UNRATE", "T10YFFM"]
def _curl(url: str, dest: Path, timeout: int) -> None:
"""Download ``url`` to ``dest`` via curl (Akamai-safe); raise on any failure/empty file."""
r = subprocess.run(
["curl", "-sSL", "--fail", "--max-time", str(timeout), "-o", str(dest), url],
capture_output=True,
text=True,
)
if r.returncode != 0 or not dest.exists() or dest.stat().st_size == 0:
raise OSError(f"curl failed for {url}: rc={r.returncode} {r.stderr.strip()[:200]}")
def _is_fredmd_csv(path: Path) -> bool:
"""True if the first line is a FRED-MD header (guards against 200-OK HTML/error pages)."""
with path.open("r", encoding="utf-8", errors="replace") as fh:
return fh.readline().lstrip("").strip().lower().startswith(_DATE_COL)
[docs]
def download(vintages: list[str], dest: str | Path, *, timeout: int = 120) -> list[Path]:
"""Download recent FRED-MD single vintages (``YYYY-MM``) into ``dest``; return local paths.
Tries three names in order — new site ``YYYY-MM-md.csv`` (2025-04+), new site ``YYYY-MM.csv``
(2025-01..03), then legacy S3 ``YYYY-MM.csv``; earlier vintages live in the zip archives (see
:func:`download_archive`). A 200-OK response that isn't a FRED-MD CSV (e.g. an HTML placeholder
for an unpublished month) is treated as a failure. Public data (cite McCracken & Ng 2016).
"""
out_dir = Path(dest).expanduser()
out_dir.mkdir(parents=True, exist_ok=True)
paths: list[Path] = []
for v in vintages:
out = out_dir / f"{v}.csv"
errors: list[str] = []
for url in (f"{_FREDMD_NEW}/{v}-md.csv", f"{_FREDMD_NEW}/{v}.csv", f"{_FREDMD_S3}/{v}.csv"):
try:
_curl(url, out, timeout)
if not _is_fredmd_csv(out):
raise OSError(f"{url}: 200 OK but not a FRED-MD CSV")
break
except OSError as e:
errors.append(str(e))
else:
raise OSError(f"all candidate URLs failed for {v}: {'; '.join(errors)}")
paths.append(out)
return paths
[docs]
def download_archive(which: str, dest: str | Path, *, timeout: int = 600) -> Path:
"""Download a full-history zip (``which`` in :data:`HIST_ARCHIVES`, or a URL) into ``dest``."""
url = HIST_ARCHIVES.get(which, which)
out = Path(dest).expanduser()
if out.is_dir():
out = out / (which if which in HIST_ARCHIVES else "fred-md-archive") / "archive.zip"
out.parent.mkdir(parents=True, exist_ok=True)
_curl(url, out, timeout)
return out
[docs]
def apply_tcode(x: np.ndarray, code: int) -> np.ndarray:
"""Apply a FRED-MD stationarity transform (codes 1-7) to a level series (NaN-padded at start).
1 level, 2 diff, 3 diff^2, 4 log, 5 dlog, 6 dlog^2, 7 diff(x_t/x_{t-1} - 1). All are causal
(use only past values), so applying them per vintage is point-in-time safe.
"""
x = np.asarray(x, dtype=np.float64)
if code == 1:
return x
if code == 2:
return np.append([np.nan], np.diff(x))
if code == 3:
return np.append([np.nan, np.nan], np.diff(x, n=2))
if code == 4:
return np.log(x)
if code == 5:
return np.append([np.nan], np.diff(np.log(x)))
if code == 6:
return np.append([np.nan, np.nan], np.diff(np.log(x), n=2))
if code == 7:
g = np.append([np.nan], x[1:] / x[:-1] - 1.0)
return np.append([np.nan], np.diff(g))
raise ValueError(f"unknown FRED-MD tcode {code!r} (expected 1-7)")
[docs]
def read_vintage(path: str | Path) -> tuple[pd.DataFrame, pd.Series]:
"""Read one FRED-MD vintage file → ``(levels, tcodes)``.
``levels`` is ``(ref_date x series)`` of raw monthly levels (``ref_date`` = ``sasdate`` at
month-end); ``tcodes`` maps each series to its transform code from the ``Transform:`` row.
"""
raw = pd.read_csv(path)
if raw.columns[0] != _DATE_COL:
raise ValueError(f"{path}: expected first column {_DATE_COL!r}, got {raw.columns[0]!r}")
series_cols = list(raw.columns[1:])
tcode_row = raw.iloc[0]
if str(tcode_row[_DATE_COL]).strip() != _TRANSFORM_LABEL:
raise ValueError(f"{path}: row 2 is not the {_TRANSFORM_LABEL!r} tcode row")
tcodes = pd.Series(
{c: int(float(tcode_row[c])) for c in series_cols}, name="tcode", dtype="int64"
)
body = raw.iloc[1:].copy()
ref = pd.to_datetime(body[_DATE_COL], format="%m/%d/%Y", errors="coerce")
body = body[ref.notna()]
ref = ref[ref.notna()].dt.to_period("M").dt.to_timestamp("M")
levels = body[series_cols].apply(pd.to_numeric, errors="coerce")
levels.index = pd.DatetimeIndex(ref, name=REF_COL)
return levels, tcodes
def _vintage_label(name: str) -> str | None:
"""Extract the ``YYYY-MM`` vintage from a filename stem, or None if it carries no date."""
m = _PAT_DASH.search(name) or _PAT_M.search(name)
return f"{int(m.group(1)):04d}-{int(m.group(2)):02d}" if m else None
def _vintage_month_end(label: str) -> pd.Timestamp:
"""``"2025-02"`` → that month's end (the vintage's identity as a date)."""
return pd.Period(label, freq="M").to_timestamp("M")
[docs]
def build_table(
paths: Sequence[str | Path],
*,
transform: bool = True,
series: list[str] | None = None,
) -> pd.DataFrame:
"""Stack FRED-MD vintage files into a wide point-in-time table ``[ref_date, vintage, series]``.
Each file's name must carry its vintage (``YYYY-MM``, ``FRED-MD_YYYYmMM``, or ``YYYY-MM-md``).
``transform`` applies each series' tcode **per vintage** (causal, PIT-safe); ``transform=False``
keeps raw levels. ``series`` subsets/orders the columns (missing ones become NaN). Rows are the
ragged edge as-is.
"""
frames: list[pd.DataFrame] = []
for p in paths:
p = Path(p)
label = _vintage_label(p.stem)
if label is None:
raise ValueError(f"no vintage YYYY-MM found in filename {p.name!r}")
levels, tcodes = read_vintage(p)
if series is not None:
levels = levels.reindex(columns=series)
tcodes = tcodes.reindex(series)
if transform:
cols = {}
for c in levels.columns:
tc = tcodes.get(c)
arr = levels[c].to_numpy()
cols[c] = apply_tcode(arr, int(tc)) if pd.notna(tc) else arr
levels = pd.DataFrame(cols, index=levels.index)
frame = levels.reset_index()
frame.insert(1, VINTAGE_COL, _vintage_month_end(label))
frames.append(frame)
table = pd.concat(frames, ignore_index=True)
return table.sort_values([VINTAGE_COL, REF_COL]).reset_index(drop=True)
[docs]
def build_from_dir(
directory: str | Path, *, transform: bool = True, series: list[str] | None = None
) -> pd.DataFrame:
"""Build from every vintage CSV under ``directory`` (recurses; non-vintage files skipped)."""
paths = [
p for p in sorted(Path(directory).glob("**/*.csv")) if _vintage_label(p.stem) is not None
]
if not paths:
raise ValueError(f"no vintage CSVs found under {directory}")
return build_table(paths, transform=transform, series=series)
def build_from_zip(
zip_path: str | Path, *, transform: bool = True, series: list[str] | None = None
) -> pd.DataFrame:
"""Extract a FRED-MD history zip to a temp dir and build the table from its vintage CSVs."""
with tempfile.TemporaryDirectory() as td:
with zipfile.ZipFile(zip_path) as zf:
zf.extractall(td)
return build_from_dir(td, transform=transform, series=series)
def tcode_sidecar(path: str | Path, *, series: list[str] | None = None) -> pd.DataFrame:
"""``[series, tcode]`` metadata from one vintage (tcodes are stable across vintages)."""
_, tcodes = read_vintage(path)
if series is not None:
tcodes = tcodes.reindex(series)
return tcodes.rename_axis("series").reset_index()
def build_fredmd(
sources: Sequence[str | Path],
*,
series: list[str] | None = None,
transform: bool = True,
out: str | Path | None = None,
) -> pd.DataFrame:
"""Assemble one FRED-MD point-in-time table from mixed sources → ``[ref_date, vintage, …]``.
Each source is a zip archive, a directory of vintage CSVs, or a single vintage CSV. Vintages
are deduplicated (same ``(ref_date, vintage)`` kept once) so overlapping archives + recent
single downloads stitch cleanly. ``series`` subsets columns (see :data:`REPRESENTATIVE_SERIES`).
If ``out`` is given, the table is written there (``.parquet`` suffix → parquet, else CSV).
"""
frames: list[pd.DataFrame] = []
for raw in sources:
src = Path(raw)
if src.suffix.lower() == ".zip":
frames.append(build_from_zip(src, transform=transform, series=series))
elif src.is_dir():
frames.append(build_from_dir(src, transform=transform, series=series))
else:
frames.append(build_table([src], transform=transform, series=series))
table = (
pd.concat(frames, ignore_index=True)
.drop_duplicates([REF_COL, VINTAGE_COL])
.sort_values([VINTAGE_COL, REF_COL])
.reset_index(drop=True)
)
if out is not None:
dest = Path(out).expanduser()
dest.parent.mkdir(parents=True, exist_ok=True)
if dest.suffix == ".parquet":
table.to_parquet(dest, index=False)
else:
table.to_csv(dest, index=False)
return table
def load_representative_sample() -> pd.DataFrame:
"""The tiny bundled FRED-MD point-in-time sample (:data:`REPRESENTATIVE_SERIES`, ~5y vintages).
Ships with this package so examples/tests run offline; raw levels, ``[ref_date, vintage, …]``.
"""
path = files("numeraire_dataset").joinpath("_samples/fredmd_representative.csv.gz")
with path.open("rb") as f:
return pd.read_csv(f, compression="gzip", parse_dates=[REF_COL, VINTAGE_COL])
def _cache_path(series: list[str], *, transform: bool, cache_dir: str | Path | None) -> Path:
"""Deterministic cache file for this request (keyed by series set + transform)."""
key = hashlib.sha1((",".join(series) + f"|t={transform}").encode()).hexdigest()[:10]
name = f"fredmd_{len(series)}s_{'tx' if transform else 'lvl'}_{key}.parquet"
return data_home(cache_dir) / "fredmd" / name
def _recent_vintages(start: tuple[int, int] = (2025, 1)) -> list[str]:
"""``YYYY-MM`` labels from ``start`` to the current month (post-archive, downloaded singly)."""
today = dt.date.today()
y, m = start
out: list[str] = []
while (y, m) <= (today.year, today.month):
out.append(f"{y:04d}-{m:02d}")
y, m = (y + 1, 1) if m == 12 else (y, m + 1)
return out
def _fetch_all_sources(dest: Path) -> list[Path]:
"""Download the two history archives + each published post-2024 vintage into ``dest``."""
got: list[Path] = [download_archive(k, dest) for k in HIST_ARCHIVES]
for ym in _recent_vintages():
# A not-yet-published (or unavailable) month is skipped, not fatal.
with contextlib.suppress(OSError):
got += download([ym], dest)
return got
def build(
*,
series: Sequence[str] | None = None,
transform: bool = False,
sources: Sequence[str | Path] | None = None,
cache_dir: str | Path | None = None,
) -> Path:
"""Download the full FRED-MD history, assemble the point-in-time table, and cache it as parquet.
No arguments needed: ``build()`` fetches the archives + recent vintages itself. ``series``
defaults to :data:`REPRESENTATIVE_SERIES` (``None``); pass a list for others. ``sources`` (zip /
dir / CSV paths) skips downloading — for offline/local builds. Returns the cache path; also
writes a ``.meta.json`` provenance sidecar.
"""
resolved = list(REPRESENTATIVE_SERIES) if series is None else list(series)
dest = _cache_path(resolved, transform=transform, cache_dir=cache_dir)
dest.parent.mkdir(parents=True, exist_ok=True)
if sources is None:
with tempfile.TemporaryDirectory() as td:
got = _fetch_all_sources(Path(td))
if not got:
raise OSError(
"could not download any FRED-MD source; check network or pass sources="
)
table = build_fredmd(got, series=resolved, transform=transform)
else:
table = build_fredmd(sources, series=resolved, transform=transform)
table.to_parquet(dest, index=False)
meta = {
"dataset": "fredmd",
"series": resolved,
"transform": transform,
"n_vintages": int(table[VINTAGE_COL].nunique()),
"vintage_min": str(table[VINTAGE_COL].min().date()),
"vintage_max": str(table[VINTAGE_COL].max().date()),
"built_utc": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"),
"source": "FRED-MD (McCracken & Ng 2016), Federal Reserve Bank of St. Louis — public",
}
dest.with_suffix(".meta.json").write_text(json.dumps(meta, indent=2))
return dest
def load(
*,
series: Sequence[str] | None = None,
transform: bool = False,
sources: Sequence[str | Path] | None = None,
refresh: bool = False,
cache_dir: str | Path | None = None,
) -> pd.DataFrame:
"""Load the cached FRED-MD point-in-time table, **auto-building on first use** if absent.
One call: ``load()`` returns :data:`REPRESENTATIVE_SERIES` over the full history, downloading
and caching it the first time and reading the cache thereafter. ``refresh=True`` forces a
rebuild; ``series`` / ``cache_dir`` mirror :func:`build`.
"""
resolved = list(REPRESENTATIVE_SERIES) if series is None else list(series)
dest = _cache_path(resolved, transform=transform, cache_dir=cache_dir)
if refresh or not dest.exists():
build(series=resolved, transform=transform, sources=sources, cache_dir=cache_dir)
return pd.read_parquet(dest)