수익률곡선 모니터링: 스프레드를 매일 기록하고 임계값과 비교하기

리스크 팀에 일회성 답 외에 필요한 것

매일 커브를 지켜보는 리스크 또는 포트폴리오 팀에는 네 가지가 필요합니다. 팀이 소유한 시계열, 하루에 한 번 도는 점검, 표시를 띄우는 임계값, 그리고 나중에 각 표시를 설명해 주는 로그입니다. API는 금리를 제공하고, 저장되는 시계열과 임계값과 로그는 팀의 몫입니다. 최신 커브는 GET /v1/economics/USA/yield-curve에서 오며, maturity로 모니터가 쓰는 지점만 추립니다.

curl -s "https://api.narwhalapi.com/v1/economics/USA/yield-curve?maturity=3m,2y,10y" \
  -H "Authorization: Bearer $NARWHAL_API_KEY"

{
  "date": "2026-11-02",
  "yields": [
    {
      "maturity": "3m",
      "tenor": "P3M",
      "yield_percent": "4.21"
    },
    {
      "maturity": "2y",
      "tenor": "P2Y",
      "yield_percent": "4.59"
    },
    {
      "maturity": "10y",
      "tenor": "P10Y",
      "yield_percent": "4.91"
    }
  ],
  "country": "USA",
  "curve_type": "nominal_par"
}

date는 커브의 원본 날짜로 호출한 날보다 하루 전일 수 있으므로, 저장하는 시계열은 실행 날짜가 아니라 이 값을 키로 삼으세요. yield_percent는 퍼센트 단위의 십진수 문자열입니다. curve_type은 원본이 커브를 어떻게 정의하는지 알려 주며, 정의가 바뀌면 로그에서 드러나도록 모니터가 저장해야 합니다. 10년 금리에서 2년 금리를 빼면 0.32퍼센트포인트이고, 3개월 금리를 빼면 0.70입니다.

날짜와 쌍마다 한 행씩 있는 표 하나 두기

아래 스크립트는 히스토리 경로에서 표를 채우고 이어서 최신 커브를 추가합니다. 기본 키는 날짜와 쌍이고 쓸 때마다 행을 덮어쓰므로, 스크립트를 두 번 실행하거나 중단된 뒤 다시 시작해도 같은 상태가 됩니다.

1. 1년치 백필 backfill은 limit을 1000으로 두고 /history에 365일치를 요청하며 next_cursor를 따라가므로, 새 데이터베이스도 처음부터 1년치 시계열을 갖게 됩니다.
2. Decimal로 각 스프레드 계산하기 record는 PAIRS의 각 쌍에서 장기 금리에서 단기 금리를 뺍니다. 값은 Decimal로 유지되므로 뺄셈에 부동소수점 오차가 생기지 않으며, 텍스트로 저장됩니다.
3. 임계값으로 표시 정하기 flag는 0 미만이면 INVERTED, 0부터 WATCH까지는 WATCH, 그보다 크면 OK를 돌려줍니다. 여기서 WATCH는 0.10퍼센트포인트이며 팀이 정한 값입니다.
4. 매일 최신 커브 추가하기 daily는 단일 날짜 경로를 읽어 그 행을 씁니다. 마지막 몇 줄은 날짜 수, 최신 스프레드와 표시, 그리고 이전 날짜와 비교해 표시가 바뀌었는지를 출력합니다.
   Python, 표시가 붙은 저장된 스프레드 시계열

   import os
   import sqlite3
   from datetime import date, datetime, timedelta, timezone
   from decimal import Decimal
   import requests
   
   BASE = "https://api.narwhalapi.com/v1/economics/USA/yield-curve"
   HEAD = {"Authorization": f"Bearer {os.environ['NARWHAL_API_KEY']}"}
   PAIRS = [("2y", "10y"), ("3m", "10y")]  # long yield minus short yield
   WATCH = Decimal("0.10")  # percentage points
   
   db = sqlite3.connect("curve.db")
   db.execute("""create table if not exists spread (
       date text, pair text, spread text, flag text, curve_type text, logged_at text,
       primary key (date, pair))""")
   
   
   def flag(value):
       return "INVERTED" if value < 0 else "WATCH" if value <= WATCH else "OK"
   
   
   def record(day, yields, curve_type):
       y = {x["maturity"]: Decimal(x["yield_percent"]) for x in yields}
       for short, long in PAIRS:
           if short in y and long in y:
               s = y[long] - y[short]
               db.execute("insert or replace into spread values (?,?,?,?,?,?)",
                          (day, f"{long}-{short}", str(s), flag(s), curve_type,
                           datetime.now(timezone.utc).isoformat(timespec="seconds")))
   
   
   def backfill(start, end):
       params = {"start": start, "end": end, "maturity": "3m,2y,10y", "limit": 1000}
       while True:
           r = requests.get(BASE + "/history", params=params, headers=HEAD, timeout=15)
           r.raise_for_status()
           body = r.json()
           for obs in body["observations"]:
               record(obs["date"], obs["yields"], body["curve_type"])
           if not body.get("next_cursor"):
               return
           params["cursor"] = body["next_cursor"]
   
   
   def daily():
       r = requests.get(BASE, params={"maturity": "3m,2y,10y"}, headers=HEAD, timeout=15)
       r.raise_for_status()
       body = r.json()
       record(body["date"], body["yields"], body["curve_type"])
       return body["date"]
   
   
   backfill((date.today() - timedelta(days=365)).isoformat(), date.today().isoformat())
   latest = daily()
   db.commit()
   for pair in ("10y-2y", "10y-3m"):
       rows = db.execute("select date, spread, flag from spread where pair=? order by date", (pair,)).fetchall()
       watch = sum(1 for _, _, f in rows if f == "WATCH")
       prev, now = rows[-2], rows[-1]
       change = "flag changed" if prev[2] != now[2] else "flag unchanged"
       print(f"{pair}: {len(rows)} dates, latest {now[0]} spread {now[1]} {now[2]} ({change}), WATCH days {watch}")

첫 실행과 곧바로 이어진 두 번째 실행은 같은 두 줄을 출력했고 500행이 남았습니다. 두 쌍 각각 250일치입니다. 시계열의 날짜 수가 달력 일수보다 적은 것은 주말과 시장 휴일에는 행이 없기 때문입니다. WATCH인 날은 2025-11-26 하루뿐이었고, 그날 10y에서 3m를 뺀 값은 0.08이었습니다.

10y-2y: 250 dates, latest 2026-11-02 spread 0.32 OK (flag unchanged), WATCH days 0
10y-3m: 250 dates, latest 2026-11-02 spread 0.70 OK (flag unchanged), WATCH days 1

스프레드를 팀이 대응할 수 있는 표시로 바꾸기

표시가 같은 상태로 유지되는 동안 실행할 때마다 보내지 말고, 두 날짜 사이에 표시가 바뀔 때 메시지를 보내세요. 스크립트 출력의 마지막 줄이 그 비교를 보여 줍니다. 10y에서 2y를 뺀 값과 10y에서 3m를 뺀 값은 같은 날 서로 다른 표시에 놓일 수 있으므로 메시지에 쌍을 명시하세요.

실행 로그에 기록할 내용

• 커브의 원본 날짜와 실행한 날짜 및 시각.
• 사용한 만기와 받은 그대로의 yield_percent 값.
• 쌍별 스프레드와 표시.
• curve_type 값. 정의가 바뀌면 보이도록 합니다.
• 기록한 행 수와, HTTP 오류가 있으면 그 원인이 된 요청.

요청이 실패하면 오류가 발생하고 행은 기록되지 않습니다. 시계열에 빈 날짜가 남지만, 히스토리 경로가 빠진 날짜를 돌려주므로 다음 실행이 채웁니다.

새 날짜가 공개되기 전에 실행되면 어떻게 됩니까?

단일 날짜 경로는 가장 최근의 원본 날짜를 돌려줍니다. 그것이 아직 어제 날짜라면 같은 값으로 행이 다시 쓰일 뿐 아무것도 바뀌지 않으며, 이후 실행이 새 날짜를 가져옵니다.

같은 모니터로 다른 나라도 다룰 수 있습니까?

경로는 세 글자 국가 코드를 받습니다. 만기는 나라마다 다르므로, 먼저 maturity 없이 해당 국가를 요청해 나열된 만기를 읽으세요. 원본에 없는 만기는 코드 invalid_query의 400을 돌려주고, 커브가 없는 나라는 코드 series_not_found의 404를 돌려줍니다.

점검은 얼마나 자주 실행해야 합니까?

커브는 하루 단위로 공개되므로 새 날짜가 나온 뒤 하루에 한 번 실행하면 충분합니다. 일일 데이터 가져오기 예약하기에 나온 방식으로 일정을 잡으세요.