殖利率利差：每日記錄曲線利差並依門檻標記

風險團隊除了一次性答案還需要什麼

每天關注曲線的風險或投資組合團隊需要四樣東西：自己擁有的一條序列、每天執行一次的檢查、觸發標記的門檻，以及事後能解釋每個標記的日誌。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. 回填一年的資料 backfill 以 limit 為 1000 向 /history 請求 365 天的資料，並沿著 next_cursor 翻頁，所以全新的資料庫一開始就有完整一年的序列。
2. 用 Decimal 計算每個利差 record 對 PAIRS 中的每個組合，用長端殖利率減去短端殖利率。數值保持為 Decimal，所以減法不會引入浮點誤差，並以文字形式儲存。
3. 根據門檻設定標記 flag 在利差低於零時回傳 INVERTED，從零到 WATCH 之間回傳 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 請求該國，讀取它列出的期限。來源中沒有的期限會回傳 400，代碼為 invalid_query；沒有曲線的國家會回傳 404，代碼為 series_not_found。

檢查應該多久執行一次？

曲線是一次公布一個日期，所以在新日期出現後每天執行一次就夠了。可以依排程每日資料擷取的方法設定排程。