fred_client.py

"""FRED API client for the external-indicator sections.

The API key is passed as a query parameter, so any error carrying a URL carries the key.
Every failure path here raises a message built from the series id and status code only,
with the original exception's context suppressed: requests exceptions embed the full URL.
"""
import json
import os
import time
from datetime import datetime
from pathlib import Path

import pandas as pd
import requests

API = "https://api.stlouisfed.org/fred"
ROOT = Path(__file__).resolve().parent


def _parse_env(path):
    out = {}
    for line in Path(path).read_text(encoding="utf-8-sig").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        out[k.strip()] = v.strip().strip('"').strip("'")
    return out


def load_key(env_path=None):
    """Environment first, then the project-root .env. Never echoes the file."""
    key = os.environ.get("FRED_API_KEY")
    if key:
        return key
    path = Path(env_path) if env_path else ROOT / ".env"
    if path.exists():
        key = _parse_env(path).get("FRED_API_KEY")
    if not key:
        raise RuntimeError("FRED_API_KEY not found; put it in .env or the environment")
    return key


def _get(path, key, **params):
    params.update(api_key=key, file_type="json")
    label = params.get("series_id", path)
    try:
        r = requests.get(f"{API}/{path}", params=params, timeout=30)
    except requests.RequestException as e:
        raise RuntimeError(f"FRED {label}: request failed ({type(e).__name__})") from None
    if r.status_code != 200:
        raise RuntimeError(f"FRED {label}: HTTP {r.status_code}")
    return r.json()


def obs_frame(payload):
    """Observation payload to a [date, value] frame. FRED writes '.' for missing."""
    rows = payload.get("observations", [])
    if not rows:
        return pd.DataFrame({"date": pd.Series(dtype="datetime64[ns]"),
                             "value": pd.Series(dtype="float64")})
    df = pd.DataFrame(rows)
    df["date"] = pd.to_datetime(df["date"])
    df["value"] = pd.to_numeric(df["value"], errors="coerce")
    return df[["date", "value"]].reset_index(drop=True)


def wide(frames):
    """Outer-join per-series frames into one date-indexed table."""
    out = None
    for sid, df in frames.items():
        s = df.set_index("date")["value"].rename(sid)
        out = s.to_frame() if out is None else out.join(s, how="outer")
    return out.sort_index() if out is not None else pd.DataFrame()


def series_meta(series_id, key):
    s = _get("series", key, series_id=series_id)["seriess"][0]
    return {k: s.get(k) for k in (
        "id", "title", "frequency", "frequency_short", "units", "units_short",
        "seasonal_adjustment_short", "observation_start", "observation_end", "last_updated")}


def observations(series_id, key):
    return obs_frame(_get("series/observations", key, series_id=series_id))


def snapshot(series_ids, out_dir, key, pause=0.5, stamp=None):
    """Pull metadata and observations for each series; write CSVs plus a meta JSON.

    Files are date-stamped because these series revise and the notebook must be able to
    reproduce a figure from the snapshot that produced it.
    """
    out_dir = Path(out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    stamp = stamp or datetime.now().strftime("%Y%m%d")

    frames, meta = {}, {}
    for sid in series_ids:
        m = series_meta(sid, key)
        time.sleep(pause)
        df = observations(sid, key)
        time.sleep(pause)
        df.to_csv(out_dir / f"{sid}_{stamp}.csv", index=False)
        frames[sid] = df
        last = df.dropna(subset=["value"])
        meta[sid] = {**m, "rows": len(df),
                     "last_value_date": (last.date.max().strftime("%Y-%m-%d")
                                         if len(last) else None)}
        print(f"{sid:24s} {m['frequency_short']:3s} {str(m['units_short'])[:22]:22s} "
              f"{len(df):>6,} rows  last {meta[sid]['last_value_date']}  {m['title'][:46]}")

    w = wide(frames)
    w.to_csv(out_dir / f"fred_wide_{stamp}.csv")
    (out_dir / f"fred_meta_{stamp}.json").write_text(
        json.dumps({"pulled_at": datetime.now().isoformat(timespec="seconds"),
                    "stamp": stamp, "series": meta}, indent=1), encoding="utf-8")
    return w


def latest_snapshot(out_dir="data/fred"):
    """Most recent stamp on disk, with its metadata. Used by the notebook, offline."""
    out_dir = Path(out_dir)
    if not out_dir.exists():
        out_dir = Path("..") / out_dir
    stamps = sorted(p.stem.split("_")[-1] for p in out_dir.glob("fred_meta_*.json"))
    if not stamps:
        raise FileNotFoundError(
            "no data/fred/fred_meta_<date>.json; run: uv run python pull_fred.py")
    stamp = stamps[-1]
    meta = json.loads((out_dir / f"fred_meta_{stamp}.json").read_text(encoding="utf-8"))
    return out_dir, stamp, meta


def load_series(series_id, out_dir="data/fred", stamp=None):
    """Read one cached series as a date-indexed Series. No network."""
    base, latest, _ = latest_snapshot(out_dir)
    df = pd.read_csv(base / f"{series_id}_{stamp or latest}.csv", parse_dates=["date"])
    return df.set_index("date")["value"].rename(series_id)