OpenFin/backend/services/fred_bootstrap.py

"""
FRED historical bootstrap — fetches monthly/quarterly/weekly release data
from FRED's public CSV endpoint (no API key required) and stores it in
the economic_events table with rolling z-score surprises.
"""
import logging
from datetime import date
from typing import Any, Dict, List, Optional, Tuple

import httpx
import pandas as pd

logger = logging.getLogger(__name__)

# ── Series catalog ────────────────────────────────────────────────────────────
#
# transform options:
#   None            — store raw FRED value as-is
#   "yoy_pct"       — compute year-over-year %: (val / val_12m_ago - 1) * 100
#   "qoq_annualized"— compute annualized QoQ growth: ((val/val_prev)^4 - 1)*100
#   "div1000"       — divide by 1000 (FRED gives raw count, display in K)
#
# delta_absolute: if True, surprise_pct = val - prev (absolute pp change)
#                 if False, surprise_pct = (val-prev)/|prev| * 100 (% change)

FRED_SERIES: Dict[str, Dict[str, Any]] = {
    "PAYEMS": {
        "name": "Non-Farm Payrolls",
        "unit": "K",
        "freq": "monthly",
        "category": "employment",
        "higher_is_bullish": True,
        "assets": ["SPY", "QQQ", "EURUSD=X", "TLT"],
        "zscore_window": 12,
        "transform": None,
        "delta_absolute": False,
    },
    "UNRATE": {
        "name": "Unemployment Rate",
        "unit": "%",
        "freq": "monthly",
        "category": "employment",
        "higher_is_bullish": False,
        "assets": ["SPY", "QQQ", "EURUSD=X"],
        "zscore_window": 12,
        "transform": None,
        "delta_absolute": True,   # pp change (e.g. 4.1 → 4.0 = -0.1 pp)
    },
    "CPIAUCSL": {
        "name": "CPI All Items (YoY %)",
        "unit": "%",
        "freq": "monthly",
        "category": "inflation",
        "higher_is_bullish": False,
        "assets": ["TLT", "GLD", "EURUSD=X", "SPY"],
        "zscore_window": 12,
        "transform": "yoy_pct",   # FRED gives index level → compute YoY
        "delta_absolute": True,   # pp change between consecutive YoY readings
    },
    "CPILFESL": {
        "name": "Core CPI (YoY %)",
        "unit": "%",
        "freq": "monthly",
        "category": "inflation",
        "higher_is_bullish": False,
        "assets": ["TLT", "GLD", "EURUSD=X"],
        "zscore_window": 12,
        "transform": "yoy_pct",
        "delta_absolute": True,
    },
    "PCEPILFE": {
        "name": "Core PCE (YoY %)",
        "unit": "%",
        "freq": "monthly",
        "category": "inflation",
        "higher_is_bullish": False,
        "assets": ["TLT", "GLD", "SPY"],
        "zscore_window": 12,
        "transform": "yoy_pct",
        "delta_absolute": True,
    },
    "FEDFUNDS": {
        "name": "Fed Funds Rate (FOMC)",
        "unit": "%",
        "freq": "monthly",
        "category": "monetary",
        "higher_is_bullish": False,
        "assets": ["TLT", "SPY", "EURUSD=X", "GLD"],
        "zscore_window": 12,
        "transform": None,
        "delta_absolute": True,   # bp/pp move
    },
    "ICSA": {
        "name": "Initial Jobless Claims",
        "unit": "K",
        "freq": "weekly",
        "category": "employment",
        "higher_is_bullish": False,
        "assets": ["SPY", "QQQ"],
        "zscore_window": 52,
        "transform": "div1000",   # FRED gives raw count → display in K
        "delta_absolute": False,
    },
    "GDPC1": {
        # Real GDP level in billions (chained 2017$) → compute QoQ annualized growth
        "name": "GDP Growth (QoQ Ann. %)",
        "unit": "%",
        "freq": "quarterly",
        "category": "growth",
        "higher_is_bullish": True,
        "assets": ["SPY", "QQQ", "EURUSD=X", "TLT"],
        "zscore_window": 8,
        "transform": "qoq_annualized",
        "delta_absolute": True,   # pp change between quarterly readings
    },
    "BAMLH0A0HYM2": {
        "name": "HY Credit Spread (OAS)",
        "unit": "pp",
        "freq": "weekly",
        "category": "credit",
        "higher_is_bullish": False,
        "assets": ["HYG", "LQD", "SPY"],
        "zscore_window": 52,
        "transform": None,
        "delta_absolute": True,   # pp change in spread
    },
    "T10Y2Y": {
        "name": "Yield Spread 10Y-2Y",
        "unit": "pp",
        "freq": "weekly",
        "category": "rates",
        "higher_is_bullish": True,
        "assets": ["TLT", "IEF", "SPY", "HYG"],
        "zscore_window": 52,
        "transform": None,
        "delta_absolute": True,
    },
    "T10Y3M": {
        "name": "Yield Spread 10Y-3M",
        "unit": "pp",
        "freq": "weekly",
        "category": "rates",
        "higher_is_bullish": True,
        "assets": ["TLT", "IEF", "SPY"],
        "zscore_window": 52,
        "transform": None,
        "delta_absolute": True,
    },
}

# Series that were renamed/replaced — will be purged from DB on bootstrap
DEPRECATED_SERIES = ["A191RL1Q225SBEA"]

CATEGORIES = sorted({v["category"] for v in FRED_SERIES.values()})


# ── FRED fetch ────────────────────────────────────────────────────────────────

_FRED_CSV_URL = "https://fred.stlouisfed.org/graph/fredgraph.csv?id={series_id}"


_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0.0.0 Safari/537.36"
    ),
    "Accept": "text/csv,text/plain,*/*",
    "Accept-Language": "en-US,en;q=0.9",
}


def _fetch_fred_csv(series_id: str, from_date: str) -> Optional[pd.DataFrame]:
    url = _FRED_CSV_URL.format(series_id=series_id)
    print(f"[FRED] Fetching {series_id} from {url}", flush=True)
    try:
        with httpx.Client(timeout=httpx.Timeout(10.0, connect=10.0), follow_redirects=True) as client:
            resp = client.get(url, headers=_HEADERS)
        print(f"[FRED] {series_id} status={resp.status_code} len={len(resp.content)}", flush=True)
        resp.raise_for_status()
        from io import StringIO
        df = pd.read_csv(StringIO(resp.text), parse_dates=["DATE"], index_col="DATE")
        df.columns = ["value"]
        df["value"] = pd.to_numeric(df["value"], errors="coerce")
        df = df.dropna()
        df = df[df.index >= pd.Timestamp(from_date)]
        print(f"[FRED] {series_id}: {len(df)} rows from {from_date}", flush=True)
        return df.sort_index()
    except Exception as e:
        print(f"[FRED] ERROR {series_id}: {type(e).__name__}: {e}", flush=True)
        logger.warning(f"[FRED] Failed to fetch {series_id}: {e}")
        return None


# ── Transforms ────────────────────────────────────────────────────────────────

def _apply_transform(df: pd.DataFrame, transform: str) -> pd.DataFrame:
    """Apply a series-level transform before storing. Drops NaN rows introduced."""
    if transform == "yoy_pct":
        # Year-over-year % change from index level
        df = df.copy()
        df["value"] = (df["value"] / df["value"].shift(12) - 1) * 100
        df = df.dropna()
    elif transform == "qoq_annualized":
        # Annualized quarter-over-quarter growth rate
        df = df.copy()
        ratio = df["value"] / df["value"].shift(1)
        df["value"] = (ratio ** 4 - 1) * 100
        df = df.dropna()
    elif transform == "div1000":
        df = df.copy()
        df["value"] = df["value"] / 1000.0
    return df


# ── Z-score ───────────────────────────────────────────────────────────────────

def _compute_zscore_series(values: pd.Series, window: int) -> Tuple[pd.Series, pd.Series, pd.Series]:
    """
    Z-score using previous-window only (no look-ahead).
    Z = (value - rolling_mean_prev_N) / rolling_std_prev_N
    """
    roll_mean = values.shift(1).rolling(window, min_periods=max(4, window // 3)).mean()
    roll_std  = values.shift(1).rolling(window, min_periods=max(4, window // 3)).std()
    z = (values - roll_mean) / roll_std.replace(0, float("nan"))
    return z.round(2), roll_mean.round(4), roll_std.round(4)


def _direction(zscore: float, higher_is_bullish: bool) -> str:
    if abs(zscore) < 0.5:
        return "neutral"
    return "bullish" if (zscore > 0) == higher_is_bullish else "bearish"


def _resample_weekly(df: pd.DataFrame) -> pd.DataFrame:
    return df.resample("W-FRI").last().dropna()


# ── Main bootstrap ────────────────────────────────────────────────────────────

def bootstrap_fred(
    from_date: str = "2020-01-01",
    series_ids: Optional[List[str]] = None,
    force: bool = False,
) -> Dict[str, Any]:
    """
    Fetch and store FRED data from from_date to today.
    Transforms are applied before storing (YoY%, QoQ Ann., /1000).
    Returns summary dict with counts per series.
    """
    from services.database import get_conn
    import json

    target_series = series_ids or list(FRED_SERIES.keys())
    results: Dict[str, Any] = {}
    conn = get_conn()

    # Purge deprecated series from DB
    for dep in DEPRECATED_SERIES:
        try:
            conn.execute("DELETE FROM economic_events WHERE series_id=?", (dep,))
        except Exception:
            pass
    conn.commit()

    for sid in target_series:
        meta = FRED_SERIES.get(sid)
        if not meta:
            logger.warning(f"[FRED bootstrap] Unknown series: {sid}")
            continue

        logger.info(f"[FRED bootstrap] Fetching {sid} ({meta['name']})")

        # Warm-up period: extra years before from_date for transform + z-score
        transform = meta.get("transform")
        extra_years = 3 if transform == "qoq_annualized" else (2 if transform == "yoy_pct" else 1)
        fetch_from = str(date(int(from_date[:4]) - extra_years, 1, 1))

        df = _fetch_fred_csv(sid, from_date=fetch_from)
        if df is None or df.empty:
            results[sid] = {"status": "fetch_failed", "count": 0}
            continue

        # Apply transform on full (warm-up included) dataset
        if transform:
            df = _apply_transform(df, transform)
            if df.empty:
                results[sid] = {"status": "transform_empty", "count": 0}
                continue

        # Downsample continuous daily/weekly series to weekly
        if meta["freq"] == "weekly" and len(df) > 200:
            df = _resample_weekly(df)

        # Z-score on transformed values
        window = meta["zscore_window"]
        z_series, _, _ = _compute_zscore_series(df["value"], window)

        delta_absolute = meta.get("delta_absolute", False)
        inserted = 0
        skipped  = 0

        for dt, row in df.iterrows():
            ev_date = dt.strftime("%Y-%m-%d")
            if ev_date < from_date:
                continue  # warm-up only, don't store

            val  = float(row["value"])
            z    = float(z_series.get(dt) or 0.0)
            if pd.isna(z):
                z = 0.0

            prev_raw = df["value"].shift(1).get(dt)
            prev = float(prev_raw) if prev_raw is not None and not pd.isna(prev_raw) else None

            if delta_absolute:
                # pp / absolute change (for rates, spreads, YoY series)
                surprise_pct = round(val - prev, 4) if prev is not None else None
            else:
                # % change (for levels: NFP, ICSA in K, etc.)
                surprise_pct = round((val - prev) / abs(prev) * 100, 2) if prev and prev != 0 else None

            direction = _direction(z, meta["higher_is_bullish"])

            try:
                conn.execute(
                    """INSERT INTO economic_events
                       (event_name, series_id, event_date, actual_value, actual_unit,
                        forecast_value, previous_value, surprise_pct, surprise_zscore,
                        surprise_direction, assets_impacted, source)
                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
                       ON CONFLICT(series_id, event_date) DO """ + (
                        "UPDATE SET actual_value=excluded.actual_value, "
                        "previous_value=excluded.previous_value, "
                        "surprise_pct=excluded.surprise_pct, "
                        "surprise_zscore=excluded.surprise_zscore, "
                        "surprise_direction=excluded.surprise_direction, "
                        "fetched_at=datetime('now')"
                        if force else "NOTHING"
                    ),
                    (
                        meta["name"], sid, ev_date, round(val, 4), meta["unit"],
                        None, prev, surprise_pct, z, direction,
                        json.dumps(meta["assets"]),
                        "FRED/bootstrap",
                    ),
                )
                if conn.execute("SELECT changes()").fetchone()[0]:
                    inserted += 1
                else:
                    skipped += 1
            except Exception as e:
                logger.debug(f"[FRED bootstrap] {sid} {ev_date}: {e}")

        conn.commit()
        results[sid] = {
            "status": "ok",
            "name": meta["name"],
            "inserted": inserted,
            "skipped": skipped,
            "total_fetched": len(df[df.index >= pd.Timestamp(from_date)]),
        }
        logger.info(f"[FRED bootstrap] {sid}: {inserted} inserted, {skipped} skipped")

    conn.close()
    return results