fix: qualify stg_* table references with the staging dataset

macro_agents/src/macro_agents/defs/resources/bigquery_warehouse.py

@@ -409,6 +409,29 @@ def get_unique_categories(self, table_name: str, column: str) -> list[str]:
         return df[column].to_list() if column in df.columns else []
 
 
+def default_dataset_for_schema(schema: str) -> str:
+    """Environment-suffixed dataset name for a given dbt schema base name.
+
+    Mirrors dbt's generate_schema_name macro (dbt_project/macros/
+    generate_schema_name.sql): prod uses the schema name as-is, staging/dev
+    append a suffix. Use this to reference dbt-managed models living in a
+    schema other than a resource's own default dataset — e.g. staging-schema
+    `stg_*` views queried by a job whose BigQueryWarehouseResource defaults
+    to the raw dataset. Computed at call time (not module load) so env
+    overrides in tests take effect.
+
+    Reads DBT_TARGET first — that's what dbt's own profiles.yml
+    (`target: "{{ env_var('DBT_TARGET', 'dev') }}"`) and generate_schema_name
+    actually key off — falling back to ENVIRONMENT only when DBT_TARGET is
+    unset. The two can diverge (e.g. ENVIRONMENT=prod with DBT_TARGET=staging
+    for an ad hoc dbt run against the production deployment), and the
+    dataset a dbt model is actually materialized into follows DBT_TARGET.
+    """
+    target = os.getenv("DBT_TARGET") or os.getenv("ENVIRONMENT", "dev")
+    suffix = {"prod": "", "staging": "_staging"}.get(target, "_dev")
+    return f"{schema}{suffix}"
+
+
 _environment = os.getenv("ENVIRONMENT", "dev")
 
 # Dataset suffix mirrors the dbt generate_schema_name macro:

macro_agents/src/macro_agents/defs/signals/entropy_complexity.py

@@ -19,7 +19,10 @@
 import numpy as np
 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,
+)
 
 
 SIGNALS_GROUP = "computed_signals"
@@ -66,10 +69,14 @@ def entropy_complexity_signals(
     bq: BigQueryWarehouseResource,
 ) -> dg.MaterializeResult:
     context.log.info("Fetching SPY and QQQ prices...")
+    # stg_major_indices is a dbt staging view living in the
+    # economics_staging schema, not this resource's default
+    # (economics_raw) dataset.
+    staging_dataset = default_dataset_for_schema("economics_staging")
     prices_df = bq.execute_query(
-        """
+        f"""
         SELECT date, symbol, adj_close
-        FROM stg_major_indices
+        FROM {staging_dataset}.stg_major_indices
         WHERE symbol IN ('SPY', 'QQQ')
           AND adj_close IS NOT NULL
           AND date >= CURRENT_DATE - INTERVAL 3 YEAR

macro_agents/src/macro_agents/defs/signals/fear_greed_composite.py

@@ -18,7 +18,10 @@
 import numpy as np
 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,
+)
 
 
 SIGNALS_GROUP = "computed_signals"
@@ -56,13 +59,17 @@ def fear_greed_signals(
     # the DuckDB/Postgres quoted-string form (`INTERVAL '3 years'`).
     lookback_years = 3
 
+    # stg_* models are dbt staging views living in the economics_staging
+    # schema, not this resource's default (economics_raw) dataset.
+    staging_dataset = default_dataset_for_schema("economics_staging")
+
     # 1. Market Momentum: SPY vs 125-day MA
     context.log.info("Computing market momentum component...")
     spy_df = bq.execute_query(
         f"""
         SELECT date, adj_close,
                AVG(adj_close) OVER (ORDER BY date ROWS BETWEEN 124 PRECEDING AND CURRENT ROW) AS sma_125
-        FROM stg_major_indices
+        FROM {staging_dataset}.stg_major_indices
         WHERE symbol = 'SPY' AND adj_close IS NOT NULL
           AND date >= CURRENT_DATE - INTERVAL {lookback_years} YEAR
         ORDER BY date
@@ -81,7 +88,7 @@ def fear_greed_signals(
                 adj_close,
                 MAX(adj_close) OVER (PARTITION BY symbol ORDER BY date ROWS BETWEEN 251 PRECEDING AND CURRENT ROW) AS high_52w,
                 MIN(adj_close) OVER (PARTITION BY symbol ORDER BY date ROWS BETWEEN 251 PRECEDING AND CURRENT ROW) AS low_52w
-            FROM stg_sp500_companies_prices
+            FROM {staging_dataset}.stg_sp500_companies_prices
             WHERE adj_close IS NOT NULL
               AND date >= CURRENT_DATE - INTERVAL {lookback_years} YEAR - INTERVAL 1 YEAR
         ),
@@ -107,7 +114,7 @@ def fear_greed_signals(
         f"""
         SELECT date, literal AS vix,
                AVG(literal) OVER (ORDER BY date ROWS BETWEEN 49 PRECEDING AND CURRENT ROW) AS vix_sma_50
-        FROM stg_fred_series
+        FROM {staging_dataset}.stg_fred_series
         WHERE series_code = 'VIXCLS' AND literal IS NOT NULL
           AND date >= CURRENT_DATE - INTERVAL {lookback_years} YEAR
         ORDER BY date
@@ -121,11 +128,11 @@ def fear_greed_signals(
         f"""
         WITH spy_ret AS (
             SELECT date, adj_close / LAG(adj_close, 20) OVER (ORDER BY date) - 1 AS spy_20d_ret
-            FROM stg_major_indices WHERE symbol = 'SPY' AND adj_close IS NOT NULL
+            FROM {staging_dataset}.stg_major_indices WHERE symbol = 'SPY' AND adj_close IS NOT NULL
         ),
         govt_ret AS (
             SELECT date, adj_close / LAG(adj_close, 20) OVER (ORDER BY date) - 1 AS govt_20d_ret
-            FROM stg_fixed_income WHERE symbol = 'GOVT' AND adj_close IS NOT NULL
+            FROM {staging_dataset}.stg_fixed_income WHERE symbol = 'GOVT' AND adj_close IS NOT NULL
         )
         SELECT s.date, s.spy_20d_ret, g.govt_20d_ret,
                s.spy_20d_ret - g.govt_20d_ret AS stock_bond_diff
@@ -143,11 +150,11 @@ def fear_greed_signals(
     credit_df = bq.execute_query(
         f"""
         WITH hy AS (
-            SELECT date, literal AS hy_spread FROM stg_fred_series
+            SELECT date, literal AS hy_spread FROM {staging_dataset}.stg_fred_series
             WHERE series_code = 'BAMLH0A0HYM2' AND literal IS NOT NULL
         ),
         ig AS (
-            SELECT date, literal AS ig_spread FROM stg_fred_series
+            SELECT date, literal AS ig_spread FROM {staging_dataset}.stg_fred_series
             WHERE series_code = 'BAMLC0A0CM' AND literal IS NOT NULL
         )
         SELECT h.date, h.hy_spread, i.ig_spread, h.hy_spread - i.ig_spread AS hy_ig_diff

macro_agents/tests/test_bigquery_warehouse.py

@@ -10,7 +10,10 @@
 from google.cloud import bigquery
 
 from macro_agents.defs.resources.bigquery_query import QueryParameter
-from macro_agents.defs.resources.bigquery_warehouse import BigQueryWarehouseResource
+from macro_agents.defs.resources.bigquery_warehouse import (
+    BigQueryWarehouseResource,
+    default_dataset_for_schema,
+)
 
 
 def _make_resource() -> BigQueryWarehouseResource:
@@ -63,6 +66,41 @@ def test_closing_connection_does_not_break_cached_client(self, mock_client_cls):
         cached.close.assert_not_called()
 
 
+class TestDefaultDatasetForSchema:
+    """dbt's own schema resolution keys off DBT_TARGET (profiles.yml,
+    generate_schema_name.sql), not ENVIRONMENT — these must stay in sync
+    or cross-dataset references 404 or silently read the wrong dataset."""
+
+    @pytest.mark.parametrize(
+        ("dbt_target", "expected"),
+        [
+            ("prod", "economics_staging"),
+            ("staging", "economics_staging_staging"),
+            ("dev", "economics_staging_dev"),
+            ("unrecognized", "economics_staging_dev"),
+        ],
+    )
+    def test_follows_dbt_target(self, monkeypatch, dbt_target, expected):
+        monkeypatch.setenv("DBT_TARGET", dbt_target)
+        monkeypatch.setenv("ENVIRONMENT", "prod")  # must not win over DBT_TARGET
+        assert default_dataset_for_schema("economics_staging") == expected
+
+    def test_falls_back_to_environment_when_dbt_target_unset(self, monkeypatch):
+        monkeypatch.delenv("DBT_TARGET", raising=False)
+        monkeypatch.setenv("ENVIRONMENT", "staging")
+        assert (
+            default_dataset_for_schema("economics_staging")
+            == "economics_staging_staging"
+        )
+
+    def test_defaults_to_dev_when_neither_is_set(self, monkeypatch):
+        monkeypatch.delenv("DBT_TARGET", raising=False)
+        monkeypatch.delenv("ENVIRONMENT", raising=False)
+        assert (
+            default_dataset_for_schema("economics_staging") == "economics_staging_dev"
+        )
+
+
 class TestExecuteQueryErrorSurfacing:
     @patch("macro_agents.defs.resources.bigquery_warehouse.bigquery.Client")
     def test_query_failure_raises_instead_of_returning_empty(self, mock_client_cls):