fix: enforce BigQuery query contract (#137)

Author: C00ldudeNoonan

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

@@ -38,8 +38,11 @@ def query_data_for_findings(
         pct_change_1y,
         date_grain
     FROM agent_fred_series_latest_aggregates
-    WHERE month >= DATE_TRUNC('month', CAST(? AS DATE) - INTERVAL '12 months')
-      AND month <= CAST(? AS DATE)
+    WHERE month >= DATE_SUB(
+        DATE_TRUNC(DATE(@week_start), MONTH),
+        INTERVAL 12 MONTH
+    )
+      AND month <= DATE(@week_end)
       AND current_value IS NOT NULL
     ORDER BY series_code, month DESC
     """
@@ -99,17 +102,18 @@ def query_data_for_findings(
         weekly_avg_score,
         score_momentum_pct
     FROM agent_reddit_sentiment_trends
-    WHERE date >= CAST(? AS DATE) - INTERVAL '3 months'
-      AND date <= CAST(? AS DATE)
+    WHERE date >= DATE_SUB(DATE(@week_start), INTERVAL 3 MONTH)
+      AND date <= DATE(@week_end)
     ORDER BY date DESC
     """
 
     # Execute queries
-    economic_df = bq.execute_query(economic_query, params=[week_start, week_end])
+    date_params = {"week_start": week_start, "week_end": week_end}
+    economic_df = bq.execute_query(economic_query, params=date_params)
     market_df = bq.execute_query(market_query)
     commodity_df = bq.execute_query(commodity_query)
     correlation_df = bq.execute_query(correlation_query)
-    sentiment_df = bq.execute_query(sentiment_query, params=[week_start, week_end])
+    sentiment_df = bq.execute_query(sentiment_query, params=date_params)
 
     return {
         "economic": economic_df,

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

@@ -446,11 +446,11 @@ def _fetch_analysis_row(
         analysis_content,
         chart_manifest
     FROM economy_state_analysis
-    WHERE dagster_run_id = ?
+    WHERE dagster_run_id = @run_id
     ORDER BY analysis_timestamp DESC
     LIMIT 1
     """
-    df = bq.execute_query(query, read_only=True, params=[run_id])
+    df = bq.execute_query(query, read_only=True, params={"run_id": run_id})
     if df.is_empty():
         return None
     return df.to_dicts()[0]
@@ -558,14 +558,18 @@ def generate_economy_state_charts(
     if manifest:
         update_query = """
         UPDATE economy_state_analysis
-        SET chart_manifest = CAST(? AS JSON),
-            analysis_content = ?
-        WHERE analysis_timestamp = ?
+        SET chart_manifest = PARSE_JSON(@chart_manifest),
+            analysis_content = @analysis_content
+        WHERE analysis_timestamp = CAST(@analysis_timestamp AS TIMESTAMP)
         """
         bq.execute_query(
             update_query,
             read_only=False,
-            params=[json.dumps(manifest), updated_content, timestamp_str],
+            params={
+                "chart_manifest": json.dumps(manifest),
+                "analysis_content": updated_content,
+                "analysis_timestamp": timestamp_str,
+            },
         )
 
     return dg.MaterializeResult(

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

@@ -461,11 +461,11 @@ def _fetch_recommendations_row(
         personality,
         recommendations_content
     FROM investment_recommendations
-    WHERE dagster_run_id = ?
+    WHERE dagster_run_id = @run_id
     ORDER BY analysis_timestamp DESC
     LIMIT 1
     """
-    df = bq.execute_query(query, read_only=True, params=[run_id])
+    df = bq.execute_query(query, read_only=True, params={"run_id": run_id})
     if df.is_empty():
         return None
     return df.to_dicts()[0]
@@ -579,14 +579,18 @@ def generate_investment_recommendation_charts(
         )
         update_query = """
         UPDATE investment_recommendations
-        SET chart_manifest = CAST(? AS JSON),
-            recommendations_content = ?
-        WHERE analysis_timestamp = ?
+        SET chart_manifest = PARSE_JSON(@chart_manifest),
+            recommendations_content = @recommendations_content
+        WHERE analysis_timestamp = CAST(@analysis_timestamp AS TIMESTAMP)
         """
         bq.execute_query(
             update_query,
             read_only=False,
-            params=[json.dumps(manifest), updated_content, timestamp_str],
+            params={
+                "chart_manifest": json.dumps(manifest),
+                "recommendations_content": updated_content,
+                "analysis_timestamp": timestamp_str,
+            },
         )
 
     return dg.MaterializeResult(