تسجيل Yield Curve Spread يوميًا ووسمه مقابل حد

ما يحتاجه فريق المخاطر إلى جانب جواب لمرة واحدة

يحتاج فريق المخاطر أو المحفظة الذي يراقب المنحنى كل يوم إلى أربعة أشياء: سلسلة يملكها، وفحصًا يعمل مرة في اليوم، وحدًا يطلق الوسم، وسجلًا يشرح كل وسم لاحقًا. توفر الـ 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 سنوات ناقص عائد سنتين هو 0.32 نقطة مئوية، وناقص عائد 3 أشهر هو 0.70.

جدول واحد بصف واحد لكل تاريخ وزوج

يملأ السكربت أدناه الجدول من مسار السجل التاريخي ثم يضيف أحدث منحنى. المفتاح الأساسي هو التاريخ والزوج، وكل عملية كتابة تستبدل صفًا، فيؤدي تشغيل السكربت مرتين أو إعادة تشغيله بعد انهيار إلى الحالة نفسها.

1. ملء سنة من الماضي يطلب backfill من /history مدة 365 يومًا مع ضبط limit على 1000 ويتتبع next_cursor، فتبدأ قاعدة البيانات الجديدة بسلسلة سنة كاملة.
2. حساب كل فارق باستخدام Decimal يطرح record العائد القصير من العائد الطويل لكل زوج في PAIRS. وتبقى القيم من النوع Decimal فلا يدخل الطرحَ ضجيج الفاصلة العائمة، وتُخزَّن نصوصًا.
3. ضبط الوسم من الحد تعيد flag القيمة INVERTED تحت الصفر، و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.

كم مرة ينبغي أن يعمل الفحص؟

يُنشر المنحنى لتاريخ واحد في كل مرة، لذا يكفي تشغيل واحد في اليوم بعد ظهور التاريخ الجديد. جدوِل التشغيل كما في جدولة سحب البيانات اليومي.