feat: add commodity ETF ingestion universe

dbt_project/macros/data_quality_helpers.sql

@@ -2,6 +2,7 @@
     {% set tables = [
         'us_sector_etfs_raw',
         'currency_etfs_raw',
+        'commodity_etfs_raw',
         'major_indices_raw',
         'fixed_income_etfs_raw',
         'global_markets_raw',

dbt_project/models/markets/technical/schema.yml

@@ -25,6 +25,7 @@ models:
                   - major_index
                   - fixed_income_etf
                   - currency_etf
+                  - commodity_etf
                   - global_market
       - name: symbol
         tests:

dbt_project/models/markets/technical/technical_price_universe.sql

@@ -23,6 +23,7 @@
     {'model': 'stg_major_indices', 'universe': 'major_index'},
     {'model': 'stg_fixed_income', 'universe': 'fixed_income_etf'},
     {'model': 'stg_currency', 'universe': 'currency_etf'},
+    {'model': 'stg_commodity_etfs', 'universe': 'commodity_etf'},
     {'model': 'stg_global_markets', 'universe': 'global_market'},
 ] %}
 

dbt_project/models/semantic_layer/current_data_coverage.sql

@@ -6,6 +6,7 @@ WITH source_specs AS (
         STRUCT('sp500_companies_prices_raw' AS source_name, 'markets' AS source_domain, 'daily_market_prices' AS grain, 31 AS lookback_days, 5 AS freshness_warn_days, 10 AS freshness_error_days),
         STRUCT('us_sector_etfs_raw' AS source_name, 'markets' AS source_domain, 'daily_market_prices' AS grain, 31 AS lookback_days, 5 AS freshness_warn_days, 10 AS freshness_error_days),
         STRUCT('currency_etfs_raw' AS source_name, 'markets' AS source_domain, 'daily_market_prices' AS grain, 31 AS lookback_days, 5 AS freshness_warn_days, 10 AS freshness_error_days),
+        STRUCT('commodity_etfs_raw' AS source_name, 'markets' AS source_domain, 'daily_market_prices' AS grain, 31 AS lookback_days, 5 AS freshness_warn_days, 10 AS freshness_error_days),
         STRUCT('major_indices_raw' AS source_name, 'markets' AS source_domain, 'daily_market_prices' AS grain, 31 AS lookback_days, 5 AS freshness_warn_days, 10 AS freshness_error_days),
         STRUCT('fixed_income_etfs_raw' AS source_name, 'markets' AS source_domain, 'daily_market_prices' AS grain, 31 AS lookback_days, 5 AS freshness_warn_days, 10 AS freshness_error_days),
         STRUCT('global_markets_raw' AS source_name, 'markets' AS source_domain, 'daily_market_prices' AS grain, 31 AS lookback_days, 5 AS freshness_warn_days, 10 AS freshness_error_days),
@@ -44,6 +45,15 @@ raw_source_observations AS (
 
     UNION ALL
 
+    SELECT
+        'commodity_etfs_raw' AS source_name,
+        symbol AS entity_id,
+        date AS observation_date
+    FROM {{ ref('stg_commodity_etfs') }}
+    WHERE symbol IS NOT NULL AND date IS NOT NULL
+
+    UNION ALL
+
     SELECT
         'major_indices_raw' AS source_name,
         symbol AS entity_id,

dbt_project/models/sources.yml

@@ -53,6 +53,12 @@ sources:
           freshness:
             warn_after: {count: 5, period: day}
             error_after: {count: 10, period: day}
+      - name: commodity_etfs_raw
+        config:
+          loaded_at_field: date
+          freshness:
+            warn_after: {count: 5, period: day}
+            error_after: {count: 10, period: day}
       - name: major_indices_raw
         config:
           loaded_at_field: date

dbt_project/models/staging/schema.yml

@@ -78,6 +78,25 @@ models:
         tests:
           - not_null
 
+  - name: stg_commodity_etfs
+    description: >
+      raw data from MarketStack API for Commodity ETFs
+    tests:
+      - ohlc_consistency:
+          config:
+            severity: warn
+      - usd_currency_only:
+          column_name: price_currency
+          config:
+            severity: warn
+    columns:
+      - name: date
+        tests:
+          - not_null
+      - name: symbol
+        tests:
+          - not_null
+
   - name: stg_major_indices
     description: >
       raw data from MarketStack API for Major Indices

dbt_project/models/staging/stg_commodity_etfs.sql

@@ -0,0 +1,21 @@
+SELECT
+    open,
+    high,
+    low,
+    close,
+    volume,
+    adj_high,
+    adj_low,
+    adj_close,
+    adj_open,
+    adj_volume,
+    split_factor,
+    dividend,
+    name,
+    exchange_code,
+    asset_type,
+    price_currency,
+    symbol,
+    exchange,
+    SAFE_CAST(SUBSTR(CAST(date AS STRING), 1, 10) AS DATE) AS date
+FROM {{ source('staging', 'commodity_etfs_raw') }}

macro_agents/src/macro_agents/defs/constants/market_stack_constants.py

@@ -54,6 +54,22 @@
     # IBIT.INDX removed - MarketStack returns 422 (ticker dot notation unsupported)
 ]
 
+# Commodity ETFs spanning precious metals, energy, agriculture, and broad baskets.
+COMMODITY_ETFS = [
+    "GLD",  # SPDR Gold Shares
+    "SLV",  # iShares Silver Trust
+    "USO",  # United States Oil Fund
+    "UNG",  # United States Natural Gas Fund
+    "DBA",  # Invesco DB Agriculture Fund
+    "DBC",  # Invesco DB Commodity Index Tracking Fund
+    "CORN",  # Teucrium Corn Fund
+    "WEAT",  # Teucrium Wheat Fund
+    "SOYB",  # Teucrium Soybean Fund
+    "CPER",  # United States Copper Index Fund
+    "PPLT",  # abrdn Physical Platinum Shares ETF
+    "PALL",  # abrdn Physical Palladium Shares ETF
+]
+
 # Energy commodities (requires premium plan)
 ENERGY_COMMODITIES = [
     "brent",
@@ -123,6 +139,7 @@
     MAJOR_INDICES_TICKERS,
     US_SECTOR_ETFS,
     FIXED_INCOME_ETFS,
+    COMMODITY_ETFS,
 ]
 
 ALL_COMMODITIES = ENERGY_COMMODITIES + INPUT_COMMODITIES + AGRICULTURE_COMMODITIES

macro_agents/src/macro_agents/defs/domains/markets/assets.py

@@ -11,6 +11,7 @@
     COMMODITIES_GROUP,
     MARKETS_GROUP,
     agriculture_commodities_partitions,
+    commodity_etfs_partitions,
     currency_etfs_partitions,
     energy_commodities_partitions,
     fixed_income_etfs_partitions,
@@ -495,6 +496,24 @@ def currency_etfs_raw(
     return _fetch_ticker_partitions(context, bq, marketstack, "currency_etfs_raw")
 
 
+@dg.asset(
+    group_name=MARKETS_GROUP,
+    kinds={"polars", "duckdb"},
+    partitions_def=commodity_etfs_partitions,
+    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=50),
+    automation_condition=dg.AutomationCondition.on_cron(
+        "45 18 * * 5", "America/New_York"
+    ),
+    description="Raw data from MarketStack API for Commodity ETFs",
+)
+def commodity_etfs_raw(
+    context: dg.AssetExecutionContext,
+    bq: BigQueryWarehouseResource,
+    marketstack: MarketStackResource,
+) -> dg.MaterializeResult:
+    return _fetch_ticker_partitions(context, bq, marketstack, "commodity_etfs_raw")
+
+
 @dg.asset(
     group_name=MARKETS_GROUP,
     kinds={"polars", "duckdb"},

macro_agents/src/macro_agents/defs/domains/markets/checks.py

@@ -4,6 +4,7 @@
 
 from macro_agents.defs.domains.markets.assets import (
     agriculture_commodities_raw,
+    commodity_etfs_raw,
     currency_etfs_raw,
     energy_commodities_raw,
     fixed_income_etfs_raw,
@@ -44,19 +45,16 @@ def check_weekly_data_coverage(
     query = f"""
     WITH date_range AS (
         SELECT DISTINCT
-            DATE_TRUNC('week', date_col) as week_start
-        FROM (
-            SELECT CAST('{one_year_ago}' AS DATE) + (INTERVAL '1 day' * days) as date_col
-            FROM (SELECT * FROM generate_series(0, 365)) AS t(days)
-        )
+            DATE_TRUNC(date_col, WEEK(MONDAY)) AS week_start
+        FROM UNNEST(GENERATE_DATE_ARRAY(DATE('{one_year_ago}'), DATE('{today}'))) AS date_col
     ),
     data_weeks AS (
         SELECT DISTINCT
             {partition_column},
-            DATE_TRUNC('week', CAST({date_column} AS DATE)) as week_start
+            DATE_TRUNC(SAFE_CAST({date_column} AS DATE), WEEK(MONDAY)) AS week_start
         FROM {table_name}
-        WHERE CAST({date_column} AS DATE) >= '{one_year_ago}'
-            AND CAST({date_column} AS DATE) <= '{today}'
+        WHERE SAFE_CAST({date_column} AS DATE) >= DATE('{one_year_ago}')
+            AND SAFE_CAST({date_column} AS DATE) <= DATE('{today}')
     ),
     expected_weeks AS (
         SELECT DISTINCT
@@ -69,7 +67,7 @@ def check_weekly_data_coverage(
         SELECT 
             ew.{partition_column},
             ew.week_start,
-            CAST(ew.week_start AS VARCHAR) || ' to ' || CAST(ew.week_start + INTERVAL '6 days' AS VARCHAR) as week_range
+            CAST(ew.week_start AS STRING) || ' to ' || CAST(DATE_ADD(ew.week_start, INTERVAL 6 DAY) AS STRING) AS week_range
         FROM expected_weeks ew
         LEFT JOIN data_weeks dw 
             ON ew.{partition_column} = dw.{partition_column} 
@@ -168,6 +166,19 @@ def currency_etfs_weekly_coverage_check(
     )
 
 
+@dg.asset_check(asset=commodity_etfs_raw)
+def commodity_etfs_weekly_coverage_check(
+    bq: BigQueryWarehouseResource,
+) -> dg.AssetCheckResult:
+    """Check if every ticker has at least one value per week in the last year."""
+    return check_weekly_data_coverage(
+        "commodity_etfs_raw",
+        "symbol",
+        "date",
+        bq,
+    )
+
+
 @dg.asset_check(asset=major_indices_raw)
 def major_indices_weekly_coverage_check(
     bq: BigQueryWarehouseResource,

macro_agents/src/macro_agents/defs/domains/markets/data_quality_assets.py

@@ -98,6 +98,7 @@ def dq_apply_corrections(
         is_commodity = "commodities" in source_table and source_table not in (
             "us_sector_etfs_raw",
             "currency_etfs_raw",
+            "commodity_etfs_raw",
             "major_indices_raw",
             "fixed_income_etfs_raw",
             "global_markets_raw",

macro_agents/src/macro_agents/defs/domains/markets/defs.py

@@ -6,6 +6,7 @@
 )
 from macro_agents.defs.domains.markets.assets import (
     agriculture_commodities_raw,
+    commodity_etfs_raw,
     currency_etfs_raw,
     energy_commodities_raw,
     fixed_income_etfs_raw,
@@ -19,6 +20,7 @@
 )
 from macro_agents.defs.domains.markets.checks import (
     agriculture_commodities_weekly_coverage_check,
+    commodity_etfs_weekly_coverage_check,
     currency_etfs_weekly_coverage_check,
     energy_commodities_weekly_coverage_check,
     fixed_income_etfs_weekly_coverage_check,
@@ -30,6 +32,7 @@
 )
 from macro_agents.defs.domains.markets.jobs import (
     agriculture_commodities_ingestion_job,
+    commodity_etfs_ingestion_job,
     currency_etfs_ingestion_job,
     energy_commodities_ingestion_job,
     fixed_income_etfs_ingestion_job,
@@ -68,6 +71,7 @@
     sp500_companies_list_job,
     us_sector_etfs_ingestion_job,
     currency_etfs_ingestion_job,
+    commodity_etfs_ingestion_job,
     major_indices_ingestion_job,
     fixed_income_etfs_ingestion_job,
     global_markets_ingestion_job,
@@ -83,6 +87,7 @@
     assets=[
         us_sector_etfs_raw,
         currency_etfs_raw,
+        commodity_etfs_raw,
         major_indices_raw,
         fixed_income_etfs_raw,
         global_markets_raw,
@@ -98,6 +103,7 @@
     asset_checks=[
         us_sector_etfs_weekly_coverage_check,
         currency_etfs_weekly_coverage_check,
+        commodity_etfs_weekly_coverage_check,
         major_indices_weekly_coverage_check,
         fixed_income_etfs_weekly_coverage_check,
         global_markets_weekly_coverage_check,

macro_agents/src/macro_agents/defs/domains/markets/jobs.py

@@ -28,6 +28,13 @@
     description="Currency ETFs ingestion job - runs current month partition weekly on Fridays",
 )
 
+commodity_etfs_ingestion_job = dg.define_asset_job(
+    name="commodity_etfs_ingestion_job",
+    tags={"dagster/priority": "10", "dagster/max_runtime": 1800},
+    selection=dg.AssetSelection.assets("commodity_etfs_raw"),
+    description="Commodity ETFs ingestion job - runs current month partition weekly on Fridays",
+)
+
 major_indices_ingestion_job = dg.define_asset_job(
     name="major_indices_ingestion_job",
     tags={"dagster/priority": "10", "dagster/max_runtime": 1800},

macro_agents/src/macro_agents/defs/domains/markets/partitions.py

@@ -4,6 +4,7 @@
 
 from macro_agents.defs.constants.market_stack_constants import (
     AGRICULTURE_COMMODITIES,
+    COMMODITY_ETFS,
     CURRENCY_ETFS,
     ENERGY_COMMODITIES,
     FIXED_INCOME_ETFS,
@@ -25,6 +26,7 @@
 
 us_sector_etfs_static = dg.StaticPartitionsDefinition(US_SECTOR_ETFS)
 currency_etfs_static = dg.StaticPartitionsDefinition(CURRENCY_ETFS)
+commodity_etfs_static = dg.StaticPartitionsDefinition(COMMODITY_ETFS)
 major_indices_tickers_static = dg.StaticPartitionsDefinition(MAJOR_INDICES_TICKERS)
 fixed_income_etfs_static = dg.StaticPartitionsDefinition(FIXED_INCOME_ETFS)
 global_markets_static = dg.StaticPartitionsDefinition(GLOBAL_MARKETS)
@@ -37,6 +39,10 @@
     {"ticker": currency_etfs_static, "date": monthly_partitions}
 )
 
+commodity_etfs_partitions = dg.MultiPartitionsDefinition(
+    {"ticker": commodity_etfs_static, "date": monthly_partitions}
+)
+
 major_indices_tickers_partitions = dg.MultiPartitionsDefinition(
     {"ticker": major_indices_tickers_static, "date": monthly_partitions}
 )

macro_agents/tests/test_market_checks.py

@@ -0,0 +1,24 @@
+from unittest.mock import Mock
+
+import polars as pl
+
+from macro_agents.defs.domains.markets.checks import check_weekly_data_coverage
+
+
+def test_weekly_data_coverage_query_uses_bigquery_date_array():
+    bq = Mock()
+    bq.table_exists.return_value = True
+    bq.execute_query.return_value = pl.DataFrame()
+
+    result = check_weekly_data_coverage(
+        "commodity_etfs_raw",
+        "symbol",
+        "date",
+        bq,
+    )
+
+    assert result.passed
+    query = bq.execute_query.call_args.args[0]
+    assert "GENERATE_DATE_ARRAY" in query
+    assert "DATE_TRUNC(date_col, WEEK(MONDAY))" in query
+    assert "generate_series" not in query

macro_agents/tests/test_schedules.py

@@ -125,6 +125,7 @@ def test_job_creation(self):
         assert "interesting_data_points_job" in job_names
         assert "us_sector_etfs_ingestion_job" in job_names
         assert "currency_etfs_ingestion_job" in job_names
+        assert "commodity_etfs_ingestion_job" in job_names
         assert "major_indices_ingestion_job" in job_names
         assert "fixed_income_etfs_ingestion_job" in job_names
         assert "global_markets_ingestion_job" in job_names
@@ -169,6 +170,7 @@ def test_job_asset_selection(self):
         assert "interesting_data_points_job" in job_names
         assert "us_sector_etfs_ingestion_job" in job_names
         assert "currency_etfs_ingestion_job" in job_names
+        assert "commodity_etfs_ingestion_job" in job_names
         assert "major_indices_ingestion_job" in job_names
         assert "fixed_income_etfs_ingestion_job" in job_names
         assert "global_markets_ingestion_job" in job_names
@@ -288,6 +290,7 @@ def test_definitions_include_jobs(self):
             "interesting_data_points_job",
             "us_sector_etfs_ingestion_job",
             "currency_etfs_ingestion_job",
+            "commodity_etfs_ingestion_job",
             "major_indices_ingestion_job",
             "fixed_income_etfs_ingestion_job",
             "global_markets_ingestion_job",