""" Economic calendar router — FRED-backed historical events + Forex Factory calendar. Prefix: /api/eco """ import json import logging import os from typing import Any, Dict, List, Optional from fastapi import APIRouter, BackgroundTasks, File, HTTPException, Query, UploadFile logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/eco", tags=["Economic Calendar"]) # ── Helpers ─────────────────────────────────────────────────────────────────── def _parse_row(row: Dict) -> Dict: for field in ("assets_impacted",): val = row.get(field) if isinstance(val, str): try: row[field] = json.loads(val) if val else [] except Exception: row[field] = [] elif val is None: row[field] = [] return row # ── Bootstrap ───────────────────────────────────────────────────────────────── _bootstrap_status: Dict[str, Any] = {"running": False, "last_result": None} def _run_bootstrap(from_date: str, series_ids: Optional[List[str]], force: bool): global _bootstrap_status _bootstrap_status["running"] = True try: from services.fred_bootstrap import bootstrap_fred result = bootstrap_fred(from_date=from_date, series_ids=series_ids, force=force) _bootstrap_status["last_result"] = result total_inserted = sum(v.get("inserted", 0) for v in result.values() if isinstance(v, dict)) logger.info(f"[eco/bootstrap] Done — {total_inserted} rows inserted") except Exception as e: logger.error(f"[eco/bootstrap] Failed: {e}") _bootstrap_status["last_result"] = {"error": str(e)} finally: _bootstrap_status["running"] = False @router.get("/fred-key") def get_fred_key() -> Dict[str, Any]: """Check if a FRED API key is configured.""" from services.database import get_config key = get_config("fred_api_key") or "" return { "configured": bool(key), "preview": (key[:6] + "…" + key[-4:]) if len(key) > 10 else ("***" if key else ""), } @router.post("/fred-key") def save_fred_key(key: str = Query(..., description="FRED API key")) -> Dict[str, Any]: """Save FRED API key to DB config.""" from services.database import set_config key = key.strip() if not key: raise HTTPException(400, "Empty key") set_config("fred_api_key", key) return {"status": "saved", "preview": key[:6] + "…" + key[-4:]} @router.post("/bootstrap") def bootstrap( background_tasks: BackgroundTasks, from_date: str = Query("2020-01-01", description="Start date for historical data"), series: Optional[str] = Query(None, description="Comma-separated series IDs (blank = all)"), force: bool = Query(False, description="Overwrite existing rows"), ) -> Dict[str, Any]: """ Trigger FRED data bootstrap (runs in background). Fetches public CSV endpoint — no API key required. """ if _bootstrap_status["running"]: raise HTTPException(409, "Bootstrap already running") series_ids = [s.strip() for s in series.split(",") if s.strip()] if series else None background_tasks.add_task(_run_bootstrap, from_date, series_ids, force) return {"status": "started", "from_date": from_date, "series": series_ids or "all"} @router.get("/bootstrap/status") def bootstrap_status() -> Dict[str, Any]: return _bootstrap_status # ── Series catalog ──────────────────────────────────────────────────────────── @router.get("/series") def list_series() -> List[Dict[str, Any]]: """Return available FRED series metadata.""" from services.fred_bootstrap import FRED_SERIES, CATEGORIES return [ { "id": sid, "name": meta["name"], "category": meta["category"], "freq": meta["freq"], "unit": meta["unit"], "assets": meta["assets"], } for sid, meta in FRED_SERIES.items() ] # ── Events list ─────────────────────────────────────────────────────────────── _SORT_COLS = { "date": "ee.event_date", "zscore": "ABS(COALESCE(ee.surprise_zscore, 0))", "series": "ee.series_id", "name": "ee.event_name", } @router.get("/events") def list_eco_events( date_from: Optional[str] = Query(None), date_to: Optional[str] = Query(None), series: Optional[str] = Query(None, description="Comma-separated series IDs"), category: Optional[str] = Query(None, description="employment|inflation|growth|monetary|credit|rates"), min_zscore: float = Query(0.0, ge=0.0, description="Minimum |z-score| filter"), direction: Optional[str] = Query(None, description="bullish|bearish|neutral"), sort_by: str = Query("date", description="date|zscore|series|name"), sort_dir: str = Query("desc", description="asc|desc"), limit: int = Query(200, ge=1, le=2000), offset: int = Query(0, ge=0), ) -> Dict[str, Any]: from services.database import get_conn from services.fred_bootstrap import FRED_SERIES conn = get_conn() where_parts: List[str] = [] params: List[Any] = [] if date_from: where_parts.append("ee.event_date >= ?") params.append(date_from) if date_to: where_parts.append("ee.event_date <= ?") params.append(date_to) if series: ids = [s.strip() for s in series.split(",") if s.strip()] if ids: placeholders = ",".join("?" * len(ids)) where_parts.append(f"ee.series_id IN ({placeholders})") params.extend(ids) if category: # Map category → series IDs matching = [sid for sid, m in FRED_SERIES.items() if m["category"] == category] if matching: placeholders = ",".join("?" * len(matching)) where_parts.append(f"ee.series_id IN ({placeholders})") params.extend(matching) else: # No matching series — return empty conn.close() return {"total": 0, "offset": offset, "limit": limit, "events": []} if min_zscore > 0: where_parts.append("ABS(COALESCE(ee.surprise_zscore, 0)) >= ?") params.append(min_zscore) if direction: where_parts.append("ee.surprise_direction = ?") params.append(direction) where_sql = ("WHERE " + " AND ".join(where_parts)) if where_parts else "" sort_col = _SORT_COLS.get(sort_by, "ee.event_date") order_sql = f"ORDER BY {sort_col} {'DESC' if sort_dir == 'desc' else 'ASC'}" count_sql = f"SELECT COUNT(*) FROM economic_events ee {where_sql}" data_sql = f"SELECT ee.* FROM economic_events ee {where_sql} {order_sql} LIMIT ? OFFSET ?" try: total = conn.execute(count_sql, params).fetchone()[0] rows = conn.execute(data_sql, params + [limit, offset]).fetchall() finally: conn.close() events = [_parse_row(dict(r)) for r in rows] return {"total": total, "offset": offset, "limit": limit, "events": events} # ── Upcoming releases ──────────────────────────────────────────────────────── # Typical publication lag (days) after the end of the reference period _RELEASE_LAG: Dict[str, Dict[str, Any]] = { "PAYEMS": {"freq_days": 30, "lag_days": 35, "note": "1er vendredi du mois"}, "UNRATE": {"freq_days": 30, "lag_days": 35, "note": "Avec NFP"}, "CPIAUCSL": {"freq_days": 30, "lag_days": 15, "note": "Mi-mois"}, "CPILFESL": {"freq_days": 30, "lag_days": 15, "note": "Avec CPI"}, "PCEPILFE": {"freq_days": 30, "lag_days": 30, "note": "Fin de mois"}, "FEDFUNDS": {"freq_days": 45, "lag_days": 45, "note": "Réunion FOMC (~8/an)"}, "ICSA": {"freq_days": 7, "lag_days": 4, "note": "Chaque jeudi"}, "GDPC1": {"freq_days": 90, "lag_days": 30, "note": "Estimation avancée BEA"}, "BAMLH0A0HYM2": {"freq_days": 7, "lag_days": 1, "note": "Hebdomadaire"}, "T10Y2Y": {"freq_days": 7, "lag_days": 1, "note": "Hebdomadaire"}, "T10Y3M": {"freq_days": 7, "lag_days": 1, "note": "Hebdomadaire"}, } @router.get("/upcoming") def upcoming_events() -> List[Dict[str, Any]]: """Estimate next release dates for each series based on last stored date + typical lag.""" from services.database import get_conn from services.fred_bootstrap import FRED_SERIES from datetime import date, timedelta conn = get_conn() today = date.today() upcoming = [] for sid, lag_info in _RELEASE_LAG.items(): meta = FRED_SERIES.get(sid) if not meta: continue try: row = conn.execute( "SELECT MAX(event_date), actual_value, actual_unit, surprise_direction " "FROM economic_events WHERE series_id=?", (sid,), ).fetchone() if not row or not row[0]: # No data in DB yet — still show as "not loaded" upcoming.append({ "series_id": sid, "name": meta["name"], "note": lag_info["note"], "category": meta["category"], "last_release": None, "last_value": None, "last_unit": meta["unit"], "last_direction": None, "next_expected": None, "days_until": None, "status": "no_data", }) continue last_date = date.fromisoformat(row[0]) freq_days = lag_info["freq_days"] lag_days = lag_info["lag_days"] # Next expected = last ref period end + lag next_release = last_date + timedelta(days=freq_days + lag_days) days_until = (next_release - today).days if days_until < -freq_days: # Very overdue — likely bootstrap missing recent data status = "overdue" elif days_until < 0: status = "due_soon" # probably released, not yet in DB elif days_until <= 7: status = "imminent" elif days_until <= 30: status = "upcoming" else: status = "scheduled" upcoming.append({ "series_id": sid, "name": meta["name"], "note": lag_info["note"], "category": meta["category"], "last_release": str(last_date), "last_value": round(float(row[1]), 3) if row[1] is not None else None, "last_unit": row[2] or meta["unit"], "last_direction": row[3], "next_expected": str(next_release), "days_until": days_until, "status": status, }) except Exception as e: logger.debug(f"[eco/upcoming] {sid}: {e}") conn.close() upcoming.sort(key=lambda x: (x["next_expected"] or "9999", x["series_id"])) return upcoming # ── Forex Factory calendar ──────────────────────────────────────────────────── _ff_import_status: Dict[str, Any] = {"running": False, "last_result": None} _ff_sync_status: Dict[str, Any] = {"running": False, "last_result": None} _ff_scrape_status: Dict[str, Any] = {"running": False, "last_result": None} _te_sync_status: Dict[str, Any] = {"running": False, "last_result": None} _fxs_sync_status: Dict[str, Any] = {"running": False, "last_result": None} _te_html_status: Dict[str, Any] = {"running": False, "last_result": None} _TE_HTML_CANDIDATES = [ "/app/calendar_data.txt", "/tmp/calendar_data.txt", os.path.join(os.path.dirname(__file__), "..", "..", "calendar_data.txt"), os.path.join(os.path.dirname(__file__), "..", "calendar_data.txt"), ] def _find_te_html() -> str | None: for p in _TE_HTML_CANDIDATES: full = os.path.abspath(p) if os.path.exists(full): return full return None # CSV search order: Docker image (/app), uploaded to /tmp, local dev paths _FF_CSV_CANDIDATES = [ "/app/forex_factory_cache.csv", # Docker (backend/ copied to /app) "/tmp/forex_factory_cache.csv", # browser upload os.path.join(os.path.dirname(__file__), "..", "forex_factory_cache.csv"), # local dev (backend/) os.path.join(os.path.dirname(__file__), "..", "..", "forex_factory_cache.csv"), # project root ] def _find_csv() -> str | None: for p in _FF_CSV_CANDIDATES: full = os.path.abspath(p) if os.path.exists(full): return full return None def _run_ff_import(csv_path: str): global _ff_import_status _ff_import_status["running"] = True try: from services.ff_calendar import import_csv result = import_csv(csv_path) _ff_import_status["last_result"] = result except Exception as e: logger.error(f"[eco/ff-import] Failed: {e}") _ff_import_status["last_result"] = {"error": str(e)} finally: _ff_import_status["running"] = False def _run_ff_sync(): global _ff_sync_status _ff_sync_status["running"] = True try: from services.ff_calendar import sync_live result = sync_live() _ff_sync_status["last_result"] = result except Exception as e: logger.error(f"[eco/ff-sync] Failed: {e}") _ff_sync_status["last_result"] = {"error": str(e)} finally: _ff_sync_status["running"] = False @router.post("/ff-upload") async def ff_upload(file: UploadFile = File(...)) -> Dict[str, Any]: """Upload forex_factory_cache.csv directly from the browser (saves to /tmp).""" dest = "/tmp/forex_factory_cache.csv" try: content = await file.read() with open(dest, "wb") as f: f.write(content) size_mb = round(len(content) / 1_048_576, 1) print(f"[FF upload] Saved {size_mb} MB to {dest}", flush=True) return {"status": "uploaded", "path": dest, "size_mb": size_mb} except Exception as e: raise HTTPException(500, f"Upload failed: {e}") @router.post("/ff-import") def ff_import(background_tasks: BackgroundTasks) -> Dict[str, Any]: """Import forex_factory_cache.csv into ff_calendar table (background).""" if _ff_import_status["running"]: raise HTTPException(409, "Import already running") csv_path = _find_csv() if not csv_path: raise HTTPException(404, "CSV not found. Upload it first via POST /api/eco/ff-upload") background_tasks.add_task(_run_ff_import, csv_path) return {"status": "started", "csv": csv_path} @router.get("/ff-import/status") def ff_import_status() -> Dict[str, Any]: return _ff_import_status @router.post("/ff-sync") def ff_sync(background_tasks: BackgroundTasks) -> Dict[str, Any]: """Fetch this week + next week from faireconomy.media and upsert.""" if _ff_sync_status["running"]: raise HTTPException(409, "Sync already running") background_tasks.add_task(_run_ff_sync) return {"status": "started"} @router.get("/ff-sync/status") def ff_sync_status_ep() -> Dict[str, Any]: return _ff_sync_status def _run_ff_scrape(weeks_ahead: int = 5): global _ff_scrape_status _ff_scrape_status["running"] = True try: from services.ff_calendar import scrape_upcoming result = scrape_upcoming(weeks_ahead=weeks_ahead) _ff_scrape_status["last_result"] = result except Exception as e: logger.error(f"[eco/ff-scrape] Failed: {e}") _ff_scrape_status["last_result"] = {"error": str(e)} finally: _ff_scrape_status["running"] = False @router.post("/ff-scrape") def ff_scrape( background_tasks: BackgroundTasks, weeks: int = Query(5, ge=1, le=8, description="Number of weeks ahead to scrape"), ) -> Dict[str, Any]: """Scrape forexfactory.com HTML calendar for upcoming N weeks (incl. forecasts).""" if _ff_scrape_status["running"]: raise HTTPException(409, "Scrape already running") background_tasks.add_task(_run_ff_scrape, weeks) return {"status": "started", "weeks_ahead": weeks} @router.get("/ff-scrape/status") def ff_scrape_status_ep() -> Dict[str, Any]: return _ff_scrape_status # ── FMP calendar (upcoming weeks with consensus forecasts) ──────────────────── @router.get("/fmp-key") def get_fmp_key() -> Dict[str, Any]: from services.fmp_calendar import check_fmp_key return check_fmp_key() @router.post("/fmp-key") def save_fmp_key(key: str = Query(..., description="FMP API key")) -> Dict[str, Any]: from services.database import set_config key = key.strip() if not key: raise HTTPException(400, "Empty key") set_config("fmp_api_key", key) return {"status": "saved", "preview": key[:4] + "…" + key[-4:]} def _run_fmp_sync(weeks_ahead: int): global _te_sync_status _te_sync_status["running"] = True try: from services.fmp_calendar import fetch_upcoming result = fetch_upcoming(weeks_ahead=weeks_ahead) _te_sync_status["last_result"] = result except Exception as e: logger.error(f"[eco/fmp-sync] Failed: {e}") _te_sync_status["last_result"] = {"error": str(e)} finally: _te_sync_status["running"] = False @router.post("/fmp-sync") def fmp_sync( background_tasks: BackgroundTasks, weeks: int = Query(6, ge=1, le=12), ) -> Dict[str, Any]: """Fetch upcoming economic events + forecasts from FMP API (free, 250 req/day).""" if _te_sync_status["running"]: raise HTTPException(409, "FMP sync already running") background_tasks.add_task(_run_fmp_sync, weeks) return {"status": "started", "weeks_ahead": weeks} @router.get("/fmp-sync/status") def fmp_sync_status_ep() -> Dict[str, Any]: return _te_sync_status # ── Trading Economics HTML import ───────────────────────────────────────────── @router.post("/te-html-upload") async def te_html_upload(file: UploadFile = File(...)) -> Dict[str, Any]: """Upload calendar_data.txt (TE HTML page) from the browser.""" dest = "/tmp/calendar_data.txt" try: content = await file.read() with open(dest, "wb") as f: f.write(content) size_mb = round(len(content) / 1_048_576, 1) print(f"[TE HTML upload] Saved {size_mb} MB to {dest}", flush=True) return {"status": "uploaded", "path": dest, "size_mb": size_mb} except Exception as e: raise HTTPException(500, f"Upload failed: {e}") def _run_te_html_import(): global _te_html_status _te_html_status["running"] = True try: from services.te_html_parser import import_html_file path = _find_te_html() if not path: _te_html_status["last_result"] = {"error": "calendar_data.txt not found — upload first"} return result = import_html_file(path) _te_html_status["last_result"] = result except Exception as e: logger.error(f"[eco/te-html-import] Failed: {e}") _te_html_status["last_result"] = {"error": str(e)} finally: _te_html_status["running"] = False @router.post("/te-html-import") def te_html_import(background_tasks: BackgroundTasks) -> Dict[str, Any]: """Parse calendar_data.txt (TE HTML) and upsert into ff_calendar.""" if _te_html_status["running"]: raise HTTPException(409, "TE HTML import already running") if not _find_te_html(): raise HTTPException(404, "calendar_data.txt not found — upload via POST /api/eco/te-html-upload") background_tasks.add_task(_run_te_html_import) return {"status": "started"} @router.get("/te-html-import/status") def te_html_import_status() -> Dict[str, Any]: return _te_html_status # ── FXStreet calendar (no API key required) ─────────────────────────────────── def _run_fxs_sync(weeks_ahead: int): global _fxs_sync_status _fxs_sync_status["running"] = True try: from services.fxstreet_calendar import fetch_upcoming result = fetch_upcoming(weeks_ahead=weeks_ahead) _fxs_sync_status["last_result"] = result except Exception as e: logger.error(f"[eco/fxs-sync] Failed: {e}") _fxs_sync_status["last_result"] = {"error": str(e)} finally: _fxs_sync_status["running"] = False @router.post("/fxs-sync") def fxs_sync( background_tasks: BackgroundTasks, weeks: int = Query(6, ge=1, le=12), ) -> Dict[str, Any]: """Fetch upcoming events + forecasts from FXStreet (no API key needed).""" if _fxs_sync_status["running"]: raise HTTPException(409, "FXStreet sync already running") background_tasks.add_task(_run_fxs_sync, weeks) return {"status": "started", "weeks_ahead": weeks} @router.get("/fxs-sync/status") def fxs_sync_status_ep() -> Dict[str, Any]: return _fxs_sync_status # ── Unified calendar sync ───────────────────────────────────────────────────── def _run_calendar_sync(weeks_ahead: int): from services.calendar_sync import calendar_sync try: calendar_sync(weeks_ahead=weeks_ahead) except Exception as e: logger.error(f"[eco/calendar-sync] Failed: {e}") @router.post("/calendar-sync") def calendar_sync_ep( background_tasks: BackgroundTasks, weeks: int = Query(8, ge=1, le=12), ) -> Dict[str, Any]: """Trigger unified calendar sync (FXStreet primary → FF fallback) in background.""" from services.calendar_sync import get_sync_status if get_sync_status()["running"]: raise HTTPException(409, "Calendar sync already running") background_tasks.add_task(_run_calendar_sync, weeks) return {"status": "started", "weeks_ahead": weeks} @router.get("/calendar-sync/status") def calendar_sync_status_ep() -> Dict[str, Any]: """Get unified calendar sync status (last_run, source, last_result, next_run).""" from services.calendar_sync import get_sync_status return get_sync_status() @router.get("/calendar") def ff_calendar( period: str = Query("recent", description="recent|today|tomorrow|yesterday|this_week|next_week|previous_week|this_month|next_month|previous_month|custom"), date_from: Optional[str] = Query(None, description="YYYY-MM-DD (used when period=custom)"), date_to: Optional[str] = Query(None, description="YYYY-MM-DD (used when period=custom)"), currencies: Optional[str] = Query(None, description="Comma-separated: USD,EUR,GBP,..."), impacts: Optional[str] = Query(None, description="Comma-separated: high,medium,low"), limit: int = Query(500, ge=1, le=5000), offset: int = Query(0, ge=0), ) -> Dict[str, Any]: """Unified Forex Factory calendar — past releases + upcoming events.""" from services.ff_calendar import get_calendar cur_list = [c.strip().upper() for c in currencies.split(",") if c.strip()] if currencies else None imp_list = [i.strip().lower() for i in impacts.split(",") if i.strip()] if impacts else None return get_calendar( period=period, date_from=date_from, date_to=date_to, currencies=cur_list, impacts=imp_list, limit=limit, offset=offset, ) @router.get("/series/{series_id}/history") def series_history( series_id: str, from_date: Optional[str] = Query(None, description="YYYY-MM-DD"), to_date: Optional[str] = Query(None, description="YYYY-MM-DD"), ) -> Dict[str, Any]: """ Time series data for a FRED series (from economic_events table) combined with the matching FF calendar events (surprises, forecasts). """ from services.database import get_conn from services.fred_bootstrap import FRED_SERIES meta = FRED_SERIES.get(series_id) if not meta: raise HTTPException(404, f"Unknown series: {series_id}") conn = get_conn() try: # ── FRED time series ────────────────────────────────────────────── ts_where = ["series_id = ?"] ts_params: list = [series_id] if from_date: ts_where.append("event_date >= ?"); ts_params.append(from_date) if to_date: ts_where.append("event_date <= ?"); ts_params.append(to_date) ts_rows = conn.execute( f"SELECT event_date, actual_value, surprise_zscore, surprise_direction " f"FROM economic_events WHERE {' AND '.join(ts_where)} ORDER BY event_date ASC", ts_params, ).fetchall() # ── FF events for this series (no date ceiling — include future releases) ── ff_where = ["series_id = ?"] ff_params: list = [series_id] if from_date: ff_where.append("event_date >= ?"); ff_params.append(from_date) ff_rows = conn.execute( f"SELECT event_date, event_time, event_name, actual_value, " f"forecast_value, previous_value, impact " f"FROM ff_calendar WHERE {' AND '.join(ff_where)} ORDER BY event_date ASC", ff_params, ).fetchall() return { "series_id": series_id, "name": meta["name"], "unit": meta["unit"], "category": meta["category"], "freq": meta["freq"], "timeseries": [dict(r) for r in ts_rows], "events": [dict(r) for r in ff_rows], } finally: conn.close() @router.get("/ff-stats") def ff_stats() -> Dict[str, Any]: """Quick inventory of ff_calendar table.""" from services.database import get_conn conn = get_conn() try: total = conn.execute("SELECT COUNT(*) FROM ff_calendar").fetchone()[0] if total == 0: return {"total": 0} earliest = conn.execute("SELECT MIN(event_date) FROM ff_calendar").fetchone()[0] latest = conn.execute("SELECT MAX(event_date) FROM ff_calendar").fetchone()[0] by_currency = conn.execute( "SELECT currency, COUNT(*) AS cnt FROM ff_calendar GROUP BY currency ORDER BY cnt DESC" ).fetchall() by_impact = conn.execute( "SELECT impact, COUNT(*) AS cnt FROM ff_calendar GROUP BY impact ORDER BY cnt DESC" ).fetchall() return { "total": total, "earliest": earliest, "latest": latest, "by_currency": [dict(r) for r in by_currency], "by_impact": [dict(r) for r in by_impact], } finally: conn.close() # ── DB count summary ────────────────────────────────────────────────────────── @router.get("/status") def eco_status() -> Dict[str, Any]: from services.database import get_conn conn = get_conn() try: total = conn.execute("SELECT COUNT(*) FROM economic_events").fetchone()[0] latest = conn.execute( "SELECT MAX(event_date) FROM economic_events" ).fetchone()[0] earliest = conn.execute( "SELECT MIN(event_date) FROM economic_events" ).fetchone()[0] by_series = conn.execute( "SELECT series_id, COUNT(*) as cnt, MAX(event_date) as latest " "FROM economic_events GROUP BY series_id ORDER BY series_id" ).fetchall() return { "total": total, "earliest": earliest, "latest": latest, "by_series": [dict(r) for r in by_series], } finally: conn.close() # ── Macro Series Log ────────────────────────────────────────────────────────── @router.get("/series-log") def get_series_log( series_id: str = Query(...), from_date: str = Query("2020-01-01"), limit: int = Query(500, le=2000), ) -> List[Dict[str, Any]]: """Return logged actual/forecast snapshots for a given series (change events only).""" from services.database import get_conn conn = get_conn() try: rows = conn.execute( """SELECT series_id, event_name, event_date, actual_value, forecast_value, previous_value, currency, impact, logged_at, source, changed FROM macro_series_log WHERE series_id = ? AND event_date >= ? ORDER BY event_date ASC, logged_at ASC LIMIT ?""", (series_id, from_date, limit) ).fetchall() return [dict(r) for r in rows] finally: conn.close() @router.get("/series-log/distinct-series") def list_logged_series() -> List[Dict[str, Any]]: """Return all series_id that have entries in macro_series_log.""" from services.database import get_conn conn = get_conn() try: rows = conn.execute( """SELECT series_id, event_name, COUNT(*) as log_count, MIN(event_date) as first_date, MAX(event_date) as last_date, MAX(logged_at) as last_logged FROM macro_series_log GROUP BY series_id ORDER BY last_date DESC""" ).fetchall() return [dict(r) for r in rows] finally: conn.close() @router.post("/series-log/backfill") def backfill_series_log() -> Dict[str, Any]: """Seed macro_series_log from existing ff_calendar rows (one-time historical import).""" from services.database import get_conn from services.macro_series_log import backfill_from_ff_calendar conn = get_conn() try: result = backfill_from_ff_calendar(conn) return result finally: conn.close()