数据挖掘管道搭建：流程比模型更值钱，以电力AI赛题为例

简介：面向电力AI赛题的数据挖掘管道搭建完整示例，涵盖数据清洗、特征抽取、样本划分、模型训练与预测等关键环节，适合大数据、人工智能等专业学生在毕业设计、课程设计或项目初期立项中参考，也可作为入门到进阶的实操范本。压缩包共32个文件，包括22个Jupyter Notebook分阶段演示数据预处理与建模流程，4个Python脚本承担数据清洗、特征抽取、天气及节假日数据爬取等复用任务，另含3个CSV数据集和详细说明文档，整体仅3.72MB，便于快速下载复现，目录结构清晰，便于分步对照学习。已有45人学习，项目代码经测试运行成功，答辩评审达95分。除了管道搭建源码，还提供特征组合、规则分析、结果输出等配套脚本，并配有完整文档帮助理解整体思路；读者可在现有基础上修改扩展，用于课程设计、作业或多种数据挖掘场景的二次开发。

1. 基于大航杯“智造扬中”电力AI大赛的数据挖掘管道：值钱的不是模型，而是流程

如果你参加过电力AI类的数据挖掘赛，多半经历过这种尴尬：初赛用单文件脚本拼分数，复赛赛题稍微变化，脚本越改越乱，最后连自己都说不清预测文件是怎么跑出来的。这个基于大航杯“智造扬中”电力AI大赛的数据挖掘管道搭建示例，核心不在某个高精度模型，而在把数据从原始表变成提交结果的完整流程——读数据、清洗、特征构造、训练、验证、输出，每一步都固化成可复现的模块。管道化的价值在赛题变化时最明显：换数据集、换目标列、换评估方式，只改配置和少量注册函数就能继续跑。它适合正在备赛的团队，也适合被临时脚本逼疯的从业者。

2. 先定评估再搭管道：把数据挖掘流程拆成五层和一套配置

赛题数据到手之后，大多数人的第一反应是跑一个baseline看分数，然后一头扎进特征和调参。这么做不是不行，但缺少一个前置动作：想清楚验证方式。电力AI赛题往往只给一份训练集和一份测试集，提交后没有实时反馈，本地验证的可靠性决定你后面所有改动的方向对不对。管道设计的第一步不是写代码，而是把赛题的数据结构画出来——时间跨度多长、采样粒度多大、目标列在什么范围、有没有明显的周期性。这些信息直接决定切分策略和特征构造顺序。

2.1 先问三个问题：预测什么、按什么切、用什么指标

第一个问题：预测目标是连续值、类别还是事件序列。电力赛题常见的任务有三类，负荷与电量的回归预测、设备故障的二分类、设备状态的多分类。目标类型决定模型损失函数和评估指标，也决定特征构造的方向。第二个问题：训练集和测试集在时间上是什么关系。很多电力赛题按时间先后切分，训练集是前几个月，测试集是后几周，中间可能还隔着一段没有数据的空窗。如果随机打散样本做交叉验证，短期相关性会把验证分数推得很高，一到测试集就露馅。第三个问题：赛方用什么指标排名。有的用RMSE，有的用F1，有的用自定义得分，指标直接决定了要在验证集上优化什么。

这三个问题答完，管道的第一层约束就出来了：验证方式必须贴近赛题评估方式。我一般会先把这一条写在项目README最上面，防止后面调参调上头把验证给忘了。临时脚本最容易在这里翻车——写的时候觉得“反正只是试试”，等试了几十次之后，验证逻辑已经和业务目标脱节，线上分数一掉，回头看代码都是一团浆糊。

2.2 五层分工：数据、特征、模型、验证、输出

把管道拆成五层不是为了好看，而是为了让每一层都可以单独修改和调试。数据层负责读文件、统一列名、按时间排序，但不在这里做清洗，因为清洗策略往往要按业务场景反复调整。特征层只接受DataFrame、返回DataFrame，不访问文件系统，不读写外部状态，这样所有特征函数都变成纯函数，好测试也好回溯。模型层只负责训练和预测，不接触原始表，所有输入都是特征层产出的矩阵。验证层负责切分和指标计算，不修改任何数据。输出层把预测结果转成赛题要求的提交格式。

每层之间的数据流是单向的，上层不反向读取下层的数据结构。这样设计最直接的好处是排错快：分数不对，先看验证层有没有切对；预测文件格式不对，只看输出层；特征数量对不上，只看特征层。我曾经见过一个赛题项目，特征处理和模型训练在同一个文件里，中间用十几个全局变量传递数据，每次改动都要把所有代码通读一遍，这就是没分层的代价。分层之后，每一层都能独立替换，LightGBM换成神经网络只动模型层，加特征只动特征层，其他代码一行不用改。

2.3 配置驱动：把路径、参数、特征开关全部浮出水面

分层解决的是代码组织问题，配置驱动解决的是实验管理问题。常见做法是把所有可变参数写进一个yaml文件，代码里只读配置，不写死任何路径和超参。配置与代码分离之后，每次实验只需要复制一份配置，改几个字段，跑完把配置和分数一起存档，后面想复现哪个结果都有后悔药。

# configs/default.yaml data: train_path: data/train.csv test_path: data/test.csv id_col: task_id time_col: record_time target: load_value feat: window_sizes: [6, 12, 24, 48] lag_steps: [1, 3, 6, 12] missing_strategy: ffill_bfill keep_missing_flag: true train: model_type: lgb params: learning_rate: 0.05 num_leaves: 63 min_child_samples: 20 n_estimators: 500 subsample: 0.8 colsample_bytree: 0.8 random_state: 42 eval: tscv_folds: 5 gap_hours: 24 metric: rmse

这份配置里需要解释几个字段。time_col是时间列，特征构造和时序切分都依赖它。window_sizes是滚动窗口长度，单位按数据采样粒度来定，如果数据是小时级的，6、12、24、48分别对应6小时、半天、一天、两天。gap_hours是切分时训练集和验证集之间留出的空窗，模拟赛题里训练集与测试集之间无数据的时间段。metric是验证指标，rmse对应回归任务的均方根误差。配置加载时要做一层校验，关键分组缺失就直接报错，不要等到运行到一半才因为某个键不存在而崩溃。

3. 搭一套最小可运行管道：注册式特征工程与时序验证的骨架代码

这一章直接给一套能跑的骨架。代码量不大，但把上一章说的分层和配置驱动全部落进去。你拿到手之后，把数据路径换成自己的赛题数据，改一下目标列名，就能跑通第一条基线。之后再往里面加特征、换模型，都是在骨架上做增量。

3.1 项目组织与最小主流程：先让链路跑通再谈精度

目录结构按层拆分，保持简单。

pipeline_demo/ ├── configs/ │ └── default.yaml ├── src/ │ ├── pipeline.py │ ├── features.py │ └── evaluate.py ├── data/ │ ├── train.csv │ └── test.csv └── output/

主流程文件只做编排，不写具体实现。

# src/pipeline.py # 数据挖掘管道主流程：读配置 -> 加载数据 -> 构造特征 -> 时序验证 -> 训练输出 import sys import yaml import pandas as pd sys.path.append("src") from features import FeatureRegistry from evaluate import tscv_evaluate, train_lgb, make_submission def load_config(path="configs/default.yaml"): with open(path, "r", encoding="utf-8") as f: cfg = yaml.safe_load(f) required_keys = {"data", "feat", "train", "eval"} missing = required_keys - cfg.keys() assert not missing, f"配置文件缺少分组: {missing}" return cfg def load_data(cfg): train = pd.read_csv(cfg["data"]["train_path"], parse_dates=[cfg["data"]["time_col"]]) test = pd.read_csv(cfg["data"]["test_path"], parse_dates=[cfg["data"]["time_col"]]) for df in (train, test): df.sort_values(cfg["data"]["time_col"], inplace=True) df.reset_index(drop=True, inplace=True) return train, test def main(): cfg = load_config() train, test = load_data(cfg) registry = FeatureRegistry(cfg) train_feat = registry.transform(train) test_feat = registry.transform(test) fold_scores = tscv_evaluate(train_feat, cfg) print("fold scores:", fold_scores) model = train_lgb(train_feat, train_feat[cfg["data"]["target"]], cfg["train"]) make_submission(model, test_feat, cfg) if __name__ == "__main__": main()

主流程只有三十行左右，每一段干一件明确的事。load_config读取配置并校验必备分组，load_data负责读文件、解析时间列、按时间排序，这两个函数完成后，train和test的列结构就是一致的。FeatureRegistry是特征工程入口，下面会讲。tscv_evaluate做时序交叉验证，返回每一折的分数，方便观察稳定性。最后用全量训练集重新训练模型，对测试集预测并写出提交文件。

注意两个细节。一是load_data里重置索引，因为后续滚动特征和滞后特征都依赖行顺序，索引起伏会导致shift错位。二是训练和预测走同一个registry.transform，这是“训练预测同源”的基础，后面预测阶段不会出现特征列对不上的问题。

3.2 特征工程注册器：新增特征不需要改主流程

特征工程是数据挖掘赛里改动最频繁的部分。如果每个新特征都要在主流程里加一行调用，主流程会越来越臃肿。用注册器模式，新特征写好之后挂一个装饰器，主流程自动加载。

# src/features.py # 注册式特征工程：新特征函数加 @register_feature 装饰器即自动生效 import pandas as pd import numpy as np FEATURE_FUNCS = [] def register_feature(func): FEATURE_FUNCS.append(func) return func @register_feature def rolling_stats(df, cfg): """滚动统计特征：窗口越大越平滑，窗口越小越敏感""" t = cfg["data"]["target"] df = df.copy() df[f"{t}_is_missing"] = df[t].isna().astype(int) series = df[t].ffill().bfill() for w in cfg["feat"]["window_sizes"]: df[f"rmean_{w}"] = series.rolling(w, min_periods=1).mean() df[f"rstd_{w}"] = series.rolling(w, min_periods=1).std() return df @register_feature def lag_features(df, cfg): """滞后特征：把历史几步的目标值搬到当前行""" t = cfg["data"]["target"] df = df.copy() series = df[t].ffill().bfill() for s in cfg["feat"]["lag_steps"]: df[f"lag_{s}"] = series.shift(s) return df class FeatureRegistry: def __init__(self, cfg): self.cfg = cfg self.train_cols = None def transform(self, df): for fn in FEATURE_FUNCS: df = fn(df, self.cfg) if self.train_cols is None: self.train_cols = list(df.columns) else: assert list(df.columns) == self.train_cols, \ f"特征列与训练时不一致，缺少: {set(self.train_cols) - set(df.columns)}" return df

这个注册器的设计要点有三个。一是FEATURE_FUNCS列表在模块加载时自动收集所有带装饰器的函数，主流程不需要知道具体有哪些特征。二是每个特征函数只接受df和cfg，返回df，不读文件、不写全局变量，保持纯函数特性，调试时可以单独拿一个函数跑。三是transform里做了训练与预测的列一致性校验，第一次对训练集调用时记录列清单，之后对测试集或验证集调用时比对列是否一致，不一致立刻报错，防止静默错位。

实际使用时，新特征函数写好后加一个装饰器就能接入管道。rolloing_stats里先计算is_missing列，再对目标列做前后向填充，最后算滚动均值和滚动标准差。这里先填充再滚动是因为NaN传播会让滚动窗口断档，窗口前面一大片都变成NaN，浪费时间且没意义。lag_features同理，滞后特征在填充后的序列上shift，避免缺失值在滞后列里扩散。

3.3 时序交叉验证与模型训练：验证方式决定可信度

模型训练和验证都放在evaluate.py里，和特征层解耦。

# src/evaluate.py # 时序交叉验证 + LightGBM 训练 + 提交文件生成 import numpy as np import pandas as pd import lightgbm as lgb from sklearn.model_selection import TimeSeriesSplit from sklearn.metrics import mean_squared_error def train_lgb(X_tr, y_tr, tr_cfg): model = lgb.LGBMRegressor(**tr_cfg["params"]) model.fit(X_tr, y_tr) return model def tscv_evaluate(feat_df, cfg): t = cfg["data"]["target"] drop_cols = [cfg["data"]["id_col"], cfg["data"]["time_col"], t] X = feat_df.drop(columns=[c for c in drop_cols if c in feat_df.columns]) y = feat_df[t] tscv = TimeSeriesSplit(n_splits=cfg["eval"]["tscv_folds"]) scores = [] for fold, (tr_idx, va_idx) in enumerate(tscv.split(X)): X_tr, X_va = X.iloc[tr_idx], X.iloc[va_idx] y_tr, y_va = y.iloc[tr_idx], y.iloc[va_idx] model = train_lgb(X_tr, y_tr, cfg["train"]) pred = model.predict(X_va) rmse = mean_squared_error(y_va, pred, squared=False) scores.append(rmse) print(f"fold {fold}: rmse={rmse:.4f}") return scores def make_submission(model, feat_df, cfg): drop_cols = [cfg["data"]["time_col"]] if cfg["data"]["target"] in feat_df.columns: drop_cols.append(cfg["data"]["target"]) X_test = feat_df.drop(columns=drop_cols, errors="ignore") feat_df = feat_df.copy() feat_df["pred"] = model.predict(X_test) out = feat_df[[cfg["data"]["id_col"], "pred"]] out.to_csv("output/submission.csv", index=False)

train_lgb把LightGBM的初始化参数从配置传入，**tr_cfg["params"]解包之后，所有超参都由yaml控制。tscv_evaluate用sklearn的TimeSeriesSplit做前向切分，每一折都用更早的数据训练、更晚的数据验证，和赛题的时间顺序一致。mean_squared_error的squared=False返回RMSE，误差单位与目标列一致，更直观。

make_submission里有一个错误处理需要解释：测试集如果带了目标列就删掉，避免预测阶段把标签当特征喂进去；带errors="ignore"是为了兼容某些测试集本身没有目标列的情况。预测完成后只保留id列和pred列，输出为submission.csv，这是大多数赛题的标准提交格式。如果赛题要求的是分类概率或特定列名，在这个函数里改就行。

4. 电力数据的三个专属难点：时序切分、缺失语义与不平衡评估

通用管道骨架跑通之后，真正决定电力赛题名次的是对数据本身的理解。电力数据有三个性质，任何从通用表格数据集迁移过来的方案都会在这三个地方踩坑。第一个是强时序性，第二个是缺失值的业务含义，第三个是少数类样本的评估方式。

4.1 样本不能乱切：按业务时间切分并留出预测空窗

电力负荷、设备温度、电压电流这类序列，相邻时间点的数值高度相关。今天12点的负荷和昨天12点的负荷有关系，和今天11点的负荷关系更紧密。如果用随机K-Fold切分，训练集和验证集会互相混入前后时刻的数据，模型等于提前看到了验证集的信息，验证分数虚高一截，提交后立刻掉分。

我一般会用TimeSeriesSplit加一个手动空窗。TimeSeriesSplit本身只保证训练集索引全部小于验证集索引，但赛题场景里训练集结束到测试集开始往往隔着几天甚至几周，验证时要模拟这个空窗。

def tscv_with_gap(feat_df, cfg): t = cfg["data"]["target"] drop_cols = [cfg["data"]["id_col"], cfg["data"]["time_col"], t] X = feat_df.drop(columns=[c for c in drop_cols if c in feat_df.columns]) y = feat_df[t] tscv = TimeSeriesSplit(n_splits=cfg["eval"]["tscv_folds"]) gap_samples = cfg["eval"]["gap_hours"] # 这里假设数据按小时采样，gap_hours 直接换算成样本数 scores = [] for tr_idx, va_idx in tscv.split(X): if gap_samples > 0: tr_idx = tr_idx[: len(tr_idx) - gap_samples] X_tr, X_va = X.iloc[tr_idx], X.iloc[va_idx] y_tr, y_va = y.iloc[tr_idx], y.iloc[va_idx] model = train_lgb(X_tr, y_tr, cfg["train"]) pred = model.predict(X_va) rmse = mean_squared_error(y_va, pred, squared=False) scores.append(rmse) return scores

这段代码在TimeSeriesSplit的基础上，把每一折训练集末尾的gap_samples个样本切掉，使训练结束时间比验证开始时间早至少gap_samples步。如果数据是小时级的，gap_hours设为24就代表中间空了一天。如果采样间隔不是1小时，gap_hours要除以采样频率换算成样本数。

切分方式的选择可以参考这张表。

4.2 缺失值不是噪声，是业务事件

电力数据里出现缺失太常见了，但缺失的原因完全不一样。可能是采集设备瞬时故障，可能是线路检修停运，也可能是数据传输丢包。这三种情况对业务含义完全不同：检修停运本身就是一种状态变化，如果直接把缺失填成0或者用均值填，模型会丢掉“这一段没数据”这个强信号。

我习惯把“怎么填”和“填没填”分开处理。填充策略保证后续特征计算不中断，is_missing标志列把缺失事件本身保留下来，让模型自己决定这个信号有没有用。

# 参数化的缺失值处理策略 def apply_missing_strategy(df, col, strategy="ffill_bfill"): flag = df[col].isna().astype(int) if strategy == "ffill_bfill": df[col] = df[col].ffill().bfill() elif strategy == "zero": df[col] = df[col].fillna(0) elif strategy == "median": df[col] = df[col].fillna(df[col].median()) else: raise ValueError(f"未知缺失值策略: {strategy}") df[f"{col}_is_missing"] = flag return df

这段代码关键在两点。前向填充和反向填充组合会让缺失段用最近的正常值补上，滚动特征和滞后特征不会断档；而is_missing列单独记录了缺失的位置，模型可以学出“这段缺失意味着停运”之类的规律。更进一步，可以构造“近6小时是否发生过缺失”这类窗口聚合特征，把缺失的时间持续性也编码进去。

还有一个细节容易被忽略：测试集也可能有缺失。训练预测同源在这里同样适用，测试集的特征构造必须走同一套缺失策略，不能在测试阶段才发现训练时没处理过的NaN。

4.3 少数类才是业务重点：评估指标要加权

电力故障类任务中，故障样本占比可能只有百分之几甚至更低。这时候用准确率评估毫无意义，全预测正常也能拿到很高的准确率。赛题如果给的是故障二分类，我一般看F1和PR-AUC；如果是多分类设备状态，看macro F1，让每个类别等权重参与评分。

选对指标之后还要做后处理。二分类模型默认阈值是0.5，但故障样本极少时，0.5往往不是最优阈值。常见做法是在验证集上按F1扫描阈值，找到最佳截断点，提交时用这个阈值做最终分类。

def search_best_threshold(y_true, y_proba): best_th, best_f1 = 0.5, 0.0 for th in np.arange(0.1, 0.9, 0.05): pred = (y_proba >= th).astype(int) f1 = f1_score(y_true, pred) if f1 > best_f1: best_th, best_f1 = th, f1 return best_th, best_f1

这个扫描逻辑不复杂，但价值很高。很多团队把模型跑完就直接用0.5截断，阈值扫描往往能带来比换模型更明显的提升。

5. 管道搭建常见问题排查：五个容易翻车的细节与解决方案

管道跑通只是开始，真正耗时间的是各种莫名其妙的现象。下面五条是我在类似赛题里反复见过的坑，每一条都按现象、原因、解决来排查。

5.1 随机切分评分虚高，提交后分数断崖

现象：本地用K-Fold验证，RMSE很低，模型表现稳定，提交到测试集后分数比验证差一大截。原因：随机切分把未来信息混进了训练集，电力负荷序列的短期自相关让模型在验证集上偷看了答案。解决：改成TimeSeriesSplit，并且在训练集和验证集之间留出与赛题一致的空窗。改完之后验证分数会变差，但那个差分数才是可信的。

5.2 同一套代码重跑，结果对不上

现象：昨天跑出的预测文件和今天重跑的同配置结果完全不一样，换了数据源时更明显。原因：特征函数里用了全局变量，或者DataFrame列顺序发生了波动，又或者某一步inplace操作移位了索引。解决：特征函数统一改成纯函数，靠配置和入参驱动；FeatureRegistry里加上训练与预测的列一致性校验；每次运行前打印特征数量和列名前五列，方便对比。

5.3 缺失值填零后，模型分数不升反降

现象：把缺失值全部填充为0，训练更稳定了，但验证分数和线上分数都掉。原因：电力数据里的缺失往往有业务含义，填零等于告诉模型“这个时刻数据正常”，模型学到了错误的状态信息。解决：改用ffill加bfill组合填充，同时加is_missing标志列，保留缺失事件本身。更进一步可以在配置里把missing_strategy做成可选项，跑一组对比实验看哪个策略更适合当前数据。

5.4 测试集分布和训练集差异巨大

现象：验证分数稳定，但提交后分数崩掉，查看测试集发现目标列分布、均值、缺失率都和训练集差一大截。原因：赛题故意按不同时间段或不同工况切分，模型在训练集上学到的分布不能直接泛化。解决：在管道里加一步数据体检，对比训练集和测试集的分组均值、缺失率、时间覆盖范围，差异超过阈值就输出警告。这时候优先做鲁棒性处理，比如删除对分布敏感的特征、分位数归一化、或者用更保守的正则化参数。

5.5 改了配置参数，模型分数完全不变

现象：调整yaml里的learning_rate、num_leaves，跑出来的分数一模一样，怀疑代码没生效。原因：配置文件里字段名写错了，train层加载时没有校验，代码用了默认参数。解决：在load_config里增加深层校验，遍历train.params的键是否都在LightGBM合法参数集合里，不合法直接报错并提示字段名。我后来习惯每次改配置都在输出文件名里带上配置版本号，跑完看到文件名就知道哪份配置生效了。

6. 把管道升级成工具箱：实验留痕、AutoML接入与训练预测同源

管道骨架能跑通之后，下一步是让它长成你自己的工具箱。这一章讲三个我用着最顺手的升级动作，都不复杂，但对长期做数据挖掘的人来说价值很大。

第一个动作是实验留痕。每次跑完管道，把配置快照、特征列清单、每一折分数、预测文件一起归档到带时间戳的目录下。做这个动作之后，再也不存在“昨晚那个好分数是怎么跑出来的”这种问题。

import os, json, datetime exp_dir = f"output/exp_{datetime.datetime.now():%Y%m%d_%H%M%S}" os.makedirs(exp_dir, exist_ok=True) with open(f"{exp_dir}/cfg.yaml", "w", encoding="utf-8") as f: yaml.dump(cfg, f) with open(f"{exp_dir}/feature_cols.json", "w", encoding="utf-8") as f: json.dump(train_feat.columns.tolist(), f, indent=2) with open(f"{exp_dir}/scores.json", "w", encoding="utf-8") as f: json.dump({"scores": fold_scores, "mean": float(np.mean(fold_scores))}, f, indent=2)

这段代码放在主流程末尾，每次运行自动生成一个实验目录。后面对比实验只需要比较不同实验目录里的scores.json和cfg.yaml，哪个改动带来了提升一目了然。

第二个动作是给AutoML留接口。超参搜索不要写进主流程，把特征矩阵固化之后单独做。

# 调参脚本：特征矩阵固定，只优化超参 # train_feat 由管道产出，存成 parquet 后在这里独立加载 X = train_feat.drop(columns=["task_id", "record_time", "load_value"]) y = train_feat["load_value"] # 用 optuna 定义超参搜索空间，每次 trial 训练一个 LightGBM # 关键点：搜索过程复用同一份特征矩阵，不重复做特征工程

为什么要单独做？因为特征工程和超参搜索是两种不同的迭代节奏，混在一起会让每次实验的变量太多，根本分不清是特征起了作用还是参数起了作用。特征矩阵固定之后，调参实验每次只跑模型层，速度快且归属清晰。

第三个动作是坚持训练预测同源。预测阶段永远不手动构造特征，不走“上次存过一份中间结果直接拿来用”的捷径。哪怕预测时多花二十分钟重新构造特征，也绝不复用某个中间csv——因为一旦特征函数有改动，旧的时间文件就会造成错位，而且这种错位极难排查。我自己就因为图省事直接读取中间结果吃过亏，后来立了个规矩：训练和预测必须走同一个registry对象，预测前自动做列一致性校验，过不了就停下重跑。

管道这个方向值不值得投入时间，我的答案是值得。一套配置驱动、分层清晰、可留痕的管道，前期搭建成本大约一两天，但换来的是一次实验可以清清楚楚复盘、特征迭代不破坏旧结果、队伍协作不再靠临时拷贝文件。我后来做任何数据挖掘赛题，第一周总是先搭管道而不是先调模型，这个习惯帮我省掉的时间远比搭管道花掉的多。希望帮到你。

本文还有配套的精品资源，点击获取