بث البيانات المباشر WebSocket للأحداث

بث الأحداث الاقتصادية عبر API

يرسل بث الأحداث رسالة في كل مرة تسجّل فيها Narwhal إصدار بيانات جديدًا أو مراجعة أو تغييرًا في التقويم أو منحنى عائد. وتختار الدول وعائلات البيانات وأنواع الأحداث، وتحمل كل رسالة cursor يمكنك استخدامه لمتابعة العمل من حيث توقفت.

1. تجهيز Python ومفتاح API ثبّت حزمة Python المسماة websockets عبر pip install "websockets>=14". يدعم الإصدار 14 أو الأحدث الوسيط additional_headers المستخدم هنا. احتفظ بمفتاحك في متغيّر البيئة NARWHAL_API_KEY. تستخدم REST وMCP وكل البث المفتاح نفسه. يمكنك الحصول على مفتاح مجاني.
2. الاتصال وإرسال الاشتراك ترسل الخوادم والتطبيقات Authorization: Bearer YOUR_KEY في ترويسات مصافحة WebSocket؛ أما تذاكر البث فهي لموقع Narwhal فقط. اقرأ ready ثم أرسل subscribe بوصفه الإطار الأول خلال 5 ثوانٍ. يطلب هذا البرنامج النصي أحداث الأسعار وسوق العمل في الولايات المتحدة والمكسيك. وتشمل القائمة الفارغة كل قيم ذلك المرشّح، لذا يستقبل "event_types": [] كل أنواع الأحداث.
   Python (websockets)، الاشتراك في أحداث الأسعار وسوق العمل للولايات المتحدة والمكسيك

   import asyncio, json, os
   import websockets
   
   
   async def main(cursor=None):
       headers = {"Authorization": f"Bearer {os.environ['NARWHAL_API_KEY']}"}
       async with websockets.connect(
           "wss://api.narwhalapi.com/v1/events/stream", additional_headers=headers
       ) as ws:
           print(json.loads(await ws.recv()))  # ready
           subscribe = {
               "action": "subscribe",
               "countries": ["USA", "MEX"],
               "families": ["prices", "labour"],
               "event_types": [],
           }
           if cursor:
               subscribe["cursor"] = cursor
           await ws.send(json.dumps(subscribe))
           async for message in ws:
               frame = json.loads(message)
               if "cursor" in frame:
                   cursor = frame["cursor"]  # store this to resume after a drop
               print(frame)
   
   
   asyncio.run(main())

3. قراءة الرسائل وحفظ المؤشر يرد الخادم بـ subscribed ثم يرسل كائنات الأحداث وإطارات heartbeat وpong، ولكل منها cursor. احفظ أحدث مؤشر. وإرسال رسالة subscribe ثانية يغلق الاتصال بالخطأ resubscribe_required، لذا أعد الاتصال عندما تحتاج إلى مرشّحات مختلفة. يقبل الاشتراك حتى 60 من countries و12 من families و9 من event_types، وتُحدَّد إطارات العميل بـ 2,048 بايت.

تصل أخطاء البث بالصيغة {"type":"error","code":"..."}. والرموز هي stream_limit وrate_limited وsubscribe_required وquota_unavailable وunauthorized وslow_consumer وfeed_unavailable وrevoked وresubscribe_required وcommand_limit وinvalid_command وinvalid_frame وcursor_expired وcursor_future وshutdown.

بث احتمالات التنبؤ

يرسل بث التنبؤات احتمالات Kalshi وPolymarket للموضوعات أو الأحداث التي تختارها. اشترك بما يصل إلى 40 من topics و200 من event_ids. يرسل الخادم snapshot لكل حدث، ثم update عند تغيّر احتمالاته، بحد أقصى مرة في الدقيقة لكل حدث، ورسالة status عند إغلاق الحدث أو تسويته.

import asyncio, json, os
import websockets


async def main():
    headers = {"Authorization": f"Bearer {os.environ['NARWHAL_API_KEY']}"}
    async with websockets.connect(
        "wss://api.narwhalapi.com/v1/predictions/stream", additional_headers=headers
    ) as ws:
        print(json.loads(await ws.recv()))  # ready
        subscribe = {"action": "subscribe", "topics": ["usa-fed"], "event_ids": []}
        await ws.send(json.dumps(subscribe))
        async for message in ws:
            frame = json.loads(message)
            if frame["type"] in ("snapshot", "update", "status"):
                print(frame)


asyncio.run(main())

تصل رسالة heartbeat كل 30 ثانية. تمنحك إعادة الاتصال لقطات جديدة دون إعادة تشغيل للماضي، لذا لا يحتاج انقطاع الاتصال إلا إلى اشتراك جديد.

الخيار المناسب لمهمتك

ماذا يحدث عند انقطاع الاتصال؟

في بث الأحداث، أعد الاتصال مع تأخير تدريجي (backoff) وأرسل آخر مؤشر حفظته في رسالة subscribe. يعيد البث كل حدث بعده. يعني cursor_expired أن المؤشر خارج السجل المحتفظ به: ابدأ من دونه وأعد البناء من البيانات الحالية. ويعني cursor_future أن المؤشر أبعد من السجل، فاستخدم مؤشرًا أعاده الخادم.

في بث التنبؤات، أعد الاتصال واشترك من جديد. ستتلقى لقطة جديدة لكل حدث، ولا تُعاد التغييرات التي حدثت أثناء غيابك.

لماذا انقطع اتصالي بعد 5 ثوانٍ؟

يغلق بث الأحداث الاتصال الذي لا يرسل رسالة subscribe خلال 5 ثوانٍ من ready، مع subscribe_required. ويُغلق العميل الذي يتوقف عن القراءة مع slow_consumer. وتواصل حلقة الاستقبال في كل برنامج نصي القراءة بعد الاشتراك.

كم تكلفة البث؟

تكلّف كل عملية اتصال استدعاءً واحدًا من رصيدك، ويمكن لكل حساب البث. والاتصال المفتوح ليوم كامل يكلّف الاستدعاء الواحد نفسه. وعندما يبلغ الحساب الحد الأقصى لاتصالات البث المفتوحة، يتلقى الاتصال الجديد stream_limit ثم يُغلق. راجع الأسعار لمعرفة سعر الاستدعاء.

مثال تطبيقي: إصدار بيانات سوق العمل في المكسيك

مع الاشتراك المحدد في البرنامج النصي بلغة Python، يصل إصدار بيانات سوق العمل في المكسيك المسجَّل في 25 سبتمبر 2026 على هيئة كائن الحدث أدناه. قيمة type هي dataset.published، ويحدد resource_id مع resource_version الإصدار الذي نُشر. احفظ cursor قبل معالجة الحدث، حتى تستأنف إعادة الاتصال بعده.

{
  "cursor": "1",
  "event_id": "2cdf117d-0000-4000-8000-000000000001",
  "type": "dataset.published",
  "country": "MEX",
  "family": "labour",
  "resource_id": "775d11a3-0000-4000-8000-000000000001",
  "resource_version": "822e107c-0000-4000-8000-000000000001",
  "recorded_at": "2026-09-25T15:44:56Z"
}