feat: Forex Factory calendar — unified past+future list style Trading Economics

- New ff_calendar table (event_date, time, currency, impact, actual, forecast, previous)
- New service ff_calendar.py: bulk CSV import (83K events 2007-2025) + live sync
  from faireconomy.media JSON endpoint (this week / next week)
- New API endpoints: POST /api/eco/ff-import, POST /api/eco/ff-sync,
  GET /api/eco/calendar (period filter), GET /api/eco/ff-stats
- CalendarPage.tsx full rewrite: period tabs (Recent/Today/Tomorrow/This Week…),
  currency flags filter, impact filter, unified date-grouped table with
  Time·Flag·Currency·Impact·Event·Actual·Forecast·Previous columns,
  green/red actual vs forecast, TODAY badge, auto-refresh 60s

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

backend/routers/eco.py

@@ -1,9 +1,10 @@
 """
-Economic calendar router — FRED-backed historical events + bootstrap endpoint.
+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, HTTPException, Query
@@ -290,6 +291,125 @@ def upcoming_events() -> List[Dict[str, Any]]:
     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_CSV_PATH = os.path.join(
+    os.path.dirname(__file__), "..", "..", "forex_factory_cache.csv"
+)
+
+
+def _run_ff_import():
+    global _ff_import_status
+    _ff_import_status["running"] = True
+    try:
+        from services.ff_calendar import import_csv
+        csv_path = os.path.abspath(_FF_CSV_PATH)
+        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-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 = os.path.abspath(_FF_CSV_PATH)
+    if not os.path.exists(csv_path):
+        raise HTTPException(404, f"CSV not found: {csv_path}")
+    background_tasks.add_task(_run_ff_import)
+    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
+
+
+@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"),
+    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(300, ge=1, le=2000),
+    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,
+        currencies=cur_list,
+        impacts=imp_list,
+        limit=limit,
+        offset=offset,
+    )
+
+
+@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")

backend/services/database.py

@@ -685,6 +685,29 @@ def init_db():
     except Exception:
         pass
 
+    # ── Forex Factory Calendar ─────────────────────────────────────────────────
+    c.execute("""CREATE TABLE IF NOT EXISTS ff_calendar (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        event_date TEXT NOT NULL,
+        event_time TEXT,
+        currency TEXT NOT NULL,
+        impact TEXT,
+        event_name TEXT NOT NULL,
+        actual_value TEXT,
+        forecast_value TEXT,
+        previous_value TEXT,
+        detail TEXT,
+        source TEXT DEFAULT 'ff_cache',
+        fetched_at TEXT DEFAULT (datetime('now')),
+        UNIQUE(event_date, event_time, currency, event_name)
+    )""")
+    try:
+        c.execute("CREATE INDEX IF NOT EXISTS idx_ff_date ON ff_calendar(event_date)")
+        c.execute("CREATE INDEX IF NOT EXISTS idx_ff_currency ON ff_calendar(currency)")
+        c.execute("CREATE INDEX IF NOT EXISTS idx_ff_impact ON ff_calendar(impact)")
+    except Exception:
+        pass
+
     # ── Specialist Desks ───────────────────────────────────────────────────────
     c.execute("""CREATE TABLE IF NOT EXISTS specialist_reports (
         id TEXT PRIMARY KEY,

backend/services/ff_calendar.py

@@ -0,0 +1,316 @@
+"""
+Forex Factory calendar service.
+- CSV bulk import from forex_factory_cache.csv (2007-2025)
+- Live sync from nfs.faireconomy.media JSON endpoint (this week / next week)
+"""
+import csv
+import json
+import logging
+from datetime import datetime, timezone, timedelta
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+_FF_THIS_WEEK_URL = "https://nfs.faireconomy.media/ff_calendar_thisweek.json"
+_FF_NEXT_WEEK_URL = "https://nfs.faireconomy.media/ff_calendar_nextweek.json"
+
+# Map FF impact strings → normalized
+_IMPACT_MAP = {
+    "High Impact Expected": "high",
+    "High":                 "high",
+    "Medium Impact Expected": "medium",
+    "Medium":               "medium",
+    "Low Impact Expected":  "low",
+    "Low":                  "low",
+    "Non-Economic":         "non-economic",
+    "Holiday":              "non-economic",
+}
+
+# Currencies we care about (drop 'All' pseudo-currency)
+SUPPORTED_CURRENCIES = {"USD", "EUR", "GBP", "JPY", "AUD", "CAD", "NZD", "CHF", "CNY"}
+
+
+def _parse_ff_datetime(dt_str: str) -> tuple[str, str]:
+    """
+    Parse an ISO datetime with offset (e.g. '2007-01-05T17:00:00+03:30')
+    into (event_date YYYY-MM-DD UTC, event_time HH:MM UTC).
+    """
+    try:
+        dt = datetime.fromisoformat(dt_str)
+        dt_utc = dt.astimezone(timezone.utc)
+        return dt_utc.strftime("%Y-%m-%d"), dt_utc.strftime("%H:%M")
+    except Exception:
+        # Fallback: take the date/time literally
+        parts = dt_str[:16].split("T")
+        return parts[0], parts[1] if len(parts) > 1 else "00:00"
+
+
+def import_csv(csv_path: str, progress_cb=None) -> Dict[str, Any]:
+    """
+    Bulk-import forex_factory_cache.csv into ff_calendar table.
+    Skips non-economic events and unsupported currencies.
+    Returns {inserted, skipped, total}.
+    """
+    from services.database import get_conn
+
+    conn = get_conn()
+    inserted = 0
+    skipped = 0
+    total = 0
+
+    try:
+        with open(csv_path, encoding="utf-8", errors="replace") as f:
+            reader = csv.DictReader(f)
+            batch = []
+
+            for row in reader:
+                total += 1
+                currency = (row.get("Currency") or "").strip()
+                impact_raw = (row.get("Impact") or "").strip()
+                impact = _IMPACT_MAP.get(impact_raw, "low")
+
+                # Skip non-economic and unsupported currencies
+                if currency not in SUPPORTED_CURRENCIES:
+                    skipped += 1
+                    continue
+                if impact == "non-economic":
+                    skipped += 1
+                    continue
+
+                event_name = (row.get("Event") or "").strip()
+                if not event_name:
+                    skipped += 1
+                    continue
+
+                dt_str = (row.get("DateTime") or "").strip()
+                event_date, event_time = _parse_ff_datetime(dt_str)
+
+                actual   = (row.get("Actual") or "").strip() or None
+                forecast = (row.get("Forecast") or "").strip() or None
+                previous = (row.get("Previous") or "").strip() or None
+                detail   = (row.get("Detail") or "").strip() or None
+
+                batch.append((
+                    event_date, event_time, currency, impact, event_name,
+                    actual, forecast, previous, detail, "ff_cache",
+                ))
+
+                if len(batch) >= 500:
+                    _upsert_batch(conn, batch)
+                    inserted += len(batch)
+                    batch = []
+                    if progress_cb:
+                        progress_cb({"inserted": inserted, "total": total})
+
+            if batch:
+                _upsert_batch(conn, batch)
+                inserted += len(batch)
+
+        conn.commit()
+        print(f"[FF import] Done: {inserted} inserted, {skipped} skipped, {total} total", flush=True)
+        return {"inserted": inserted, "skipped": skipped, "total": total}
+    except Exception as e:
+        logger.error(f"[FF import] Failed: {e}")
+        print(f"[FF import] ERROR: {e}", flush=True)
+        return {"error": str(e)}
+    finally:
+        conn.close()
+
+
+def _upsert_batch(conn, batch: list):
+    conn.executemany(
+        """INSERT INTO ff_calendar
+           (event_date, event_time, currency, impact, event_name,
+            actual_value, forecast_value, previous_value, detail, source)
+           VALUES (?,?,?,?,?,?,?,?,?,?)
+           ON CONFLICT(event_date, event_time, currency, event_name)
+           DO UPDATE SET
+             actual_value   = COALESCE(excluded.actual_value, ff_calendar.actual_value),
+             forecast_value = COALESCE(excluded.forecast_value, ff_calendar.forecast_value),
+             previous_value = COALESCE(excluded.previous_value, ff_calendar.previous_value),
+             fetched_at     = datetime('now')
+        """,
+        batch,
+    )
+
+
+def sync_live() -> Dict[str, Any]:
+    """
+    Fetch this week + next week from faireconomy.media and upsert into ff_calendar.
+    Updates actual_value when it becomes available (real-time detection).
+    """
+    from services.database import get_conn
+
+    results = {}
+    conn = get_conn()
+
+    for label, url in [("thisweek", _FF_THIS_WEEK_URL), ("nextweek", _FF_NEXT_WEEK_URL)]:
+        try:
+            resp = httpx.get(url, timeout=15, follow_redirects=True)
+            if resp.status_code == 404:
+                results[label] = {"status": "not_available"}
+                continue
+            resp.raise_for_status()
+            events = resp.json()
+
+            batch = []
+            for ev in events:
+                currency = (ev.get("country") or "").strip().upper()
+                if currency not in SUPPORTED_CURRENCIES:
+                    continue
+
+                impact = _IMPACT_MAP.get(ev.get("impact") or "", "low")
+                if impact == "non-economic":
+                    continue
+
+                event_name = (ev.get("title") or "").strip()
+                if not event_name:
+                    continue
+
+                dt_str = ev.get("date") or ""
+                event_date, event_time = _parse_ff_datetime(dt_str)
+
+                actual   = (ev.get("actual")   or "").strip() or None
+                forecast = (ev.get("forecast") or "").strip() or None
+                previous = (ev.get("previous") or "").strip() or None
+
+                batch.append((
+                    event_date, event_time, currency, impact, event_name,
+                    actual, forecast, previous, None, "ff_live",
+                ))
+
+            if batch:
+                _upsert_batch(conn, batch)
+                conn.commit()
+
+            results[label] = {"status": "ok", "events": len(batch)}
+            print(f"[FF sync] {label}: {len(batch)} events upserted", flush=True)
+
+        except Exception as e:
+            logger.warning(f"[FF sync] {label} failed: {e}")
+            results[label] = {"status": "error", "error": str(e)}
+
+    conn.close()
+    return results
+
+
+def get_calendar(
+    period: str = "recent",
+    currencies: Optional[List[str]] = None,
+    impacts: Optional[List[str]] = None,
+    limit: int = 300,
+    offset: int = 0,
+) -> Dict[str, Any]:
+    """
+    Query ff_calendar with period filter.
+
+    period options:
+      recent | today | tomorrow | yesterday | this_week | next_week |
+      previous_week | this_month | next_month | previous_month | custom
+    """
+    from services.database import get_conn
+
+    today = datetime.now(timezone.utc).date()
+    monday = today - timedelta(days=today.weekday())
+    sunday = monday + timedelta(days=6)
+
+    date_from: Optional[str] = None
+    date_to:   Optional[str] = None
+
+    if period == "today":
+        date_from = date_to = str(today)
+    elif period == "tomorrow":
+        t = today + timedelta(days=1)
+        date_from = date_to = str(t)
+    elif period == "yesterday":
+        t = today - timedelta(days=1)
+        date_from = date_to = str(t)
+    elif period == "this_week":
+        date_from = str(monday)
+        date_to   = str(sunday)
+    elif period == "next_week":
+        nm = monday + timedelta(weeks=1)
+        date_from = str(nm)
+        date_to   = str(nm + timedelta(days=6))
+    elif period == "previous_week":
+        pm = monday - timedelta(weeks=1)
+        date_from = str(pm)
+        date_to   = str(pm + timedelta(days=6))
+    elif period == "this_month":
+        date_from = str(today.replace(day=1))
+        # last day of month
+        if today.month == 12:
+            last = today.replace(year=today.year + 1, month=1, day=1) - timedelta(days=1)
+        else:
+            last = today.replace(month=today.month + 1, day=1) - timedelta(days=1)
+        date_to = str(last)
+    elif period == "next_month":
+        if today.month == 12:
+            first = today.replace(year=today.year + 1, month=1, day=1)
+        else:
+            first = today.replace(month=today.month + 1, day=1)
+        if first.month == 12:
+            last = first.replace(year=first.year + 1, month=1, day=1) - timedelta(days=1)
+        else:
+            last = first.replace(month=first.month + 1, day=1) - timedelta(days=1)
+        date_from = str(first)
+        date_to   = str(last)
+    elif period == "previous_month":
+        if today.month == 1:
+            first = today.replace(year=today.year - 1, month=12, day=1)
+        else:
+            first = today.replace(month=today.month - 1, day=1)
+        last = today.replace(day=1) - timedelta(days=1)
+        date_from = str(first)
+        date_to   = str(last)
+    else:  # recent: last 7 days + next 14 days
+        date_from = str(today - timedelta(days=7))
+        date_to   = str(today + timedelta(days=14))
+
+    conn = get_conn()
+    where: List[str] = []
+    params: List[Any] = []
+
+    if date_from:
+        where.append("event_date >= ?")
+        params.append(date_from)
+    if date_to:
+        where.append("event_date <= ?")
+        params.append(date_to)
+
+    if currencies:
+        ph = ",".join("?" * len(currencies))
+        where.append(f"currency IN ({ph})")
+        params.extend(currencies)
+
+    if impacts:
+        ph = ",".join("?" * len(impacts))
+        where.append(f"impact IN ({ph})")
+        params.extend(impacts)
+
+    where_sql = ("WHERE " + " AND ".join(where)) if where else ""
+
+    count_sql = f"SELECT COUNT(*) FROM ff_calendar {where_sql}"
+    data_sql  = (
+        f"SELECT * FROM ff_calendar {where_sql} "
+        f"ORDER BY event_date ASC, event_time ASC "
+        f"LIMIT ? OFFSET ?"
+    )
+
+    try:
+        total = conn.execute(count_sql, params).fetchone()[0]
+        rows  = conn.execute(data_sql, params + [limit, offset]).fetchall()
+        events = [dict(r) for r in rows]
+        return {
+            "total": total,
+            "offset": offset,
+            "limit": limit,
+            "period": period,
+            "date_from": date_from,
+            "date_to":   date_to,
+            "events": events,
+        }
+    finally:
+        conn.close()