在业务中跑供应链气候风险评估时最大的困难不是“有没有模型”而是“数据到了模型能不能及时反映当下的异常”。尤其当评估对象是哥伦比亚农业这类强依赖降水和温度的地区时过去一周的降水异常、持续高温天数、土壤湿度变化往往比气候平均值更有决策价值。之前我做过一版基于月度气候统计的供应链风险评分上线后业务反馈“指标太钝”等月报出来香蕉产区可能已经被暴雨影响一周了。后来重新设计方案改用“临近预报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 框架整体设计我建议把整个实时气候风险评估框架拆成四层数据接入层对接气象观测、卫星降水估计、数值天气预报产品统一时间分辨率和空间粒度。特征工程层把原始气候数据转换成历史窗口特征、累积指标、距平百分位等可建模输入。模型推理层使用训练好的机器学习模型对每个产区、每个未来时间窗输出风险概率。决策应用层将风险概率映射为业务等级通过 API 或消息队列传递给供应链管理系统。这种分层设计的好处是每一层可以独立替换。比如数据源从 CHIRPS 换成 IMERG只需要改接入层模型从 LightGBM 换成时序模型只需要改推理层业务规则调整也只影响决策应用层。下面从环境准备开始逐步搭建这个框架。2. 环境准备与版本说明2.1 开发工具链本文演示环境以常见 Python 数据科学栈为基础具体版本需要根据你的项目实际情况调整。这里给出我使用的环境供参考操作系统Ubuntu 22.04Windows / 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.pyconfig/放配置参数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 天预报。做特征工程之前必须先把所有序列统一到相同的时空基准。我通常做这几步确定评估单元比如某个省份、某个市镇、某个咖啡种植区。将网格数据按区域聚合取区域平均或按种植面积加权平均。把所有数据重采样到统一时间频率例如逐日。对缺失时间窗口做插值或标记不让模型直接面对空洞序列。需要注意“实时”不等于“不等待数据校验”。气象数据本身存在延迟卫星反演产品通常有几个小时延迟。框架要设计一个“数据新鲜度”字段超过最大容忍延迟的记录应标记为 stale而不是直接参与评分。3.2 特征工程从原始气候序列到风险特征在农业风险评估里单一时间点的降水值意义有限更有用的是“累积效应”和“距平状态”。下面列出几个典型特征特征名称计算方式业务含义precip_3d过去 3 天累积降水短期湿涝压力precip_30d过去 30 天累积降水土壤水分累积temp_max_7d过去 7 天最高温均值高温热害强度heat_days_30d过去 30 天最高温超过阈值的天数高温持续影响precip_percentile当前累积降水在历史同期分布的百分位异常程度spi_3030 天标准化降水指数干旱或湿润异常对于临近预报任务还可以加入数值天气预报的前向特征比如未来 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中写入依赖pandas2.0 numpy1.24 scikit-learn1.2 lightgbm4.0 fastapi0.100 uvicorn0.23 requests2.31 pyyaml6.0 pydantic2.0安装依赖pip install -r requirements.txt4.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(startstart_date, endend_date, freqD) precip rng.gamma(shape2.0, scale3.0, sizelen(date_range)) temp_max 28 2.5 * np.sin( np.arange(len(date_range)) / 365.0 * 2 * np.pi ) rng.normal(0, 1.2, sizelen(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_periodswindow).sum() valid cum.dropna() if len(valid) 30: return pd.Series(np.nan, indexprecip_series.index) # 使用极大似然估计 Gamma 分布参数 fit_alpha, fit_loc, fit_beta gamma.fit(valid.values, floc0) # Gamma CDF 到标准正态分位数 spi_values norm.ppf(gamma.cdf(valid.values, fit_alpha, locfit_loc, scalefit_beta)) result pd.Series(spi_values, indexvalid.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_keysFalse) df[precip_3d] grouped[precip].transform( lambda x: x.rolling(3, min_periods1).sum() ) df[precip_7d] grouped[precip].transform( lambda x: x.rolling(7, min_periods3).sum() ) df[precip_30d] grouped[precip].transform( lambda x: x.rolling(30, min_periods15).sum() ) df[temp_max_7d_avg] grouped[temp_max].transform( lambda x: x.rolling(7, min_periods3).mean() ) df[heat_days_30d] grouped[temp_max].transform( lambda x: (x temp_threshold).rolling(30, min_periods15).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_periods1).max()) ) df[label] (df[future_extreme] extreme_threshold).astype(int) # 删除最后 horizon 天没有未来标签的样本 df df.dropna(subset[future_extreme]).reset_index(dropTrue) return df def train_model(df: pd.DataFrame, feature_cols: list[str]): 使用 TimeSeriesSplit 训练 LightGBM 二分类模型并输出 AUC。 model lgb.LGBMClassifier( n_estimators300, learning_rate0.05, num_leaves31, max_depth6, subsample0.8, colsample_bytree0.8, random_state42, ) X df[feature_cols] y df[label] tscv TimeSeriesSplit(n_splits3) 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_metricauc, ) pred model.predict_proba(X_valid)[:, 1] auc_scores.append(roc_auc_score(y_valid, pred)) print(fTimeSeriesSplit 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(titleColombia 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_code500, detailstr(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, host0.0.0.0, port8000)启动 APIcd 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. 常见问题与排查思路实时气候风险评估项目里跑通代码只是第一步下面这些问题几乎每个团队都会遇到。问题现象常见原因解决思路特征列大量为 NaN滑动窗口 min_periods 设置过高冷启动阶段数据不足检查最小窗口设置或对缺失值做前向填充模型 AUC 虚高但线上效果差使用普通 KFold 随机切分造成时间泄漏改用 TimeSeriesSplit并保证特征只用 t 时刻之前的数据SPI 计算报错降水序列存在大量 0 值Gamma 分布拟合不稳定加入平滑项或改用经验分位数 SPIAPI 响应慢每次请求都重新加载模型或重复构建特征模型启动时加载一次特征计算放批处理任务数据源延迟导致预测过期没有监控数据新鲜度在数据接入层增加 timestamp 字段设置 freshness 阈值预测周期不一致训练时用日粒度评分时传入小时粒度统一重采样为日粒度并校验日期范围5.1 SPI 计算异常Gamma 分布拟合要求数据非负并且至少要有一定数量的有效样本。如果某个月降水全部为 0gamma.fit可能给出极端参数。我的建议是在实际代码中先过滤全零序列或者给降水加上非常小的平滑值1e-3避免数值崩溃。5.2 时间泄漏问题时间泄漏是气候建模中最隐蔽的问题。举例来说如果滚动特征时错误地使用了未来数据比如用shift(-1)做滚动均值模型在训练集上的表现会异常好但上线后立刻失效。排查方法有两个打印特征工程前后的时间顺序确认没有负的shift。在验证集上手动检查一两个样本保证特征值只依赖历史日期。5.3 数据源切换导致字段不一致项目早期用 CHIRPS 降水产品后来切换到 IMERG可能出现字段名、时间单位、缺失值编码不一致。我建议在数据接入层统一返回标准 schema下游只依赖标准字段。常用标准字段如下标准字段含义潜在单位问题municipio评估区域编码需统一行政区划编码date日期注意 UTC 与本地时区precip日降水mm/day 还是 kg/m²temp_max日最高温摄氏度还是华氏度data_source数据源名称用于数据血缘追溯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 性能优化实时评分场景通常需要处理多个产区。如果产区数量在几百个量级模型推理压力不大但如果要做网格级评估特征量会增长很多。这时可以从两个方向优化特征计算把滑动窗口计算转成numba加速或使用 PySpark 做批处理。模型推理使用模型序列化格式如 ONNX或者把模型打包到轻量级服务中降低每次推理开销。对于当前这个框架先用 LightGBM 原生接口和 pandas 完成原型性能已经足够支撑原型验证。7. 总结与后续学习路线这篇文章的核心不是提供一个“万能模型”而是一套可落地的实时气候风险评估框架。我们从业务背景出发解释了 Nowcasting 与 Forecasting 的差异然后完成了数据接入、特征工程、模型训练、风险分级、API 发布的完整链路。你可以在本地运行这套以模拟数据为基础的工程原型再替换为真实气象数据源继续迭代。下一步可以从几个方向深化换用真实数据源比如 CHIRPS 降水和 ERA5 温度数据跑通完整回测流程。扩展多分类目标比如同时预测干旱、暴雨、高温三类风险。引入供应链网络数据把产区风险传导到运输节点和港口形成端到端影响分析。增加模型可解释性分析用 SHAP 解释高风险产区的主要驱动因子便于业务人员理解。把训练和评分流程放入 Airflow实现每日自动更新模型和定时评分。在真实项目中最需要优先关注的不是模型精度提升而是数据延迟、特征时间一致性、模型版本管理和灰度发布这些工程问题往往决定系统是否值得被业务信任。希望这套框架能帮你少踩一些坑也欢迎在实际落地中按自己的业务逻辑做裁切。如果这篇文章对你有帮助可以收藏备用后续再围绕真实数据接入、模型解释和供应链传导建模做进一步分享。