Fed Speech & FOMC Monitoring

name: fed_monitor
schedule: "0 */2 * * 1-5"
steps:
  - fetch_rss:
      url: "https://www.federalreserve.gov/feeds/speeches.xml"
      type: fed_speeches
  - fetch_rss:
      url: "https://www.federalreserve.gov/feeds/press_all.xml"
      type: fed_press
  - check_seen:
      dedup_key: item_url
      store: seen_fed.json
  - llm:
      prompt: |
        Analyze this Federal Reserve publication:

        1. HAWK/DOVE: Classify as hawkish, dovish, or neutral. Quote the key sentence.
        2. RATE SIGNAL: Any explicit or implicit signal about future rate moves?
        3. KEY LANGUAGE: List any new or changed phrasing vs recent communications.
        4. SPEAKER: Note who spoke and their known position on the policy spectrum.
        5. ONE-LINE SUMMARY: For quick-scan alerts.

        Content: {{ content }}
  - notify:
      channel: email
      subject: "🏛️ Fed Alert: {{ speaker }} — {{ one_line_summary }}"
      priority: "{{ 'high' if fomc_minutes else 'normal' }}"

import feedparser
import httpx
from bs4 import BeautifulSoup

def fetch_fed_speeches(limit: int = 10) -> list:
    """Fetch latest Fed speeches from official RSS feed."""
    feed = feedparser.parse("https://www.federalreserve.gov/feeds/speeches.xml")
    speeches = []
    for entry in feed.entries[:limit]:
        speeches.append({
            "title": entry.title,
            "url": entry.link,
            "date": entry.get("published", ""),
            "summary": entry.get("summary", ""),
        })
    return speeches

def fetch_speech_text(url: str) -> str:
    """Fetch full text of a Fed speech page."""
    r = httpx.get(url, headers={"User-Agent": "FedMonitor/1.0 contact@youremail.com"},
                  timeout=10, follow_redirects=True)
    soup = BeautifulSoup(r.text, "html.parser")
    # Fed pages use div#article or div.col-xs-12.col-sm-8.col-md-8
    content = soup.find("div", {"id": "article"}) or soup.find("div", class_="col-xs-12")
    return content.get_text(separator=" ", strip=True) if content else ""

HAWKISH_TERMS = ["inflation", "price stability", "restrictive", "above target",
                  "further increases", "not yet confident", "premature", "persistent"]
DOVISH_TERMS = ["cooling", "progress", "appropriate to begin", "labor market",
                 "balance of risks", "gradual", "easing", "below target", "sustainable"]

def score_fed_tone(text: str) -> dict:
    text_lower = text.lower()
    hawk_count = sum(text_lower.count(term) for term in HAWKISH_TERMS)
    dove_count = sum(text_lower.count(term) for term in DOVISH_TERMS)
    total = hawk_count + dove_count
    if total == 0:
        return {"tone": "neutral", "hawk_score": 0, "dove_score": 0}
    tone = "hawkish" if hawk_count > dove_count * 1.2 else "dovish" if dove_count > hawk_count * 1.2 else "neutral"
    return {"tone": tone, "hawk_score": hawk_count, "dove_score": dove_count,
            "hawk_pct": round(hawk_count / total * 100)}