From 4580e25d19347461e421f318ee9fe0a4f4d409e5 Mon Sep 17 00:00:00 2001 From: Alex Noonan <48368867+C00ldudeNoonan@users.noreply.github.com> Date: Sun, 12 Jul 2026 08:00:33 -0400 Subject: [PATCH] fix: qualify agents_preprocess model refs with analysis dataset (#141) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reddit_daily_summary and query_data_for_findings queried dbt models (agent_reddit_posts_daily, agent_reddit_sentiment_trends, etc.) by bare table name. BigQueryWarehouseResource.execute_query resolves unqualified names against its default economics_raw dataset, but agents_preprocess models materialize into economics_analysis, so every run failed with 404 Not found. Qualify these references with default_dataset_for_schema( "economics_analysis") — the same environment-suffix helper used for the staging-dataset fix in #143 — and add caller tests asserting the qualification. Closes #141 Co-Authored-By: Claude Opus 4.8 --- .../analysis/data_points/data_point_finder.py | 30 +++++++++----- .../defs/analysis/news/news_summary_assets.py | 12 +++++- .../test_bigquery_parameterized_callers.py | 41 +++++++++++++++++++ 3 files changed, 70 insertions(+), 13 deletions(-) diff --git a/macro_agents/src/macro_agents/defs/analysis/data_points/data_point_finder.py b/macro_agents/src/macro_agents/defs/analysis/data_points/data_point_finder.py index 933c9b4..4935af2 100644 --- a/macro_agents/src/macro_agents/defs/analysis/data_points/data_point_finder.py +++ b/macro_agents/src/macro_agents/defs/analysis/data_points/data_point_finder.py @@ -9,7 +9,10 @@ import polars as pl -from macro_agents.defs.resources.bigquery_warehouse import BigQueryWarehouseResource +from macro_agents.defs.resources.bigquery_warehouse import ( + BigQueryWarehouseResource, + default_dataset_for_schema, +) def query_data_for_findings( @@ -26,8 +29,13 @@ def query_data_for_findings( Returns: Dict with DataFrames for: economic, market, commodity, correlation, sentiment """ + # These agent_* tables are dbt models in the agents_preprocess group, which + # materializes into economics_analysis — not the resource's default + # (economics_raw) dataset. Qualify every reference so it resolves correctly. + analysis_dataset = default_dataset_for_schema("economics_analysis") + # Economic indicators with 3m, 6m, 1y changes - economic_query = """ + economic_query = f""" SELECT series_code, series_name, @@ -37,7 +45,7 @@ def query_data_for_findings( pct_change_6m, pct_change_1y, date_grain - FROM agent_fred_series_latest_aggregates + FROM {analysis_dataset}.agent_fred_series_latest_aggregates WHERE month >= DATE_SUB( DATE_TRUNC(DATE(@week_start), MONTH), INTERVAL 12 MONTH @@ -48,7 +56,7 @@ def query_data_for_findings( """ # Market data from summary tables (12 weeks) - market_query = """ + market_query = f""" SELECT ticker AS symbol, time_period, @@ -57,12 +65,12 @@ def query_data_for_findings( win_rate_pct, worst_day_pct_change, best_day_pct_change - FROM agent_market_performance + FROM {analysis_dataset}.agent_market_performance WHERE time_period IN ('12_weeks', '26_weeks') """ # Commodity data from summary tables - commodity_query = """ + commodity_query = f""" SELECT commodity AS symbol, time_period, @@ -71,12 +79,12 @@ def query_data_for_findings( win_rate_pct, worst_day_pct_change, best_day_pct_change - FROM agent_commodity_performance + FROM {analysis_dataset}.agent_commodity_performance WHERE time_period IN ('12_weeks', '26_weeks') """ # Correlation data showing economic indicators vs forward returns - correlation_query = """ + correlation_query = f""" SELECT symbol, series_name, @@ -88,12 +96,12 @@ def query_data_for_findings( corr_econ_q3_returns, avg_q1_return_when_econ_growing, avg_q1_return_when_econ_declining - FROM agent_leading_econ_return_indicator + FROM {analysis_dataset}.agent_leading_econ_return_indicator WHERE observation_count >= 12 """ # Sentiment trends with momentum - sentiment_query = """ + sentiment_query = f""" SELECT date, avg_score, @@ -101,7 +109,7 @@ def query_data_for_findings( total_posts, weekly_avg_score, score_momentum_pct - FROM agent_reddit_sentiment_trends + FROM {analysis_dataset}.agent_reddit_sentiment_trends WHERE date >= DATE_SUB(DATE(@week_start), INTERVAL 3 MONTH) AND date <= DATE(@week_end) ORDER BY date DESC diff --git a/macro_agents/src/macro_agents/defs/analysis/news/news_summary_assets.py b/macro_agents/src/macro_agents/defs/analysis/news/news_summary_assets.py index 3479018..2e09769 100644 --- a/macro_agents/src/macro_agents/defs/analysis/news/news_summary_assets.py +++ b/macro_agents/src/macro_agents/defs/analysis/news/news_summary_assets.py @@ -9,7 +9,10 @@ import polars as pl from macro_agents.defs.analysis.news.news_summarizer import NewsSummarizerResource -from macro_agents.defs.resources.bigquery_warehouse import BigQueryWarehouseResource +from macro_agents.defs.resources.bigquery_warehouse import ( + BigQueryWarehouseResource, + default_dataset_for_schema, +) @dg.asset( @@ -41,6 +44,11 @@ def reddit_daily_summary( context.log.info(f"Generating daily Reddit summary for {partition_date}") + # agent_reddit_posts_daily is a dbt model in the agents_preprocess group, + # which materializes into economics_analysis — not the resource's default + # (economics_raw) dataset. Qualify the reference so it resolves correctly. + analysis_dataset = default_dataset_for_schema("economics_analysis") + # Fetch all posts for this date across all subreddits query = f""" SELECT @@ -50,7 +58,7 @@ def reddit_daily_summary( subreddit, author, url - FROM agent_reddit_posts_daily + FROM {analysis_dataset}.agent_reddit_posts_daily WHERE partition_date = '{partition_date}' ORDER BY score DESC LIMIT 100 diff --git a/macro_agents/tests/test_bigquery_parameterized_callers.py b/macro_agents/tests/test_bigquery_parameterized_callers.py index 215e749..4c99624 100644 --- a/macro_agents/tests/test_bigquery_parameterized_callers.py +++ b/macro_agents/tests/test_bigquery_parameterized_callers.py @@ -2,6 +2,7 @@ from pathlib import Path from unittest.mock import Mock +import dagster as dg import polars as pl from macro_agents.defs.analysis.data_points.data_point_finder import ( @@ -13,6 +14,8 @@ from macro_agents.defs.analysis.investments.investment_recommendation_charts import ( _fetch_recommendations_row, ) +from macro_agents.defs.analysis.news.news_summary_assets import reddit_daily_summary +from macro_agents.defs.resources.bigquery_warehouse import default_dataset_for_schema def test_raw_query_calls_do_not_pass_positional_parameters() -> None: @@ -98,3 +101,41 @@ def test_data_point_queries_use_named_date_parameters() -> None: assert "DATE(@week_start)" in sentiment_call.args[0] assert "DATE(@week_end)" in sentiment_call.args[0] assert sentiment_call.kwargs["params"] == expected_params + + +def test_data_point_queries_qualify_agents_preprocess_dataset() -> None: + """agent_* models live in economics_analysis, not the resource's default + economics_raw dataset — every reference must be dataset-qualified (issue #141). + """ + bq = Mock() + bq.execute_query.return_value = pl.DataFrame() + + query_data_for_findings(bq, "2026-07-06", "2026-07-12") + + analysis_dataset = default_dataset_for_schema("economics_analysis") + expected_tables = { + f"{analysis_dataset}.agent_fred_series_latest_aggregates", + f"{analysis_dataset}.agent_market_performance", + f"{analysis_dataset}.agent_commodity_performance", + f"{analysis_dataset}.agent_leading_econ_return_indicator", + f"{analysis_dataset}.agent_reddit_sentiment_trends", + } + queries = "\n".join(call.args[0] for call in bq.execute_query.call_args_list) + for table in expected_tables: + assert table in queries, f"unqualified reference to {table}" + + +def test_reddit_daily_summary_qualifies_agents_preprocess_dataset() -> None: + """reddit_daily_summary reads the agent_reddit_posts_daily dbt model, which + materializes into economics_analysis rather than the resource default (#141). + """ + bq = Mock() + bq.execute_query.return_value = pl.DataFrame() # empty -> early return + news_summarizer = Mock() + + context = dg.build_asset_context(partition_key="2026-07-11") + reddit_daily_summary(context, news_summarizer, bq) + + analysis_dataset = default_dataset_for_schema("economics_analysis") + query = bq.execute_query.call_args.args[0] + assert f"FROM {analysis_dataset}.agent_reddit_posts_daily" in query