批量下载历史预报数据

需求：批量下载一次或多个历史预报运行的结果数据，用于预报检验、模型评估等场景。

历史预报数据模型：archive_ifs（25km，360小时）
支持的历史预报数据源及要素：历史预报数据API
全部支持的逐日数据要素查看：逐日要素支持
以 Python 代码为例，分四步完成：

---

多起报范围下载

仅历史预报模型支持多起报范围模式。下面的请求会选择 2026-01-01 至 2026-01-10（含）每天 UTC 00

和 12 的起报场，共 20 个 run；每个 run 读取起报时刻开始、最长 72 小时的逐小时数据。

{
  "domain": "archive_ifs",
  "hourly": ["temperature_2m", "precipitation"],
  "daily": [],
  "monthly": [],
  "time_range": {
    "type": "forecast",
    "start": "2026-01-01",
    "end": "2026-01-10",
    "start_hour": "00,12",
    "forecast_hours": 72
  },
  "area": {
    "type": "point",
    "locations": [
      {"id": "1", "name": "北京", "lon": 116.4, "lat": 39.9},
      {"id": "2", "name": "上海", "lon": 121.5, "lat": 31.2}
    ]
  },
  "timezone": "Asia/Shanghai",
  "area_average": false
}

start 和 end 是包含首尾日期的 UTC 日期；start_hour 仅允许由 00、06、12、18 组成的逗号分隔列表。日期按升序生成，小时按同日升序生成。空列表、重复小时、01、24、ISO 日期时间或非历史预报模型都会导致参数错误。

多起报范围模式只支持 hourly 或 daily，不能选择 monthly。forecast_hours 省略时，每个 run 使用该 run 的可用预报时长；指定时，系统会分别校验每个 run 是否支持请求的时效。任何一个 run 的时效超限或读取失败，整个任务都会失败，并报告对应的起报时间，不会静默跳过。

输出结构

多起报范围模式会把所有 run 写入同一个按地点和分辨率组织的输出文件，不会为每个 run 额外生成文件。

• CSV：在现有列中加入 init_time，位置在 time 之前：id,name,lat,lon,init_time,time,<variables>。init_time 是 UTC ISO-8601 起报时间；time 是逐小时有效时间或逐日有效日期。
• NetCDF：增加 init_time 维度和坐标变量，单位为 seconds since 1970-01-01 00:00:00；time 坐标表示从 0 开始的预报时效秒数，并使用 standard_name=forecast_period。数据变量维度顺序为 init_time, time, <空间维度>，空间维度沿用 latitude, longitude、y, x 或 point。daily 的 lead time 为 0、86400、172800……。不同 run 的可用时长不同时，尾部缺失值写为 NaN。
• 估算与进度：输出大小和进度总量按所有 run 以及每个 run 的时间步数计算，estimate 的选择和估算逻辑与正式导出一致。

单起报模式和非历史预报导出不增加 init_time，继续使用当前的 time × 空间维度 结构和 CSV 列格式。

兼容的单起报模式

只需将 start_hour 设置为 UTC ISO 日期时间，并省略 start、end，即可继续下载单个历史预报场：

"time_range": {
  "type": "forecast",
  "start_hour": "2025-05-15T00:00",
  "forecast_hours": 48
}

该模式保持原有输出结构；也可以省略 forecast_hours，由模型使用该起报时间对应的可用预报时长。

---

方式一：使用行政区划

适合按省/市/区县行政区划批量下载，adcode 可从平台提供的 CSV 文件中查询：省/直辖市使用 province.csv，地级市使用 city.csv，区县使用 district.csv。

第一步：获取区划 adcode 列表

import io
import time
import zipfile
import requests
import pandas as pd

API_KEY = "your_api_key"
BASE_URL = "https://api.mirror-earth.com"

# 下载区县表（也可替换为 province.csv / city.csv）
res = requests.get("https://open.mirror-earth.com/district.csv")
res.encoding = "utf-8"
df = pd.read_csv(io.StringIO(res.text))
adcodes = df["adcode"].astype(str).tolist()

print(f"共 {len(adcodes)} 个区县")

第二步：创建批量下载任务

headers = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json",
}

payload = {
    "domain": "archive_ifs",
    "hourly": ["temperature_2m", "precipitation"],
    "daily": [],
    "monthly": [],
    "time_range": {
        "type": "forecast",
        "start_hour": "2025-05-15T00:00",   # 起报时间（世界时）
        "forecast_hours": 48,               # 预报时效（小时数）
    },
    "area": {
        "type": "point",
        "adcodes": adcodes,
    },
    "timezone": "Asia/Shanghai",
    "area_average": False,
}

# 先预估，确认当前用户限额并获取有效期 10 分钟的 estimate_id
estimate_res = requests.post(f"{BASE_URL}/api/tasks/estimate", json=payload, headers=headers)
estimate_res.raise_for_status()
estimate_data = estimate_res.json()["data"]
if estimate_data.get("error") or not estimate_data.get("estimate_id"):
    raise RuntimeError(estimate_data.get("error_msg") or "预估失败")
payload["estimate_id"] = estimate_data["estimate_id"]

res = requests.post(f"{BASE_URL}/api/tasks/export", json=payload, headers=headers)
res.raise_for_status()
task_id = res.json()["data"]["task_id"]
print(f"任务已创建，task_id: {task_id}")

---

方式二：使用自定义点位

适合已有一批经纬度坐标的场景，id 和 name 为可选字段，用于结果关联和标识。

第一步：准备自定义点位列表

import time
import zipfile
import requests

API_KEY = "your_api_key"
BASE_URL = "https://api.mirror-earth.com"

locations = [
    {"id": "1", "name": "北京", "lon": 116.4, "lat": 39.9},
    {"id": "2", "name": "上海", "lon": 121.5, "lat": 31.2},
    {"id": "3", "name": "广州", "lon": 113.3, "lat": 23.1},
]

print(f"共 {len(locations)} 个自定义点位")

第二步：创建批量下载任务

headers = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json",
}

payload = {
    "domain": "archive_ifs",
    "hourly": ["temperature_2m", "precipitation"],
    "daily": [],
    "monthly": [],
    "time_range": {
        "type": "forecast",
        "start_hour": "2025-05-15T00:00",
        "forecast_hours": 48,
    },
    "area": {
        "type": "point",
        "locations": locations,
    },
    "timezone": "Asia/Shanghai",
    "area_average": False,
}

# 先预估，确认当前用户限额并获取有效期 10 分钟的 estimate_id
estimate_res = requests.post(f"{BASE_URL}/api/tasks/estimate", json=payload, headers=headers)
estimate_res.raise_for_status()
estimate_data = estimate_res.json()["data"]
if estimate_data.get("error") or not estimate_data.get("estimate_id"):
    raise RuntimeError(estimate_data.get("error_msg") or "预估失败")
payload["estimate_id"] = estimate_data["estimate_id"]

res = requests.post(f"{BASE_URL}/api/tasks/export", json=payload, headers=headers)
res.raise_for_status()
task_id = res.json()["data"]["task_id"]
print(f"任务已创建，task_id: {task_id}")

---

方式三：使用区域平均

将行政区划内所有格点的数值取空间平均，输出为一行一个区划的时序数据，适合区域级预报精度评估。仅支持以 adcodes 指定范围。

第一步：创建批量下载任务

import time
import zipfile
import requests

API_KEY = "your_api_key"
BASE_URL = "https://api.mirror-earth.com"

# 以各省为例（可从 province.csv 中获取全部 adcode）
adcodes = ["110000", "310000", "440000"]  # 北京、上海、广东

headers = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json",
}

payload = {
    "domain": "archive_ifs",
    "hourly": ["temperature_2m", "precipitation"],
    "daily": [],
    "monthly": [],
    "time_range": {
        "type": "forecast",
        "start_hour": "2025-05-15T00:00",
        "forecast_hours": 48,
    },
    "area": {
        "type": "point",
        "adcodes": adcodes,
    },
    "timezone": "Asia/Shanghai",
    "area_average": True,   # 开启区域平均
}

# 先预估，确认当前用户限额并获取有效期 10 分钟的 estimate_id
estimate_res = requests.post(f"{BASE_URL}/api/tasks/estimate", json=payload, headers=headers)
estimate_res.raise_for_status()
estimate_data = estimate_res.json()["data"]
if estimate_data.get("error") or not estimate_data.get("estimate_id"):
    raise RuntimeError(estimate_data.get("error_msg") or "预估失败")
payload["estimate_id"] = estimate_data["estimate_id"]

res = requests.post(f"{BASE_URL}/api/tasks/export", json=payload, headers=headers)
res.raise_for_status()
task_id = res.json()["data"]["task_id"]
print(f"任务已创建，task_id: {task_id}")

---

第三步：轮询等待任务完成

以上三种方式创建任务后，均使用相同方式轮询进度：

output_file = None

while True:
    res = requests.get(f"{BASE_URL}/api/tasks/{task_id}/status", headers=headers)
    res.raise_for_status()
    data = res.json()["data"]

    status = data["status"]
    progress = data.get("progress_percent", 0)
    print(f"任务状态：{status}，进度：{progress}%")

    if status == "completed":
        output_file = data["output_file"]
        break
    elif status in ("failed", "cancelled"):
        raise RuntimeError(f"任务异常，状态：{status}")

    time.sleep(10)

---

第四步：下载并解压任务结果

download_url = f"{BASE_URL}/user_downloads/{output_file}"
res = requests.get(download_url, headers=headers, stream=True)
res.raise_for_status()

zip_path = "archive_forecast_result.zip"
with open(zip_path, "wb") as f:
    for chunk in res.iter_content(chunk_size=8192):
        f.write(chunk)

with zipfile.ZipFile(zip_path, "r") as z:
    z.extractall("archive_forecast_result")

print("下载完成，数据已解压到 archive_forecast_result/")

---

切换其他历史预报模型

只需修改 domain 字段即可切换数据源，全部历史预报模型请查看 可用数据模型：