Datawhale AI春训营学习笔记

数据竞赛Baseline代码全解析：从数据加载到结果输出

一、环境配置与数据加载

1.1 依赖库导入

from netCDF4 import Dataset  # 处理气象.nc格式数据
import numpy as np
import pandas as pd
from sklearn.model_selection import KFold
from sklearn.metrics import mean_squared_error
import lightgbm as lgb
from tqdm import tqdm  # 进度条工具

1.2 气象数据读取

def get_data(path_template, date):"""气象数据读取核心函数Args:path_template: 文件路径模板（需包含{}占位符）date: 日期字符串Returns:pd.DataFrame: 处理后的气象数据"""dataset = Dataset(path_template.format(date), mode='r')channel = dataset.variables["channel"][:]  # 获取特征名称data = dataset.variables["data"][:]        # 原始数据矩阵# 空间维度聚合（11x11网格取均值）mean_values = np.array([np.mean(data[:, :, i, :, :][0], axis=(1, 2)) for i in range(8)]).Treturn pd.DataFrame(mean_values, columns=channel)

数据维度处理示意图：

原始数据维度：(1, 24h, 8通道, 11x11网格)
处理过程：沿空间维度取平均
输出维度：(24h, 8特征)

二、数据预处理关键代码

2.1 时间对齐处理

# 原始功率数据读取
target = pd.read_csv("fact_data/1_normalization_train.csv")# 每15分钟数据保留整点数据（每小时第1条）
target = target[96:]  # 首日数据对齐
target = target[target['时间'].str.endswith('00:00')]

数据频率说明表：

数据类型	原始频率	处理后频率
气象数据	每小时	保持不变
功率数据	15分钟	每小时

2.2 缺失值处理

# 简单缺失值处理
train = train.dropna().reset_index(drop=True)# 高级处理建议：
# 1. 时间序列插值（线性/样条插值）
# 2. 特征重要性筛选后处理

三、特征工程实现

3.1 基础特征扩展

def feature_combine(df):"""特征衍生函数包含：- 向量风速计算- 时间周期特征- 物理量比值特征"""df_copy = df.copy()# 向量风速计算df_copy["wind_speed"] = np.sqrt(df['u100']**2 + df['v100']**2)# 时间周期特征df_copy["hour"] = df_copy.index % 24df_copy["hour_sin"] = np.sin(2 * np.pi * df_copy["hour"] / 24)df_copy["hour_cos"] = np.cos(2 * np.pi * df_copy["hour"] / 24)# 物理量交互特征df_copy["ghi_poai_ratio"] = df["ghi"] / (df["poai"] + 1e-7)df_copy["tcc_t2m_interact"] = df["tcc"] * df["t2m"]return df_copy

3.2 特征重要性分析

# 训练完成后获取特征重要性
feature_importance = pd.DataFrame({'feature': cols,'importance': model.feature_importance()
}).sort_values('importance', ascending=False)plt.figure(figsize=(12,6))
sns.barplot(x='importance', y='feature', data=feature_importance[:15])
plt.title('Top 15 Feature Importance')

四、模型训练核心代码

4.1 LightGBM参数配置

lgb_params = {'boosting_type': 'gbdt',       # 梯度提升树'objective': 'regression',     # 回归任务'metric': 'rmse',              # 评估指标'num_leaves': 256,             # 最大叶子数'learning_rate': 0.05,         # 学习率'feature_fraction': 0.8,       # 特征采样比例'bagging_freq': 4,             # 每4次迭代执行bagging'lambda_l2': 10,               # L2正则化'min_data_in_leaf': 20         # 叶子节点最小样本数
}

4.2 交叉验证实现

def cv_model(clf, train_x, train_y, test_x, seed=2024):"""5折交叉验证框架Args:clf: 模型对象train_x: 训练特征train_y: 训练标签test_x: 测试特征Returns:tuple: (oof预测值, 测试集预测均值)"""folds = 5kf = KFold(n_splits=folds, shuffle=True, random_state=seed)oof = np.zeros(train_x.shape[0])test_predict = np.zeros(test_x.shape[0])for i, (train_idx, valid_idx) in enumerate(kf.split(train_x)):# 数据划分trn_x, trn_y = train_x.iloc[train_idx], train_y[train_idx]val_x, val_y = train_x.iloc[valid_idx], train_y[valid_idx]# 模型训练train_set = lgb.Dataset(trn_x, label=trn_y)model = lgb.train(params, train_set, valid_sets=[train_set])# 预测与结果聚合oof[valid_idx] = model.predict(val_x)test_predict += model.predict(test_x) / foldsreturn oof, test_predict

训练过程监控：

Training until validation scores don't improve for 100 rounds
[100]	training's rmse: 0.245632	valid_1's rmse: 0.267891
[200]	training's rmse: 0.198743	valid_1's rmse: 0.253214
Early stopping, best iteration is: 258

五、结果后处理与提交

5.1 预测结果处理

# 结果重复4次对齐15分钟粒度
lgb_test = [item for item in lgb_test for _ in range(4)]# 应用滑动平均滤波
window_size = 4
output['power'] = output['power'].rolling(window=window_size, min_periods=1).mean()# 物理约束处理
output['power'] = output['power'].clip(lower=0)  # 功率非负

5.2 提交文件生成

output.to_csv('submission.csv', index=False)

文件格式示例：

timestamp,power
2024-01-01 00:00:00,12.34
2024-01-01 00:15:00,12.56
...

六、代码优化建议

6.1 内存优化技巧

# 使用分类数据类型
dtype_mapping = {'ghi': 'float32','poai': 'float32','hour': 'int8'
}
train = train.astype(dtype_mapping)# 分批读取数据
chunksize = 100000
data_chunks = pd.read_csv('big_data.csv', chunksize=chunksize)

6.2 并行计算加速

# LightGBM并行设置
params.update({'num_threads': 8,      # 使用8线程'device_type': 'gpu'   # GPU加速（需编译GPU版本）
})# 交叉验证并行化
from joblib import Parallel, delayedresults = Parallel(n_jobs=4)(delayed(train_model)(fold) for fold in kf.split(X)
)

---

最佳实践建议：

1. 使用版本控制管理不同实验（Git）
2. 添加详细的日志记录
3. 使用MLflow等工具进行实验追踪
4. 对关键代码模块进行单元测试

完整项目代码结构示例：

├── data/
├── notebooks/        # Jupyter实验记录
├── src/
│   ├── features/     # 特征工程
│   ├── models/       # 模型代码
│   └── utils/        # 工具函数
└── requirements.txt  # 依赖列表