收益率曲线监控：每日记录利差并按阈值标记

风险团队除了一次性答案还需要什么

每天关注曲线的风险或投资组合团队需要四样东西：自己拥有的一条序列、每天运行一次的检查、触发标记的阈值，以及事后能解释每个标记的日志。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。

检查应该多久运行一次？

曲线是一次发布一个日期，所以在新日期出现后每天运行一次就够了。可以按安排每日数据拉取的方法设置定时。