マクロイベントの市場オッズを表示するダッシュボードを作る

curl -s -H "Authorization: Bearer $NARWHAL_API_KEY" \
  "https://api.narwhalapi.com/v1/predictions/topics?group=us_jobs"

{
  "topics": [
    {
      "topic_id": "usa-payrolls",
      "group": "us_jobs",
      "title": "US nonfarm payrolls change",
      "shape": "scalar",
      "unit": "jobs",
      "unit_key": "job",
      "venues": [
        "kalshi"
      ],
      "resolution_kind": "release",
      "calendar": {
        "country": "USA",
        "family": "labour"
      },
      "next_event_id": "usa-payrolls-2026-10"
    },
    {
      "topic_id": "usa-unemployment",
      "group": "us_jobs",
      "title": "US unemployment rate",
      "shape": "scalar",
      "unit": "percent",
      "unit_key": "percent",
      "venues": [
        "kalshi",
        "polymarket"
      ],
      "resolution_kind": "release",
      "calendar": {
        "country": "USA",
        "family": "labour"
      },
      "next_event_id": "usa-unemployment-2026-10"
    }
  ]
}

curl -s -H "Authorization: Bearer $NARWHAL_API_KEY" \
  "https://api.narwhalapi.com/v1/predictions/binary/usa-recession-2026"

{
  "event_id": "usa-recession-2026",
  "topic_id": "usa-recession-2026",
  "question": "US recession by end of 2026?",
  "status": "trading",
  "closes_at": "2026-12-31T00:00:00Z",
  "resolves_at": "2026-12-31T00:00:00Z",
  "as_of": "2026-11-03T02:21:00Z",
  "stale": false,
  "resolution": "Two consecutive negative quarters of US real GDP from Q2 2025 to Q4 2026, or an NBER recession call for 2025 or 2026",
  "venues": [
    {
      "venue": "kalshi",
      "url": "https://kalshi.com/markets/kxrecssnber/kxrecssnber-26",
      "rules_url": "https://assets.kalshi.com/contract_terms/NBERRECSS.pdf",
      "settlement_source": "Bureau of Economic Analysis",
      "excluded_reason": "rules",
      "match": {
        "differences": [
          "rules"
        ],
        "note": "Kalshi resolves on two negative GDP quarters only; Polymarket also counts an NBER recession call"
      },
      "volume_24h_usd": "123.64",
      "open_interest_usd": "17994.22"
    },
    {
      "venue": "polymarket",
      "url": "https://polymarket.com/event/us-recession-by-end-of-2026",
      "settlement_source": "Bureau of Economic Analysis; NBER",
      "weight": "1.0000",
      "volume_24h_usd": "2090.24"
    }
  ],
  "probability": "0.0550",
  "quotes": [
    {
      "venue": "kalshi",
      "probability": "0.0250",
      "market_ids": [
        "KXRECSSNBER-26"
      ]
    },
    {
      "venue": "polymarket",
      "probability": "0.0550",
      "market_ids": [
        "100379208559626151022751801118534484742123694725746262280150222742563282755057"
      ]
    }
  ]
}

import json
import os
import urllib.request

def get(path):
    req = urllib.request.Request(
        "https://api.narwhalapi.com" + path,
        headers={"Authorization": "Bearer " + os.environ["NARWHAL_API_KEY"]},
    )
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.load(r)

def headline(shape, e):
    if shape == "binary":
        if "probability" not in e:
            return "no usable quote"
        return f"yes {float(e['probability']):.1%}"
    if shape == "scalar":
        priced = [o for o in e["outcomes"] if "probability" in o]
        if not priced:
            return "no usable quote"
        top = max(priced, key=lambda o: float(o["probability"]))
        return f"{top['label']} {float(top['probability']):.1%}"
    if shape == "timing":
        step = e["steps"][0]
        if "probability" not in step:
            return "no usable quote"
        return f"by {step['label']} {float(step['probability']):.1%}"
    if shape == "extreme":
        if not e["up"] or "probability" not in e["up"][0]:
            return "no usable quote"
        step = e["up"][0]
        return f"reach {step['level']} {float(step['probability']):.1%}"

for group in ("us_prices", "us_jobs", "us_activity"):
    for topic in get("/v1/predictions/topics?group=" + group)["topics"]:
        if not topic.get("next_event_id") or topic["shape"] == "scalar" and topic["unit"] == "count":
            continue
        e = get(f"/v1/predictions/{topic['shape']}/{topic['next_event_id']}")
        print(group, e["event_id"], topic["shape"], headline(topic["shape"], e),
              "stale" if e["stale"] else "live")