scripts/fetch_bea_schema.py

#!/usr/bin/env python3
"""
Fetch raw BEA (Bureau of Economic Analysis) data and output the schema.
Shows raw API response structure for NIPA GDP data.
"""

import asyncio
import json
import os
from datetime import datetime
from pathlib import Path

from dotenv import load_dotenv
import httpx

# Load environment variables
env_paths = [
    Path.home() / ".env",
    Path(__file__).parent.parent / ".env",
]
for env_path in env_paths:
    if env_path.exists():
        load_dotenv(env_path)
        break

BEA_API_KEY = os.getenv("BEA_API_KEY")
BEA_BASE_URL = "https://apps.bea.gov/api/data"


def print_table(title: str, rows: list, col_widths: list = None):
    """Print ASCII table."""
    if not rows:
        return

    # Calculate column widths
    if col_widths is None:
        col_widths = []
        for col in range(len(rows[0])):
            width = max(len(str(row[col])) for row in rows)
            col_widths.append(width)

    # Print header
    print(f"\n{title}")

    # Top border
    line = "┌" + "┬".join("─" * (w + 2) for w in col_widths) + "┐"
    print(line)

    # Header row
    header = rows[0]
    row_str = "│" + "│".join(f" {str(header[i]).ljust(col_widths[i])} " for i in range(len(header))) + "│"
    print(row_str)

    # Separator
    line = "├" + "┼".join("─" * (w + 2) for w in col_widths) + "┤"
    print(line)

    # Data rows
    for row in rows[1:]:
        row_str = "│" + "│".join(f" {str(row[i]).ljust(col_widths[i])} " for i in range(len(row))) + "│"
        print(row_str)

    # Bottom border
    line = "└" + "┴".join("─" * (w + 2) for w in col_widths) + "┘"
    print(line)


async def fetch_gdp_data() -> dict:
    """Fetch GDP data from BEA NIPA dataset."""
    if not BEA_API_KEY:
        return {"error": "BEA_API_KEY not configured"}

    try:
        async with httpx.AsyncClient() as client:
            params = {
                "UserID": BEA_API_KEY,
                "method": "GetData",
                "datasetname": "NIPA",
                "TableName": "T10101",  # Percent Change From Preceding Period in Real GDP
                "Frequency": "Q",        # Quarterly
                "Year": "X",             # All recent years
                "ResultFormat": "JSON"
            }
            response = await client.get(BEA_BASE_URL, params=params, timeout=15)
            return response.json()
    except Exception as e:
        return {"error": str(e)}


async def main():
    print("BEA Data Schema")
    print("=" * 60)
    print()
    print("Endpoint: https://apps.bea.gov/api/data")
    print("Dataset: NIPA (National Income and Product Accounts)")
    print("Table: T10101 (Percent Change From Preceding Period in Real GDP)")
    print()

    if not BEA_API_KEY:
        print("ERROR: BEA_API_KEY not configured")
        print("Add BEA_API_KEY to ~/.env file")
        print("Get free key at: https://apps.bea.gov/api/signup/")
        return

    print("Fetching GDP data...")
    data = await fetch_gdp_data()

    if "error" in data:
        print(f"ERROR: {data}")
        return

    beaapi = data.get("BEAAPI", {})
    results = beaapi.get("Results", {})

    if not results:
        print("ERROR: No results returned")
        return

    print()
    print("=" * 60)
    print()

    # Print raw API response structure
    print("Raw API Response Structure")
    print("-" * 40)

    # Request metadata
    request = beaapi.get("Request", {})
    rows = [["field", "value"]]
    rows.append(["RequestParam.DataSetName", request.get("RequestParam", [{}])[0].get("ParameterValue", "") if request.get("RequestParam") else ""])
    rows.append(["RequestParam.TableName", "T10101"])
    rows.append(["RequestParam.Frequency", "Q"])
    print_table("BEAAPI.Request", rows)

    # Results metadata
    rows = [["field", "description"]]
    rows.append(["Statistic", results.get("Statistic", "")])
    rows.append(["UTCProductionTime", results.get("UTCProductionTime", "")])
    rows.append(["Notes[]", "Array of data notes/descriptions"])
    rows.append(["Data[]", "Array of data observations"])
    print_table("BEAAPI.Results", rows)

    # Data row structure
    data_rows = results.get("Data", [])
    if data_rows:
        # Get a recent GDP row (LineNumber = 1 is Real GDP)
        gdp_rows = [r for r in data_rows if r.get("LineNumber") == "1"]
        gdp_rows.sort(key=lambda x: x.get("TimePeriod", ""), reverse=True)

        if gdp_rows:
            sample = gdp_rows[0]
            rows = [["field", "value"]]
            rows.append(["TableName", sample.get("TableName", "")])
            rows.append(["SeriesCode", sample.get("SeriesCode", "")])
            rows.append(["LineNumber", sample.get("LineNumber", "")])
            rows.append(["LineDescription", sample.get("LineDescription", "")])
            rows.append(["TimePeriod", sample.get("TimePeriod", "")])
            rows.append(["METRIC_NAME", sample.get("METRIC_NAME", "")])
            rows.append(["CL_UNIT", sample.get("CL_UNIT", "")])
            rows.append(["UNIT_MULT", sample.get("UNIT_MULT", "")])
            rows.append(["DataValue", sample.get("DataValue", "")])
            rows.append(["NoteRef", sample.get("NoteRef", "")])
            print_table("Data[0] (Row Structure)", rows)

    # Field descriptions
    print()
    print()
    print("Field Descriptions")
    print("-" * 40)

    rows = [["field", "description"]]
    rows.append(["TableName", "NIPA table identifier (T10101)"])
    rows.append(["SeriesCode", "BEA series code for the metric"])
    rows.append(["LineNumber", "Row number in the table (1 = Real GDP)"])
    rows.append(["LineDescription", "Human-readable metric name"])
    rows.append(["TimePeriod", "Time period (YYYYQN format, e.g., 2025Q3)"])
    rows.append(["METRIC_NAME", "Metric type (e.g., Percent Change)"])
    rows.append(["CL_UNIT", "Classification unit"])
    rows.append(["UNIT_MULT", "Unit multiplier"])
    rows.append(["DataValue", "The actual data value"])
    rows.append(["NoteRef", "Reference to notes array"])
    print_table("Field Descriptions", rows)

    # Recent GDP values
    if gdp_rows:
        print()
        print()
        print("Recent GDP Growth Data")
        print("-" * 40)

        rows = [["TimePeriod", "DataValue", "LineDescription"]]
        for row in gdp_rows[:6]:
            rows.append([
                row.get("TimePeriod", ""),
                row.get("DataValue", ""),
                row.get("LineDescription", "")[:40]
            ])
        print_table("Real GDP % Change (Latest)", rows)

    # Save raw JSON
    output_path = Path(__file__).parent.parent / "docs" / "bea_raw.json"
    with open(output_path, 'w') as f:
        json.dump(data, f, indent=2, default=str)
    print(f"\nRaw JSON saved to: {output_path}")


if __name__ == "__main__":
    asyncio.run(main())