OpenFin/backend/services/ff_calendar.py

"""
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"}

# ── FF event name → FRED series_id mapping ────────────────────────────────────
# Exact FF event names → our FRED series catalog
FF_TO_FRED: dict[str, str] = {
    # Employment
    "Non-Farm Employment Change":       "PAYEMS",
    "ADP Non-Farm Employment Change":   "PAYEMS",
    "Unemployment Rate":                "UNRATE",
    "Unemployment Claims":              "ICSA",
    # Inflation — CPI
    "CPI m/m":                          "CPIAUCSL",
    "CPI y/y":                          "CPIAUCSL",
    "Core CPI m/m":                     "CPILFESL",
    "Core CPI y/y":                     "CPILFESL",
    # Inflation — PCE
    "Core PCE Price Index m/m":         "PCEPILFE",
    "Core PCE Price Index y/y":         "PCEPILFE",
    # Monetary policy
    "Federal Funds Rate":               "FEDFUNDS",
    "FOMC Statement":                   "FEDFUNDS",
    # Growth — GDP
    "Advance GDP q/q":                  "GDPC1",
    "Prelim GDP q/q":                   "GDPC1",
    "Final GDP q/q":                    "GDPC1",
    "GDP q/q":                          "GDPC1",
}


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

                series_id = FF_TO_FRED.get(event_name)
                batch.append((
                    event_date, event_time, currency, impact, event_name,
                    actual, forecast, previous, series_id, 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, series_id, 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),
             series_id      = COALESCE(excluded.series_id, ff_calendar.series_id),
             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

                series_id = FF_TO_FRED.get(event_name)
                batch.append((
                    event_date, event_time, currency, impact, event_name,
                    actual, forecast, previous, series_id, 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()