feat: causal lab

backend/services/database.py

@@ -1238,6 +1238,16 @@ def init_db():
     except Exception:
         pass
 
+    # Purge macro_series_log entries from multi-country contamination.
+    # USD-series (UNRATE, PAYEMS…) were logging both US and foreign events → flip-flop.
+    # Safe to truncate: all data is synthetic backfill, no real live captures yet.
+    try:
+        c.execute(
+            "DELETE FROM macro_series_log WHERE source IN ('backfill', 'calendar_sync')"
+        )
+    except Exception:
+        pass
+
     conn.commit()
     conn.close()
 

backend/services/macro_series_log.py

@@ -83,8 +83,10 @@ def backfill_from_ff_calendar(conn) -> dict:
     """
     Seed macro_series_log from existing ff_calendar rows (historical).
     Only inserts rows that don't already exist (idempotent).
-    Skips events without series_id.
+    Filters by ff_currency to avoid multi-country contamination.
     """
+    currency_map = _series_currency_map()
+
     rows = conn.execute(
         """SELECT series_id, event_name, event_date, actual_value, forecast_value,
                   previous_value, currency, impact, fetched_at
@@ -96,6 +98,9 @@ def backfill_from_ff_calendar(conn) -> dict:
 
     inserted = 0
     for r in rows:
+        expected = currency_map.get(r['series_id'])
+        if expected and r['currency'] != expected:
+            continue
         already = conn.execute(
             "SELECT 1 FROM macro_series_log WHERE series_id=? AND event_date=? AND source='backfill' LIMIT 1",
             (r['series_id'], r['event_date'])
@@ -135,11 +140,22 @@ def backfill_from_ff_calendar(conn) -> dict:
     return {"inserted": inserted, "scanned": len(rows)}
 
 
+def _series_currency_map() -> dict:
+    """Return {series_id: ff_currency} for series that have a currency restriction."""
+    try:
+        from services.fred_bootstrap import FRED_SERIES
+        return {sid: meta['ff_currency'] for sid, meta in FRED_SERIES.items() if meta.get('ff_currency')}
+    except Exception:
+        return {}
+
+
 def scan_and_log_all(conn, source: str = 'calendar_sync') -> dict:
     """
     After a calendar sync, scan all ff_calendar rows with series_id and log any changes.
-    Typically called at end of each 6h sync cycle.
+    Filters by ff_currency to avoid multi-country flip-flop (e.g. US vs CA unemployment).
     """
+    currency_map = _series_currency_map()
+
     rows = conn.execute(
         """SELECT series_id, event_name, event_date, actual_value, forecast_value,
                   previous_value, currency, impact
@@ -150,6 +166,9 @@ def scan_and_log_all(conn, source: str = 'calendar_sync') -> dict:
 
     logged = 0
     for r in rows:
+        expected = currency_map.get(r['series_id'])
+        if expected and r['currency'] != expected:
+            continue
         did_log = log_if_changed(
             conn,
             series_id=r['series_id'],