beanflows/transform/sqlmesh_beanflows/models/serving/obt_commodity_metrics.sql

MODEL (
  name serving.commodity_metrics,
  kind INCREMENTAL_BY_TIME_RANGE (
    time_column ingest_date
  ),
  start '2006-08-01',
  cron '@daily'
);

/* CTE to calculate country-level derived metrics */
WITH country_metrics AS (
  SELECT
    commodity_code,
    commodity_name,
    country_code,
    country_name,
    market_year,
    ingest_date,
    Production,
    Imports,
    Exports,
    Total_Distribution,
    Ending_Stocks,
    (
      Production + Imports - Exports
    ) AS Net_Supply, /* Derived metrics per country, mirroring Python script */
    (
      Exports - Imports
    ) AS Trade_Balance,
    (
      Production + Imports - Exports
    ) - Total_Distribution AS Supply_Demand_Balance,
    (
      Ending_Stocks / NULLIF(Total_Distribution, 0)
    ) /* Handle division by zero for Stock-to-Use Ratio */ * 100 AS Stock_to_Use_Ratio_pct,
    (
      Production - LAG(Production, 1, 0) OVER (PARTITION BY commodity_code, country_code ORDER BY market_year, ingest_date)
    ) /* Calculate Production YoY percentage change using a window function */ / NULLIF(
      LAG(Production, 1, 0) OVER (PARTITION BY commodity_code, country_code ORDER BY market_year, ingest_date),
      0
    ) * 100 AS Production_YoY_pct
  FROM cleaned.psdalldata__commodity_pivoted
), global_aggregates AS (
  SELECT
    commodity_code,
    commodity_name,
    NULL::TEXT AS country_code, /* Use NULL for global aggregates */
    'Global' AS country_name,
    market_year,
    ingest_date,
    SUM(Production) AS Production,
    SUM(Imports) AS Imports,
    SUM(Exports) AS Exports,
    SUM(Total_Distribution) AS Total_Distribution,
    SUM(Ending_Stocks) AS Ending_Stocks
  FROM cleaned.psdalldata__commodity_pivoted
  GROUP BY
    commodity_code,
    commodity_name,
    market_year,
    ingest_date
), global_metrics /* CTE to calculate derived metrics for global aggregates */ AS (
  SELECT
    commodity_code,
    commodity_name,
    country_code,
    country_name,
    market_year,
    ingest_date,
    Production,
    Imports,
    Exports,
    Total_Distribution,
    Ending_Stocks,
    (
      Production + Imports - Exports
    ) AS Net_Supply,
    (
      Exports - Imports
    ) AS Trade_Balance,
    (
      Production + Imports - Exports
    ) - Total_Distribution AS Supply_Demand_Balance,
    (
      Ending_Stocks / NULLIF(Total_Distribution, 0)
    ) * 100 AS Stock_to_Use_Ratio_pct,
    (
      Production - LAG(Production, 1, 0) OVER (PARTITION BY commodity_code ORDER BY market_year, ingest_date)
    ) / NULLIF(
      LAG(Production, 1, 0) OVER (PARTITION BY commodity_code ORDER BY market_year, ingest_date),
      0
    ) * 100 AS Production_YoY_pct
  FROM global_aggregates
)
/* Combine country-level and global-level data into a single output */
SELECT
  commodity_code,
  commodity_name,
  country_code,
  country_name,
  market_year,
  ingest_date,
  Production,
  Imports,
  Exports,
  Total_Distribution,
  Ending_Stocks,
  Net_Supply,
  Trade_Balance,
  Supply_Demand_Balance,
  Stock_to_Use_Ratio_pct,
  Production_YoY_pct
FROM (
  SELECT
    *
  FROM country_metrics
  UNION ALL
  SELECT
    *
  FROM global_metrics
) AS combined_data
ORDER BY
  commodity_name,
  country_name,
  market_year,
  ingest_date