Warrior_EA/research/altdata/fred.py

"""FRED / ALFRED collector - free API key (https://fred.stlouisfed.org/docs/api/api_key.html).

Why ALFRED and not plain FRED: most macro series are REVISED. Training on
today's FRED values leaks the revision (same trap as the economic calendar's
post-revision actual_value - see calendar recorder memory). The ALFRED
`realtime_start/realtime_end` window returns each observation AS FIRST
PUBLISHED, and `realtime_start` of each vintage row IS the publication date,
which becomes our `published` column.

Series set (daily unless noted) - chosen for SP500/gold/FX/oil relevance:
    VIXCLS    VIX close
    DGS2      2y treasury yield
    DGS10     10y treasury yield
    T10Y2Y    10y-2y spread
    T5YIE     5y breakeven inflation
    DTWEXBGS  broad dollar index
    DCOILWTICO WTI spot
    DFF       fed funds effective
    NFCI      Chicago Fed financial conditions (weekly)
    WALCL     Fed balance sheet (weekly)

Usage:
    python -m altdata.fred                  # all series, full history
    python -m altdata.fred --series VIXCLS
Key: env ALTDATA_FRED_KEY or "fred" in Market Data/altdata/keys.json.
"""
import argparse
import datetime as dt

import pandas as pd

from .common import DATA_ROOT, api_key, ensure_dirs, get

API = "https://api.stlouisfed.org/fred/series/observations"

SERIES = ["VIXCLS", "DGS2", "DGS10", "T10Y2Y", "T5YIE",
          "DTWEXBGS", "DCOILWTICO", "DFF", "NFCI", "WALCL"]


def fetch_unrevised(series_id: str, key: str) -> pd.DataFrame:
    """Plain FRED for series with no ALFRED archive (market data, never revised).

    For these the current value IS the first print. `published` = observed + 1
    day: daily market series post after their session close; +1 day makes the
    as-of join conservative by construction.
    """
    print(f"  ({series_id}: no ALFRED archive -> unrevised path)")
    params = {"series_id": series_id, "api_key": key, "file_type": "json",
              "limit": 100000, "offset": 0}
    rows = []
    while True:
        j = get(API, params=params).json()
        rows.extend(j["observations"])
        params["offset"] += params["limit"]
        if params["offset"] >= j["count"]:
            break
    df = pd.DataFrame(rows)
    df["value"] = pd.to_numeric(df["value"], errors="coerce")
    df = df.dropna(subset=["value"])
    df["observed"] = pd.to_datetime(df["date"])
    df["published"] = df["observed"] + pd.Timedelta(days=1)
    return df[["observed", "published", "value"]].reset_index(drop=True)


def fetch_vintaged(series_id: str, key: str) -> pd.DataFrame:
    """First prints via ALFRED output_type=4 (initial release only).

    Each observation appears once, valued as FIRST PUBLISHED, and its
    realtime_start IS the publication date. Avoids the 2000-vintage cap that a
    full-vintage query hits on long daily series.
    """
    # the API caps vintage dates per realtime period at 2000, so chunk the
    # realtime axis: everything pre-2000 (sparse archives) then 3y windows.
    # realtime_end MUST NOT be after today or FRED 400s - which silently made
    # every REVISED series (unemployment, CPI, GDP: the ones that actually need
    # the vintage path) unreachable, while unrevised series never noticed
    # because they bail to fetch_unrevised before the window is used.
    today = dt.date.today().isoformat()
    windows = [("1776-07-04", "1999-12-31")]
    for y in range(2000, 2100, 3):
        start = f"{y}-01-01"
        if start > today:
            break
        windows.append((start, min(f"{y + 2}-12-31", today)))
    rows = []
    for rt_start, rt_end in windows:
        params = {
            "series_id": series_id, "api_key": key, "file_type": "json",
            "realtime_start": rt_start, "realtime_end": rt_end,
            "output_type": 4, "limit": 100000, "offset": 0,
        }
        while True:
            r = get(API, params=params, ok_codes=(200, 400))
            if r.status_code == 400:
                if "does not exist in ALFRED" in r.text:
                    return fetch_unrevised(series_id, key)
                raise RuntimeError(f"FRED 400 for {series_id}: {r.text[:200]}")
            j = r.json()
            rows.extend(j["observations"])
            params["offset"] += params["limit"]
            if params["offset"] >= j["count"]:
                break
    df = pd.DataFrame(rows)
    df["value"] = pd.to_numeric(df["value"], errors="coerce")
    df = df.dropna(subset=["value"])
    df["observed"] = pd.to_datetime(df["date"])
    df["published"] = pd.to_datetime(df["realtime_start"])
    df = df.sort_values("observed").drop_duplicates("observed", keep="first")
    # a publication date earlier than the observed period is a data error;
    # clamp to observed so the as-of join can never look ahead of the period
    df["published"] = df[["published", "observed"]].max(axis=1)
    return df[["observed", "published", "value"]].reset_index(drop=True)


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--series", nargs="*", default=SERIES)
    args = ap.parse_args()
    key = api_key("fred")
    if not key:
        raise SystemExit("No FRED key. Set ALTDATA_FRED_KEY or add 'fred' to "
                         f"{DATA_ROOT / 'keys.json'} (free: fred.stlouisfed.org)")
    ensure_dirs()
    out_dir = DATA_ROOT / "fred"
    out_dir.mkdir(parents=True, exist_ok=True)
    for sid in args.series:
        df = fetch_vintaged(sid, key)
        dest = out_dir / f"{sid}.csv"
        df.to_csv(dest, index=False)
        print(f"{sid}: {len(df)} first-print rows "
              f"{df['observed'].min():%Y-%m-%d} -> {df['observed'].max():%Y-%m-%d} => {dest.name}")


if __name__ == "__main__":
    main()