diff --git a/dbt_project/macros/data_quality_helpers.sql b/dbt_project/macros/data_quality_helpers.sql index 5465551..eec2f3e 100644 --- a/dbt_project/macros/data_quality_helpers.sql +++ b/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', diff --git a/dbt_project/models/markets/technical/schema.yml b/dbt_project/models/markets/technical/schema.yml index 31a0284..2f32494 100644 --- a/dbt_project/models/markets/technical/schema.yml +++ b/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: diff --git a/dbt_project/models/markets/technical/technical_price_universe.sql b/dbt_project/models/markets/technical/technical_price_universe.sql index 76e1021..113a5fb 100644 --- a/dbt_project/models/markets/technical/technical_price_universe.sql +++ b/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'}, ] %} diff --git a/dbt_project/models/semantic_layer/current_data_coverage.sql b/dbt_project/models/semantic_layer/current_data_coverage.sql index e903db5..8a6427d 100644 --- a/dbt_project/models/semantic_layer/current_data_coverage.sql +++ b/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, diff --git a/dbt_project/models/sources.yml b/dbt_project/models/sources.yml index 11d09f5..b6ba519 100644 --- a/dbt_project/models/sources.yml +++ b/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 diff --git a/dbt_project/models/staging/schema.yml b/dbt_project/models/staging/schema.yml index 87a9b4c..9e98b78 100644 --- a/dbt_project/models/staging/schema.yml +++ b/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 diff --git a/dbt_project/models/staging/stg_commodity_etfs.sql b/dbt_project/models/staging/stg_commodity_etfs.sql new file mode 100644 index 0000000..a05fabe --- /dev/null +++ b/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') }} diff --git a/macro_agents/src/macro_agents/defs/constants/market_stack_constants.py b/macro_agents/src/macro_agents/defs/constants/market_stack_constants.py index a3a8811..e2becb3 100644 --- a/macro_agents/src/macro_agents/defs/constants/market_stack_constants.py +++ b/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 diff --git a/macro_agents/src/macro_agents/defs/domains/markets/assets.py b/macro_agents/src/macro_agents/defs/domains/markets/assets.py index c539bb5..0585e0a 100644 --- a/macro_agents/src/macro_agents/defs/domains/markets/assets.py +++ b/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"}, diff --git a/macro_agents/src/macro_agents/defs/domains/markets/checks.py b/macro_agents/src/macro_agents/defs/domains/markets/checks.py index db2ff08..d7ff9ec 100644 --- a/macro_agents/src/macro_agents/defs/domains/markets/checks.py +++ b/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, diff --git a/macro_agents/src/macro_agents/defs/domains/markets/data_quality_assets.py b/macro_agents/src/macro_agents/defs/domains/markets/data_quality_assets.py index 46e5256..7873b75 100644 --- a/macro_agents/src/macro_agents/defs/domains/markets/data_quality_assets.py +++ b/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", diff --git a/macro_agents/src/macro_agents/defs/domains/markets/defs.py b/macro_agents/src/macro_agents/defs/domains/markets/defs.py index b067f81..3987659 100644 --- a/macro_agents/src/macro_agents/defs/domains/markets/defs.py +++ b/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, diff --git a/macro_agents/src/macro_agents/defs/domains/markets/jobs.py b/macro_agents/src/macro_agents/defs/domains/markets/jobs.py index 1e01bd7..423cd53 100644 --- a/macro_agents/src/macro_agents/defs/domains/markets/jobs.py +++ b/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}, diff --git a/macro_agents/src/macro_agents/defs/domains/markets/partitions.py b/macro_agents/src/macro_agents/defs/domains/markets/partitions.py index 81e5f2b..1ddc7da 100644 --- a/macro_agents/src/macro_agents/defs/domains/markets/partitions.py +++ b/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} ) diff --git a/macro_agents/tests/test_market_checks.py b/macro_agents/tests/test_market_checks.py new file mode 100644 index 0000000..8baf803 --- /dev/null +++ b/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 diff --git a/macro_agents/tests/test_schedules.py b/macro_agents/tests/test_schedules.py index 25ad059..419465d 100644 --- a/macro_agents/tests/test_schedules.py +++ b/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",