scripts/build_bls_labor_indicators.py

import csv
import json
import urllib.request
from pathlib import Path

OUT_DIR = Path("data/bls")
OUT_DIR.mkdir(parents=True, exist_ok=True)

START_YEAR = "2020"
END_YEAR = "2026"

# Small useful starter set for business/HR/workforce AI tools.
SERIES = {
    "LNS14000000": "Unemployment Rate - Civilian labor force",
    "LNS11300000": "Labor Force Participation Rate",
    "LNS12300000": "Employment-Population Ratio",
    "CES0000000001": "Total Nonfarm Employment",
    "CES0500000003": "Average Hourly Earnings - Total Private",
    "JTS000000000000000JOL": "Job Openings - Total Nonfarm",
    "JTS000000000000000HIL": "Hires - Total Nonfarm",
    "JTS000000000000000QUL": "Quits - Total Nonfarm",
}

URL = "https://api.bls.gov/publicAPI/v2/timeseries/data/"

payload = {
    "seriesid": list(SERIES.keys()),
    "startyear": START_YEAR,
    "endyear": END_YEAR,
}

print("Downloading BLS labor market indicators...")
request = urllib.request.Request(
    URL,
    data=json.dumps(payload).encode("utf-8"),
    headers={
        "Content-Type": "application/json",
        "User-Agent": "Realigns Inc support@realignsinc.com",
    },
    method="POST",
)

with urllib.request.urlopen(request, timeout=60) as response:
    raw = response.read().decode("utf-8", errors="replace")

data = json.loads(raw)

if data.get("status") != "REQUEST_SUCCEEDED":
    print(json.dumps(data, indent=2))
    raise SystemExit("BLS API request failed.")

def to_float(value):
    if value in (None, "", "-", "N/A", "null"):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None

records = []

for series in data.get("Results", {}).get("series", []):
    series_id = series.get("seriesID")
    series_name = SERIES.get(series_id, series_id)

    for item in series.get("data", []):
        period = item.get("period")
        year = item.get("year")
        value = item.get("value")

        records.append({
            "series_id": series_id,
            "series_name": series_name,
            "year": int(year) if year else None,
            "period": period,
            "period_name": item.get("periodName"),
            "value": to_float(value),
            "footnotes": "; ".join(
                note.get("text", "") for note in item.get("footnotes", []) if note.get("text")
            ),
            "source": "U.S. Bureau of Labor Statistics Public Data API",
            "license": "U.S. Government public data"
        })

jsonl_path = OUT_DIR / "bls_labor_market_indicators_2020_2026.jsonl"
csv_path = OUT_DIR / "bls_labor_market_indicators_2020_2026.csv"

with jsonl_path.open("w", encoding="utf-8") as f:
    for record in records:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

with csv_path.open("w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=list(records[0].keys()))
    writer.writeheader()
    writer.writerows(records)

print(f"Done. Records: {len(records)}")
print(f"Saved: {jsonl_path}")
print(f"Saved: {csv_path}")