bls-data/dc_md_va_unemployment.py

"""
DC/MD/VA Unemployment Dashboard
Pulls LAUS data for DC, Maryland, Virginia (state) plus the DC-Arlington-Alexandria metro.
Outputs a formatted console table and saves results to CSV.
"""

import requests
import csv
import sys
from datetime import datetime
from config import BLS_API_KEY, BLS_API_BASE

# ---------------------------------------------------------------------------
# Series definitions
# Format: LAUS[S=SA / U=unadj][area code 15 chars][measure 2 chars]
# Measures: 03=rate  04=unemployed  05=employed  06=labor force
# ---------------------------------------------------------------------------
SERIES = {
    # DC
    "DC_rate":        "LAUST110000000000003",
    "DC_unemployed":  "LAUST110000000000004",
    "DC_employed":    "LAUST110000000000005",
    "DC_laborforce":  "LAUST110000000000006",
    # Maryland
    "MD_rate":        "LAUST240000000000003",
    "MD_unemployed":  "LAUST240000000000004",
    "MD_employed":    "LAUST240000000000005",
    "MD_laborforce":  "LAUST240000000000006",
    # Virginia
    "VA_rate":        "LAUST510000000000003",
    "VA_unemployed":  "LAUST510000000000004",
    "VA_employed":    "LAUST510000000000005",
    "VA_laborforce":  "LAUST510000000000006",
    # DC-Arlington-Alexandria MSA (state FIPS 11 + CBSA 47900)
    "DCMSA_rate":      "LAUMT114790000000003",
    "DCMSA_unemployed":"LAUMT114790000000004",
    "DCMSA_employed":  "LAUMT114790000000005",
    "DCMSA_laborforce":"LAUMT114790000000006",
}

LABELS = {
    "DC":    "District of Columbia",
    "MD":    "Maryland",
    "VA":    "Virginia",
    "DCMSA": "DC Metro (MSA)",
}

MEASURES = ["rate", "unemployed", "employed", "laborforce"]
MEASURE_LABELS = {
    "rate":       "Unemp Rate %",
    "unemployed": "Unemployed",
    "employed":   "Employed",
    "laborforce": "Labor Force",
}

# ---------------------------------------------------------------------------

def fetch(series_ids: list[str], start_year: int, end_year: int) -> dict:
    """POST to BLS API v2, return dict keyed by seriesID."""
    payload = {
        "seriesid": series_ids,
        "startyear": str(start_year),
        "endyear": str(end_year),
        "registrationkey": BLS_API_KEY,
        "catalog": True,
        "calculations": True,
        "annualaverage": False,
    }
    r = requests.post(f"{BLS_API_BASE}/timeseries/data/", json=payload, timeout=30)
    r.raise_for_status()
    body = r.json()
    if body["status"] != "REQUEST_SUCCEEDED":
        raise RuntimeError(f"BLS API error: {body['message']}")
    return {s["seriesID"]: s["data"] for s in body["Results"]["series"]}


def latest(data: list[dict]) -> dict | None:
    """Return the most recent non-null observation."""
    for obs in data:
        if obs["value"] != "-":
            return obs
    return None


def get_calc(obs: dict, window: str) -> str:
    """Pull a net_change or pct_change value safely."""
    try:
        return obs["calculations"]["net_changes"].get(window, "—")
    except (KeyError, TypeError):
        return "—"


def fmt_num(val: str, is_rate: bool = False) -> str:
    if val in (None, "—", "-"):
        return "—"
    try:
        n = float(val)
        return f"{n:.1f}%" if is_rate else f"{n:,.0f}"
    except ValueError:
        return val


def fmt_chg(val: str, is_rate: bool = False) -> str:
    if val in (None, "—", ""):
        return "—"
    try:
        n = float(val)
        sign = "+" if n > 0 else ""
        return f"{sign}{n:.1f}pp" if is_rate else f"{sign}{n:,.0f}"
    except ValueError:
        return val


def print_dashboard(results: dict):
    geo_keys = ["DC", "MD", "VA", "DCMSA"]
    now = datetime.now().strftime("%Y-%m-%d %H:%M")

    print()
    print("=" * 72)
    print(f"  DC / MD / VA  UNEMPLOYMENT DASHBOARD     pulled {now}")
    print("=" * 72)

    for geo in geo_keys:
        rate_series = SERIES[f"{geo}_rate"]
        rate_data   = results.get(rate_series, [])
        obs         = latest(rate_data)
        if not obs:
            print(f"\n  {LABELS[geo]}: no data\n")
            continue

        period = f"{obs['periodName']} {obs['year']}"
        rate   = fmt_num(obs["value"], is_rate=True)
        mom    = fmt_chg(get_calc(obs, "1"),  is_rate=True)
        yoy    = fmt_chg(get_calc(obs, "12"), is_rate=True)

        print()
        print(f"  {LABELS[geo]}  ({period})")
        print(f"  {'─' * 60}")
        print(f"  {'Unemployment Rate:':<28} {rate:>8}   MoM: {mom:>8}   YoY: {yoy:>8}")

        for measure in ["unemployed", "employed", "laborforce"]:
            sid  = SERIES[f"{geo}_{measure}"]
            data = results.get(sid, [])
            o    = latest(data)
            if not o:
                continue
            label = MEASURE_LABELS[measure]
            val   = fmt_num(o["value"])
            m     = fmt_chg(get_calc(o, "1"))
            y     = fmt_chg(get_calc(o, "12"))
            print(f"  {label + ':':<28} {val:>10}   MoM: {m:>8}   YoY: {y:>8}")

    print()
    print("=" * 72)
    print("  Note: Not seasonally adjusted. MoM/YoY = net change.")
    print("=" * 72)
    print()


def save_csv(results: dict, path: str):
    rows = []
    geo_keys = ["DC", "MD", "VA", "DCMSA"]

    for geo in geo_keys:
        for measure in MEASURES:
            key = f"{geo}_{measure}"
            sid = SERIES[key]
            data = results.get(sid, [])
            for obs in data:
                if obs["value"] == "-":
                    continue
                rows.append({
                    "geography": LABELS[geo],
                    "geo_code": geo,
                    "measure": measure,
                    "series_id": sid,
                    "year": obs["year"],
                    "period": obs["period"],
                    "period_name": obs["periodName"],
                    "value": obs["value"],
                    "mom_chg": get_calc(obs, "1"),
                    "qoq_chg": get_calc(obs, "3"),
                    "yoy_chg": get_calc(obs, "12"),
                })

    if not rows:
        print("No data to save.")
        return

    fieldnames = list(rows[0].keys())
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames)
        w.writeheader()
        w.writerows(rows)
    print(f"  Saved {len(rows)} rows → {path}")


def main():
    current_year = datetime.now().year
    start_year   = current_year - 4  # 5 years of history (within 20-yr v2 limit)

    print(f"Fetching {len(SERIES)} LAUS series ({start_year}–{current_year})...")

    # API allows 50 series per call; we have 16, so one call is fine
    results = fetch(list(SERIES.values()), start_year, current_year)

    print_dashboard(results)

    out_csv = "dc_md_va_unemployment.csv"
    save_csv(results, out_csv)


if __name__ == "__main__":
    main()