AI 农业数据分析气象数据 产量预测的跨界融合案例一、农业数据分析的跨界吸引力做数据分析久了容易陷入电商GMV、广告CTR、用户LTV的舒适区。但去年一个机会让我接触到了农业数据分析——帮一个智慧农业项目做产量预测模型这才发现农业数据的复杂度和趣味性远超互联网业务。互联网数据大多是人造数据用户行为、交易记录而农业数据是天造数据——天气、土壤、病虫害这些变量你控制不了只能预测和适应。这种不确定性让建模挑战更大但也让结果更有价值预测准确了农民能提前调整种植策略减少损失、增加收益。项目的核心目标是融合气象数据温度、降水、日照和土壤数据湿度、pH值、有机质预测某区域主要农作物的产量并为种植决策提供优化建议。二、气象数据的采集与特征工程气象数据是产量预测最重要的外生变量。我们用的是县级气象站的日级观测数据涵盖温度、降水、日照时数、相对湿度等指标。多源气象数据的融合气象数据来源多、格式杂国家气象局的历史数据是CSV格式、实时数据是API接口、卫星遥感数据是GeoTIFF格式。融合的第一步是统一到空间网格上。import pandas as pd import numpy as np # 加载县级气象站日级观测数据 station_weather pd.read_csv(weather_station_daily.csv) # 数据包含: station_id, date, temp_max, temp_min, precipitation, sunshine_hours, humidity # 加载卫星遥感降水数据网格化空间分辨率0.25° satellite_precip pd.read_csv(satellite_precipitation_grid.csv) # 数据包含: grid_lat, grid_lon, date, precip_estimate # 将站点数据映射到县级网格 # 每个县对应一个0.25°网格单元 county_grid_mapping pd.read_csv(county_grid_mapping.csv) # 包含: county_id, grid_lat, grid_lon # 融合站点观测与卫星估计数据 # 策略: 有站点观测的县用站点数据没有的用卫星估计数据 def fuse_weather_data(county_id, date): 融合站点与卫星气象数据 # 查找该县是否有气象站观测 station_data station_weather[ (station_weather[county_id] county_id) (station_weather[date] date) ] if len(station_data) 0: return station_data.iloc[0] # 有站点数据优先使用 else: # 没有站点数据用卫星估计值 grid_info county_grid_mapping[county_grid_mapping[county_id] county_id] sat_data satellite_precip[ (satellite_precip[grid_lat] grid_info[grid_lat].values[0]) (satellite_precip[grid_lon] grid_info[grid_lon].values[0]) (satellite_precip[date] date) ] return sat_data.iloc[0] # 批量融合所有县所有日期的气象数据 counties county_grid_mapping[county_id].unique() dates pd.date_range(2025-01-01, 2025-12-31) fused_weather [] for county in counties: for date in dates: row fuse_weather_data(county, date.strftime(%Y-%m-%d)) fused_weather.append(row) fused_weather_df pd.DataFrame(fused_weather) print(f融合后气象数据: {len(fused_weather_df)} 条记录)农业气象特征工程原始气象指标日温度、日降水太细粒度了需要聚合到与作物生长周期对应的特征窗口def create_agrometeorological_features(weather_df, crop_growth_stages): 创建农业气象特征按作物生长阶段聚合 参数: weather_df: 日级气象数据 crop_growth_stages: 作物生长阶段定义播种期、苗期、拔节期、抽穗期、成熟期 features {} # 按生长阶段聚合关键气象指标 for stage_name, (start_month, end_month) in crop_growth_stages.items(): stage_data weather_df[ (weather_df[month] start_month) (weather_df[month] end_month) ] # 温度特征 features[f{stage_name}_avg_temp] stage_data[temp_avg].mean() features[f{stage_name}_max_temp] stage_data[temp_max].max() features[f{stage_name}_min_temp] stage_data[temp_min].min() features[f{stage_name}_temp_range] ( stage_data[temp_max].max() - stage_data[temp_min].min() ) # 积温作物生长的累计温度指标 features[f{stage_name}_accumulated_temp] ( (stage_data[temp_avg] - 10).clip(lower0).sum() # 有效积温日均温-10℃以上部分累计 ) # 降水特征 features[f{stage_name}_total_precip] stage_data[precipitation].sum() features[f{stage_name}_precip_days] (stage_data[precipitation] 0).sum() features[f{stage_name}_dry_days] (stage_data[precipitation] 1).sum() # 日照特征 features[f{stage_name}_total_sunshine] stage_data[sunshine_hours].sum() features[f{stage_name}_avg_sunshine] stage_data[sunshine_hours].mean() # 跨阶段衍生特征 # 生长季总积温 features[growing_season_accumulated_temp] sum( features[f{s}_accumulated_temp] for s in crop_growth_stages.keys() ) # 抽穗期-苗期温差温差大有利于某些作物品质提升 features[heading_seedling_temp_diff] ( features.get(heading_avg_temp, 0) - features.get(seedling_avg_temp, 0) ) return features # 定义水稻生长阶段华北地区 rice_growth_stages { sowing: (4, 5), # 播种期: 4-5月 seedling: (5, 6), # 苗期: 5-6月 jointing: (6, 7), # 拔节期: 6-7月 heading: (7, 8), # 抽穗期: 7-8月 maturity: (8, 9), # 成熟期: 8-9月 } # 批量创建所有县的特征 all_features [] for county in counties: county_weather fused_weather_df[fused_weather_df[county_id] county] feat create_agrometeorological_features(county_weather, rice_growth_stages) feat[county_id] county all_features.append(feat) features_df pd.DataFrame(all_features) print(f农业气象特征矩阵: {features_df.shape})三、产量预测模型的构建与评估产量预测的核心挑战是气象条件对产量的影响不是线性的——降水太少当然不好但降水太多也不好涝灾温度同理积温不够产量低但极端高温也会减产。随机森林 XGBoost 双模型对比from sklearn.ensemble import RandomForestRegressor from xgboost import XGBRegressor from sklearn.model_selection import cross_val_score, TimeSeriesSplit from sklearn.metrics import mean_absolute_error, r2_score # 准备训练数据 X features_df.drop(columns[county_id]) y pd.read_csv(historical_yield.csv)[yield_ton_per_hectare] # 时间序列交叉验证不能用随机split因为年份有顺序性 tscv TimeSeriesSplit(n_splits5) # 模型1: 随机森林 rf_model RandomForestRegressor( n_estimators200, max_depth15, min_samples_leaf5, random_state42 ) rf_scores cross_val_score(rf_model, X, y, cvtscv, scoringneg_mean_absolute_error) rf_mae -rf_scores.mean() print(f随机森林 MAE: {rf_mae:.3f} 吨/公顷) # 模型2: XGBoost xgb_model XGBRegressor( n_estimators300, max_depth8, learning_rate0.05, subsample0.8, random_state42 ) xgb_scores cross_val_score(xgb_model, X, y, cvtscv, scoringneg_mean_absolute_error) xgb_mae -xgb_scores.mean() print(fXGBoost MAE: {xgb_mae:.3f} 吨/公顷) # 选择表现更好的模型 best_model xgb_model if xgb_mae rf_mae else rf_model best_model.fit(X, y) # 特征重要性分析 importances best_model.feature_importances_ feature_importance_df pd.DataFrame({ feature: X.columns, importance: importances }).sort_values(importance, ascendingFalse) print(Top10 重要特征:) print(feature_importance_df.head(10).to_string(indexFalse))模型结果XGBoost的MAE为0.35吨/公顷在水稻平均产量6.5吨/公顷的背景下误差约5.4%算是可以接受的精度。特征重要性排名前三的是抽穗期总降水、生长季总积温、拔节期日均温度。这跟农学经验完全吻合——抽穗期是水稻最需水的阶段积温决定生长速度拔节期温度影响分蘖。非线性效应的验证import matplotlib.pyplot as plt # 验证降水对产量的非线性效应倒U型曲线 # 按抽穗期降水量分组计算每组平均产量 precip_bins pd.qcut(features_df[heading_total_precip], q10) yield_by_precip pd.DataFrame({ precip_bin: precip_bins, yield: y }).groupby(precip_bin)[yield].mean() plt.figure(figsize(8, 5)) yield_by_precip.plot(kindbar) plt.xlabel(抽穗期降水量分组) plt.ylabel(平均产量吨/公顷) plt.title(抽穗期降水量与产量的非线性关系) plt.tight_layout() plt.savefig(images/precip_yield_nonlinear.png, dpi150)结果确实呈现倒U型降水量在适中范围内产量最高过多或过少都降低产量。线性模型会错过这个关键非线性效应。四、种植决策优化从预测到行动预测产量只是第一步更关键的是基于预测给出种植优化建议——如果调整播种日期或改种品种产量能提升多少反事实模拟假设性情景分析def simulate_yield_under_scenario(model, base_features, scenario_changes): 反事实模拟在假设气象条件下预测产量变化 参数: model: 已训练的产量预测模型 base_features: 当前气象特征 scenario_changes: 假设性变化如温度2℃、降水-20% modified_features base_features.copy() for feature, change in scenario_changes.items(): if change[type] absolute: modified_features[feature] base_features[feature] change[value] elif change[type] relative: modified_features[feature] base_features[feature] * change[factor] # 预测新情景下的产量 base_yield model.predict([base_features])[0] modified_yield model.predict([modified_features])[0] yield_change modified_yield - base_yield return { base_yield: base_yield, modified_yield: modified_yield, yield_change: yield_change, change_pct: yield_change / base_yield * 100 } # 模拟情景抽穗期降水增加30%模拟灌溉增强的效果 scenario { heading_total_precip: {type: relative, factor: 1.3}, heading_avg_temp: {type: absolute, value: 1.5}, # 同时温度上升1.5℃ } results [] for idx, row in features_df.iterrows(): base_feat row.drop(county_id).values.reshape(1, -1) result simulate_yield_under_scenario(best_model, base_feat.flatten(), scenario) result[county_id] row[county_id] results.append(result) scenario_df pd.DataFrame(results) print(f灌溉增强温度上升情景下:) print(f 平均产量变化: {scenario_df[change_pct].mean():.1f}%) print(f 受益县数: {(scenario_df[change_pct] 0).sum()}) print(f 受损县数: {(scenario_df[change_pct] 0).sum()})种植品种推荐不同品种对气象条件的适应性不同。我们可以用模型为每个县推荐最适合当前气象预测的品种# 不同水稻品种对积温和降水的要求不同 variety_params { 早稻品种A: {min_accumulated_temp: 2800, optimal_precip_heading: 120, growth_days: 110}, 中稻品种B: {min_accumulated_temp: 3200, optimal_precip_heading: 150, growth_days: 130}, 晚稻品种C: {min_accumulated_temp: 3500, optimal_precip_heading: 180, growth_days: 150}, } def recommend_variety(county_features, variety_params): 为每个县推荐最适合的品种 参数: county_features: 该县的气象特征 variety_params: 各品种的适应性参数 recommendations [] for variety, params in variety_params.items(): # 计算品种与气象条件的匹配度 temp_match min(county_features[growing_season_accumulated_temp] / params[min_accumulated_temp], 1.2) precip_match min( county_features[heading_total_precip] / params[optimal_precip_heading], 1.5 ) # 降水偏离最优值的惩罚 precip_penalty abs(precip_match - 1.0) * 0.5 # 综合匹配度评分 match_score temp_match * 0.6 (1 - precip_penalty) * 0.4 recommendations.append({ variety: variety, match_score: match_score, temp_match: temp_match, precip_match: precip_match }) # 按匹配度排序推荐最佳品种 recommendations.sort(keylambda x: x[match_score], reverseTrue) return recommendations[0] # 返回最佳品种五、总结AI农业数据分析这个跨界项目给了我一堆意料之外的收获农业数据的不确定性远超互联网数据天气你控制不了病虫害也控制不了模型必须学会在不确定性下给出有参考价值的预测而非精确数字非线性效应是农业建模的核心降水过多和过少都减产温度过高和过低都减产线性模型完全不适合农业场景生长阶段特征比日级特征更有预测力把气象数据按作物生长阶段聚合比用全年平均更有意义——抽穗期降水比全年降水重要得多反事实模拟比预测更有决策价值告诉农民今年预计产量6.2吨/公顷不如告诉他如果在抽穗期加强灌溉产量可以提升8%农学经验是特征工程的宝藏积温、有效降水、日照时数这些农学概念远比原始温度、降水量更有预测力跨界做数据分析最难的不是技术而是理解另一个领域的语言——把积温分蘖抽穗这些农学概念翻译成特征工程这个翻译过程才是核心挑战。