用 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。

发布于。

后续步骤