イールドカーブ スプレッドを毎日記録し、しきい値に照らして警告を付ける

一度きりの回答のほかにリスクチームが必要とするもの

毎日カーブを見るリスクチームまたはポートフォリオチームには、4つのものが必要です。自分が持つ系列、1日1回走るチェック、警告を出すしきい値、そしてあとで各警告を説明できるログです。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 はカーブの参照元の日付で、呼び出した日より1日前のことがあるので、保存する系列のキーは実行日ではなくこの日付にしてください。yield_percent はパーセントを表す10進数の文字列です。curve_type は参照元がカーブをどう定義しているかを示し、定義の変更がログで分かるようにモニターが保存しておくべき値です。10年利回りから2年利回りを引くと 0.32 パーセントポイント、3か月利回りを引くと 0.70 です。

日付とペアごとに1行ずつの表を1つ持つ

以下のスクリプトは、履歴ルートから表を埋め、続いて最新のカーブを追加します。主キーは日付とペアで、書き込みのたびに行を置き換えるので、スクリプトを2回実行しても、クラッシュ後に再開しても、同じ状態に落ち着きます。

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}")

最初の実行と直後の2回目の実行は同じ2行を出力し、500行が残りました。2つのペアそれぞれ250日分です。系列の日付数が暦日より少ないのは、週末と市場休日に行がないためです。WATCH の日は 2025-11-26 の1日だけで、その日は 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

スプレッドをチームが動ける警告にする

メッセージは、警告が同じまま続く間は毎回の実行ではなく、2つの日付の間で警告が変わったときに送ります。スクリプト出力の最終行にその比較が出ています。10y から 2y を引いた値と 10y から 3m を引いた値は同じ日に別の警告になることがあるので、メッセージにはペアを明記してください。

実行ログに書く内容

• カーブの参照元の日付と、実行した日付・時刻。
• 使った満期と、受け取ったままの yield_percent の値。
• ペアごとのスプレッドと警告。
• curve_type の値。定義の変更が分かるようにします。
• 書き込んだ行数と、HTTP エラーがあればその原因となったリクエスト。

リクエストが失敗するとエラーが上がり、行は書き込まれません。系列には欠けた日が残りますが、履歴ルートが欠けた日付を返すので次の実行で埋まります。

新しい日付が公表される前に実行されたらどうなりますか。

単一日付のルートは最新の参照元の日付を返します。それがまだ昨日の日付なら、同じ値で行が書き直されるだけで何も変わらず、あとの実行で新しい日付が取り込まれます。

同じモニターで別の国もカバーできますか。

ルートは3文字の国コードを受け取ります。満期は国によって異なるので、まず maturity を付けずにその国を取得し、載っている満期を読んでください。参照元にない満期はコード invalid_query の 400 を返し、カーブのない国はコード series_not_found の 404 を返します。

どのくらいの頻度でチェックを実行すべきですか。

カーブは1日分ずつ公表されるので、新しい日付が出た後に1日1回実行すれば十分です。毎日のデータ取得をスケジュールするの方法で組んでください。