fix: complete BigQuery cutover for SEC search and query interface (#130)

.env.dagster.example

@@ -6,7 +6,7 @@
 # PostgreSQL - Dagster Storage Backend
 # ============================================================================
 # This database stores Dagster's run history, event logs, and schedule state
-# It is separate from your data warehouse (MotherDuck)
+# It is separate from your data warehouse (BigQuery)
 DAGSTER_PG_PASSWORD=change-me-to-secure-password
 
 # ============================================================================
@@ -22,22 +22,14 @@ DAGSTER_EVENT_RETENTION_HOURS=24
 DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS=3600
 
 # ============================================================================
-# MotherDuck - Primary Data Warehouse
-# ============================================================================
-# Get token from: https://motherduck.com/
-MOTHERDUCK_TOKEN=your-motherduck-token-here
-MOTHERDUCK_DATABASE=your-database-name
-MOTHERDUCK_PROD_SCHEMA=your-schema-name
-
-# ============================================================================
-# Google Cloud Platform
+# Google Cloud Platform — Primary Data Warehouse (BigQuery)
 # ============================================================================
 # GCS Bucket for DSPy models and FOMC data
 GCS_BUCKET_NAME=your-gcs-bucket-name
 
 # BigQuery - Analytics Warehouse
-BIGQUERY_PROJECT_ID=your-gcp-project-id
-BIGQUERY_DATASET=economic_data
+BIGQUERY_PROJECT=your-gcp-project-id
+BIGQUERY_DATASET=economics_raw
 BIGQUERY_LOCATION=US
 
 # Note: GCP service account JSON should be placed at ./gcp-service-account.json

.env.example

@@ -126,7 +126,7 @@ DBT_PLATFORM_OBSERVE_ONLY=false
 # CLAUDE CODE / MCP SERVERS
 # =============================================================================
 # MotherDuck MCP is no longer used — queries now go to BigQuery via the
-# MotherDuck MCP server or the BigQuery console.
+# dbt-index MCP server's `warehouse` tool (see .mcp.json) or the BigQuery console.
 
 # Claude API - For context documentation maintainer skill
 # Get your key from: https://console.anthropic.com/

.mcp.json

@@ -6,13 +6,6 @@
         "--auto-reingest",
         "serve"
       ]
-    },
-    "motherduck": {
-      "type": "http",
-      "url": "https://mcp.motherduck.com/mcp",
-      "headers": {
-        "Authorization": "Bearer ${MOTHERDUCK_TOKEN}"
-      }
     }
   }
 }

Readme.md

@@ -1,17 +1,17 @@
 # Economic Data Project
 
-An end-to-end data platform that ingests, transforms, and analyzes economic and financial market data. Pipelines run on self-hosted Dagster OSS (Docker) and store everything in MotherDuck (DuckDB cloud). Data is queried and explored via AI agents using the MotherDuck MCP server.
+An end-to-end data platform that ingests, transforms, and analyzes economic and financial market data. Pipelines run on self-hosted Dagster OSS (Docker) and store everything in Google BigQuery. Data is queried and explored via AI agents using the dbt-index MCP server's warehouse tools.
 
 ## Architecture
 
 ```
-External APIs → Dagster (ingestion) → MotherDuck/DuckDB
+External APIs → Dagster (ingestion) → Google BigQuery
                                           ↓
                               dbt (SQL transformations)
                                           ↓
                          DSPy AI agents (analysis & signals)
                                           ↓
-                         Claude + MotherDuck MCP (query interface)
+                         Claude + dbt-index MCP (query interface)
 ```
 
 ## Project Structure
@@ -40,7 +40,7 @@ economic-data-project/
 │   │   │   └── news/                # Reddit + FOMC weekly summaries
 │   │   ├── transformation/          # dbt orchestration + FCI
 │   │   ├── backtesting/             # Strategy backtesting
-│   │   ├── resources/               # Shared resources (MotherDuck, GCS, etc.)
+│   │   ├── resources/               # Shared resources (BigQuery, GCS, etc.)
 │   │   └── telemetry/               # Pipeline monitoring assets
 │   └── tests/                       # 440+ test suite
 ├── dbt_project/                     # SQL transformations
@@ -60,7 +60,7 @@ economic-data-project/
 ### Prerequisites
 - Docker Desktop
 - API keys (see `.env.example`)
-- MotherDuck account
+- Google Cloud project with BigQuery enabled + a service account (see `GOOGLE_APPLICATION_CREDENTIALS`)
 
 ### Setup
 
@@ -113,7 +113,7 @@ uv run ruff format .
 
 ## Querying the Data
 
-The platform is designed to be queried via AI agents using the MotherDuck MCP server. Configure Claude Code with your MotherDuck token (see `.mcp.json`) to query any table directly in conversation.
+The platform is designed to be queried via AI agents using the dbt-index MCP server (see `.mcp.json`), whose `warehouse` tool runs SQL directly against BigQuery. With that server configured, Claude Code can query any table directly in conversation.
 
 ```bash
 # Example: ask Claude about the data
@@ -134,8 +134,9 @@ Copy `.env.example` to `.env`. Required keys:
 
 | Variable | Description |
 |----------|-------------|
-| `MOTHERDUCK_TOKEN` | MotherDuck auth token |
-| `MOTHERDUCK_DATABASE` | Target database name |
+| `GOOGLE_APPLICATION_CREDENTIALS` | Path to a GCP service-account JSON file (or inline JSON) for BigQuery |
+| `BIGQUERY_PROJECT` | GCP project ID hosting the BigQuery datasets |
+| `BIGQUERY_DATASET` | Default BigQuery dataset (defaults to `economics_raw*` per environment) |
 | `FRED_API_KEY` | FRED API key |
 | `MARKETSTACK_API_KEY` | MarketStack API key |
 | `OPENAI_API_KEY` or `ANTHROPIC_API_KEY` | LLM provider for AI agents |
@@ -167,7 +168,7 @@ uv run dg check defs                       # Validate Dagster definitions
 
 - **[Dagster](https://docs.dagster.io)** — asset orchestration, schedules, sensors
 - **[dbt](https://docs.getdbt.com)** — SQL transformations
-- **[DuckDB](https://duckdb.org/docs) / [MotherDuck](https://motherduck.com/docs)** — analytical database + cloud sync
+- **[Google BigQuery](https://cloud.google.com/bigquery/docs)** — serverless analytical data warehouse
 - **[DSPy](https://dspy.ai)** — LLM pipeline framework for AI agents
 - **[Polars](https://docs.pola.rs)** — dataframe processing
 - **[uv](https://docs.astral.sh/uv/)** — Python package management

macro_agents/.env.example

@@ -1,8 +1,5 @@
 CENSUS_API_KEY=
 FRED_API_KEY=
-MOTHERDUCK_TOKEN=
-MOTHERDUCK_DATABASE=
-MOTHERDUCK_PROD_SCHEMA=
 GOOGLE_DRIVE_FOLDER_ID=
 GOOGLE_SERVICE_ACCOUNT_JSON=
 ENVIRONMENT=
@@ -16,6 +13,6 @@ GEMINI_API_KEY=
 ANTHROPIC_API_KEY=
 GOOGLE_APPLICATION_CREDENTIALS=/path/to/your/service-account-key.json
 GCS_BUCKET_NAME=
-BIGQUERY_PROJECT_ID=
+BIGQUERY_PROJECT=
 BIGQUERY_LOCATION=
 BIGQUERY_DATASET=

macro_agents/src/macro_agents/defs/analysis/ai/ai_models_fetcher.py

@@ -170,7 +170,7 @@ def _get_gemini_models(api_key: str | None = None) -> list[str]:
 
 
 @dg.asset(
-    kinds={"duckdb"},
+    kinds={"bigquery"},
     group_name="agents",
     description="Fetches and stores current available chat/completion AI models from OpenAI, Anthropic, and Gemini APIs suitable for economic analysis",
 )

macro_agents/src/macro_agents/defs/analysis/ai/nl_to_sql_module.py

@@ -26,17 +26,20 @@ class NaturalLanguageToSQLSignature(dspy.Signature):
     )
 
     sql_query: str = dspy.OutputField(
-        desc="""DuckDB-compatible SQL query (SELECT only, no DROP/DELETE/UPDATE).
+        desc="""BigQuery Standard SQL (GoogleSQL) query (SELECT only, no DROP/DELETE/UPDATE).
         CRITICAL RULES:
         1. Only SELECT queries allowed
         2. Respect column additivity:
            - ADDITIVE: Use SUM() for aggregation
-           - NON_ADDITIVE: Use AVG(), MIN(), MAX(), or FIRST() - NEVER SUM
+           - NON_ADDITIVE: Use AVG(), MIN(), MAX(), or ANY_VALUE() - NEVER SUM
            - SEMI_ADDITIVE: Use for GROUP BY, ORDER BY, WHERE only
         3. Always include proper GROUP BY for aggregations
-        4. Use date filters intelligently (WHERE date >= '2023-01-01')
+        4. Use date filters intelligently (WHERE date >= DATE '2023-01-01')
         5. Table and column names must exactly match data_dictionary
-        6. LIMIT results to max 1000 rows if not specified
+        6. Use BigQuery syntax: backtick-quote identifiers when needed, DATE_SUB/
+           DATE_TRUNC/INTERVAL for date math, SAFE_CAST for casts. Do NOT use
+           DuckDB-only constructs (e.g. list_*/QUALIFY-less window tricks, `::` casts).
+        7. LIMIT results to max 1000 rows if not specified
         """
     )
 
@@ -96,10 +99,10 @@ def forward(
 class SQLValidator:
     """Validates SQL queries for safety and correctness.
 
-    Uses sqlglot to parse the query into an AST and verify it is a single
-    read-only SELECT. The prior implementation matched forbidden keywords
-    as substrings of the raw string, which was bypassable by comments,
-    mixed case, and embedded newlines.
+    Uses sqlglot to parse the query into an AST (BigQuery dialect) and verify
+    it is a single read-only SELECT. The prior implementation matched
+    forbidden keywords as substrings of the raw string, which was bypassable
+    by comments, mixed case, and embedded newlines.
     """
 
     # AST node types that mutate or escape data — reject if present anywhere.
@@ -139,7 +142,7 @@ def validate_query(sql: str) -> tuple[bool, str]:
             )
 
         try:
-            statements = sqlglot.parse(sql, dialect="duckdb")
+            statements = sqlglot.parse(sql, dialect="bigquery")
         except sqlglot.errors.ParseError as e:
             return False, f"SQL parse error: {e}"
 

macro_agents/src/macro_agents/defs/analysis/asset_relationships/asset_class_relationship_analyzer.py

@@ -174,7 +174,7 @@ def extract_relationship_summary(analysis_content: str) -> dict[str, Any]:
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="economic_analysis",
     description="Understand relationships between asset classes and the economic cycle",
     deps=[

macro_agents/src/macro_agents/defs/analysis/data_points/interesting_data_points_assets.py

@@ -61,7 +61,7 @@ def get_latest_economic_context(bq: BigQueryWarehouseResource) -> str:
 
 @dg.asset(
     group_name="economic_analysis",
-    kinds={"ai", "duckdb"},
+    kinds={"ai", "bigquery"},
     partitions_def=dg.WeeklyPartitionsDefinition(start_date="2024-01-01"),
     deps=[
         dg.AssetKey(["agent_fred_series_latest_aggregates"]),

macro_agents/src/macro_agents/defs/analysis/economy_state/assets.py

@@ -20,7 +20,7 @@
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="economic_analysis",
     description="Analyze current economic indicators and commodity data to determine the state of the economy",
     deps=[
@@ -261,7 +261,7 @@ def analyze_economy_state(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="economic_analysis",
     description="Analyze economy state using specialized domain sub-agents (v2)",
     deps=[

macro_agents/src/macro_agents/defs/analysis/economy_state/economy_state_charts.py

@@ -457,7 +457,7 @@ def _fetch_analysis_row(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb", "gcs"},
+    kinds={"dspy", "bigquery", "gcs"},
     group_name="economic_analysis",
     description="Generate chart images for the economy state analysis and store a chart manifest",
     deps=[dg.AssetKey(["analyze_economy_state"])],

macro_agents/src/macro_agents/defs/analysis/fed_sentiment/assets.py

@@ -38,7 +38,7 @@ def _generate_id(prefix: str, *components: str) -> str:
 
 @dg.asset(
     group_name="fed_sentiment",
-    kinds={"nlp", "duckdb"},
+    kinds={"nlp", "bigquery"},
     deps=[dg.AssetKey(["process_fomc_transcripts"])],
     description="Dictionary-based hawkish/dovish scoring of FOMC transcript sections",
 )
@@ -156,7 +156,7 @@ def score_fed_sentiment_dictionary(
 
 @dg.asset(
     group_name="fed_sentiment",
-    kinds={"ai", "duckdb"},
+    kinds={"ai", "bigquery"},
     deps=[dg.AssetKey(["process_fomc_transcripts"])],
     description="LLM-based hawkish/dovish scoring of FOMC transcript sections",
 )
@@ -300,7 +300,7 @@ def score_fed_sentiment_llm(
 
 @dg.asset(
     group_name="fed_sentiment",
-    kinds={"analysis", "duckdb"},
+    kinds={"analysis", "bigquery"},
     deps=[
         dg.AssetKey(["score_fed_sentiment_dictionary"]),
         dg.AssetKey(["fomc_transcript_schema"]),

macro_agents/src/macro_agents/defs/analysis/investments/investment_recommendation_charts.py

@@ -472,7 +472,7 @@ def _fetch_recommendations_row(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb", "gcs"},
+    kinds={"dspy", "bigquery", "gcs"},
     group_name="economic_analysis",
     description="Generate chart images for investment recommendations and store a chart manifest",
     deps=[dg.AssetKey(["generate_investment_recommendations"])],

macro_agents/src/macro_agents/defs/analysis/investments/investment_recommendations.py

@@ -219,7 +219,7 @@ def extract_recommendations_summary(recommendations_content: str) -> dict[str, A
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="economic_analysis",
     description="Generate actionable investment recommendations based on economy state and asset class relationships",
     deps=[

macro_agents/src/macro_agents/defs/analysis/narratives/economic_narrative_assets.py

@@ -234,7 +234,7 @@ def create_indicator_forecast_table_if_not_exists(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="economic_narratives",
     description="Generate plain-English narratives for economic indicator releases and store in database",
     deps=[
@@ -423,7 +423,7 @@ def generate_economic_narratives(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="economic_narratives",
     description="Generate short-term forecasts for economic indicators",
     deps=[

macro_agents/src/macro_agents/defs/analysis/news/news_summary_assets.py

@@ -14,7 +14,7 @@
 
 @dg.asset(
     group_name="news_summaries",
-    kinds={"ai", "duckdb"},
+    kinds={"ai", "bigquery"},
     partitions_def=dg.DailyPartitionsDefinition(start_date="2024-01-01"),
     deps=[dg.AssetKey(["agent_reddit_posts_daily"])],
     description="AI-generated daily summary of Reddit posts from financial/economic subreddits",
@@ -130,7 +130,7 @@ def reddit_daily_summary(
 
 @dg.asset(
     group_name="news_summaries",
-    kinds={"ai", "duckdb"},
+    kinds={"ai", "bigquery"},
     partitions_def=dg.WeeklyPartitionsDefinition(start_date="2024-01-01"),
     description="AI-generated weekly cross-source summary combining Reddit and FOMC data",
 )

macro_agents/src/macro_agents/defs/backtesting/backtest_asset_class_relationship_analyzer.py

@@ -57,7 +57,7 @@ def get_latest_backtest_economy_state_analysis(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="backtesting",
     description="Backtest asset class relationship analysis with historical data cutoff",
     deps=[

macro_agents/src/macro_agents/defs/backtesting/backtest_economy_state_analyzer.py

@@ -101,7 +101,7 @@ def get_backtest_dates(config: BacktestConfig) -> list[str]:
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="backtesting",
     description="Backtest economy state analysis with historical data cutoff",
     deps=[

macro_agents/src/macro_agents/defs/backtesting/backtest_evaluator.py

@@ -131,7 +131,7 @@ def _period_outperformance(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="backtesting",
     description="Evaluate backtest investment recommendations using DSPy evaluation framework",
     deps=[

macro_agents/src/macro_agents/defs/backtesting/backtest_investment_recommendations.py

@@ -92,7 +92,7 @@ def get_latest_backtest_relationship_analysis(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="backtesting",
     description="Backtest investment recommendations generation with historical data cutoff",
     deps=[

macro_agents/src/macro_agents/defs/backtesting/backtest_optimizer.py

@@ -122,7 +122,7 @@ def create_model_versions_table(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb"},
+    kinds={"dspy", "bigquery"},
     group_name="optimization",
     description="Prepare training data from backtest evaluation results for DSPy optimization",
     deps=[
@@ -302,7 +302,7 @@ def prepare_optimization_training_data(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb", "gcs"},
+    kinds={"dspy", "bigquery", "gcs"},
     group_name="optimization",
     description="Optimize DSPy modules using MIPROv2 optimizer with backtest data",
     deps=[
@@ -734,7 +734,7 @@ def optimize_dspy_modules(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb", "gcs"},
+    kinds={"dspy", "bigquery", "gcs"},
     group_name="optimization",
     description="Promote optimized model to production if improvement threshold is met",
     deps=[optimize_dspy_modules],
@@ -832,7 +832,7 @@ def promote_optimized_model_to_production(
 
 
 @dg.asset(
-    kinds={"dspy", "duckdb", "gcs"},
+    kinds={"dspy", "bigquery", "gcs"},
     group_name="optimization",
     description="Automatically promote the best optimized model to production for each module",
     deps=[optimize_dspy_modules],