Robust Extraction: FRED, API Keys, Errors and Retries

9.1 FRED's CSV Endpoint (23 min)

The notebook uses FRED's graph CSV endpoint. It needs no key and returns plain CSV text. Create day09/fred_csv.py.

Example 1 — Look before you parse

import requests

url = "https://fred.stlouisfed.org/graph/fredgraph.csv"
r = requests.get(url, params={"id": "FEDFUNDS"}, timeout=30)
r.raise_for_status()

print(r.headers.get("Content-Type"))
print(r.text[:200])

Example 2 — CSV text → DataFrame without saving a file

io.StringIO wraps text so pandas can read it as if it were a file:

import io
import pandas as pd


def parse_fred_csv(text, series_id):
    """Parse FRED graph CSV text into a long DataFrame."""
    df = pd.read_csv(io.StringIO(text))
    df = df.iloc[:, :2]                       # header names vary: DATE / observation_date
    df.columns = ["date", "value"]
    df["date"] = pd.to_datetime(df["date"], errors="coerce")
    df["value"] = pd.to_numeric(df["value"], errors="coerce")   # "." or "" → NaN
    df["series_id"] = series_id
    return df.dropna(subset=["date", "value"])[["date", "series_id", "value"]]


fed = parse_fred_csv(r.text, "FEDFUNDS")
print(fed.tail())

The output has the same three columns as parse_valet. One schema for every source is the key design idea of Day 10.

Example 3 — Several series

frames = []
for sid in ["FEDFUNDS", "DCOILWTICO", "CANLOLITOAASTSAM"]:
    r = requests.get(url, params={"id": sid}, timeout=30)
    r.raise_for_status()
    frames.append(parse_fred_csv(r.text, sid))

fred_long = pd.concat(frames, ignore_index=True)
print(fred_long.groupby("series_id")["date"].agg(["min", "max", "count"]))

The graph CSV endpoint is convenient but not a formally documented API, and it can change without notice. For production work, the official API (next block) is safer.

9.2 The Official FRED API and Keeping Secrets Safe (23 min)

The official FRED API returns JSON with metadata, supports date filters and is documented. It requires a free API key. Register at fred.stlouisfed.org, then request a key under My Account → API Keys.

Rule number one: never put a key in your code

Code is committed to GitHub, and anything committed is effectively public forever. Keys belong in environment variables.

Option A — Codespaces secrets (recommended)

1. On GitHub: Settings → Codespaces → Secrets → New secret.
2. Name: FRED_API_KEY. Value: your key. Repository access: macro-data-course.
3. Restart your Codespace. The secret now exists as an environment variable.

echo ${FRED_API_KEY:0:4}...      # prints only the first 4 characters

Option B — A local .env file

pip install python-dotenv
echo 'FRED_API_KEY=paste_your_key_here' > .env
echo '.env' >> .gitignore            # critical: never commit .env
pip freeze > requirements.txt

Reading the key in Python

import os
from dotenv import load_dotenv    # only needed for Option B

load_dotenv()                      # harmless if there is no .env file
key = os.environ.get("FRED_API_KEY")
if not key:
    raise RuntimeError("FRED_API_KEY is not set. See Day 9, section 9.2.")

Using the official endpoint

Create day09/fred_api.py:

import os
import pandas as pd
import requests
from dotenv import load_dotenv

load_dotenv()
KEY = os.environ["FRED_API_KEY"]

r = requests.get(
    "https://api.stlouisfed.org/fred/series/observations",
    params={"series_id": "FEDFUNDS", "api_key": KEY, "file_type": "json",
            "observation_start": "2020-01-01"},
    timeout=30)
r.raise_for_status()
payload = r.json()

print(list(payload.keys()))
obs = pd.DataFrame(payload["observations"])[["date", "value"]]
obs["value"] = pd.to_numeric(obs["value"], errors="coerce")   # FRED uses "." for missing
print(obs.tail())

Careful with logs: print(r.url) would show your key. Never print or save URLs that contain secrets.

This is a first taste of secrets management, which becomes central with dlt on Days 21–23. If you prefer not to register, keep using the CSV endpoint. The rest of the course works either way.

9.3 What Goes Wrong: Errors and Exceptions (30 min)

Pipelines fail at 3 a.m. for boring reasons. Let's cause each failure on purpose and learn to recognise it. Create day09/failures.py.

Failure 1 — Wrong series ID (HTTP error)

import requests

r = requests.get("https://fred.stlouisfed.org/graph/fredgraph.csv",
                 params={"id": "NOT_A_REAL_SERIES"}, timeout=30)
print(r.status_code)
try:
    r.raise_for_status()
except requests.HTTPError as e:
    print("HTTPError:", e)

Failure 2 — The server is too slow (timeout)

try:
    requests.get("https://www.bankofcanada.ca/valet/lists/series/json",
                 timeout=0.001)          # deliberately impossible
except requests.Timeout as e:
    print("Timeout:", type(e).__name__)

Failure 3 — The host doesn't exist (connection error)

try:
    requests.get("https://www.bankofcanada-typo.ca/valet", timeout=10)
except requests.ConnectionError as e:
    print("ConnectionError:", type(e).__name__)

Failure 4 — The response isn't what you expected

r = requests.get("https://www.bankofcanada.ca/valet/observations/V39079/csv",
                 params={"recent": 2}, timeout=20)
try:
    r.json()                 # we asked for CSV, then tried to read JSON
except ValueError as e:
    print("Not JSON:", type(e).__name__)

The requests exception family

requests.RequestException          ← catches all of the below
├── requests.ConnectionError
├── requests.Timeout
└── requests.HTTPError             ← raised by raise_for_status()

Handling them deliberately

def safe_get(url, **kwargs):
    """GET with clear, specific error messages."""
    try:
        r = requests.get(url, timeout=30, **kwargs)
        r.raise_for_status()
        return r
    except requests.Timeout:
        print(f"✗ Timed out: {url}")
    except requests.ConnectionError:
        print(f"✗ Could not connect: {url}")
    except requests.HTTPError as e:
        print(f"✗ HTTP {e.response.status_code}: {url}")
    return None

**kwargs passes any extra keyword arguments (such as params=...) straight through to requests.get.

Design question: should one failed series stop the whole pipeline? The notebook says no: each fetch is wrapped in try/except and the loop continues. That's usually right for dashboards. For a regulatory report you might prefer to stop. Decide consciously.

9.4 Retries and Backoff (23 min)

Temporary failures (timeouts, 5xx, 429) often succeed a few seconds later. Exponential backoff means waiting longer after each failure (1 s, 2 s, 4 s…) so you don't hammer a struggling server.

Version 1 — Hand-written, so you see the logic

Create day09/retries.py:

import time
import requests

RETRYABLE = {429, 500, 502, 503, 504}


def get_with_retries(url, params=None, max_attempts=4, base_wait=1.0):
    for attempt in range(1, max_attempts + 1):
        try:
            r = requests.get(url, params=params, timeout=30)
            if r.status_code in RETRYABLE:
                raise requests.HTTPError(f"retryable status {r.status_code}")
            r.raise_for_status()          # 4xx other than 429: fail immediately
            return r
        except (requests.Timeout, requests.ConnectionError, requests.HTTPError) as e:
            is_retryable = not isinstance(e, requests.HTTPError) or "retryable" in str(e)
            if attempt == max_attempts or not is_retryable:
                raise
            wait = base_wait * 2 ** (attempt - 1)
            print(f"  attempt {attempt} failed ({e}); retrying in {wait:.0f}s")
            time.sleep(wait)


r = get_with_retries("https://www.bankofcanada.ca/valet/observations/V39079/json",
                     params={"recent": 2})
print(r.status_code)

Version 2 — The professional way: a Session with automatic retries

requests can retry for you through urllib3, which is already installed as a dependency:

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,                       # 1s, 2s, 4s, ...
        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


session = make_session()
r = session.get("https://www.bankofcanada.ca/valet/observations/V39079/json",
                params={"recent": 2}, timeout=30)
print(r.json()["observations"])

A Session also reuses connections, which makes many calls to the same host faster. The User-Agent header politely identifies who is calling.

Being a good API citizen

• Fetch only what you need (start_date, recent).
• Add a short pause between many calls: time.sleep(0.5).
• Cache results (9.6) so you don't download the same data twice a day.

Orchestrators like Prefect (Days 54–57) add retries at the task level on top of this.

9.5 Validating Responses and Logging (18 min)

A successful request can still contain bad data

A 200 OK response might be empty, be missing columns, or have duplicated dates. Validate before passing data on. Create day09/validation.py:

import pandas as pd

REQUIRED = ["date", "series_id", "value"]


def validate(df, source, min_rows=1):
    """Raise ValueError if a long DataFrame breaks basic rules."""
    missing = [c for c in REQUIRED if c not in df.columns]
    if missing:
        raise ValueError(f"{source}: missing columns {missing}")
    if len(df) < min_rows:
        raise ValueError(f"{source}: only {len(df)} rows (expected ≥ {min_rows})")
    dupes = df.duplicated(["series_id", "date"]).sum()
    if dupes:
        raise ValueError(f"{source}: {dupes} duplicate (series_id, date) rows")
    if df["value"].isna().any():
        raise ValueError(f"{source}: missing values after parsing")
    return df

Compare with Day 7's check_panel: same idea, applied at the extraction stage.

Logging instead of print

print is fine for exercises. Pipelines use logging, which adds timestamps and severity levels and can write to files:

import logging

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

log.info("Fetching FEDFUNDS")
log.warning("DCOILWTICO: 12 missing values dropped")
log.error("NOT_A_REAL_SERIES: HTTP 404")

9.6 The Raw Landing Zone + Mini-Project (20 min)

Professional pipelines save the untouched response before parsing it. If your parser has a bug, you fix it and re-parse without re-downloading. You also keep a record of exactly what the agency published on a given day, which matters because macro data gets revised.

data/raw/fred/FEDFUNDS/2026-09-29T14-05-12.csv
data/raw/boc/V39079/2026-09-29T14-05-10.json

import os
from datetime import datetime, timezone


def save_raw(content, source, series_id, extension):
    """Save a raw API response under data/raw/<source>/<series>/<timestamp>."""
    stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H-%M-%S")
    folder = os.path.join("data", "raw", 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

Mini-project

Create day09/fred_robust.py with fetch_fred(series_id, session, start="2006-01-01") that:

1. Uses a retrying session (9.4).
2. Saves the raw CSV text with save_raw.
3. Parses with parse_fred_csv and filters to date >= start.
4. Validates with validate.
5. Logs the row count and latest value.

Then loop over ["FEDFUNDS", "DCOILWTICO", "NOT_A_REAL_SERIES", "A191RL1Q225SBEA"], logging failures without stopping, and print a final summary of successes and failures.

import logging
import pandas as pd

from fred_csv import parse_fred_csv      # move its demo code under __main__
from retries import make_session         # (same for retries.py)
from validation import validate
# save_raw: paste the function from 9.6 here or into its own file

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

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


def fetch_fred(series_id, session, start="2006-01-01"):
    r = session.get(URL, params={"id": series_id}, timeout=30)
    r.raise_for_status()
    path = save_raw(r.text, "fred", series_id, "csv")
    df = parse_fred_csv(r.text, series_id)
    df = df[df["date"] >= start]
    validate(df, f"fred:{series_id}")
    last = df.iloc[-1]
    log.info(f"{series_id}: {len(df)} rows, latest {last['value']} "
             f"({last['date']:%Y-%m-%d}) · raw → {path}")
    return df


session = make_session()
ok, failed, frames = [], [], []
for sid in ["FEDFUNDS", "DCOILWTICO", "NOT_A_REAL_SERIES", "A191RL1Q225SBEA"]:
    try:
        frames.append(fetch_fred(sid, session))
        ok.append(sid)
    except Exception as e:
        log.error(f"{sid}: {e}")
        failed.append(sid)

fred_long = pd.concat(frames, ignore_index=True)
log.info(f"Done: {len(ok)} succeeded, {len(failed)} failed {failed}")

Day 9 checklist

• I can parse CSV responses with io.StringIO
• My API keys live in environment variables, never in code or Git
• I can tell timeouts, connection errors and HTTP errors apart and handle each
• I retry only what's worth retrying, with exponential backoff
• I validate parsed data, log progress, and keep raw responses

git add .
git status            # make sure .env is NOT listed!
git commit -m "Day 9: FRED, secrets, errors, retries, validation"
git push