fix: qualify agents_preprocess model refs with analysis dataset (#141)

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,20 +96,20 @@ 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,
         median_score,
         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

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

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