用 DuckDB API 查詢 Narwhal 資料

藉由 DuckDB 和 Narwhal API，你可以用 SQL 查詢國定假日、發布日期、經濟數據、大宗商品、外匯和預測市場機率。一個小型 Python 函式會用你的 API 金鑰取得 JSON，把每個回應儲存為檔案，再由 DuckDB 讀入資料表。

你將完成的設定

DuckDB 用 read_json 讀取本機 JSON，這套設定會產生一個 narwhal.duckdb 資料庫，每個資料領域對應一張資料表。Python 負責發送 API 請求並寫出回應檔案，SQL 再讀取這些檔案。完整範例包含這些查詢：國定假日和發布日期的數量、CPI 觀測值、Brent 價格、匯率以及聯準會政策機率。Narwhal 用一個帳號和一把 API 金鑰涵蓋這些 API，REST 和 MCP 共用該金鑰。

設定你的 API 金鑰

  1. 建立免費金鑰 在 narwhalapi.com/signup 建立免費金鑰。範例中的各資料領域使用同一把 Narwhal API 金鑰；請求輔助函式會把它放在 Authorization 標頭中，格式為 Bearer YOUR_KEY。
  2. 把金鑰存入環境變數 為 Python 檔案設定一個名為 NARWHAL_API_KEY 的環境變數。Python 設定部分透過 os.environ 讀取它，並放進 fetch 使用的標頭，因此金鑰不會出現在 SQL 文字中。
  3. 安裝 Python 套件 為 Python 檔案安裝 duckdb 和 requests。範例會匯入這兩個套件，用 requests 取得每個回應，並用 duckdb.connect 開啟資料庫。
    安裝 DuckDB 和 requests
    pip install duckdb requests

用程式碼取得資料

把 Python 範例儲存為檔案，並在環境中提供 NARWHAL_API_KEY 後執行。每次呼叫 fetch 都會寫出一個命名的 JSON 檔案，隨後的 SQL 會把該檔案讀入對應的資料表。輔助函式向 https://api.narwhalapi.com 發送 GET 請求，並在寫入回應前呼叫 raise_for_status。最後的查詢透過同一個 DuckDB 連線讀取產生的資料表。如果只想拉取某個領域，請搭配這個輔助函式和連線使用較短的領域範例。

DuckDB：六個領域匯入資料表
import os
import duckdb
import requests

API = "https://api.narwhalapi.com"
HEADERS = {"Authorization": f"Bearer {os.environ['NARWHAL_API_KEY']}"}


def fetch(name, path, **params):
    r = requests.get(API + path, params=params, headers=HEADERS, timeout=10)
    r.raise_for_status()
    with open(f"{name}.json", "w") as file:
        file.write(r.text)


con = duckdb.connect("narwhal.duckdb")

fetch("holidays", "/v1/calendars/holidays/USA/2026")
con.sql(
    "CREATE OR REPLACE TABLE holidays AS "
    "SELECT unnest(holidays, recursive := true) "
    "FROM read_json('holidays.json')"
)

fetch(
    "releases",
    "/v1/economics/calendar",
    countries="USA",
    start_date="2026-11-01",
    end_date="2026-11-30",
    limit=100,
)
con.sql(
    "CREATE OR REPLACE TABLE releases AS "
    "SELECT unnest(releases, recursive := true) "
    "FROM read_json('releases.json')"
)

fetch("cpi", "/v1/economics/USA/cpi/observations", start="2025-09")
con.sql(
    "CREATE OR REPLACE TABLE cpi AS "
    "SELECT unnest(observations, recursive := true) "
    "FROM read_json('cpi.json')"
)

fetch("brent", "/v1/commodities/spot/BRENT/USD")
con.sql(
    "CREATE OR REPLACE TABLE brent AS SELECT price, unit FROM read_json('brent.json')"
)

fetch("fx", "/v1/fx/rates/USD", currencies="EUR,IDR")
con.sql(
    "CREATE OR REPLACE TABLE fx AS "
    "SELECT unnest(rates, recursive := true) "
    "FROM read_json('fx.json')"
)

fetch("fed", "/v1/predictions/policy-paths/federal-reserve")
con.sql(
    "CREATE OR REPLACE TABLE fed AS "
    "SELECT unnest(meetings, recursive := true) "
    "FROM read_json('fed.json')"
)

print(
    con.sql("SELECT count(*) FROM holidays").fetchall(),
    con.sql("SELECT count(*) FROM releases").fetchall(),
)
print(con.sql("SELECT period, value FROM cpi ORDER BY period DESC LIMIT 2").fetchall())
print(
    con.sql("SELECT * FROM brent").fetchall(),
    con.sql("SELECT currency, rate FROM fx").fetchall(),
)
print(
    con.sql(
        "SELECT p_hold, expected_change_bps FROM fed ORDER BY resolves_at LIMIT 1"
    ).fetchall()
)

按排程定時執行

要重複拉取資料，就用 cron 執行這個 Python 檔案。請求標頭從環境變數讀取金鑰，所以要讓排程執行的 Python 檔案能存取 NARWHAL_API_KEY。每次呼叫都會重新取得回應，並在 DuckDB 讀取之前寫出 JSON 檔案。CREATE OR REPLACE TABLE 讓每次執行都可重複：同名的資料表會換成從這些檔案讀到的資料。做排程拉取時，請把 fetch 呼叫和建表語句放在同一個 Python 檔案裡。

在 DuckDB 中查詢國定假日

取得美國 2026 年的國定假日日曆，再把其中的 holidays 欄位讀入 holidays 資料表。範例選取 date 和 name，所以 SQL 結果裡每個假日的日期和名稱並列顯示。2026年10月24日，這個回應包含 11 個國定假日。如果只想要數量而不是逐列的假日，完整的 Python 範例裡也有針對該資料表的計數查詢。

讀取國定假日的日期和名稱
fetch("holidays", "/v1/calendars/holidays/USA/2026")
con.sql(
    "CREATE OR REPLACE TABLE holidays AS "
    "SELECT unnest(holidays, recursive := true) "
    "FROM read_json('holidays.json')"
)
print(con.sql("SELECT date, name FROM holidays").fetchall())

在 DuckDB 中查詢發布日期

用給定的開始日期、結束日期和筆數上限，拉取美國 2026 年 11 月的經濟日曆。回應的 releases 欄位包含 at、period、topic 和 status，範例從資料表中讀取期間、主題和狀態這幾欄。2026年10月24日，回應包含 31 筆發布記錄。這些是經濟數據的發布日期，而下一個範例中的觀測值請求讀取的是按期間排列的 CPI 值。

讀取發布期間和主題
fetch(
    "releases",
    "/v1/economics/calendar",
    countries="USA",
    start_date="2026-11-01",
    end_date="2026-11-30",
    limit=100,
)
con.sql(
    "CREATE OR REPLACE TABLE releases AS "
    "SELECT unnest(releases, recursive := true) "
    "FROM read_json('releases.json')"
)
print(con.sql("SELECT period, topic, status FROM releases").fetchall())

在 DuckDB 中查詢經濟數據

讀取美國 CPI 觀測值，並把 start 設為 2025-09。JSON 欄位名為 observations，產生的 cpi 資料表包含 period 和 value 兩欄，供查詢選取。2026年10月24日，2026-09 的 CPI 值為 336.940。範例按期間遞減排列觀測值並限制回傳筆數，使用的查詢與完整 Python 範例相同。

讀取 CPI 觀測值
fetch("cpi", "/v1/economics/USA/cpi/observations", start="2025-09")
con.sql(
    "CREATE OR REPLACE TABLE cpi AS "
    "SELECT unnest(observations, recursive := true) "
    "FROM read_json('cpi.json')"
)
print(con.sql("SELECT period, value FROM cpi ORDER BY period DESC LIMIT 2").fetchall())

在 DuckDB 中查詢大宗商品

取得以 USD 計價的 Brent 現貨價格，直接從儲存的 JSON 中讀取 price 和 unit。這條建表語句不做清單展開，直接選取這兩個欄位。2026年10月24日，Brent 為每桶 116.8809 美元。範例用 SELECT * 讀取 brent 資料表，所以回傳給 Python 的結果同時包含價格和單位。

讀取 Brent 價格和單位
fetch("brent", "/v1/commodities/spot/BRENT/USD")
con.sql(
    "CREATE OR REPLACE TABLE brent AS SELECT price, unit FROM read_json('brent.json')"
)
print(con.sql("SELECT * FROM brent").fetchall())

在 DuckDB 中查詢外匯

以 USD 為基準貨幣，在 currencies 參數中請求 EUR 和 IDR。把回應的 rates 欄位讀入 fx 資料表，然後一起選取 currency 和 rate。2026年10月24日，USD/EUR 為 0.89271，USD/IDR 為 18107.8。這些匯率以文字字串形式儲存。查詢把資料列讀回 Python 時，每種貨幣都和它的匯率保持對應。

讀取匯率
fetch("fx", "/v1/fx/rates/USD", currencies="EUR,IDR")
con.sql(
    "CREATE OR REPLACE TABLE fx AS "
    "SELECT unnest(rates, recursive := true) "
    "FROM read_json('fx.json')"
)
print(con.sql("SELECT currency, rate FROM fx").fetchall())

在 DuckDB 中查詢預測市場

把回應 meetings 欄位中的聯準會政策路徑讀入 fed 資料表。2026年10月24日，2026年10月28日那次會議的維持不變機率為 0.9188，預期變動為 1.7 個基點。查詢按 resolves_at 排序，並讀取該次會議的 p_hold 和 expected_change_bps，把數值欄回傳給 Python。

讀取維持不變機率和預期變動
fetch("fed", "/v1/predictions/policy-paths/federal-reserve")
con.sql(
    "CREATE OR REPLACE TABLE fed AS "
    "SELECT unnest(meetings, recursive := true) "
    "FROM read_json('fed.json')"
)
print(
    con.sql(
        "SELECT p_hold, expected_change_bps FROM fed ORDER BY resolves_at LIMIT 1"
    ).fetchall()
)

為什麼對清單欄位使用 unnest？

清單嵌在單一 JSON 欄位裡。unnest(list_column, recursive := true) 會把它展開成每個項目一列，於是國定假日、發布、觀測值、匯率和會議這些清單就成了各自資料表中的資料列。在 unnest 裡要寫回應的欄位名：例如 CPI 的回應用的是 observations，儘管它儲存的檔案和資料表都叫 cpi。

如何透過 Python 選取時間戳記？

透過 Python 選取 at 或 resolves_at 時間戳記之前，請先安裝 pytz。你也可以視需要在 SQL 中轉換這些時間戳記的型別。預測市場範例在 SQL 內按 resolves_at 排序，並把數值欄回傳給 Python；發布日期範例選取的是 period、topic 和 status。

發布於 。

後續步驟