Add economic calendar module to filter trades near high-impact events

Introduces a full Trading Economics API pipeline that fetches, stores, and
queries economic events (NFP, CPI, rate decisions, etc.) so the backtester
can block trade entries within a configurable buffer window of high-impact
releases — reducing slippage and false signals.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/backtester.py

@@ -18,6 +18,78 @@ from config_loader import load_config, get_project_root
 from data_engine import detect_engulfing, add_pivot_points
 
 
+# ---------------------------------------------------------------------------
+# Economic calendar helpers
+# ---------------------------------------------------------------------------
+
+def fetch_calendar_for_backtest(instrument, supabase_client, cfg):
+    """
+    Load calendar events from Supabase filtered by the currencies in
+    the instrument and the configured impact threshold.
+    Returns a DataFrame or None if calendar is disabled/empty.
+    """
+    from economic_calendar import fetch_calendar_from_supabase
+
+    cal_cfg = cfg.get("economic_calendar", {})
+    if not cal_cfg.get("enabled", False):
+        return None
+
+    table = cal_cfg.get("table", "economic_calendar")
+    threshold = cal_cfg.get("impact_threshold", "High")
+    impact_map = {"Low": 1, "Medium": 2, "High": 3}
+    impact_min = impact_map.get(threshold, 3)
+
+    # Derive currencies from instrument (e.g. EUR_USD -> [EUR, USD])
+    currencies = instrument.split("_")
+
+    print(f"  Loading economic calendar events (currencies={currencies}, impact>={threshold})...")
+    cal_df = fetch_calendar_from_supabase(
+        supabase_client,
+        table=table,
+        currencies=currencies,
+        impact_min=impact_min,
+    )
+    if cal_df.empty:
+        print("  No calendar events found.")
+        return None
+
+    print(f"  Loaded {len(cal_df)} calendar events.")
+    return cal_df
+
+
+def is_near_event(bar_time, calendar_df, buffer_minutes=30):
+    """
+    O(log n) check whether bar_time is within buffer_minutes of any
+    calendar event, using DatetimeIndex slicing.
+    """
+    if calendar_df is None or calendar_df.empty:
+        return False
+
+    # Ensure we have a DatetimeIndex for fast slicing
+    if not isinstance(calendar_df.index, pd.DatetimeIndex):
+        return False
+
+    bar_ts = pd.Timestamp(bar_time, tz="UTC")
+    window_start = bar_ts - pd.Timedelta(minutes=buffer_minutes)
+    window_end = bar_ts + pd.Timedelta(minutes=buffer_minutes)
+
+    nearby = calendar_df.loc[window_start:window_end]
+    return len(nearby) > 0
+
+
+def _prepare_calendar_index(calendar_df):
+    """
+    Set event_datetime as a sorted DatetimeIndex for O(log n) lookups.
+    Returns the prepared DataFrame or None.
+    """
+    if calendar_df is None or calendar_df.empty:
+        return None
+    cal = calendar_df.copy()
+    cal["event_datetime"] = pd.to_datetime(cal["event_datetime"], utc=True)
+    cal = cal.set_index("event_datetime").sort_index()
+    return cal
+
+
 # ---------------------------------------------------------------------------
 # Data fetching
 # ---------------------------------------------------------------------------
@@ -314,7 +386,8 @@ def _find_tp_levels_short(level_values, entry_lv_idx):
 # Dual take-profit backtest engine
 # ---------------------------------------------------------------------------
 
-def run_backtest_dual_tp(df, strategy_cfg):
+def run_backtest_dual_tp(df, strategy_cfg, calendar_df=None,
+                         event_buffer_minutes=30):
     """
     Backtest engine supporting per-trade SL/TP with partial closes.
 
@@ -324,6 +397,9 @@ def run_backtest_dual_tp(df, strategy_cfg):
     - On TP2 hit: close remaining 50%.
     - On SL hit: close full remaining position.
 
+    If calendar_df is provided, entries within event_buffer_minutes of a
+    high-impact event are blocked.
+
     Returns dict with equity_curve, trades, metrics (same interface as run_backtest).
     """
     trade_size_pct = strategy_cfg.get("trade_size_pct_of_equity", 0.01)
@@ -344,6 +420,10 @@ def run_backtest_dual_tp(df, strategy_cfg):
     position_size = 0.0
     half_closed = False
 
+    # Calendar event filter
+    cal_indexed = _prepare_calendar_index(calendar_df)
+    blocked_by_calendar = 0
+
     equity_curve = []
     trades = []
 
@@ -494,6 +574,11 @@ def run_backtest_dual_tp(df, strategy_cfg):
         if not in_position and not stopped:
             sig = signals[i]
             if sig in (1, -1) and not np.isnan(sl_col[i]):
+                # Block entry if near a high-impact economic event
+                if is_near_event(bar_time, cal_indexed, event_buffer_minutes):
+                    blocked_by_calendar += 1
+                    equity_curve.append(equity)
+                    continue
                 direction = sig
                 entry_price = bar_close
                 sl_price = sl_col[i]
@@ -546,6 +631,10 @@ def run_backtest_dual_tp(df, strategy_cfg):
     equity_series = pd.Series(equity_curve, index=df.index, name="equity")
     metrics = compute_metrics(equity_series, trades, starting_equity)
 
+    if blocked_by_calendar > 0:
+        print(f"  Calendar filter blocked {blocked_by_calendar} trade entries.")
+        metrics["blocked_by_calendar"] = blocked_by_calendar
+
     return {
         "equity_curve": equity_series,
         "trades": trades,
@@ -557,9 +646,11 @@ def run_backtest_dual_tp(df, strategy_cfg):
 # Backtest engine (SMA cross — original)
 # ---------------------------------------------------------------------------
 
-def run_backtest(df, strategy_cfg):
+def run_backtest(df, strategy_cfg, calendar_df=None, event_buffer_minutes=30):
     """
     Walk through bars, track position, equity, trades.
+    If calendar_df is provided, entries within event_buffer_minutes of a
+    high-impact event are blocked.
     Returns dict with equity_curve (Series), trades (list of dicts), metrics (dict).
     """
     trade_size_pct = strategy_cfg.get("trade_size_pct_of_equity", 0.01)
@@ -572,6 +663,10 @@ def run_backtest(df, strategy_cfg):
     position_size = 0.0
     stopped = False
 
+    # Calendar event filter
+    cal_indexed = _prepare_calendar_index(calendar_df)
+    blocked_by_calendar = 0
+
     equity_curve = []
     trades = []
 
@@ -622,6 +717,11 @@ def run_backtest(df, strategy_cfg):
         # Position change
         if sig != position:
             if sig == 1 and position == 0:
+                # Block entry if near a high-impact economic event
+                if is_near_event(bar_time, cal_indexed, event_buffer_minutes):
+                    blocked_by_calendar += 1
+                    equity_curve.append(equity)
+                    continue
                 # Enter long
                 position_size = equity * trade_size_pct
                 trades.append({
@@ -650,6 +750,10 @@ def run_backtest(df, strategy_cfg):
     equity_series = pd.Series(equity_curve, index=df.index, name="equity")
     metrics = compute_metrics(equity_series, trades, starting_equity)
 
+    if blocked_by_calendar > 0:
+        print(f"  Calendar filter blocked {blocked_by_calendar} trade entries.")
+        metrics["blocked_by_calendar"] = blocked_by_calendar
+
     return {
         "equity_curve": equity_series,
         "trades": trades,
@@ -840,7 +944,17 @@ def main():
     print(f"Instruments: {instruments}")
     print(f"Granularities: {granularities}\n")
 
+    # Economic calendar config
+    cal_cfg = cfg.get("economic_calendar", {})
+    cal_enabled = cal_cfg.get("enabled", False)
+    event_buffer_minutes = cal_cfg.get("event_buffer_minutes", 30)
+
     for instrument in instruments:
+        # Load calendar data once per instrument (shared across granularities)
+        calendar_df = None
+        if cal_enabled:
+            calendar_df = fetch_calendar_for_backtest(instrument, sb, cfg)
+
         for granularity in granularities:
             print(f"=== {instrument} / {granularity} ===")
 
@@ -860,9 +974,17 @@ def main():
                 continue
 
             if strategy_cfg["rule"] == "pivot_retest_engulfing":
-                results = run_backtest_dual_tp(df, strategy_cfg)
+                results = run_backtest_dual_tp(
+                    df, strategy_cfg,
+                    calendar_df=calendar_df,
+                    event_buffer_minutes=event_buffer_minutes,
+                )
             else:
-                results = run_backtest(df, strategy_cfg)
+                results = run_backtest(
+                    df, strategy_cfg,
+                    calendar_df=calendar_df,
+                    event_buffer_minutes=event_buffer_minutes,
+                )
             save_results(instrument, granularity, results, results["metrics"])
 
     print("Backtesting complete.")

src/data_engine.py

@@ -159,12 +159,114 @@ def add_pivot_points(df):
     return df
 
 
-def build_all_features(df, config=None):
+def add_calendar_features(df, calendar_df, cfg=None):
+    """
+    Add economic calendar feature columns to a candle DataFrame using merge_asof.
+
+    Columns added:
+    - minutes_to_next_event: minutes until the next event of any impact
+    - minutes_to_next_high_impact: minutes until the next High-impact event
+    - upcoming_high_impact_event: boolean, True if within buffer window of a High event
+    - minutes_since_last_event: minutes since the most recent event
+
+    Parameters
+    ----------
+    df : DataFrame with DatetimeIndex (candle bars)
+    calendar_df : DataFrame with 'event_datetime' and 'impact_numeric' columns
+    cfg : dict, optional economic_calendar config section
+    """
+    if cfg is None:
+        cfg = {}
+
+    buffer_minutes = cfg.get("event_buffer_minutes", 30)
+
+    if calendar_df is None or calendar_df.empty:
+        df["minutes_to_next_event"] = np.nan
+        df["minutes_to_next_high_impact"] = np.nan
+        df["upcoming_high_impact_event"] = False
+        df["minutes_since_last_event"] = np.nan
+        return df
+
+    # Prepare calendar: ensure sorted datetime index
+    cal = calendar_df[["event_datetime"]].copy()
+    cal["event_datetime"] = pd.to_datetime(cal["event_datetime"], utc=True)
+    cal = cal.drop_duplicates(subset=["event_datetime"]).sort_values("event_datetime")
+
+    cal_high = calendar_df.loc[
+        calendar_df["impact_numeric"] >= 3, ["event_datetime"]
+    ].copy()
+    cal_high["event_datetime"] = pd.to_datetime(cal_high["event_datetime"], utc=True)
+    cal_high = cal_high.drop_duplicates(subset=["event_datetime"]).sort_values("event_datetime")
+
+    # Prepare candle times
+    candle_times = df.index.to_series().dt.tz_localize("UTC") if df.index.tz is None else df.index.to_series()
+    candle_times = candle_times.reset_index(drop=True)
+
+    # --- Forward merge: next event after each candle ---
+    candle_frame = pd.DataFrame({"candle_time": candle_times})
+
+    # Next event (any impact)
+    merged_fwd = pd.merge_asof(
+        candle_frame.sort_values("candle_time"),
+        cal.rename(columns={"event_datetime": "next_event_time"}),
+        left_on="candle_time",
+        right_on="next_event_time",
+        direction="forward",
+    )
+    df["minutes_to_next_event"] = (
+        (merged_fwd["next_event_time"] - merged_fwd["candle_time"])
+        .dt.total_seconds()
+        .values / 60.0
+    )
+
+    # Next high-impact event
+    if not cal_high.empty:
+        merged_high = pd.merge_asof(
+            candle_frame.sort_values("candle_time"),
+            cal_high.rename(columns={"event_datetime": "next_high_time"}),
+            left_on="candle_time",
+            right_on="next_high_time",
+            direction="forward",
+        )
+        df["minutes_to_next_high_impact"] = (
+            (merged_high["next_high_time"] - merged_high["candle_time"])
+            .dt.total_seconds()
+            .values / 60.0
+        )
+    else:
+        df["minutes_to_next_high_impact"] = np.nan
+
+    # Upcoming high impact: True if within buffer_minutes of a High event
+    df["upcoming_high_impact_event"] = (
+        df["minutes_to_next_high_impact"].notna()
+        & (df["minutes_to_next_high_impact"] <= buffer_minutes)
+    )
+
+    # --- Backward merge: last event before each candle ---
+    merged_bwd = pd.merge_asof(
+        candle_frame.sort_values("candle_time"),
+        cal.rename(columns={"event_datetime": "prev_event_time"}),
+        left_on="candle_time",
+        right_on="prev_event_time",
+        direction="backward",
+    )
+    df["minutes_since_last_event"] = (
+        (merged_bwd["candle_time"] - merged_bwd["prev_event_time"])
+        .dt.total_seconds()
+        .values / 60.0
+    )
+
+    return df
+
+
+def build_all_features(df, config=None, calendar_df=None):
     """
     Run a standard set of features. config is optional dict matching the
     'features' section of system.yaml.  Supports lists of windows for
     SMA and EMA so multiple columns are produced (e.g. sma_3, sma_20).
 
+    If calendar_df is provided, also adds economic calendar features.
+
     Returns DataFrame with new columns.
     """
     if config is None:
@@ -196,4 +298,9 @@ def build_all_features(df, config=None):
     # Drop intermediate helper columns
     df.drop(columns=["pv", "typical_price", "tr"], inplace=True, errors="ignore")
 
+    # Economic calendar features (optional)
+    if calendar_df is not None:
+        cal_cfg = config.get("economic_calendar", {}) if config else {}
+        df = add_calendar_features(df, calendar_df, cfg=cal_cfg)
+
     return df

src/economic_calendar.py

@@ -0,0 +1,380 @@
+# src/economic_calendar.py
+"""
+Economic calendar pipeline using Trading Economics API.
+Fetches economic events, uploads to Supabase, and provides query helpers
+for the backtester and data engine.
+
+Usage:
+    python src/economic_calendar.py              # full load (~400 days)
+    python src/economic_calendar.py --incremental  # fetch only recent missing data
+"""
+
+import os
+import sys
+import math
+import time as time_mod
+from datetime import datetime, timezone, timedelta
+
+import pandas as pd
+import numpy as np
+import tradingeconomics as te
+from supabase import create_client
+
+from config_loader import load_config, get_project_root
+
+cfg = load_config()
+
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_KEY = os.getenv("SUPABASE_KEY")
+TE_API_KEY = os.getenv("TRADING_ECONOMICS_API_KEY")
+
+IMPACT_MAP = {"Low": 1, "Medium": 2, "High": 3}
+
+
+# ---------------------------------------------------------------------------
+# Country / currency helpers
+# ---------------------------------------------------------------------------
+
+def get_countries_for_instruments(instruments, currency_country_map):
+    """
+    Derive the set of countries to fetch from configured instrument pairs.
+    E.g. ['EUR_USD'] -> {'euro area', 'united states'}
+    """
+    currencies = set()
+    for inst in instruments:
+        parts = inst.split("_")
+        currencies.update(parts)
+
+    countries = set()
+    for ccy in currencies:
+        country = currency_country_map.get(ccy)
+        if country:
+            countries.add(country)
+        else:
+            print(f"  Warning: no country mapping for currency {ccy}")
+    return countries
+
+
+# ---------------------------------------------------------------------------
+# Fetch from Trading Economics
+# ---------------------------------------------------------------------------
+
+def fetch_calendar_chunk(country, init_date, end_date):
+    """
+    Fetch calendar events for one country and one date window.
+    Returns a list of dicts from the TE API.
+    """
+    try:
+        data = te.getCalendarData(
+            country=country,
+            initDate=init_date,
+            endDate=end_date,
+            output_type="df",
+        )
+        if isinstance(data, pd.DataFrame) and not data.empty:
+            return data.to_dict(orient="records")
+        return []
+    except Exception as e:
+        print(f"  Error fetching {country} ({init_date} -> {end_date}): {e}")
+        return []
+
+
+def fetch_all_calendar_data(countries, days_back=400, chunk_days=28,
+                            sleep_between_chunks=1.0):
+    """
+    Paginate through Trading Economics calendar in chunk_days windows.
+    Returns a deduplicated DataFrame of all events.
+    """
+    end = datetime.now(timezone.utc)
+    start = end - timedelta(days=days_back)
+
+    all_records = []
+    for country in sorted(countries):
+        cursor = start
+        chunk_num = 0
+        while cursor < end:
+            chunk_end = min(cursor + timedelta(days=chunk_days), end)
+            chunk_num += 1
+            init_str = cursor.strftime("%Y-%m-%d")
+            end_str = chunk_end.strftime("%Y-%m-%d")
+            print(f"  {country}: chunk {chunk_num} ({init_str} -> {end_str}) ...", end=" ")
+
+            records = fetch_calendar_chunk(country, init_str, end_str)
+            print(f"{len(records)} events")
+            all_records.extend(records)
+
+            cursor = chunk_end
+            time_mod.sleep(sleep_between_chunks)
+
+    if not all_records:
+        print("  No calendar events fetched.")
+        return pd.DataFrame()
+
+    df = pd.DataFrame(all_records)
+    print(f"  Total raw events: {len(df)}")
+
+    # Normalize column names (TE API returns mixed-case)
+    col_map = {}
+    for col in df.columns:
+        col_map[col] = col.lower().replace(" ", "_")
+    df = df.rename(columns=col_map)
+
+    # Parse datetime
+    date_col = None
+    for candidate in ["date", "datetime", "event_datetime"]:
+        if candidate in df.columns:
+            date_col = candidate
+            break
+    if date_col is None:
+        print("  Warning: no date column found in TE response.")
+        return pd.DataFrame()
+
+    df["event_datetime"] = pd.to_datetime(df[date_col], utc=True, errors="coerce")
+    df = df.dropna(subset=["event_datetime"])
+
+    # Map category/event name
+    if "category" in df.columns and "event" not in df.columns:
+        df["event"] = df["category"]
+    event_col = "event" if "event" in df.columns else "category"
+
+    # Map impact to numeric
+    if "importance" in df.columns and "impact" not in df.columns:
+        # TE uses 'importance' with values like 1,2,3 or Low/Medium/High
+        importance = df["importance"]
+        if importance.dtype in (int, float, np.int64, np.float64):
+            df["impact_numeric"] = importance.astype(int)
+            reverse_map = {1: "Low", 2: "Medium", 3: "High"}
+            df["impact"] = df["impact_numeric"].map(reverse_map).fillna("Low")
+        else:
+            df["impact"] = importance.astype(str)
+            df["impact_numeric"] = df["impact"].map(IMPACT_MAP).fillna(1).astype(int)
+    elif "impact" in df.columns:
+        df["impact_numeric"] = df["impact"].map(IMPACT_MAP).fillna(1).astype(int)
+    else:
+        df["impact"] = "Low"
+        df["impact_numeric"] = 1
+
+    # Standardize column names for our schema
+    rename = {}
+    if event_col != "event_name":
+        rename[event_col] = "event_name"
+    for te_col, our_col in [("actual", "actual"), ("forecast", "forecast"),
+                             ("previous", "previous"), ("revised", "revised"),
+                             ("teforecast", "forecast"), ("teprevious", "previous"),
+                             ("reference", "reference"), ("source", "source"),
+                             ("currency", "currency"), ("country", "country")]:
+        if te_col in df.columns and te_col != our_col:
+            rename[te_col] = our_col
+    df = df.rename(columns=rename)
+
+    # Ensure required columns exist
+    for col in ["country", "currency", "event_name", "actual", "forecast",
+                "previous", "revised", "reference", "source"]:
+        if col not in df.columns:
+            df[col] = None
+
+    # Select only the columns we need
+    keep = ["event_datetime", "country", "currency", "event_name",
+            "impact", "impact_numeric", "actual", "forecast",
+            "previous", "revised", "reference", "source"]
+    df = df[[c for c in keep if c in df.columns]]
+
+    # Deduplicate on composite key
+    df = df.drop_duplicates(subset=["event_datetime", "country", "event_name"], keep="last")
+    df = df.sort_values("event_datetime").reset_index(drop=True)
+    print(f"  After dedup: {len(df)} events")
+
+    return df
+
+
+# ---------------------------------------------------------------------------
+# Supabase upload
+# ---------------------------------------------------------------------------
+
+def calendar_df_to_records(df):
+    """
+    Clean a calendar DataFrame for Supabase upsert.
+    Mirrors supabase_upload.df_to_records sanitization patterns.
+    """
+    df2 = df.copy()
+
+    # Convert timestamps to ISO strings
+    df2["event_datetime"] = df2["event_datetime"].apply(
+        lambda t: pd.to_datetime(t).isoformat() if pd.notna(t) else None
+    )
+    df2["last_updated"] = datetime.now(timezone.utc).isoformat()
+
+    # Replace inf/-inf with NaN
+    df2 = df2.replace([np.inf, -np.inf], np.nan)
+
+    def _convert(v):
+        if pd.isna(v):
+            return None
+        if isinstance(v, (bool,)):
+            return bool(v)
+        if isinstance(v, (np.integer, int)):
+            return int(v)
+        if isinstance(v, (np.floating, float)):
+            fv = float(v)
+            return fv if math.isfinite(fv) else None
+        if isinstance(v, str):
+            return v
+        try:
+            return str(v)
+        except Exception:
+            return None
+
+    for col in df2.columns:
+        df2[col] = df2[col].apply(_convert)
+
+    records = df2.to_dict(orient="records")
+
+    # Final NaN/inf sweep
+    def _sanitize(v):
+        if isinstance(v, float) and (math.isnan(v) or math.isinf(v)):
+            return None
+        return v
+
+    records = [{k: _sanitize(v) for k, v in row.items()} for row in records]
+    return records
+
+
+def upload_calendar(df, supabase_client, table="economic_calendar", chunk_size=500):
+    """
+    Upsert calendar records to Supabase in batches.
+    """
+    records = calendar_df_to_records(df)
+    total = len(records)
+    if total == 0:
+        print("  No calendar rows to upload.")
+        return
+
+    chunks = math.ceil(total / chunk_size)
+    print(f"  Uploading {total} calendar rows in {chunks} chunk(s)...")
+
+    for i in range(chunks):
+        start = i * chunk_size
+        end = start + chunk_size
+        batch = records[start:end]
+
+        resp = supabase_client.table(table).upsert(batch).execute()
+        if hasattr(resp, "data") and resp.data is not None:
+            pass  # success
+        elif getattr(resp, "status_code", None) not in (200, 201, None):
+            print(f"  Upload chunk error: {resp}")
+            raise SystemExit("Calendar upsert failed")
+        print(f"  Chunk {i + 1}/{chunks} uploaded ({len(batch)} rows).")
+
+
+# ---------------------------------------------------------------------------
+# Supabase read (for backtester + data_engine)
+# ---------------------------------------------------------------------------
+
+def fetch_calendar_from_supabase(supabase_client, table="economic_calendar",
+                                 currencies=None, impact_min=None,
+                                 start_time=None, end_time=None):
+    """
+    Read calendar events from Supabase with optional filters.
+    Returns a DataFrame sorted by event_datetime.
+    """
+    all_rows = []
+    page_size = 1000
+    offset = 0
+
+    while True:
+        query = supabase_client.table(table).select("*")
+
+        if currencies:
+            # Filter by currency (case-insensitive via ilike would be ideal,
+            # but supabase-py .in_ works for exact match)
+            query = query.in_("currency", list(currencies))
+        if impact_min is not None:
+            query = query.gte("impact_numeric", impact_min)
+        if start_time:
+            query = query.gte("event_datetime", start_time)
+        if end_time:
+            query = query.lte("event_datetime", end_time)
+
+        query = query.order("event_datetime", desc=False)
+        query = query.range(offset, offset + page_size - 1)
+        resp = query.execute()
+
+        rows = resp.data or []
+        all_rows.extend(rows)
+        if len(rows) < page_size:
+            break
+        offset += page_size
+
+    if not all_rows:
+        return pd.DataFrame()
+
+    df = pd.DataFrame(all_rows)
+    df["event_datetime"] = pd.to_datetime(df["event_datetime"], utc=True)
+    df = df.sort_values("event_datetime").reset_index(drop=True)
+
+    # Convert numeric columns
+    for col in ["impact_numeric", "actual", "forecast", "previous", "revised"]:
+        if col in df.columns:
+            df[col] = pd.to_numeric(df[col], errors="coerce")
+
+    return df
+
+
+# ---------------------------------------------------------------------------
+# CLI entry point
+# ---------------------------------------------------------------------------
+
+def main():
+    if not TE_API_KEY:
+        raise SystemExit("Missing TRADING_ECONOMICS_API_KEY in config/.env")
+    if not SUPABASE_URL or not SUPABASE_KEY:
+        raise SystemExit("Missing SUPABASE_URL or SUPABASE_KEY in config/.env")
+
+    # Authenticate with Trading Economics
+    te.login(TE_API_KEY)
+
+    sb = create_client(SUPABASE_URL, SUPABASE_KEY)
+
+    cal_cfg = cfg.get("economic_calendar", {})
+    table = cal_cfg.get("table", "economic_calendar")
+    days_back = cal_cfg.get("days_back", 400)
+    chunk_days = cal_cfg.get("chunk_days", 28)
+    sleep_s = cal_cfg.get("sleep_between_chunks", 1.0)
+    currency_country_map = cal_cfg.get("currency_country_map", {})
+
+    instruments = cfg["brokers"][0]["instruments"]
+    countries = get_countries_for_instruments(instruments, currency_country_map)
+
+    if not countries:
+        raise SystemExit("No countries derived from instruments. Check currency_country_map.")
+
+    incremental = "--incremental" in sys.argv
+
+    if incremental:
+        # Fetch only the last 30 days
+        days_back = 30
+        print(f"Incremental mode: fetching last {days_back} days")
+    else:
+        print(f"Full load: fetching last {days_back} days")
+
+    print(f"Countries: {sorted(countries)}")
+    print(f"Instruments: {instruments}\n")
+
+    df = fetch_all_calendar_data(
+        countries=countries,
+        days_back=days_back,
+        chunk_days=chunk_days,
+        sleep_between_chunks=sleep_s,
+    )
+
+    if df.empty:
+        print("No events to upload.")
+        return
+
+    print(f"\nUploading to Supabase table '{table}'...")
+    upload_calendar(df, sb, table=table)
+    print("Economic calendar load complete.")
+
+
+if __name__ == "__main__":
+    main()