使用 Python 获取世界银行数据

操作指南由 Narwhal API 提供

Narwhal API 收录了 1,489 个采用 CC BY 4.0 许可的世界银行 WDI 年度指标。要在 Python 中获取世界银行数据，请先搜索指标，在其 ID 模板中填入小写国家代码，再使用 API 密钥请求观测记录。响应包含时期、十进制字符串数值、署名信息和许可 URL。套餐决定可访问的历史时间范围，以及每日未缓存数据的拉取额度。

使用 API 完成操作

  1. 获取免费密钥并准备 Python 环境 在注册页面创建账户，获取 API 密钥，并在运行 Python 的 shell 中设置 NARWHAL_API_KEY。将占位内容替换为你的密钥。安装 requests 用于 HTTP 调用，安装 pandas 用于 DataFrame。每次请求都在 Authorization 请求头中使用密钥；最后一步的完整脚本会从环境变量中读取密钥。
    安装依赖并设置密钥
    python -m pip install requests pandas
export NARWHAL_API_KEY="replace-with-your-key"
  2. 搜索指标 使用 source=wdi 在目录中搜索 gdp per capita。打印每条结果的 id_template 和标题，再选择所需指标。目录搜索向所有套餐开放，不会调用数据来源。params 参数会处理搜索短语中的空格。搜索结果描述的是指标；国家代码应填入下一步构建的序列 ID。
    查找世界银行指标模板
    import os
import requests

BASE_URL = "https://api.narwhalapi.com"
headers = {
    "Authorization": f"Bearer {os.environ['NARWHAL_API_KEY']}"
}
response = requests.get(
    f"{BASE_URL}/v1/economics/catalogue",
    headers=headers,
    params={"q": "gdp per capita", "source": "wdi"},
    timeout=10,
)
response.raise_for_status()
catalogue = response.json()
for hit in catalogue["results"]:
    print(hit["id_template"], hit["title"])
  3. 在模板中填入国家代码 将所选模板中的 {country} 替换为小写的 ISO 3166-1 alpha-3 代码。印度尼西亚使用 idn。世界银行 ID 的格式为 wdi:<indicator>/<iso3>，指标和国家之间包含斜杠。通过 id 查询参数传入完整 ID。代码片段选择第一条搜索结果；使用前请检查打印出的标题。
    构建指定国家的序列 ID
    if not catalogue["results"]:
    raise SystemExit("No matching indicators found.")
selected = catalogue["results"][0]
series_id = selected["id_template"].replace("{country}", "idn")
print(series_id)
  4. 读取年度观测记录 请求该 ID 的观测记录。年度时期使用 YYYY；可选的 start 和 end 均包含边界。响应上限最多为 1,000 条观测记录，默认为 100。如果 next_cursor 有值，请将其作为 cursor 传入以继续获取。提高上限不会扩大套餐的历史时间范围，缺失年份也不会出现在响应中。
    读取所选序列
    response = requests.get(
    f"{BASE_URL}/v1/economics/catalogue/series/observations",
    headers=headers,
    params={"id": series_id, "limit": 1000},
    timeout=10,
)
response.raise_for_status()
observations = response.json()["observations"]
  5. 创建 DataFrame 使用观测记录构建 DataFrame，将 period 转为整数，将 value 转为浮点数。返回的数值采用十进制字符串。需要十进制运算时，请使用 Decimal 代替 float()。分析时请保留响应中的单位标签：指标标题说明了数值衡量的内容。此转换不会填补缺口，也不会为缺失年份创建观测记录。
    转换时期和数值
    import pandas as pd

frame = pd.DataFrame(observations, columns=["period", "value"])
frame["period"] = frame["period"].astype(int)
frame["value"] = frame["value"].map(float)
print(frame)
  6. 批量查询国家并处理请求错误 运行此完整脚本，获取示例 GDP 序列，并批量查询印度尼西亚和加拿大的最新值。使用逗号连接 ID；最新值端点接受 1 至 50 个不同 ID。按返回的每个序列计一次调用，并一次性扣除。脚本打印每个国家的时期和数值，报告缺失的 ID，并在收到 400、404 或 429 fetch_limit_reached 响应时显示错误信息并停止。
    读取 GDP 历史并批量获取最新值
    import os
import requests
import pandas as pd

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

def get_json(path, params):
    response = requests.get(
        f"{BASE_URL}{path}",
        headers=HEADERS,
        params=params,
        timeout=10,
    )
    try:
        response.raise_for_status()
    except requests.HTTPError:
        problem = response.json()
        code = problem.get("code", "unknown_error")
        if response.status_code == 429 and code == "fetch_limit_reached":
            retry_after = response.headers.get("Retry-After")
            raise SystemExit(
                f"Uncached fetch allowance used. Retry-After: {retry_after}"
            )
        if response.status_code in (400, 404):
            raise SystemExit(f"{code}: {problem.get('detail', '')}")
        raise
    return response.json()

series_id = "wdi:ny.gdp.mktp.cd/idn"
rows = []
params = {"id": series_id, "limit": 1000}
while True:
    payload = get_json(
        "/v1/economics/catalogue/series/observations", params
    )
    rows.extend(payload["observations"])
    cursor = payload.get("next_cursor")
    if not cursor:
        break
    params["cursor"] = cursor

frame = pd.DataFrame(rows, columns=["period", "value"])
frame["period"] = frame["period"].astype(int)
frame["value"] = frame["value"].map(float)
print(frame)

countries = ("idn", "can")
ids = [f"wdi:ny.gdp.mktp.cd/{country}" for country in countries]
batch = get_json(
    "/v1/economics/catalogue/latest",
    {"ids": ",".join(ids)},
)
for observation in batch["observations"]:
    country = observation["id"].rsplit("/", 1)[1]
    print(country, observation["period"], observation["value"])
for missing in batch["missing"]:
    print("Missing:", missing)

如何理解印度尼西亚 GDP 示例？

序列 wdi:ny.gdp.mktp.cd/idn 表示按现价美元计算的印度尼西亚 GDP。使用 Pro 密钥于 27 October 2026 发起的 limit=3 请求返回了这些观测记录。此示例使用 GDP 总量；如需人均 GDP，请使用目录搜索得到的模板。

从 GDP 响应中保存的观测记录
[
  {"period": "1967", "value": "5667756627.92093"},
  {"period": "1968", "value": "7076465299.81141"},
  {"period": "1969", "value": "8337423312.88344"}
]

对于 1967 年，float("5667756627.92093") 将数值转换为可在 DataFrame 中进行数值运算的形式。如果还需要其十进制表示，请保留原始字符串。响应包含世界银行 WDI 署名信息和 CC BY 4.0 许可 URL。使用数据时，请同时保留署名和许可信息。

为什么我的 Free 密钥只返回几年的数据？

Free 的历史时间范围为 12 个月，年度数据则至少提供最新的 5 条观测记录。Narwhal 会在套餐时间范围与该最低数量之间，采用能提供更多观测记录的规则。因此，即使大多数记录超出常规时间范围，年度序列仍会保留多条观测记录。年度最低数量规则适用于所有套餐。

我的套餐适用哪种限制：

套餐每账户每 UTC 日的未缓存数据获取次数历史数据范围
免费10最近 12 个月，且至少包含最新的 5 条年度观测记录
Basic100最近 5 年，且至少包含最新的 5 条年度观测记录
Pro1,000完整历史数据

如果请求范围完全早于允许的时间范围，将返回 403 plan_upgrade_required。对于年度序列，Narwhal 会在加载序列后检查该范围，以便计入最低数量规则。请参阅套餐详情，选择分析所需的历史时间范围。

429 fetch_limit_reached 错误是什么意思？

你的账户已用完每日额度，而该请求需要拉取未缓存的数据。重试前请读取 Retry-After；额度在 00:00 UTC 重置。达到上限后，已缓存的序列仍可返回结果。世界银行数据的缓存时长为 24 小时至 7 天，因此再次请求不一定需要拉取数据。

遇到 400 invalid_series_id 时，请检查指标、斜杠和小写国家代码。400 invalid_query 表示查询无效。404 not_found 可能表示索引中没有该指标；Narwhal 会将该结果保留 24 小时。重复请求前，请根据目录结果核对 ID。

过期缓存结果以及 400、404 和 503 响应不计费。如果来源不可用且没有缓存副本，Narwhal 将返回 503 upstream_busy，并附带 Retry-After。重试处理方式详见缓存与 429 处理。

发布于。

后续步骤