""" FRED API fetcher — récupère les dernières releases macro US. Séries suivies : CPIAUCSL → CPI mensuel (YoY calculé) PAYEMS → Non-Farm Payrolls (variation mensuelle k emplois) UNRATE → Taux de chômage FEDFUNDS → Fed Funds Rate GDP → PIB US trimestriel (croissance %) ICSA → Initial Jobless Claims (hebdo) T10Y2Y → Spread 10Y-2Y (courbe des taux) DEXUSEU → EUR/USD (proxy macro) """ from __future__ import annotations import logging from datetime import datetime, timedelta from typing import Any, Dict, List, Optional logger = logging.getLogger("fred_fetcher") # Series config : (id, label, unit, asset_impact, direction_interpretation) FRED_SERIES = [ ("CPIAUCSL", "US CPI (inflation)", "index", ["metals", "rates", "forex"], "higher_bearish"), ("PAYEMS", "US Non-Farm Payrolls", "k jobs", ["indices", "forex", "rates"], "higher_bullish"), ("UNRATE", "US Unemployment Rate", "%", ["indices", "forex"], "higher_bearish"), ("FEDFUNDS", "Fed Funds Rate", "%", ["bonds", "forex", "indices"], "higher_bearish_growth"), ("GDP", "US GDP Growth", "bn$", ["indices", "forex"], "higher_bullish"), ("ICSA", "US Initial Jobless Claims","k claims",["indices", "forex"], "higher_bearish"), ("T10Y2Y", "10Y-2Y Spread (courbe)", "%", ["bonds", "indices"], "positive_bullish"), ] def _get_fred_key() -> Optional[str]: try: from services.database import get_config return get_config("fred_api_key") or None except Exception: return None def _fetch_series_observations(series_id: str, api_key: str, count: int = 3) -> List[Dict]: """Fetch last N observations from FRED for a given series.""" import urllib.request import urllib.parse import json params = urllib.parse.urlencode({ "series_id": series_id, "api_key": api_key, "file_type": "json", "sort_order": "desc", "limit": count, "observation_start": (datetime.utcnow() - timedelta(days=730)).strftime("%Y-%m-%d"), }) url = f"https://api.stlouisfed.org/fred/series/observations?{params}" try: with urllib.request.urlopen(url, timeout=8) as resp: data = json.loads(resp.read()) obs = data.get("observations", []) return [{"date": o["date"], "value": o["value"]} for o in obs if o.get("value") not in (".", None)] except Exception as e: logger.warning(f"[FRED] {series_id} fetch failed: {e}") return [] def _parse_value(v: str) -> Optional[float]: try: return float(v) except (ValueError, TypeError): return None def _surprise_direction(series_id: str, current: float, previous: float, direction_hint: str) -> str: """Returns 'bullish', 'bearish', or 'neutral' based on change direction.""" delta = current - previous if abs(delta) < 0.01: return "neutral" if direction_hint == "higher_bullish": return "bullish" if delta > 0 else "bearish" elif direction_hint in ("higher_bearish", "higher_bearish_growth"): return "bearish" if delta > 0 else "bullish" elif direction_hint == "positive_bullish": return "bullish" if current > 0 else "bearish" return "neutral" def _yoy_change(obs: List[Dict]) -> Optional[float]: """For monthly series, compute rough YoY % if we have ≥12 month history.""" if len(obs) < 2: return None v_latest = _parse_value(obs[0]["value"]) v_prev = _parse_value(obs[-1]["value"]) if v_latest is None or v_prev is None or v_prev == 0: return None return round((v_latest - v_prev) / abs(v_prev) * 100, 2) def get_fred_recent_releases() -> List[Dict[str, Any]]: """ Fetch latest FRED data for key macro series. Returns a list of dicts, one per series, with: series_id, label, unit, latest_date, latest_value, previous_value, change, change_pct, direction, asset_impact """ api_key = _get_fred_key() if not api_key: return [] results = [] for series_id, label, unit, asset_impact, direction_hint in FRED_SERIES: obs = _fetch_series_observations(series_id, api_key, count=13) # 13 for YoY if not obs: continue latest = obs[0] previous = obs[1] if len(obs) > 1 else None v_latest = _parse_value(latest["value"]) v_prev = _parse_value(previous["value"]) if previous else None if v_latest is None: continue change = round(v_latest - v_prev, 4) if v_prev is not None else None change_pct = round((v_latest - v_prev) / abs(v_prev) * 100, 2) if v_prev and v_prev != 0 else None direction = _surprise_direction(series_id, v_latest, v_prev, direction_hint) if v_prev is not None else "neutral" # For CPI, compute YoY display_value = v_latest display_unit = unit if series_id == "CPIAUCSL" and len(obs) >= 13: yoy = _yoy_change(obs[:13]) if yoy is not None: display_value = yoy display_unit = "% YoY" # Re-compute change vs previous month's YoY if len(obs) >= 14: yoy_prev = _yoy_change(obs[1:14]) if yoy_prev is not None: change = round(yoy - yoy_prev, 3) direction = "bearish" if change > 0 else "bullish" # higher CPI = bearish markets entry: Dict[str, Any] = { "series_id": series_id, "label": label, "unit": display_unit, "latest_date": latest["date"], "latest_value": display_value, "previous_value": v_prev, "change": change, "direction": direction, "asset_impact": asset_impact, } results.append(entry) return results def build_fred_context_block(releases: List[Dict]) -> str: """Format FRED releases as a prompt-ready string.""" if not releases: return "" lines = ["## 📈 DONNÉES MACRO RÉCENTES (FRED — dernières releases)"] for r in releases: val_str = f"{r['latest_value']:.2f}{r['unit']}" if isinstance(r.get("latest_value"), float) else str(r.get("latest_value", "N/A")) chg_str = "" if r.get("change") is not None: arrow = "▲" if r["change"] > 0 else "▼" chg_str = f" ({arrow}{abs(r['change']):.3f} vs release précédente)" dir_emoji = {"bullish": "🟢", "bearish": "🔴", "neutral": "⚪"}.get(r.get("direction", "neutral"), "⚪") lines.append( f" {dir_emoji} {r['label']} [{r['latest_date']}] : {val_str}{chg_str} → {r['direction'].upper()}" ) lines.append("⚠️ Compare ces chiffres au consensus attendu pour évaluer si le marché a déjà intégré la surprise.") return "\n".join(lines)