当前位置：首页 > news >正文

AI 农业数据分析：气象数据 + 产量预测的跨界融合案例

news 2026/7/27 1:45:22

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(lower=0).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_splits=5) # 模型1: 随机森林 rf_model = RandomForestRegressor( n_estimators=200, max_depth=15, min_samples_leaf=5, random_state=42 ) rf_scores = cross_val_score(rf_model, X, y, cv=tscv, scoring='neg_mean_absolute_error') rf_mae = -rf_scores.mean() print(f"随机森林 MAE: {rf_mae:.3f} 吨/公顷") # 模型2: XGBoost xgb_model = XGBRegressor( n_estimators=300, max_depth=8, learning_rate=0.05, subsample=0.8, random_state=42 ) xgb_scores = cross_val_score(xgb_model, X, y, cv=tscv, scoring='neg_mean_absolute_error') xgb_mae = -xgb_scores.mean() print(f"XGBoost 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', ascending=False) print("Top10 重要特征:") print(feature_importance_df.head(10).to_string(index=False))

模型结果：XGBoost的MAE为0.35吨/公顷，在水稻平均产量6.5吨/公顷的背景下，误差约5.4%，算是可以接受的精度。

特征重要性排名前三的是：抽穗期总降水、生长季总积温、拔节期日均温度。这跟农学经验完全吻合——抽穗期是水稻最需水的阶段，积温决定生长速度，拔节期温度影响分蘖。

非线性效应的验证

import matplotlib.pyplot as plt # 验证降水对产量的非线性效应（倒U型曲线） # 按抽穗期降水量分组，计算每组平均产量 precip_bins = pd.qcut(features_df['heading_total_precip'], q=10) 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(kind='bar') plt.xlabel('抽穗期降水量分组') plt.ylabel('平均产量（吨/公顷）') plt.title('抽穗期降水量与产量的非线性关系') plt.tight_layout() plt.savefig('images/precip_yield_nonlinear.png', dpi=150)

结果确实呈现倒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(key=lambda x: x['match_score'], reverse=True) return recommendations[0] # 返回最佳品种