Yield Curve Monitor

# Install the required library
pip install fredapi pandas pyyaml

# Verify it works
python3 -c "
from fredapi import Fred
import os
fred = Fred(api_key='your_key_here')
print('10-year yield:', fred.get_series('DGS10').iloc[-1], '%')
print('✅ FRED connection working')
"

# fi-config.yaml — yield curve section
fred:
  api_key: "your_fred_api_key_here"

yield_curve:
  maturities:
    - { label: "2-Year",  series: "DGS2"  }
    - { label: "5-Year",  series: "DGS5"  }
    - { label: "10-Year", series: "DGS10" }
    - { label: "30-Year", series: "DGS30" }
  alerts:
    inversion: true           # Alert when 2yr yield > 10yr yield
    spread_threshold: -0.25   # Alert if 2s10s drops below -0.25%
    steepening_move: 0.50     # Alert if spread widens 0.50% in a week

alerts:
  email: "you@example.com"
  daily_digest: true

# fi_yield_curve.py — OpenClaw Yield Curve Monitor
# Fetches Treasury yields from FRED and checks for inversion

import yaml
from fredapi import Fred
import pandas as pd
from datetime import datetime, timedelta

# ── Step 1: Load your config ──────────────────────────────────────
def load_config(path: str = "fi-config.yaml") -> dict:
    with open(path) as f:
        return yaml.safe_load(f)

cfg  = load_config()
fred = Fred(api_key=cfg["fred"]["api_key"])
YC   = cfg["yield_curve"]

print(f"🦞 OpenClaw Yield Curve Monitor")
print(f"   {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print("─" * 45)

# ── Step 2: Fetch current yields from FRED ────────────────────────
def fetch_yields() -> dict:
    """
    Fetches the latest available yield for each maturity.
    FRED updates daily — we take the most recent non-null value.
    """
    yields = {}
    for item in YC["maturities"]:
        series = fred.get_series(item["series"], observation_start="2020-01-01")
        latest = series.dropna().iloc[-1]   # most recent value
        yields[item["label"]] = round(float(latest), 3)
        print(f"   {item['label']:<10} {latest:.3f}%")
    return yields

yields = fetch_yields()

# ── Step 3: Calculate the key spread ─────────────────────────────
two_year  = yields.get("2-Year",  0)
ten_year  = yields.get("10-Year", 0)
spread_2s10s = round(ten_year - two_year, 3)

print(f"\n   2s10s Spread: {spread_2s10s:+.3f}%")
if spread_2s10s < 0:
    print(f"   ⚠️  INVERTED — short-term rates exceed long-term")
else:
    print(f"   ✅ Normal — curve is positively sloped")

# ── Step 4: Check alert conditions ───────────────────────────────
def check_alerts(yields: dict, spread: float) -> list:
    """Check configured alert thresholds. Returns list of alert messages."""
    alerts_cfg = YC.get("alerts", {})
    fired = []

    # Inversion alert
    if alerts_cfg.get("inversion") and spread < 0:
        fired.append({
            "type": "INVERSION",
            "message": f"2s10s spread is INVERTED at {spread:+.3f}% — 2yr ({yields['2-Year']}%) > 10yr ({yields['10-Year']}%)",
            "severity": "HIGH"
        })

    # Threshold alert
    threshold = alerts_cfg.get("spread_threshold", -0.25)
    if spread < threshold:
        fired.append({
            "type": "SPREAD_BELOW_THRESHOLD",
            "message": f"2s10s spread {spread:+.3f}% is below your threshold of {threshold:+.3f}%",
            "severity": "MEDIUM"
        })

    return fired

alerts = check_alerts(yields, spread_2s10s)
if alerts:
    print(f"\n   🚨 {len(alerts)} alert(s) triggered:")
    for a in alerts:
        print(f"   [{a['severity']}] {a['message']}")
else:
    print(f"\n   ✓ No alerts triggered today")

def fetch_spread_history(days: int = 90) -> pd.Series:
    """
    Fetches 2s10s spread history for the past N days.
    Useful for seeing whether the inversion is deepening or recovering.
    """
    start = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")

    two_yr = fred.get_series("DGS2",  observation_start=start).dropna()
    ten_yr = fred.get_series("DGS10", observation_start=start).dropna()

    # Align the two series (some days one may be missing)
    spread = (ten_yr - two_yr).dropna()
    spread.name = "2s10s_spread"

    return spread

history = fetch_spread_history(90)

# Show last 10 readings
print(f"\n📅 2s10s Spread — Last 10 trading days:")
print(f"   {'Date':<14} {'Spread':>8} {'Status'}")
print("   " + "─" * 35)
for date, val in history.tail(10).items():
    status = "⚠️  Inverted" if val < 0 else "✅ Normal"
    print(f"   {str(date.date()):<14} {val:>+7.3f}%  {status}")

# Trend check: is it getting better or worse?
recent_avg  = history.tail(5).mean()
previous_avg = history.tail(15).head(10).mean()
trend = "improving" if recent_avg > previous_avg else "worsening"
print(f"\n   5-day avg: {recent_avg:+.3f}% | 10-day prior avg: {previous_avg:+.3f}%")
print(f"   Trend: {trend.upper()}")

# crontab -e
# Run every weekday at 7:00 AM local time
0 7 * * 1-5 cd /path/to/openclaw && python3 fi_yield_curve.py >> fi_brief.log 2>&1