Global Inflation Tracker

import requests
import pandas as pd

from pyspark.sql.functions import (
    col,
    current_timestamp,
    to_date
)

# COMMAND ----------

SERIES = [
    "FPCPITOTLZGBRA",  # Brazil
    "FPCPITOTLZGUSA",  # United States
    "FPCPITOTLZGJPN",  # Japan
    "FPCPITOTLZGGBR"   # United Kingdom
]

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

all_data = []

# COMMAND ----------

for series_id in SERIES:

    params = {
        "series_id": series_id,
        "api_key": API_KEY,
        "file_type": "json"
    }

    response = requests.get(BASE_URL, params=params, timeout=30)
    response.raise_for_status()

    observations = response.json().get("observations", [])

    for row in observations:
        all_data.append({
            "series_id": series_id,
            "date": row.get("date"),
            "value": row.get("value")
        })

# COMMAND ----------

pdf = pd.DataFrame(all_data)
spark_df = spark.createDataFrame(pdf)

# COMMAND ----------

bronze_df = (
    spark_df
    .filter(col("value") != ".")
    .withColumn("date", to_date(col("date")))
    .withColumn("value", col("value").cast("double"))
    .withColumn("ingestion_timestamp", current_timestamp())
)

# COMMAND ----------

bronze_df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable("inflation.bronze.fred_global_inflation_raw")

from pyspark.sql.functions import (
    col,
    min as spark_min,
    max as spark_max,
    count,
    current_timestamp
)

# COMMAND ----------

BRONZE_TABLE = "inflation.bronze.fred_global_inflation_raw"
SILVER_TABLE = "inflation.silver.fred_global_inflation_silver"

spark.sql("CREATE SCHEMA IF NOT EXISTS inflation.silver")

bronze_df = spark.table(BRONZE_TABLE)

# COMMAND ----------

# Duplicate check
duplicate_check_df = (
    bronze_df
    .groupBy("series_id", "date")
    .agg(count("*").alias("duplicate_count"))
    .filter(col("duplicate_count") > 1)
)

dedup_df = bronze_df.dropDuplicates(["series_id", "date"])

# COMMAND ----------

# Align date ranges across all series
series_date_limits_df = (
    dedup_df
    .groupBy("series_id")
    .agg(
        spark_min("date").alias("series_min_date"),
        spark_max("date").alias("series_max_date"),
        count("*").alias("total_rows")
    )
)

global_date_limits = (
    series_date_limits_df
    .agg(
        spark_max("series_min_date").alias("global_min_date"),
        spark_min("series_max_date").alias("global_max_date")
    )
    .collect()[0]
)

global_min_date = global_date_limits["global_min_date"]
global_max_date = global_date_limits["global_max_date"]

# COMMAND ----------

silver_df = (
    dedup_df
    .filter(
        (col("date") >= global_min_date) &
        (col("date") <= global_max_date)
    )
    .withColumn("silver_processed_timestamp", current_timestamp())
)

# COMMAND ----------

silver_df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable(SILVER_TABLE)

from pyspark.sql.functions import col, when, current_timestamp

# COMMAND ----------

SILVER_TABLE = "inflation.silver.fred_global_inflation_silver"
GOLD_TABLE   = "inflation.gold.global_inflation_analytics"

spark.sql("CREATE SCHEMA IF NOT EXISTS inflation.gold")

silver_df = spark.table(SILVER_TABLE)

# COMMAND ----------

# Map FRED series IDs to country names
gold_df = (
    silver_df
    .withColumn(
        "countries",
        when(col("series_id") == "FPCPITOTLZGBRA", "Brazil")
        .when(col("series_id") == "FPCPITOTLZGUSA", "United States")
        .when(col("series_id") == "FPCPITOTLZGJPN", "Japan")
        .when(col("series_id") == "FPCPITOTLZGGBR", "United Kingdom")
        .otherwise("Unknown")
    )
    .withColumnRenamed("value", "inflation")
    .drop("series_id")
)

# COMMAND ----------

gold_df = (
    gold_df
    .select(
        "countries",
        "date",
        "inflation",
        "ingestion_timestamp",
        "silver_processed_timestamp"
    )
    .orderBy("countries", "date")
)

# COMMAND ----------

gold_df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable(GOLD_TABLE)