功率预测数据接入

数据流程

训练集使用历史预报，而不是 ERA5 再分析。再分析数据包含观测同化后的"事后真值"，用它训练的模型上线后会遇到真实预报误差，精度通常明显下降。

变量选择

风电

轮毂高度不是 100 m 时，选择有对应高度层的模型：

光伏

tilt 为组件倾角（0–90°），azimuth 为组件方位（0 为正南，-90 为正东，90 为正西）。例如格尔木 35° 正南安装：

日内与超短期预测可叠加葵花卫星辐射（10 分钟）作为实况输入。

电价与负荷

电价和负荷模型通常按省或地市建模，使用区域平均得到面平均气象序列：

构建训练集

以两个场站、2026 年一季度、每天 00 和 12 UTC 两个起报场为例，导出 72 小时、15 分钟步长的历史预报。

第一步：创建任务

import time
import requests

API_KEY = "your_api_key"
BASE_URL = "https://api.mirror-earth.com"
headers = {"X-API-Key": API_KEY, "Content-Type": "application/json"}

payload = {
    "domain": "archive_hres",
    "hourly": [
        "wind_speed_100m", "wind_direction_100m", "wind_gusts_10m",
        "temperature_2m", "surface_pressure", "relative_humidity_2m",
        "shortwave_radiation", "direct_normal_irradiance", "diffuse_radiation",
        "cloud_cover_low",
    ],
    "daily": [],
    "monthly": [],
    "time_range": {
        "type": "forecast",
        "start": "2026-01-01",
        "end": "2026-03-31",
        "start_hour": "00,12",
        "forecast_hours": 72,
    },
    "area": {
        "type": "point",
        "locations": [
            {"id": "WF01", "name": "张北风场", "lon": 114.70, "lat": 41.15},
            {"id": "PV01", "name": "格尔木光伏", "lon": 94.90, "lat": 36.40},
        ],
    },
    "time_resolution": "minutely_15",
    "timezone": "UTC",
    "area_average": False,
}

estimate = requests.post(f"{BASE_URL}/api/tasks/estimate", json=payload, headers=headers)
estimate.raise_for_status()
data = estimate.json()["data"]
if data.get("error") or not data.get("estimate_id"):
    raise RuntimeError(data.get("error_msg") or "预估失败")
payload["estimate_id"] = 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"]

timezone 设为 UTC，便于下一步与 init_time 计算预报时效。场站较多时，locations 可直接由场站台账 CSV 生成。

第二步：等待并下载

import io
import zipfile

while True:
    res = requests.get(f"{BASE_URL}/api/tasks/{task_id}/status", headers=headers)
    res.raise_for_status()
    status = res.json()["data"]
    if status["status"] == "completed":
        output_file = status["output_file"]
        break
    if status["status"] in ("failed", "cancelled"):
        raise RuntimeError(status["status"])
    time.sleep(10)

zip_bytes = requests.get(f"{BASE_URL}/user_downloads/{output_file}", headers=headers).content
zipfile.ZipFile(io.BytesIO(zip_bytes)).extractall("archive_hres_2026q1")

第三步：按申报时刻切片

CSV 列为 id,name,lat,lon,init_time,time,<variables>。日前申报只能使用申报截止前已入库的起报场，训练集也应只保留这些起报场与对应时效：

import glob
import pandas as pd

df = pd.concat(pd.read_csv(f) for f in glob.glob("archive_hres_2026q1/**/*.csv", recursive=True))
df["init_time"] = pd.to_datetime(df["init_time"], utc=True)
df["time"] = pd.to_datetime(df["time"], utc=True)

# 日前申报使用 D-2 日 12 UTC 起报场，运行日为北京时间 D 日 00:00–24:00
day_ahead = df[df["init_time"].dt.hour == 12].copy()
day_ahead["run_date_bj"] = (day_ahead["time"] + pd.Timedelta(hours=8)).dt.date
day_ahead["lead_h"] = (day_ahead["time"] - day_ahead["init_time"]).dt.total_seconds() / 3600
day_ahead = day_ahead[day_ahead["lead_h"].between(28, 51.75)]  # 96 个点

起报时次怎么选，见起报时次与日前申报。

上线推理

上线时调用同一模型的实时预报，变量和步长与训练集一致：

多模型集成时，对每个模型分别构建训练集，再按 time 对齐合并特征。

相关文档

• 批量下载历史预报数据
• 历史预报数据 API
• 逐小时变量目录
• 数据更新状态