企业官网建设流程全解析

在业务中跑供应链气候风险评估时，最大的困难不是“有没有模型”，而是“数据到了模型能不能及时反映当下的异常”。尤其当评估对象是哥伦比亚农业这类强依赖降水和温度的地区时，过去一周的降水异常、持续高温天数、土壤湿度变化，往往比气候平均值更有决策价值。之前我做过一版基于月度气候统计的供应链风险评分，上线后业务反馈“指标太钝”，等月报出来，香蕉产区可能已经被暴雨影响一周了。后来重新设计方案，改用“临近预报（Nowcasting）”思路，结合历史气候窗口、近实时观测和快速推断模型，才算真正把风险评估从复盘变成了预警。

这篇文章会围绕一个完整框架展开，取名可以叫colombia-agri-climate-nowcast，它面向哥伦比亚农业供应链，覆盖数据接入、特征工程、模型训练、风险评分与 API 发布全流程。内容不是纯理论，我会把可直接运行的 Python 代码拆开讲解，并给出常见报错与排查建议。适合三部分读者：一是做供应链风险建模的数据工程师，二是关注农业气象应用的算法工程师，三是需要在业务侧落地实时评分系统的后端开发。

1. 背景与核心概念

1.1 为什么农业供应链需要“实时”气候风险评估

哥伦比亚是全球重要的咖啡、香蕉、鲜花和可可出口国，但这些作物的产区高度依赖特定气候条件。以咖啡为例，开花期降水过多会降低授粉质量，成熟期持续阴雨又会增加病害风险；香蕉种植对风速和累积降水极其敏感，一场飓风或连续三天强降水就可能影响整周出货量。

传统气候风险评估通常依赖月度或季度气候公报。这类数据适合年度产能规划，却无法支撑供应链的动态调度。比如某港口的香蕉装船计划是三天后，如果那时产区将出现强降水，运输道路可能中断，正确的决策应该提前调整装船窗口。这种“未来几小时到几天的风险变化”，就需要实时数据和临近预报来支撑。

供应链气候风险评估的关键不是预测“这个月会不会干旱”，而是回答几个具体问题：

• 未来 1 到 7 天，某产区面临极端降水还是持续高温？
• 当前土壤水分状况对作物关键生育期是否构成压力？
• 风险等级达到什么水平，业务系统需要触发什么级别响应？
• 风险信号能在数据到达后多久进入业务决策流程？

1.2 Nowcasting 与 Forecasting 的区别

“Nowcasting”直译是临近预报或现时预报，通常指对未来 0 到 6 小时、有时扩展到 24 到 72 小时的短临预测。它和传统天气预报（Forecasting）的核心区别在于：

• 天气预报侧重未来较长时间的趋势，比如未来一周、一个月的气温降水趋势。
• Nowcasting 则强调“当前状态的外推”，它利用最新观测数据，比如过去几小时的雷达回波、卫星降水估计、地面自动气象站数据，通过快速算法判断接下来极短时间内天气会如何演变。

在供应链风险评估中，我们不需要精确预报 14 天后的降水毫米数，更需要知道“未来 72 小时该产区是否进入高风险区间”。所以框架中的预测目标不是具体天气值，而是风险等级和风险触发概率。

从数据处理角度看，Nowcasting 会频繁使用滚动窗口特征，例如：

• 过去 3 天累积降水；
• 过去 7 天超过 35℃ 的天数；
• 过去 14 天降水距平百分率；
• 当前土壤湿度指数与历史同期百分位。

这些特征以“最近状态”为核心，而不是以“长期平均”为核心，这正好符合供应链预警场景。

1.3 框架整体设计

我建议把整个实时气候风险评估框架拆成四层：

1. 数据接入层：对接气象观测、卫星降水估计、数值天气预报产品，统一时间分辨率和空间粒度。
2. 特征工程层：把原始气候数据转换成历史窗口特征、累积指标、距平百分位等可建模输入。
3. 模型推理层：使用训练好的机器学习模型，对每个产区、每个未来时间窗输出风险概率。
4. 决策应用层：将风险概率映射为业务等级，通过 API 或消息队列传递给供应链管理系统。

这种分层设计的好处是每一层可以独立替换。比如数据源从 CHIRPS 换成 IMERG，只需要改接入层；模型从 LightGBM 换成时序模型，只需要改推理层；业务规则调整也只影响决策应用层。下面从环境准备开始，逐步搭建这个框架。

2. 环境准备与版本说明

2.1 开发工具链

本文演示环境以常见 Python 数据科学栈为基础，具体版本需要根据你的项目实际情况调整。这里给出我使用的环境，供参考：

• 操作系统：Ubuntu 22.04，Windows / macOS 也可运行，但要留意路径分隔符。
• Python 版本：3.10 或 3.11。
• 包管理：pip 或 poetry。
• 核心依赖：pandas、numpy、scikit-learn、lightgbm、fastapi、uvicorn、requests、pydantic。
• 可选依赖：apache-airflow（用于编排定时任务）、docker（用于部署 API）、prometheus-client（用于监控）。

版本方面不建议盲目追求最新。LightGBM 的 API 在不同大版本间有细微变化，FastAPI 和 Pydantic 的兼容性也需要留意。如果你在安装时遇到依赖冲突，可以用虚拟环境隔离。

2.2 项目结构设计

实际项目中我习惯把代码按功能拆开，而不是堆在一个main.py里。下面是一个适合中小型团队的目录结构：

colombia-agri-climate-nowcast/ ├── README.md ├── requirements.txt ├── config/ │ ├── config.yaml │ └── model_params.json ├── data/ │ ├── raw/ │ ├── processed/ │ └── predictions/ ├── src/ │ ├── __init__.py │ ├── data_ingestion.py │ ├── feature_engineering.py │ ├── model_training.py │ ├── risk_scoring.py │ ├── api_server.py │ └── utils.py ├── notebooks/ │ └── eda_demo.ipynb └── tests/ ├── test_feature_engineering.py └── test_risk_scoring.py

config/放配置参数，src/放核心代码，data/区分原始数据和中间数据，notebooks/用于探索性分析，tests/放单元测试。这样分工明确，后续接 CI/CD 也方便。

2.3 安装依赖

创建一个虚拟环境并安装依赖：

python3 -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install pandas numpy scikit-learn lightgbm fastapi uvicorn requests pyyaml pydantic

如果你想用 Airflow 编排每日训练和实时评分任务，可以额外安装：

pip install apache-airflow

需要注意，Airflow 对 Python 版本有明确限制，安装前建议先查看官方支持的版本组合。若只是本地实验，可以先用简单的schedule或APScheduler替代。

3. 核心原理拆解

3.1 数据接入与时间对齐

实时气候风险评估的数据源通常来自多个渠道，比如地面气象站、卫星降水反演、再分析资料、数值天气预报。这些数据的时间频率和空间网格各不相同。卫星降水产品可能是半小时或逐日数据，地面气象站可能是小时级，数值天气预报可能每 6 小时输出一次未来 10 天预报。

做特征工程之前，必须先把所有序列统一到相同的时空基准。我通常做这几步：

1. 确定评估单元，比如某个省份、某个市镇、某个咖啡种植区。
2. 将网格数据按区域聚合，取区域平均或按种植面积加权平均。
3. 把所有数据重采样到统一时间频率，例如逐日。
4. 对缺失时间窗口做插值或标记，不让模型直接面对空洞序列。

需要注意，“实时”不等于“不等待数据校验”。气象数据本身存在延迟，卫星反演产品通常有几个小时延迟。框架要设计一个“数据新鲜度”字段，超过最大容忍延迟的记录应标记为 stale，而不是直接参与评分。

3.2 特征工程：从原始气候序列到风险特征

在农业风险评估里，单一时间点的降水值意义有限，更有用的是“累积效应”和“距平状态”。下面列出几个典型特征：

对于临近预报任务，还可以加入数值天气预报的前向特征，比如未来 24 小时、48 小时降水预报值。这些特征可以来自公开模型或商业气象服务，具体以你接入的数据源为准。

标准化降水指数 SPI 是一种比较通用的干旱指数。它先把降水序列按指定窗口滑动累积，再做 Gamma 分布拟合，最后映射到标准正态分布。SPI 为负表示偏旱，为正表示偏湿。下面会给出简化实现。

3.3 模型选型：为什么推荐 LightGBM

在历史项目里，农业气候风险任务的样本量通常不大，可能只有几百个产区乘以几年的日数据，而且特征之间往往有复杂交互。LightGBM 这类梯度提升树模型有几个优势：

• 对特征尺度不敏感，不需要严格标准化；
• 能捕捉非线性关系和特征交互；
• 自带缺失值处理逻辑；
• 推理速度较快，适合实时评分。

深度学习模型虽然表达能力更强，但对数据量要求高，训练调试成本也更高。在“快速上线 + 可解释性”要求下，树模型是更稳妥的第一步。

如果你的目标不是分类而是输出风险概率，可以让 LightGBM 输出predict_proba，然后用阈值划分风险等级。本文示例会先构建二分类目标：未来 5 天内是否发生极端降水事件，再映射为风险分数。

3.4 风险等级映射

模型输出概率后，业务系统需要可执行的等级。我常用的映射规则是：

• 低风险：概率小于 0.3，不触发额外动作。
• 中风险：概率 0.3 到 0.6，提醒供应链运营关注。
• 高风险：概率 0.6 到 0.8，启动备选运输路线评估。
• 极高风险：概率大于 0.8，触发应急预案。

这里的阈值并非固定不变。不同作物、不同季节对相同概率的承受能力不同，阈值应当放进配置中心，而不是写死在代码里。

4. 完整实战案例

4.1 创建项目结构

假设我们已经在本地建好目录。先创建项目骨架：

mkdir -p colombia-agri-climate-nowcast/{config,data/{raw,processed,predictions},src,notebooks,tests} cd colombia-agri-climate-nowcast

然后在requirements.txt中写入依赖：

pandas>=2.0 numpy>=1.24 scikit-learn>=1.2 lightgbm>=4.0 fastapi>=0.100 uvicorn>=0.23 requests>=2.31 pyyaml>=6.0 pydantic>=2.0

安装依赖：

pip install -r requirements.txt

4.2 数据接入层实现

为了避免代码依赖无法验证的私有数据源，这里用模拟数据演示核心逻辑。真实场景中，你可以把load_weather_data内部替换为 CHIRPS、ERA5、IMERG 或哥伦比亚官方气象机构 IDEAM 的下载逻辑。

新建文件src/data_ingestion.py：

# 文件路径：src/data_ingestion.py import pandas as pd import numpy as np def generate_demo_weather_data( municipio: str = "armenia", start_date: str = "2020-01-01", end_date: str = "2024-12-31", seed: int = 42, ) -> pd.DataFrame: """ 生成模拟天气数据，用于演示流程。 真实项目中应替换为 CHIRPS / ERA5 / IMERG / IDEAM 数据。 """ rng = np.random.default_rng(seed) date_range = pd.date_range(start=start_date, end=end_date, freq="D") precip = rng.gamma(shape=2.0, scale=3.0, size=len(date_range)) temp_max = 28 + 2.5 * np.sin( np.arange(len(date_range)) / 365.0 * 2 * np.pi ) + rng.normal(0, 1.2, size=len(date_range)) df = pd.DataFrame( { "municipio": municipio, "date": date_range, "precip": np.round(precip, 2), "temp_max": np.round(temp_max, 2), } ) return df if __name__ == "__main__": df = generate_demo_weather_data() print(df.head()) print(f"总记录数: {len(df)}")

这段代码会生成包含每日降水和最高温的模拟序列。gamma分布用来模拟降水，带季节项的正态噪声模拟温度。运行后你应该看到几行数据预览，方便确认列名和日期范围。

4.3 特征工程实现

新建src/feature_engineering.py，核心是滑动窗口特征和 SPI 计算：

# 文件路径：src/feature_engineering.py import pandas as pd import numpy as np from scipy.stats import gamma, norm def compute_spi(precip_series: pd.Series, window: int = 30) -> pd.Series: """ 简化版标准化降水指数 SPI。 对指定窗口的滑动累积降水做 Gamma 拟合，再转换为标准正态分数。 实际工程中可改用更严格的分位数映射或经验 SPI。 """ cum = precip_series.rolling(window, min_periods=window).sum() valid = cum.dropna() if len(valid) < 30: return pd.Series(np.nan, index=precip_series.index) # 使用极大似然估计 Gamma 分布参数 fit_alpha, fit_loc, fit_beta = gamma.fit(valid.values, floc=0) # Gamma CDF 到标准正态分位数 spi_values = norm.ppf(gamma.cdf(valid.values, fit_alpha, loc=fit_loc, scale=fit_beta)) result = pd.Series(spi_values, index=valid.index) return result.reindex(precip_series.index) def build_features(df: pd.DataFrame, temp_threshold: float = 34.0) -> pd.DataFrame: """ 根据原始天气日数据构建模型特征。 """ df = df.sort_values(["municipio", "date"]).copy() # 按地区分组计算滚动特征，避免不同地区混合 grouped = df.groupby("municipio", group_keys=False) df["precip_3d"] = grouped["precip"].transform( lambda x: x.rolling(3, min_periods=1).sum() ) df["precip_7d"] = grouped["precip"].transform( lambda x: x.rolling(7, min_periods=3).sum() ) df["precip_30d"] = grouped["precip"].transform( lambda x: x.rolling(30, min_periods=15).sum() ) df["temp_max_7d_avg"] = grouped["temp_max"].transform( lambda x: x.rolling(7, min_periods=3).mean() ) df["heat_days_30d"] = grouped["temp_max"].transform( lambda x: (x > temp_threshold).rolling(30, min_periods=15).sum() ) df["spi_30"] = grouped["precip"].transform(lambda x: compute_spi(x, 30)) # 添加日期特征，便于模型感知季节性 df["month"] = df["date"].dt.month df["day_of_year"] = df["date"].dt.dayofyear return df

这段代码里面有几个关键点：

• groupby("municipio")保证不同产区的特征不会互相串。
• rolling(min_periods=...)允许冷启动阶段使用不完整窗口，避免前期大量缺失。
• SPI 计算做了简化，适合快速原型，如果要用于严格农业分析，建议用更完善的气象库实现。

4.4 构建训练样本与标签

新建src/model_training.py。核心思路是构建有监督分类任务：根据截至t日的特征，预测未来 5 天内是否出现极端降水，比如日降水大于 40mm。

# 文件路径：src/model_training.py import pandas as pd import numpy as np from sklearn.model_selection import TimeSeriesSplit from sklearn.metrics import roc_auc_score import lightgbm as lgb def add_label(df: pd.DataFrame, precip_col: str = "precip", extreme_threshold: float = 40.0, horizon: int = 5) -> pd.DataFrame: """ 构建未来 horizon 天内是否发生极端降水事件的标签。 """ df = df.sort_values(["municipio", "date"]).copy() df["future_extreme"] = ( df.groupby("municipio")[precip_col] .transform(lambda x: x.shift(-horizon).rolling(horizon, min_periods=1).max()) ) df["label"] = (df["future_extreme"] >= extreme_threshold).astype(int) # 删除最后 horizon 天没有未来标签的样本 df = df.dropna(subset=["future_extreme"]).reset_index(drop=True) return df def train_model(df: pd.DataFrame, feature_cols: list[str]): """ 使用 TimeSeriesSplit 训练 LightGBM 二分类模型，并输出 AUC。 """ model = lgb.LGBMClassifier( n_estimators=300, learning_rate=0.05, num_leaves=31, max_depth=6, subsample=0.8, colsample_bytree=0.8, random_state=42, ) X = df[feature_cols] y = df["label"] tscv = TimeSeriesSplit(n_splits=3) auc_scores = [] for train_idx, valid_idx in tscv.split(X): X_train, X_valid = X.iloc[train_idx], X.iloc[valid_idx] y_train, y_valid = y.iloc[train_idx], y.iloc[valid_idx] model.fit( X_train, y_train, eval_set=[(X_valid, y_valid)], eval_metric="auc", ) pred = model.predict_proba(X_valid)[:, 1] auc_scores.append(roc_auc_score(y_valid, pred)) print(f"TimeSeriesSplit AUC: {np.mean(auc_scores):.4f} (+/- {np.std(auc_scores):.4f})") return model

这里用TimeSeriesSplit而不是普通 KFold，原因是天气数据存在强时间自相关，随机切分会造成未来信息泄漏，让验证结果虚高。这是很多新手容易踩的坑。

具体的特征列，在主流程中这样组织：

feature_cols = [ "precip_3d", "precip_7d", "precip_30d", "temp_max_7d_avg", "heat_days_30d", "spi_30", "month", "day_of_year", ]

4.5 实时评分与风险分级

训练好的模型需要封装成可复用的评分模块。新建src/risk_scoring.py：

# 文件路径：src/risk_scoring.py import pandas as pd import numpy as np RISK_LEVELS = [ (0.3, "low"), (0.6, "medium"), (0.8, "high"), (1.01, "critical"), ] def score_risk(model, feature_df: pd.DataFrame) -> pd.DataFrame: """ 输入特征表，输出风险概率和风险等级。 """ proba = model.predict_proba(feature_df)[:, 1] levels = [] for p in proba: for threshold, level in RISK_LEVELS: if p < threshold: levels.append(level) break result = feature_df.copy() result["risk_prob"] = proba result["risk_level"] = levels return result

为了便于快速演示，还可以写一个简单的批量评分入口：

def score_recent(model, df: pd.DataFrame, feature_cols: list[str]): latest = df.groupby("municipio").tail(1).copy() return score_risk(model, latest[feature_cols])

4.6 发布实时评分 API

为了方便业务系统调用，我们用 FastAPI 暴露一个接口。新建src/api_server.py：

# 文件路径：src/api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import joblib import pandas as pd from risk_scoring import score_risk app = FastAPI(title="Colombia Agriculture Climate Risk API") MODEL_PATH = "../models/lgb_model.joblib" FEATURE_COLS = [ "precip_3d", "precip_7d", "precip_30d", "temp_max_7d_avg", "heat_days_30d", "spi_30", "month", "day_of_year", ] model = joblib.load(MODEL_PATH) class RiskRequest(BaseModel): municipio: str precip_3d: float precip_7d: float precip_30d: float temp_max_7d_avg: float heat_days_30d: int spi_30: float month: int day_of_year: int @app.post("/risk") def get_risk(req: RiskRequest): input_dict = req.dict() df = pd.DataFrame([input_dict]) try: result = score_risk(model, df) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) return { "municipio": req.municipio, "risk_prob": float(result["risk_prob"].iloc[0]), "risk_level": result["risk_level"].iloc[0], } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

启动 API：

cd src uvicorn api_server:app --host 0.0.0.0 --port 8000

调用示例：

curl -X POST "http://localhost:8000/risk" \ -H "Content-Type: application/json" \ -d '{ "municipio": "armenia", "precip_3d": 45.2, "precip_7d": 96.8, "precip_30d": 210.5, "temp_max_7d_avg": 27.8, "heat_days_30d": 4, "spi_30": 1.2, "month": 4, "day_of_year": 112 }'

预期会返回类似：

{ "municipio": "armenia", "risk_prob": 0.67, "risk_level": "high" }

实际概率值取决于训练数据和模型，这里只是演示输出格式。

4.7 主流程串联

为了方便本地验证，在src/main.py中把数据生成、特征构建、训练、保存模型串联起来：

# 文件路径：src/main.py import joblib from data_ingestion import generate_demo_weather_data from feature_engineering import build_features from model_training import add_label, train_model FEATURE_COLS = [ "precip_3d", "precip_7d", "precip_30d", "temp_max_7d_avg", "heat_days_30d", "spi_30", "month", "day_of_year", ] def main(): print("Step 1: 加载天气数据") df = generate_demo_weather_data() print("Step 2: 构建特征") df_feat = build_features(df) print("Step 3: 添加标签") df_label = add_label(df_feat) print("Step 4: 训练模型") model = train_model(df_label, FEATURE_COLS) print("Step 5: 保存模型") joblib.dump(model, "../models/lgb_model.joblib") print("模型已保存到 ../models/lgb_model.joblib") if __name__ == "__main__": main()

运行：

mkdir -p models python src/main.py

你会看到类似这样的输出，数据量示意：

Step 1: 加载天气数据 Step 2: 构建特征 Step 3: 添加标签 Step 4: 训练模型 TimeSeriesSplit AUC: 0.7342 (+/- 0.0215) Step 5: 保存模型 模型已保存到 ../models/lgb_model.joblib

由于这里用的是模拟数据，AUC 只用来检验流程是否跑通，不能说明真实业务效果。

5. 常见问题与排查思路

实时气候风险评估项目里，跑通代码只是第一步，下面这些问题几乎每个团队都会遇到。

5.1 SPI 计算异常

Gamma 分布拟合要求数据非负，并且至少要有一定数量的有效样本。如果某个月降水全部为 0，gamma.fit可能给出极端参数。我的建议是：在实际代码中先过滤全零序列，或者给降水加上非常小的平滑值1e-3，避免数值崩溃。

5.2 时间泄漏问题

时间泄漏是气候建模中最隐蔽的问题。举例来说，如果滚动特征时错误地使用了未来数据，比如用shift(-1)做滚动均值，模型在训练集上的表现会异常好，但上线后立刻失效。排查方法有两个：

1. 打印特征工程前后的时间顺序，确认没有负的shift。
2. 在验证集上手动检查一两个样本，保证特征值只依赖历史日期。

5.3 数据源切换导致字段不一致

项目早期用 CHIRPS 降水产品，后来切换到 IMERG，可能出现字段名、时间单位、缺失值编码不一致。我建议在数据接入层统一返回标准 schema，下游只依赖标准字段。常用标准字段如下：

5.4 模型漂移

气候本身存在年际变化，模型训练时的数据分布可能和当前年份不一致。即使模型上线时表现不错，半年后也可能出现评分偏移。建议在预测表里保存模型版本号和数据源版本号，方便回溯。

6. 最佳实践与工程建议

6.1 数据质量优先于模型复杂度

实时气候风险评估过程中，业务方问得最多的不是“你用了什么模型”，而是“为什么这个产区突然变成高风险”。如果数据源本身有延迟或错误，那么再复杂的模型也会输出不可信结果。

工程上建议做三件事：

• 记录每个样本的数据时间戳和到达时间戳，计算数据延迟。
• 对原始数据做异常值检测，例如降水为负、温度超过物理上限等。
• 保存原始数据快照，这样后续重算特征时可以追溯。

6.2 特征与模型版本管理

模型的复现性比模型精度更重要。每天训练出的模型如果无法复现，风险评分就失去审计价值。建议把config/model_params.json和requirements.txt一起归档，并在模型文件中写入训练日期、训练数据范围和特征列表。下面是一个简单配置示例：

{ "model_name": "lgbm_extreme_precip_classifier", "train_date": "2025-01-15", "train_start": "2020-01-01", "train_end": "2024-12-31", "target": "extreme_precip_5d", "threshold": 40.0, "features": [ "precip_3d", "precip_7d", "precip_30d", "temp_max_7d_avg", "heat_days_30d", "spi_30", "month", "day_of_year" ] }

6.3 部署与灰度发布

实时评分 API 上线时，不要直接全量切流。建议先跑 shadow 模式，让新模型和旧模型同时接收请求，但只返回旧模型的分数，对比一段时间后再切换。这样可以提前发现特征分布差异、数据源延迟等问题。

另外，FastAPI 服务可以放入 Docker 容器中部署，环境一致性更有保障。Dockerfile 的一个参考写法：

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY src/ ./src/ COPY models/ ./models/ EXPOSE 8000 CMD ["uvicorn", "src.api_server:app", "--host", "0.0.0.0", "--port", "8000"]

注意，这里的模型必须在容器构建前已经存在，或者通过挂载卷方式提供。

6.4 安全与权限

涉及生产环境的数据接入和生产模型发布时，必须强调最小权限原则。气象数据源通常需要 API Key，不要把密钥写死在代码或 Dockerfile 中，建议使用环境变量或专门的密钥管理服务。同样，数据库只读账号、模型存储的只读权限，都应遵循最小授权。

农业供应链的风险评估结果会直接影响发货计划，因此上线前必须做充分测试。建议在测试环境用历史回放方式验证，比如取过去某个下雨事件前的数据，看模型是否能提前给出高风险信号。

6.5 性能优化

实时评分场景通常需要处理多个产区。如果产区数量在几百个量级，模型推理压力不大；但如果要做网格级评估，特征量会增长很多。这时可以从两个方向优化：

1. 特征计算：把滑动窗口计算转成numba加速，或使用 PySpark 做批处理。
2. 模型推理：使用模型序列化格式如 ONNX，或者把模型打包到轻量级服务中，降低每次推理开销。

对于当前这个框架，先用 LightGBM 原生接口和 pandas 完成原型，性能已经足够支撑原型验证。

7. 总结与后续学习路线

这篇文章的核心不是提供一个“万能模型”，而是一套可落地的实时气候风险评估框架。我们从业务背景出发，解释了 Nowcasting 与 Forecasting 的差异，然后完成了数据接入、特征工程、模型训练、风险分级、API 发布的完整链路。你可以在本地运行这套以模拟数据为基础的工程原型，再替换为真实气象数据源继续迭代。

下一步可以从几个方向深化：

• 换用真实数据源，比如 CHIRPS 降水和 ERA5 温度数据，跑通完整回测流程。
• 扩展多分类目标，比如同时预测干旱、暴雨、高温三类风险。
• 引入供应链网络数据，把产区风险传导到运输节点和港口，形成端到端影响分析。
• 增加模型可解释性分析，用 SHAP 解释高风险产区的主要驱动因子，便于业务人员理解。
• 把训练和评分流程放入 Airflow，实现每日自动更新模型和定时评分。

在真实项目中，最需要优先关注的不是模型精度提升，而是数据延迟、特征时间一致性、模型版本管理和灰度发布，这些工程问题往往决定系统是否值得被业务信任。希望这套框架能帮你少踩一些坑，也欢迎在实际落地中按自己的业务逻辑做裁切。如果这篇文章对你有帮助，可以收藏备用，后续再围绕真实数据接入、模型解释和供应链传导建模做进一步分享。