diff --git a/config/.env.example b/config/.env.example index e619964..df87e4a 100644 --- a/config/.env.example +++ b/config/.env.example @@ -4,3 +4,4 @@ OANDA_ENV=practice DASHBOARD_PASSWORD=changeme SUPABASE_URL=https://your-project.supabase.co SUPABASE_KEY=your-supabase-anon-key-here +TRADING_ECONOMICS_API_KEY=your-trading-economics-api-key-here diff --git a/config/system.yaml b/config/system.yaml index fac2256..ac7407a 100644 --- a/config/system.yaml +++ b/config/system.yaml @@ -62,6 +62,24 @@ execution: canary_size_pct: 0.01 max_positions: 5 interval_seconds: 60 +economic_calendar: + enabled: true + table: economic_calendar + days_back: 400 + chunk_days: 28 + sleep_between_chunks: 1.0 + impact_threshold: "High" + event_buffer_minutes: 30 + feature_lookforward_minutes: 240 + currency_country_map: + EUR: "euro area" + USD: "united states" + GBP: "united kingdom" + JPY: "japan" + AUD: "australia" + CAD: "canada" + CHF: "switzerland" + NZD: "new zealand" supabase: url: https://.supabase.co key_env_name: SUPABASE_KEY diff --git a/requirements.txt b/requirements.txt index 9865792..2f547d6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,3 +14,4 @@ python-dotenv>=1.0 PyYAML>=6.0 flask>=3.0 flask-httpauth>=4.8 +tradingeconomics diff --git a/sql/create_economic_calendar.sql b/sql/create_economic_calendar.sql new file mode 100644 index 0000000..43e990e --- /dev/null +++ b/sql/create_economic_calendar.sql @@ -0,0 +1,20 @@ +CREATE TABLE economic_calendar ( + event_datetime TIMESTAMPTZ NOT NULL, + country TEXT NOT NULL, + event_name TEXT NOT NULL, + currency TEXT, + impact TEXT, + impact_numeric INTEGER, + actual DOUBLE PRECISION, + forecast DOUBLE PRECISION, + previous DOUBLE PRECISION, + revised DOUBLE PRECISION, + reference TEXT, + source TEXT, + last_updated TIMESTAMPTZ DEFAULT now(), + PRIMARY KEY (event_datetime, country, event_name) +); + +CREATE INDEX idx_ec_datetime ON economic_calendar (event_datetime DESC); +CREATE INDEX idx_ec_currency ON economic_calendar (currency); +CREATE INDEX idx_ec_impact ON economic_calendar (impact_numeric); diff --git a/src/backtester.py b/src/backtester.py index e76f4a3..144f3ba 100644 --- a/src/backtester.py +++ b/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.") diff --git a/src/data_engine.py b/src/data_engine.py index b765ac1..1b537d1 100644 --- a/src/data_engine.py +++ b/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 diff --git a/src/economic_calendar.py b/src/economic_calendar.py new file mode 100644 index 0000000..c76aaec --- /dev/null +++ b/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()