nb7.2 — The Multi-Source Data-Pull Harness

Companion to Ch 7.2, “Data Acquisition in Practice.”

A question is not a dataset. Between “I want HMDA data” and “I have a file on disk I can defend to a referee” lies the craft of reproducible acquisition — and this notebook is the runnable spine of that craft. We build one small harness that every source in the chapter plugs into:

  1. A cache-or-fetch helper: look in a local raw-cache (Parquet) first, hit the network only on a miss, and content-hash whatever comes back.
  2. A log_pull audit logger writing one JSON line per pull (timestamp, source, query, row count, content hash) — the data-acquisition twin of the Ch 6.5 API audit log.
  3. Per-source pull functions for FRED, SEC EDGAR (XBRL + full-text), HMDA (CFPB), USPTO PatentsView, yfinance, and a WRDS pattern — each reading its key from os.environ, each gated so it does nothing offline.
  4. An offline-runnable demo: a synthetic data generator standing in for a real source, pushed through the exact same cache+log harness, so the whole notebook executes with no network and no API keys. You will see the cache miss on the first call, the cache hit on the second, and the audit log it leaves behind.

The whole notebook runs offline. Every real-source function is defined and explained but skipped unless the matching environment variable is set — so there are no network calls, no keys, and nothing licensed ever touches disk. Two rules from CONVENTIONS.md §5 are load-bearing here and we keep saying them: secrets live in environment variables, never in code, and licensed data stays on GMU infrastructure and is never committed to git.

1. Setup

We fix one seeded generator (rng = np.random.default_rng(42)) so the synthetic demo produces the identical sample on every run — the reproducibility rule from CONVENTIONS.md §5. We force matplotlib’s non-interactive Agg backend so the notebook runs headless anywhere, and we import only the standard scientific stack plus requests (used only inside the gated real-source functions, never at import time). Nothing in this setup cell touches the network.

We also make a temporary working directory for this demo’s cache and logs. In a real project the cache lives in a git-ignored data/raw/ and the log in logs/pulls.jsonl (see §8); here we write to a throwaway temp dir so running the notebook leaves nothing behind and never risks committing a raw pull.

import os
import json
import time
import hashlib
import datetime
import pathlib
import tempfile

import matplotlib
matplotlib.use("Agg")  # headless, non-interactive backend — safe anywhere

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

rng = np.random.default_rng(42)  # one seeded generator for the whole notebook

# A throwaway working area for THIS demo, so we never write into the real repo or
# accidentally cache a raw pull where it could be committed. Real projects use data/raw/.
WORKDIR = pathlib.Path(tempfile.mkdtemp(prefix="nb72_harness_"))
RAW_CACHE = WORKDIR / "data" / "raw"      # cached raw pulls live here (git-ignored in real life)
LOG_PATH = WORKDIR / "logs" / "pulls.jsonl"  # one JSON line per pull
RAW_CACHE.mkdir(parents=True, exist_ok=True)
LOG_PATH.parent.mkdir(parents=True, exist_ok=True)

print("numpy", np.__version__, "| pandas", pd.__version__)
print("temp workdir:", WORKDIR)
print("raw cache   :", RAW_CACHE)
print("audit log   :", LOG_PATH)

2. The content hash — “did the data change?” in one line

The trick that makes reproducibility checkable is a content hash: a short fingerprint of the data such that two byte-identical tables produce the same string and any change — a revised value, a dropped row, a re-stated record — produces a different one. We use SHA-256 over a canonical CSV rendering of the DataFrame. Canonical matters: we sort nothing and we always write the index the same way, so the same data always hashes the same way across machines and runs.

Why hash the CSV text rather than the Parquet bytes? Because Parquet files embed metadata (timestamps, library versions, compression details) that can differ between two writes of identical data, which would make the hash unstable. The CSV rendering is just the values, so it is the honest fingerprint of the content. Six months from now, comparing this hash against a fresh pull’s hash is a one-line “did the source move?” check.

def content_hash(df: pd.DataFrame) -> str:
    """SHA-256 of a canonical CSV rendering of a DataFrame — stable across runs/machines.

    Two byte-identical tables hash identically; any change to the values changes the hash.
    We hash the CSV TEXT (just the values), not the Parquet bytes, because Parquet embeds
    timestamps/metadata that differ between writes of identical data.
    """
    payload = df.to_csv(index=False).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


# Sanity check: identical data -> identical hash; a changed value -> different hash.
_a = pd.DataFrame({"x": [1, 2, 3]})
_b = pd.DataFrame({"x": [1, 2, 3]})
_c = pd.DataFrame({"x": [1, 2, 4]})
print("identical tables match :", content_hash(_a) == content_hash(_b))
print("changed value differs  :", content_hash(_a) != content_hash(_c))
print("example hash (truncated):", content_hash(_a)[:16], "...")

3. log_pull — the JSONL audit logger

Every pull appends one JSON line to a log file: when it happened, which source, the exact request (URL or SQL — with any key stripped), how many rows and columns came back, and the content hash. JSON Lines (one object per line) is the right format because it is append-only — you never rewrite the file, you just add a line — and any later tool can read it back row by row.

Two things to notice, both straight from the chapter’s three rules:

  • The query field stores the request, never a secret. You log the URL or SQL with the key removed, so the log is safe to commit even when the underlying data is not. We pass requests through a tiny _redact guard that strips anything that looks like a key before it is written.
  • The content_sha256 makes the log a change detector: if two pulls of the “same” thing log different hashes, the source moved, and you want to know before your results quietly move with it.
import re

# Patterns that look like an embedded secret inside a logged URL/query. Defense in depth:
# real source functions already keep keys in HEADERS (never in the query string), but if a key
# ever leaked into a query param, this scrubs it before it can be written to disk.
_SECRET_PARAM = re.compile(
    r"(?i)(api[_-]?key|apikey|token|password|secret|access[_-]?key)=([^&\s]+)"
)

def _redact(query: str) -> str:
    """Strip anything that looks like a key from a query string before logging."""
    return _SECRET_PARAM.sub(r"\1=***REDACTED***", str(query))


def log_pull(logfile, source: str, query: str, df: pd.DataFrame) -> dict:
    """Append one JSON line describing a data pull: what, when, how big, content hash.

    `query` is the request (URL or SQL) with NO secrets — we redact defensively anyway.
    Returns the record so callers can inspect/print it.
    """
    record = {
        "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "source": source,
        "query": _redact(query),
        "n_rows": int(len(df)),
        "n_cols": int(df.shape[1]),
        "content_sha256": content_hash(df),
    }
    logfile = pathlib.Path(logfile)
    logfile.parent.mkdir(parents=True, exist_ok=True)
    with open(logfile, "a") as f:
        f.write(json.dumps(record) + "\n")
    return record


# Quick demonstration of the redactor (NOT a real key — illustrative anti-pattern, then scrubbed):
print(_redact("https://api.example.com/data?series=UNRATE&api_key=abc123SECRET&fmt=json"))

4. cache_or_fetch — check the cache first, hit the network only on a miss

This is the spine of every reproducible pull in Chapter 7.2. Given a cache filename and a zero-argument fetch function, it:

  1. Checks the cache. If data/raw/<name>.parquet already exists, read and return it — the network is never touched. This makes re-runs instant, independent of whether the source is up, and — critically — frozen: the analysis reads the same bytes every time.
  2. On a miss, calls fetch() to get the DataFrame, writes it to the cache as Parquet, logs the pull, and returns it.

The fetch argument is the seam that lets one harness serve every source: FRED, EDGAR, HMDA, PatentsView, yfinance, WRDS, or — in our offline demo — a synthetic generator all look identical from here. They differ only in the little function you hand in. We return a (df, hit) pair so the caller can see whether it was a cache hit; that visible hit on the second call is the whole point.

Notice the discipline: the cache write and the log entry happen together, only on a miss, so the log records exactly the pulls that actually went to the source — not the cache reads.

def cache_or_fetch(cache_path, source: str, query: str, fetch):
    """Return (df, was_cache_hit). Read Parquet cache if present; else fetch, cache, and log.

    `fetch` is a zero-arg callable returning a DataFrame. This is the seam that lets one
    harness serve every source — only `fetch` changes from source to source.
    """
    cache_path = pathlib.Path(cache_path)
    if cache_path.exists():
        df = pd.read_parquet(cache_path)        # CACHE HIT — never hit the source
        return df, True

    df = fetch()                                # CACHE MISS — go to the source exactly once
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    df.to_parquet(cache_path, index=False)      # freeze the raw response
    log_pull(LOG_PATH, source, query, df)       # record the pull (what/when/size/hash)
    return df, False

5. Gating real sources so the notebook runs offline

The six real-source functions below would each hit the network and (most) need a key. We never want them to fire when a student, a referee, or a CI server runs the notebook with no keys set. So each one is gated: it reads its key from os.environ and, if the key (or required setting) is missing, it raises a clear, friendly error instead of making a request.

We wrap that gate in one helper, require_env, and we will check the gate before ever calling the function (§7). The contract is simple and defensible:

  • Key present in the environment -> the function will run against the live API.
  • Key absent (the offline default, and what happens right now) -> the function is skipped; it never touches the network and nothing is written.

This is the same provider-switch idea as nb6.5: a research notebook that requires a paid key to run is not reproducible. Ours requires none.

class MissingKey(RuntimeError):
    """Raised by a gated source function when its required environment variable is absent."""


def require_env(var_name: str) -> str:
    """Return os.environ[var_name], or raise MissingKey with a helpful message.

    Keys are NEVER hard-coded — they come only from the environment, set out-of-band
    (e.g. `export FRED_API_KEY=...` in your shell, or a git-ignored .env file).
    """
    val = os.environ.get(var_name)
    if not val:
        raise MissingKey(
            f"{var_name} is not set. Set it in your shell or a git-ignored .env file; "
            f"this function requires a key/network and is skipped offline."
        )
    return val


# Which env var (or setting) gates each real source. SEC needs a User-Agent string, not a secret,
# but we gate on it the same way so the EDGAR functions also stay dormant offline.
GATES = {
    "FRED":        "FRED_API_KEY",
    "EDGAR":       "SEC_USER_AGENT",      # SEC fair-access: a name+email, NOT a secret
    "HMDA":        "HMDA_ENABLE",         # public API; we still gate so it stays offline by default
    "PATENTSVIEW": "USPTO_ODP_API_KEY",   # USPTO Open Data Portal; PatentsView migrated to ODP on 2026-03-20
    "YFINANCE":    "YFINANCE_ENABLE",     # no key, but network; gate so it stays offline by default
    "WRDS":        "WRDS_USERNAME",       # licensed — GMU infrastructure only
}
for name, var in GATES.items():
    print(f"{name:12s} gated on {var:20s} -> {'SET' if os.environ.get(var) else 'absent (offline)'}")

6. Per-source pull functions (all gated; all skipped offline)

Each function below has the same shape: read its key/setting from the environment via require_env (raising MissingKey if absent), then make a bounded, specific request and return a tidy DataFrame. None of them runs unless its gate is set — which it is not, here — so reading these is like reading the recipe without turning on the stove.

6.1 FRED — macro and financial time series

Maya’s macro control. FRED’s API caps you near 120 requests/minute; the subtle gotcha is data revision (today’s 2020-GDP value is not the one published in 2021), so anything that pretends to use only past information should pull an ALFRED vintage rather than the latest series. The key is read from FRED_API_KEY; the URL we log has no key in it (the key rides in the params dict, which we do not log verbatim).

def pull_fred(series_id: str, start: str, end: str) -> pd.DataFrame:
    """Fetch one FRED series as a tidy DataFrame. REQUIRES key/network; skipped offline.

    Pin: record the vintage (or 'latest, pulled <date>'). Key from FRED_API_KEY (env only).
    """
    api_key = require_env("FRED_API_KEY")          # raises MissingKey offline -> never runs
    import requests                                # imported lazily, inside the gate
    url = "https://api.stlouisfed.org/fred/series/observations"
    params = {
        "series_id": series_id,
        "observation_start": start,
        "observation_end": end,
        "file_type": "json",
        "api_key": api_key,                        # key in PARAMS only, never in the logged query
    }
    resp = requests.get(url, params=params, timeout=30)
    resp.raise_for_status()
    obs = resp.json()["observations"]
    df = pd.DataFrame(obs)[["date", "value"]].copy()
    df["date"] = pd.to_datetime(df["date"])
    df["value"] = pd.to_numeric(df["value"], errors="coerce")
    return df

# The query string we would LOG — note there is no key in it (safe to commit):
FRED_LOG_QUERY = "GET https://api.stlouisfed.org/fred/series/observations?series_id=UNRATE&..."
print("pull_fred defined. Log query (no key):", FRED_LOG_QUERY)

6.2 SEC EDGAR — XBRL company facts and full-text search

Priya’s 10-K numbers and Devon’s S-1 live here. EDGAR is free and public (no license, no GMU rule) but enforces a fair-access policy: every request must carry a User-Agent header that identifies you (name + email), or you get a 403. The limit is 10 requests/second aggregated across your scripts. We gate on SEC_USER_AGENT (a contact string, not a secret), and the pin that matters is the accession number — 0000320193-23-000106 is an exact filing, while “Apple’s FY2023 10-K” is ambiguous once an amendment exists.

We define two EDGAR functions: one for the XBRL companyfacts numbers, one for full-text search (EFTS).

def pull_edgar_companyfacts(cik: str) -> pd.DataFrame:
    """Fetch XBRL company facts for one CIK. REQUIRES network + SEC User-Agent; skipped offline.

    Pin: the filing accession numbers behind each fact. cik is zero-padded to 10 digits.
    """
    user_agent = require_env("SEC_USER_AGENT")     # e.g. "Maya Rodriguez maya@gmu.edu" (env only)
    import requests
    headers = {"User-Agent": user_agent}           # MANDATORY for the SEC; from env, not hard-coded
    cik10 = str(cik).zfill(10)
    url = f"https://data.sec.gov/api/xbrl/companyfacts/CIK{cik10}.json"
    resp = requests.get(url, headers=headers, timeout=30)
    resp.raise_for_status()
    facts = resp.json()
    # Flatten one common fact (Revenues, USD units) into a tidy frame for the example.
    rows = []
    units = facts.get("facts", {}).get("us-gaap", {}).get("Revenues", {}).get("units", {})
    for unit, items in units.items():
        for it in items:
            rows.append({"cik": cik10, "metric": "Revenues", "unit": unit,
                         "end": it.get("end"), "val": it.get("val"),
                         "accn": it.get("accn"), "fy": it.get("fy"), "fp": it.get("fp")})
    return pd.DataFrame(rows)


def pull_edgar_fulltext(phrase: str, forms: str = "10-K") -> pd.DataFrame:
    """EDGAR full-text search (EFTS) for a phrase. REQUIRES network + SEC User-Agent; skipped offline.

    Returns matching filings (CIK, accession, form, date) — candidates you then pull and pin.
    """
    user_agent = require_env("SEC_USER_AGENT")
    import requests
    headers = {"User-Agent": user_agent}
    url = "https://efts.sec.gov/LATEST/search-index"
    params = {"q": f'"{phrase}"', "forms": forms}
    resp = requests.get(url, headers=headers, params=params, timeout=30)
    resp.raise_for_status()
    hits = resp.json().get("hits", {}).get("hits", [])
    rows = [{"accn": h["_id"], "form": h["_source"].get("root_form"),
             "date": h["_source"].get("file_date")} for h in hits]
    return pd.DataFrame(rows)

print("pull_edgar_companyfacts + pull_edgar_fulltext defined (both gated on SEC_USER_AGENT).")

6.3 HMDA — fair-lending data via the CFPB Data Browser

Maya’s core dataset. HMDA is huge — a single year’s national file is several gigabytes — so the cardinal rule is aggregate or filter on the server, not after downloading the universe. The CFPB Data Browser’s aggregations endpoint returns a small summary table (counts by the variables you group on) instead of tens of millions of rows. It is public and unrestricted, but it contains sensitive demographic data, so handle it respectfully and never attempt re-identification. We gate on HMDA_ENABLE purely so the public API stays dormant offline by default.

def pull_hmda_aggregations(year: str, state: str, actions_taken: str = "1,3") -> pd.DataFrame:
    """Fetch a SMALL aggregated HMDA summary (counts), not the multi-GB row file. Skipped offline.

    Pin: the data YEAR and the exact filter — the request IS the modeling decision, so log it.
    """
    require_env("HMDA_ENABLE")                     # gate: stays offline unless explicitly enabled
    import requests
    url = "https://ffiec.cfpb.gov/v2/data-browser-api/view/aggregations"
    params = {"years": year, "states": state, "actions_taken": actions_taken}
    resp = requests.get(url, params=params, timeout=60)
    resp.raise_for_status()
    return pd.json_normalize(resp.json()["aggregations"])

print("pull_hmda_aggregations defined (gated on HMDA_ENABLE; aggregates server-side).")

6.4 USPTO PatentsView — innovation data

Leah’s patent counts. PatentsView migrated to the USPTO Open Data Portal (ODP) on March 20, 2026; the legacy api.patentsview.org host was discontinued in May 2025 and the interim search.patentsview.org/api/v1 PatentSearch API was paused at the ODP cutover. The current home is https://api.uspto.gov, the header is X-API-KEY (lowercase), and the free key is read from USPTO_ODP_API_KEY. For whole-database extracts you still use the bulk PatentsView TSV/parquet downloads on ODP rather than millions of API calls. Two gotchas to respect, not paper over: disambiguation uncertainty (“the same assignee” is a statistical guess) and grant lag (a 2023 application may not grant until 2026, so recent years undercount). Pin the data release. Keys go in the X-API-KEY header (never in the logged URL).

[CHECK] The exact ODP patent-search endpoint path under ${base}/api/v1/... is being finalized post-cutover (March 20, 2026). The targeted-query call below uses /api/v1/patent/applications per the current ODP docs; confirm against data.uspto.gov/support/transition-guide/patentsview before a large run. The bulk PatentsView tables are stable on ODP today.

def pull_patentsview(assignee: str, start: str, end: str, size: int = 100) -> pd.DataFrame:
    """Fetch patents for one assignee in a date window. REQUIRES key/network; skipped offline.

    USPTO Open Data Portal (ODP) — PatentsView migrated to api.uspto.gov on 2026-03-20.
    Pin: the PatentsView data RELEASE (updates ~quarterly). Key in HEADER, never in the URL.
    """
    api_key = require_env("USPTO_ODP_API_KEY")     # env only; raises MissingKey offline
    import requests
    base = "https://api.uspto.gov"                 # ODP; pin in config.py
    headers = {"X-API-KEY": api_key}               # lowercase header name per ODP docs
    # [CHECK] the post-cutover patent-search endpoint path under ${base}/api/v1/... ;
    # the path below follows the current ODP docs (`/api/v1/patent/applications`) and the
    # data card. Bulk PatentsView TSV/parquet tables on ODP are the right tool for whole-
    # decade extracts; the API is for targeted queries like this one.
    params = {
        "q": f'patentDate:[{start} TO {end}] AND assigneeName:"{assignee}"',
        "rows": size,
    }
    resp = requests.get(
        f"{base}/api/v1/patent/applications",
        params=params,
        headers=headers,                           # key in HEADER, not in the logged URL
        timeout=60,
    )
    resp.raise_for_status()
    payload = resp.json()
    # ODP wraps result rows under a top-level key (commonly "patentBag"/"results"); the
    # exact key is part of the [CHECK] above. Try a couple of likely shapes defensively.
    rows = payload.get("patentBag") or payload.get("results") or payload.get("patents") or []
    return pd.json_normalize(rows)

print("pull_patentsview defined (gated on USPTO_ODP_API_KEY; ODP api.uspto.gov, X-API-KEY header).")

6.5 yfinance — a free price series (prototype-grade)

Sam’s quick price series before committing to CRSP. yfinance needs no key but does hit the network, and it is unofficial: it scrapes an undocumented endpoint Yahoo can change without warning, and its data carries survivorship bias (delisted tickers vanish) and no point-in-time guarantee. Fine for learning and prototyping; not what you submit a paper on when CRSP is available. Its very instability makes caching more important, not less — your cached copy is the only guarantee the numbers will not move. We gate on YFINANCE_ENABLE so it stays offline by default.

def pull_yfinance(ticker: str, start: str, end: str) -> pd.DataFrame:
    """Fetch a daily price series via yfinance. REQUIRES network (+ the yfinance pkg); skipped offline.

    Pin: pulled-on date AND auto_adjust setting. Prototype-grade: has survivorship bias.
    """
    require_env("YFINANCE_ENABLE")                 # gate: stays offline unless explicitly enabled
    import yfinance as yf                          # imported lazily, inside the gate
    prices = yf.download(ticker, start=start, end=end, auto_adjust=True, progress=False)
    return prices.reset_index()

print("pull_yfinance defined (gated on YFINANCE_ENABLE; unofficial, prototype-grade).")

6.6 WRDS — the licensed-data pattern (CRSP / Compustat)

This is the one that makes Rule 2 exist. WRDS serves licensed data (CRSP, Compustat, IBES, TRACE) that GMU pays for; the license permits use on GMU systems only. So the rule is absolute: you may not download CRSP to your laptop, email it off-campus, or commit it to git. The cached DataFrame lives only in a git-ignored data/raw/ on GMU infrastructure, and your repo commits the query code and the log — the recipe — never the bytes.

The wrds package manages credentials via ~/.pgpass (no password in code); we read the username from WRDS_USERNAME only to gate the function. The query is bounded — named columns, a date range — never SELECT * on a 100M-row table.

def pull_wrds_crsp_msf(start: str, end: str) -> pd.DataFrame:
    """Pull bounded CRSP monthly returns. LICENSED — GMU infrastructure only; skipped offline.

    Pin: the WRDS snapshot date. NEVER commit the result; cache only under a git-ignored
    data/raw/ on GMU systems. The wrds pkg reads credentials from ~/.pgpass (no password in code).
    """
    username = require_env("WRDS_USERNAME")        # env only; gates the function offline
    import wrds                                     # imported lazily, inside the gate
    db = wrds.Connection(wrds_username=username)
    try:
        df = db.raw_sql(
            """
            SELECT permno, date, ret, prc, shrout
            FROM crsp.msf
            WHERE date BETWEEN %(start)s AND %(end)s
            """,
            params={"start": start, "end": end},
            date_cols=["date"],
        )
    finally:
        db.close()
    return df

print("pull_wrds_crsp_msf defined (gated on WRDS_USERNAME; LICENSED, GMU-only, never committed).")

7. The source registry — confirm every real source is skipped offline

Here is the payoff of gating uniformly. We register each real source with its gate and a tiny zero-arg fetch closure. Then we check the gate before calling — if the environment variable is absent, we mark the source SKIPPED and never invoke its fetch function, so no network call is ever made. Run with no keys (the default) and you will see every real source skipped. Set a key out-of-band and that one source would activate — but the notebook never requires you to.

This is the defensible posture: the recipe is fully present and readable, the execution is fully gated, and the offline run is provably network-free.

# Each entry: a human label, its gate var, and a zero-arg fetch closure (NOT called unless gated on).
REAL_SOURCES = [
    ("FRED:UNRATE",            GATES["FRED"],        lambda: pull_fred("UNRATE", "2010-01-01", "2020-12-31")),
    ("EDGAR:companyfacts",     GATES["EDGAR"],       lambda: pull_edgar_companyfacts("0000320193")),
    ("EDGAR:fulltext",         GATES["EDGAR"],       lambda: pull_edgar_fulltext("climate-related risks")),
    ("HMDA:aggregations",      GATES["HMDA"],        lambda: pull_hmda_aggregations("2020", "VA")),
    ("PatentsView:IBM",        GATES["PATENTSVIEW"], lambda: pull_patentsview("International Business Machines Corporation", "2015-01-01", "2015-12-31")),
    ("yfinance:AAPL",          GATES["YFINANCE"],    lambda: pull_yfinance("AAPL", "2010-01-01", "2020-12-31")),
    ("WRDS:crsp.msf [LICENSED]", GATES["WRDS"],      lambda: pull_wrds_crsp_msf("2010-01-01", "2020-12-31")),
]

skipped, ran = [], []
for label, gate_var, fetch in REAL_SOURCES:
    if not os.environ.get(gate_var):
        skipped.append((label, gate_var))
        print(f"SKIPPED  {label:30s} (no {gate_var}; requires key/network — not called offline)")
        continue
    # This branch only runs if a key is set out-of-band; offline it is never reached.
    cache = RAW_CACHE / (label.split()[0].replace(":", "_").replace("/", "_") + ".parquet")
    df, hit = cache_or_fetch(cache, label, f"<live request for {label}>", fetch)
    ran.append((label, len(df), hit))
    print(f"RAN      {label:30s} rows={len(df)} cache_hit={hit}")

print()
print(f"Offline result: {len(skipped)} real sources SKIPPED, {len(ran)} ran.")
assert len(ran) == 0, "Offline run must call NO real source. A gate failed."
print("Confirmed: no real source was called — the notebook is network-free as run.")

8. An offline-runnable demo — a synthetic source through the real harness

To exercise the whole harness with no network, we stand in a synthetic data generator for a real source. It plays the role of, say, a FRED series: a daily series with a date column and a value column, produced from our seeded rng so it is identical on every run. The crucial point is that it flows through the exact same cache_or_fetch + log_pull the real sources use — we are testing the real harness, not a toy of it.

We also count how many times the generator is actually invoked, so on the second call we can prove the cache served the data and the source was not re-hit.

# A counter so we can PROVE the second call hits the cache and does not regenerate.
_SYNTH_CALLS = {"n": 0}

def synthetic_fetch():
    """Stand-in for a real source's network call. Deterministic via the seeded rng.

    Plays the role of e.g. a FRED daily series. Counts its own invocations so we can show
    that the second cache_or_fetch call does NOT reach it.
    """
    _SYNTH_CALLS["n"] += 1
    n = 250
    dates = pd.bdate_range("2020-01-01", periods=n)          # 250 business days
    # A gentle random walk standing in for, say, a daily rate series.
    steps = rng.normal(0.0, 0.05, size=n)
    values = 3.5 + np.cumsum(steps)
    return pd.DataFrame({"date": dates, "value": np.round(values, 4)})

SYNTH_SOURCE = "SYNTHETIC:demo-series [offline stand-in, pulled 2026-05-26]"
SYNTH_QUERY = "synthetic_fetch() :: daily series, seed=42, n=250"
SYNTH_CACHE = RAW_CACHE / "synthetic_demo_series.parquet"

print("synthetic_fetch defined. Calls so far:", _SYNTH_CALLS["n"])
print("cache target:", SYNTH_CACHE)

9. First call — a cache MISS (the source is hit once, cached, and logged)

The cache file does not exist yet, so this call goes to synthetic_fetch() exactly once, writes the result to Parquet, and appends one line to the audit log. We confirm the miss three ways: the fetch counter ticks to 1, cache_or_fetch reports hit=False, and a Parquet file now exists on disk.

assert not SYNTH_CACHE.exists(), "cache should not exist before the first call"

df1, hit1 = cache_or_fetch(SYNTH_CACHE, SYNTH_SOURCE, SYNTH_QUERY, synthetic_fetch)

print("cache_hit         :", hit1, "(expected False — a miss)")
print("synthetic calls   :", _SYNTH_CALLS["n"], "(expected 1 — the source was hit once)")
print("cache file exists :", SYNTH_CACHE.exists())
print("rows x cols       :", df1.shape)
print("content hash      :", content_hash(df1)[:16], "...")
df1.head()

10. Second call — a cache HIT (the source is NOT touched)

Now we call cache_or_fetch again with the identical arguments. This time the Parquet cache exists, so it is read straight from disk: hit=True, and — the proof that matters — the fetch counter does not move (still 1). The cached and freshly-loaded tables are byte-identical, which we verify by comparing their content hashes. This is reproducibility you can see: the second run of an analysis reads exactly the same bytes the first one froze.

calls_before = _SYNTH_CALLS["n"]
df2, hit2 = cache_or_fetch(SYNTH_CACHE, SYNTH_SOURCE, SYNTH_QUERY, synthetic_fetch)

print("cache_hit            :", hit2, "(expected True — a hit)")
print("synthetic calls      :", _SYNTH_CALLS["n"], f"(expected {calls_before} — UNCHANGED; source not re-hit)")
print("hashes match (frozen):", content_hash(df1) == content_hash(df2))

assert hit2 is True, "second call must be a cache hit"
assert _SYNTH_CALLS["n"] == calls_before, "second call must NOT re-invoke the source"
assert content_hash(df1) == content_hash(df2), "cached data must be byte-identical"
print("Confirmed: second call served from cache; the source was not touched.")

11. The audit log — what got written

The harness left exactly one line in logs/pulls.jsonl (the miss logged; the hit did not, which is correct — only real pulls are recorded). We read it back and pretty-print it. Notice it carries the timestamp, the source label with its pin, the query (no secrets), the row/column counts, and the content hash — everything a reviewer needs to ask “where did this number come from?” and get an answer.

log_lines = LOG_PATH.read_text().strip().splitlines()
print(f"audit log has {len(log_lines)} line(s) (expected 1 — only the miss was logged)\n")
for line in log_lines:
    rec = json.loads(line)
    print(json.dumps(rec, indent=2))

assert len(log_lines) == 1, "exactly one pull (the miss) should be logged"

12. A look at the cached series

Purely to make the demo concrete, we plot the synthetic series we just pulled-and-cached. Because it comes from the seeded rng, this figure is identical on every run — the same reproducibility guarantee that makes the cache trustworthy. (We use the Agg backend, so this renders headless.)

fig, ax = plt.subplots(figsize=(8, 3.2))
ax.plot(df1["date"], df1["value"], lw=1.2)
ax.set_title("Synthetic 'rate' series pulled through the harness (seed=42)")
ax.set_xlabel("date")
ax.set_ylabel("value")
ax.grid(alpha=0.3)
fig.tight_layout()
fig.savefig(WORKDIR / "synthetic_series.png", dpi=90)
plt.close(fig)
print("figure saved to", WORKDIR / "synthetic_series.png")

9. The repo discipline — .gitignore, licensed data, and what you commit

The directory layout that falls out of the harness is the one your capstone repo should use. The licensed bytes stay on GMU infrastructure and never enter git; you commit the recipe (the pull code) and the log, which is exactly enough to make public-source work fully re-runnable and licensed-source work recipe-reproducible.

your-project/
├── .gitignore            # excludes data/raw/  and  .env
├── .env                  # your API keys — NEVER committed
├── code/pull_data.py     # the bounded queries; reads keys from os.environ
├── data/raw/             # cached raw pulls — git-ignored; LICENSED data lives ONLY here, on GMU
└── logs/pulls.jsonl      # one line per pull: what, when, how big, content hash

The cell below writes the .gitignore that this discipline requires into our temp dir, so you can read the exact lines. The two that carry the rules: data/ (and *.parquet/*.csv) keeps every raw pull — especially licensed CRSP/Compustat — out of git, and .env keeps your keys out.

gitignore_text = """# Data — licensed data must NOT be committed; raw pulls stay git-ignored
data/
*.parquet
*.csv
!data-cards/**
# Secrets — keys live in the environment / a git-ignored .env, never in code
.env
*.key
# Python / notebook cruft
__pycache__/
*.pyc
.ipynb_checkpoints/
"""
(WORKDIR / ".gitignore").write_text(gitignore_text)
print((WORKDIR / ".gitignore").read_text())
print("Rule restated: licensed data (CRSP/Compustat/IBES/TRACE) stays on GMU infra and is NEVER committed.")

10. Recap and cleanup

What we built, and why each piece is load-bearing:

  • content_hash — a stable fingerprint that turns “did the data change?” into a one-line check.
  • log_pull — an append-only JSONL audit trail (timestamp, source, redacted query, size, hash).
  • cache_or_fetch — check the cache first, hit the source only on a miss, then freeze + log it.
  • Six gated source functions (FRED, EDGAR XBRL + full-text, HMDA, PatentsView, yfinance, WRDS) — each reads its key from os.environ, each skipped offline; no secret is ever hard-coded.
  • A synthetic demo that runs the real harness offline, showing a cache miss, then a cache hit that does not re-touch the source, then the one-line audit log it leaves.

We finish by removing the temporary working directory, so this notebook leaves nothing behind — no cache, no log, no figure — which is also a small reminder that raw pulls do not belong in your repo. In a real project you would instead keep data/raw/ (git-ignored) and logs/pulls.jsonl (committed).

Your Turn. Pick your capstone source. (1) Write its gated pull_* function in the shape above, reading its key from os.environ. (2) Name the one pin that would freeze a re-run a year from now (FRED vintage, EDGAR accession, WRDS snapshot, PatentsView release) and put it in the source label you log. (3) If the source is licensed, state in one sentence what stays on GMU infrastructure and what you could hand an outside collaborator instead (answer: the code and the log — the recipe).

import shutil
shutil.rmtree(WORKDIR, ignore_errors=True)
print("removed temp workdir:", WORKDIR, "-> exists?", WORKDIR.exists())
print("Done. The notebook ran fully offline: 0 network calls, 0 keys, nothing left on disk.")