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 datetime, date
from typing import Any, Dict, List, Optional, Tuple

import httpx
import pandas as pd

logger = logging.getLogger(__name__)

# ── Series catalog ────────────────────────────────────────────────────────────

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,
    },
    "UNRATE": {
        "name": "Unemployment Rate",
        "unit": "%",
        "freq": "monthly",
        "category": "employment",
        "higher_is_bullish": False,
        "assets": ["SPY", "QQQ", "EURUSD=X"],
        "zscore_window": 12,
    },
    "CPIAUCSL": {
        "name": "CPI (All Items YoY)",
        "unit": "%",
        "freq": "monthly",
        "category": "inflation",
        "higher_is_bullish": False,
        "assets": ["TLT", "GLD", "EURUSD=X", "SPY"],
        "zscore_window": 12,
    },
    "CPILFESL": {
        "name": "Core CPI",
        "unit": "%",
        "freq": "monthly",
        "category": "inflation",
        "higher_is_bullish": False,
        "assets": ["TLT", "GLD", "EURUSD=X"],
        "zscore_window": 12,
    },
    "PCEPILFE": {
        "name": "Core PCE",
        "unit": "%",
        "freq": "monthly",
        "category": "inflation",
        "higher_is_bullish": False,
        "assets": ["TLT", "GLD", "SPY"],
        "zscore_window": 12,
    },
    "FEDFUNDS": {
        "name": "Fed Funds Rate (FOMC)",
        "unit": "%",
        "freq": "monthly",
        "category": "monetary",
        "higher_is_bullish": False,
        "assets": ["TLT", "SPY", "EURUSD=X", "GLD"],
        "zscore_window": 12,
    },
    "ICSA": {
        "name": "Initial Jobless Claims",
        "unit": "K",
        "freq": "weekly",
        "category": "employment",
        "higher_is_bullish": False,
        "assets": ["SPY", "QQQ"],
        "zscore_window": 52,
    },
    "A191RL1Q225SBEA": {
        "name": "GDP Growth (Quarterly)",
        "unit": "%",
        "freq": "quarterly",
        "category": "growth",
        "higher_is_bullish": True,
        "assets": ["SPY", "QQQ", "EURUSD=X", "TLT"],
        "zscore_window": 8,
    },
    "BAMLH0A0HYM2": {
        "name": "HY Credit Spread (OAS)",
        "unit": "pp",
        "freq": "weekly",
        "category": "credit",
        "higher_is_bullish": False,
        "assets": ["HYG", "LQD", "SPY"],
        "zscore_window": 52,
    },
    "T10Y2Y": {
        "name": "Yield Spread 10Y-2Y",
        "unit": "pp",
        "freq": "weekly",
        "category": "rates",
        "higher_is_bullish": True,
        "assets": ["TLT", "IEF", "SPY", "HYG"],
        "zscore_window": 52,
    },
    "T10Y3M": {
        "name": "Yield Spread 10Y-3M",
        "unit": "pp",
        "freq": "weekly",
        "category": "rates",
        "higher_is_bullish": True,
        "assets": ["TLT", "IEF", "SPY"],
        "zscore_window": 52,
    },
}

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


def _fetch_fred_csv(series_id: str, from_date: str = "2019-01-01") -> Optional[pd.DataFrame]:
    """
    Download FRED series as CSV. Returns DataFrame with DATE index and VALUE column.
    Uses public endpoint — no API key required.
    """
    url = _FRED_CSV_URL.format(series_id=series_id)
    try:
        resp = httpx.get(url, timeout=30, follow_redirects=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)]
        df = df.sort_index()
        return df
    except Exception as e:
        logger.warning(f"[FRED] Failed to fetch {series_id}: {e}")
        return None


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

def _compute_zscore_series(values: pd.Series, window: int) -> Tuple[pd.Series, pd.Series, pd.Series]:
    """
    Returns (z_scores, rolling_mean, rolling_std) for a value series.
    Z-score = (value - rolling_mean_prev) / rolling_std_prev
    Uses previous window to avoid look-ahead.
    """
    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"
    positive = zscore > 0
    return "bullish" if positive == higher_is_bullish else "bearish"


# ── Downsample for noisy daily/weekly series ──────────────────────────────────

def _resample_weekly(df: pd.DataFrame) -> pd.DataFrame:
    """Resample to weekly (last value of each week, Friday)."""
    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.
    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()

    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']}) from {from_date}")

        # Fetch one extra year before from_date for z-score warm-up
        fetch_from = str(date(int(from_date[:4]) - 1, 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

        # For weekly-sampled continuous series, resample
        if meta["freq"] == "weekly" and len(df) > 200:
            df = _resample_weekly(df)

        window = meta["zscore_window"]
        z_series, mean_series, _ = _compute_zscore_series(df["value"], window)

        inserted = 0
        skipped  = 0

        for dt, row in df.iterrows():
            ev_date = dt.strftime("%Y-%m-%d")
            # Skip rows before the real from_date (warm-up period)
            if ev_date < from_date:
                continue

            val      = float(row["value"])
            z        = z_series.get(dt)
            prev_val = df["value"].shift(1).get(dt)

            if z is None or pd.isna(z):
                z = 0.0
            z = float(z)

            prev = float(prev_val) if prev_val is not None and not pd.isna(prev_val) else None
            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, val, meta["unit"],
                        None,   # no consensus forecast available
                        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),
        }
        logger.info(f"[FRED bootstrap] {sid}: {inserted} inserted, {skipped} skipped")

    conn.close()
    return results