案例 9：城市用电序列上的一维卷积与感受野#
学习目标与记号#
本案例把一维卷积核沿序列移动并进行局部加权求和的方法，用于预测城市下一小时的用电量。主要讨论以下内容：
把按时间排列的多个观测变量整理为
(批量, 通道, 时间)的张量；解释核大小、层数和空洞率如何共同决定感受野；
用严格的时间顺序划分避免未来信息泄漏；
比较持续性基线、普通一维卷积和空洞一维卷积；
使用平均绝对误差（mean absolute error，MAE）、均方根误差（root mean squared error，RMSE）以及相对于基线的改进幅度解释预测结果，而不是只看训练损失。
记批量大小、输入通道数和序列长度分别为 \(B\)、\(C\) 和 \(T\)，第 \(l\) 层卷积核大小和空洞率分别记为 \(k_l\) 和 \(r_l\)。本案例只用于教学，不应直接用于真实电网调度。
数据来源与任务边界#
数据来自 UCI Power Consumption of Tetouan City，包含 2017 年每 10 分钟的气象量、辐射量及三个区域的用电量。
说明页：https://archive.ics.uci.edu/dataset/849/power%2Bconsumption%2B；
匿名 HTTPS 下载地址：https://archive.ics.uci.edu/static/public/849/power+consumption+of+tetouan+city.zip；
许可：CC BY 4.0。
程序只把压缩包缓存到环境变量 AI_COURSE_DATA_DIR 指定的目录；若未设置该变量，则使用用户缓存目录。我们把每 10 分钟一次的记录取小时平均值，并使用过去 48 小时的信息预测下一小时。输入窗口只包含预测时点之前的记录，因此不会使用未来信息。
from __future__ import annotations
import os
import random
import urllib.request
import zipfile
from pathlib import Path
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.metrics import mean_absolute_error, mean_squared_error
SEED = 20260812
FAST_MODE = os.getenv("FAST_MODE", "1") != "0"
random.seed(SEED)
np.random.seed(SEED)
try:
import torch
from torch import nn
from torch.utils.data import DataLoader, TensorDataset
except ImportError as exc:
raise ImportError("本案例需要 PyTorch：请先安装 requirements 中的 torch。") from exc
torch.manual_seed(SEED)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(SEED)
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
CACHE_ROOT = Path(os.getenv("AI_COURSE_DATA_DIR", Path.home() / ".cache" / "ai-course"))
CACHE_DIR = CACHE_ROOT / "uci_tetouan_power"
CACHE_DIR.mkdir(parents=True, exist_ok=True)
print({"快速模式": FAST_MODE, "计算设备": str(DEVICE), "缓存目录": str(CACHE_DIR)})
{'快速模式': True, '计算设备': 'cpu', '缓存目录': '/private/tmp/ai-course-case-data/uci_tetouan_power'}
DATA_URL = "https://archive.ics.uci.edu/static/public/849/power+consumption+of+tetouan+city.zip"
DATA_ZIP = CACHE_DIR / "power_consumption_tetouan.zip"
if not DATA_ZIP.exists():
print("首次运行：下载公开数据……")
urllib.request.urlretrieve(DATA_URL, DATA_ZIP)
assert DATA_ZIP.stat().st_size > 100_000, "下载文件异常地小"
with zipfile.ZipFile(DATA_ZIP) as zf:
csv_names = [name for name in zf.namelist() if name.lower().endswith(".csv")]
assert len(csv_names) == 1, f"预期一个 CSV，实际为 {csv_names}"
with zf.open(csv_names[0]) as handle:
raw = pd.read_csv(handle)
raw.columns = [str(c).strip() for c in raw.columns]
time_col = next(c for c in raw.columns if "date" in c.lower())
raw["timestamp"] = pd.to_datetime(
raw[time_col], format="%m/%d/%Y %H:%M", errors="coerce"
)
assert raw["timestamp"].notna().all(), "存在无法解析的时间戳"
raw = raw.drop(columns=[time_col]).sort_values("timestamp").drop_duplicates("timestamp")
numeric_cols = [c for c in raw.columns if c != "timestamp"]
raw[numeric_cols] = raw[numeric_cols].apply(pd.to_numeric, errors="coerce")
assert raw[numeric_cols].notna().mean().min() > 0.99
print(f"原始维度={raw.shape}，时间范围={raw.timestamp.min()} 至 {raw.timestamp.max()}")
原始维度=(52416, 9)，时间范围=2017-01-01 00:00:00 至 2017-12-30 23:50:00
hourly = (
raw.set_index("timestamp")[numeric_cols]
.resample("h")
.mean()
.interpolate(method="time", limit_direction="both")
)
target_col = next(c for c in hourly.columns if "zone 1" in c.lower())
assert hourly.index.is_monotonic_increasing and hourly.notna().all().all()
assert len(hourly) > 8_000
summary = hourly[[target_col]].describe(percentiles=[0.05, 0.5, 0.95]).T
summary_display = summary.rename(
index={target_col: "一区用电量"},
columns={"count": "样本数", "mean": "均值", "std": "标准差",
"min": "最小值", "5%": "百分之五分位数",
"50%": "中位数", "95%": "百分之九十五分位数", "max": "最大值"},
)
summary_display.index.name = "指标"
display(summary_display)
print("小时数：", len(hourly), "；数值通道：", len(hourly.columns))
sample = hourly[target_col].iloc[:24 * 14]
ax = sample.plot(figsize=(11, 3), title="一区用电量：前两周小时均值")
ax.set_ylabel("用电量")
plt.tight_layout()
| 样本数 | 均值 | 标准差 | 最小值 | 百分之五分位数 | 中位数 | 百分之九十五分位数 | 最大值 | |
|---|---|---|---|---|---|---|---|---|
| 指标 | ||||||||
| 一区用电量 | 8736.0 | 32344.970564 | 7068.919876 | 14329.113923 | 21895.62004 | 32342.303347 | 44559.171777 | 51844.261932 |
小时数： 8736 ；数值通道： 8
# 先切时间，再估计标准化参数；绝不能用验证集或测试集计算均值和标准差。
n_time = len(hourly)
train_end = int(n_time * 0.70)
valid_end = int(n_time * 0.85)
assert 48 < train_end < valid_end < n_time
feature_cols = list(hourly.columns)
train_frame = hourly.iloc[:train_end]
x_mean = train_frame[feature_cols].mean().to_numpy(np.float32)
x_std = train_frame[feature_cols].std().replace(0, 1).to_numpy(np.float32)
y_mean = np.float32(train_frame[target_col].mean())
y_std = np.float32(train_frame[target_col].std())
assert y_std > 0
x_scaled = (hourly[feature_cols].to_numpy(np.float32) - x_mean) / x_std
y_scaled = (hourly[target_col].to_numpy(np.float32) - y_mean) / y_std
LOOKBACK = 48
def make_windows(x: np.ndarray, y: np.ndarray, lookback: int):
X, Y, target_positions = [], [], []
for t in range(lookback, len(x)):
X.append(x[t - lookback:t])
Y.append(y[t])
target_positions.append(t)
return np.stack(X), np.asarray(Y, np.float32), np.asarray(target_positions)
X_all, y_all, target_pos = make_windows(x_scaled, y_scaled, LOOKBACK)
train_mask = target_pos < train_end
valid_mask = (target_pos >= train_end) & (target_pos < valid_end)
test_mask = target_pos >= valid_end
X_train, y_train = X_all[train_mask], y_all[train_mask]
X_valid, y_valid = X_all[valid_mask], y_all[valid_mask]
X_test, y_test = X_all[test_mask], y_all[test_mask]
assert not (train_mask & valid_mask).any() and not (valid_mask & test_mask).any()
assert X_train.shape[1:] == (LOOKBACK, len(feature_cols))
print("窗口维度：", {"训练集": X_train.shape, "验证集": X_valid.shape, "测试集": X_test.shape})
print("每个样本在 PyTorch 中转置为 (通道, 时间) =", (len(feature_cols), LOOKBACK))
窗口维度： {'训练集': (6067, 48, 8), '验证集': (1310, 48, 8), '测试集': (1311, 48, 8)}
每个样本在 PyTorch 中转置为 (通道, 时间) = (8, 48)
# 持续性基线：把窗口最后一小时的一区负荷直接作为下一小时预测。
# 这是短期负荷预测必须击败的朴素基线。
target_channel = feature_cols.index(target_col)
persistence_scaled = X_test[:, -1, target_channel]
persistence = persistence_scaled * y_std + y_mean
y_test_real = y_test * y_std + y_mean
def regression_metrics(y_true, y_pred):
return {
"MAE": mean_absolute_error(y_true, y_pred),
"RMSE": mean_squared_error(y_true, y_pred) ** 0.5,
}
baseline_metrics = regression_metrics(y_test_real, persistence)
print("持续性基线：", {"平均绝对误差": baseline_metrics["MAE"], "均方根误差": baseline_metrics["RMSE"]})
持续性基线： {'平均绝对误差': 1748.2161865234375, '均方根误差': 2303.0529737719885}
一维卷积与感受野#
输入张量的维度为 \((B,C,T)\)：\(B\) 是批量大小，\(C\) 是气象与用电量的通道数，\(T=48\) 是使用的历史小时数。同一个卷积核沿时间方向重复使用，因此能够在不同时间位置识别相同的局部变化模式。
当步幅都为 1 时，多层卷积的理论感受野为
其中，\(k_l\) 和 \(r_l\) 分别是第 \(l\) 层的卷积核大小和空洞率。两个核大小均为 3 的普通卷积具有 5 小时的理论感受野；当两层空洞率分别为 1 和 4 时，理论感受野扩大到 11 小时，而参数量不会按相同比例增加。本案例不在输入窗口右侧补充未来值，并且整个输入窗口都早于观测标签对应的时间。
def receptive_field(kernel_sizes, dilations):
assert len(kernel_sizes) == len(dilations)
return 1 + sum((k - 1) * d for k, d in zip(kernel_sizes, dilations))
print("普通卷积感受野：", receptive_field([3, 3], [1, 1]))
print("空洞卷积感受野：", receptive_field([3, 3], [1, 4]))
class ConvForecaster(nn.Module):
def __init__(self, in_channels: int, dilations=(1, 1)):
super().__init__()
self.features = nn.Sequential(
nn.Conv1d(in_channels, 32, kernel_size=3, dilation=dilations[0]),
nn.ReLU(),
nn.Conv1d(32, 32, kernel_size=3, dilation=dilations[1]),
nn.ReLU(),
nn.AdaptiveAvgPool1d(1),
)
self.head = nn.Linear(32, 1)
def forward(self, x):
assert x.ndim == 3 # (B, C, T)
z = self.features(x).squeeze(-1)
return self.head(z).squeeze(-1)
probe = torch.zeros(4, len(feature_cols), LOOKBACK)
assert ConvForecaster(len(feature_cols), (1, 4))(probe).shape == (4,)
普通卷积感受野： 5
空洞卷积感受野： 11
def as_tensor_dataset(X, y):
# NumPy 为 (B,T,C)，Conv1d 需要 (B,C,T)。
xt = torch.from_numpy(X).permute(0, 2, 1).contiguous()
yt = torch.from_numpy(y)
return TensorDataset(xt, yt)
if FAST_MODE:
# 快速模式只缩短训练集和轮数；验证、测试定义不变。
keep = min(4_000, len(X_train))
X_fit, y_fit = X_train[-keep:], y_train[-keep:]
epochs = 3
else:
X_fit, y_fit, epochs = X_train, y_train, 15
train_ds = as_tensor_dataset(X_fit, y_fit)
valid_ds = as_tensor_dataset(X_valid, y_valid)
generator = torch.Generator().manual_seed(SEED)
train_loader = DataLoader(train_ds, batch_size=128, shuffle=True, generator=generator)
valid_loader = DataLoader(valid_ds, batch_size=256, shuffle=False)
def fit_model(dilations):
torch.manual_seed(SEED)
model = ConvForecaster(len(feature_cols), dilations).to(DEVICE)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
history = []
for epoch in range(epochs):
model.train()
train_losses = []
for xb, yb in train_loader:
xb, yb = xb.to(DEVICE), yb.to(DEVICE)
optimizer.zero_grad()
loss = nn.functional.mse_loss(model(xb), yb)
assert torch.isfinite(loss)
loss.backward()
nn.utils.clip_grad_norm_(model.parameters(), max_norm=5.0)
optimizer.step()
train_losses.append(loss.item())
model.eval()
with torch.no_grad():
val_losses = [
nn.functional.mse_loss(model(xb.to(DEVICE)), yb.to(DEVICE)).item()
for xb, yb in valid_loader
]
history.append((np.mean(train_losses), np.mean(val_losses)))
return model, np.asarray(history)
models, histories = {}, {}
model_labels = {"ordinary": "普通卷积", "dilated": "空洞卷积"}
for name, dilations in {"ordinary": (1, 1), "dilated": (1, 4)}.items():
models[name], histories[name] = fit_model(dilations)
print(model_labels[name], "最后一轮训练集/验证集均方误差：", histories[name][-1])
普通卷积 最后一轮训练集/验证集均方误差： [0.90456015 0.9887039 ]
空洞卷积 最后一轮训练集/验证集均方误差： [0.52707343 0.62789288]
for name, hist in histories.items():
plt.plot(hist[:, 0], "--", label=f"{model_labels[name]}－训练集")
plt.plot(hist[:, 1], label=f"{model_labels[name]}－验证集")
plt.xlabel("训练轮次")
plt.ylabel("标准化均方误差")
plt.title("其他条件相同，只改变空洞率进行比较")
plt.legend()
plt.tight_layout()
test_loader = DataLoader(as_tensor_dataset(X_test, y_test), batch_size=256, shuffle=False)
def predict(model):
model.eval()
parts = []
with torch.no_grad():
for xb, _ in test_loader:
parts.append(model(xb.to(DEVICE)).cpu().numpy())
return np.concatenate(parts) * y_std + y_mean
rows = [{
"model": "persistence", **baseline_metrics,
"MAE_improvement_vs_persistence": 0.0,
}]
for name, model in models.items():
pred = predict(model)
metrics = regression_metrics(y_test_real, pred)
metrics["MAE_improvement_vs_persistence"] = 1 - metrics["MAE"] / baseline_metrics["MAE"]
rows.append({"model": name, **metrics})
results = pd.DataFrame(rows).set_index("model")
result_model_labels = {"persistence": "持续性基线", "ordinary": "普通卷积", "dilated": "空洞卷积"}
results_display = results.rename(
index=result_model_labels,
columns={"MAE": "平均绝对误差", "RMSE": "均方根误差",
"MAE_improvement_vs_persistence": "相对持续性基线的平均绝对误差改进率"},
)
results_display.index.name = "模型"
display(results_display)
best_name = results["MAE"].idxmin()
assert np.isfinite(results.to_numpy()).all()
print(f"测试集平均绝对误差最低的是 {result_model_labels[best_name]}。若卷积未击败持续性基线，应先检查训练预算、窗口和泄漏，而不是选择性报告。")
| 平均绝对误差 | 均方根误差 | 相对持续性基线的平均绝对误差改进率 | |
|---|---|---|---|
| 模型 | |||
| 持续性基线 | 1748.216187 | 2303.052974 | 0.000000 |
| 普通卷积 | 6182.355469 | 7361.819884 | -2.536379 |
| 空洞卷积 | 5643.028809 | 6747.478047 | -2.227878 |
测试集平均绝对误差最低的是 持续性基线。若卷积未击败持续性基线，应先检查训练预算、窗口和泄漏，而不是选择性报告。
逐步理解本案例#
1. 为什么先计算小时平均值#
原始数据每 10 分钟记录一次。若直接使用连续 48 个记录作为输入，模型只能看到过去 8 小时；计算小时平均值后，48 个位置对应过去两天，更容易覆盖一天内的周期变化。计算小时平均值还能减小短时间测量波动，但会丢失短时峰值，因此实验结论只适用于小时级预测。程序在重新计算时间间隔后检查时间顺序、缺失值和样本数，用来尽早发现日期解析或数据缺失方面的问题。
2. 为什么必须按照时间顺序划分数据#
预测时刻 \(t\) 的用电量时，只能使用 \(t\) 之前已经发生的信息。若随机划分时间窗口，彼此相邻且大量重叠的窗口可能同时进入训练集和测试集，模型便会在训练时看到测试窗口中的大部分内容，使测试指标显得过于乐观。本案例按照时间顺序，把目标时点的前 \(70\%\)、中间 \(15\%\) 和最后 \(15\%\) 分别作为训练集、验证集和测试集。验证窗口可以使用训练期末尾已经发生的历史记录，但标准化所需的均值和标准差仍只能根据训练时段计算。
窗口建立后的维度为 \((样本数,48,通道数)\)，其中第二个轴表示时间。PyTorch 的一维卷积要求输入维度为 \((批量,通道,时间)\)，所以在送入网络之前要交换后两个轴。网络为每个样本输出一个数，输出维度为 \((批量,)\)，与观测标签向量一致。改变特征数或窗口长度后，应重新检查这些维度，不能依赖广播机制自动处理。
3. 持续性基线回答什么问题#
下一小时的用电量通常与当前小时接近，因此可以把窗口最后一个时点的用电量直接作为下一小时预测，这称为持续性基线。若卷积网络不能稳定优于这个简单方法，就没有必要仅因为结构更复杂而采用它。MAE 使用原始单位表示误差大小，RMSE 对少数较大误差更加敏感，因此应同时报告二者。计算相对于基线的改进幅度时，还应明确以哪个基线误差作为分母。
4. 卷积核与感受野怎样对应预测问题#
卷积核大小为 \(k\) 时，第一层在每个位置读取连续 \(k\) 小时。堆叠多层后，较深层的一个输出可以间接利用更长的历史。步幅均为 1 时，感受野满足 \(R_l=R_{l-1}+(k_l-1)r_l\)，其中 \(r_l\) 是第 \(l\) 层的空洞率；普通卷积的空洞率为 1。空洞卷积可以在没有明显增加参数量的情况下读取距离更远的时间位置。理论感受野覆盖两天，只表示网络能够使用两天内的信息，并不表示网络一定学会了日周期，还需要通过比较实验和分时段误差进行判断。
比较普通卷积与空洞卷积时，应保持输入、训练轮数、评价数据和其他训练条件相同，只改变空洞率。若同时改变通道数、优化器或学习率，就无法判断结果变化主要来自哪个因素。快速模式用于检查程序能否顺利运行；较完整的实验应使用多个随机种子重复训练，并报告结果的均值和标准差。
5. 怎样判断训练过程是否正常#
先查看训练损失是否下降，再观察验证损失是否也随之改善。训练损失继续下降而验证损失变差，说明模型可能过度贴合训练数据；两者都不下降时，应检查学习率、观测标签的数值范围和梯度；出现无穷大或无效数值时，应立即停止并检查输入与参数更新。模型结构和停止轮数确定后，测试集只用于一次最终评价。还可以绘制误差随时间变化的曲线，检查模型是否在高峰期、节假日或季节转换时持续出现较大误差，而不能只报告一个全年平均数。
6. 与正文知识点的联系#
本案例的重点不是得到最低误差，而是把一维卷积中的通道、卷积核、填充、空洞率和感受野用于一个可以逐步检查的时间序列问题。填充影响输出长度，空洞率决定卷积核所读取位置之间的间隔，多通道使气象变量和多个区域的用电量能够共同参与预测。如果任务需要在线预测，还必须确认每个输入变量在预测时刻确实已经获得；只能事后得到或事后修订的数据会造成信息泄漏。
7. 复现与输出格式#
固定随机种子可以控制数据加载顺序和参数初始化，但不同硬件上的结果仍可能有细小差异。发布结果时应记录数据摘要、时间划分边界、PyTorch 版本、计算设备、是否采用快速模式、训练时间和全部超参数。图题、坐标轴、表格列名和运行提示均使用中文；英文只保留官方数据集名称、Python 接口、变量名和数学缩写。
本案例小结与局限#
计算小时平均值会丢失 10 分钟尺度上的用电峰值，因此结论只适用于当前时间分辨率；
2017 年一个城市的数据不能代表其他年份、气候或城市，本案例尚未检验时间变化带来的影响；
普通卷积和空洞卷积不仅感受野不同，有效输出长度也会略有不同。本实验使用自适应池化得到相同维度的最终输出，但仍应谨慎解释模型差异；
验证窗口和测试窗口可以使用预测时点之前的真实观测，这符合逐时预测的实际过程；它们不能读取预测时点之后的数据；
测试集只用于一次最终评价。若根据测试结果继续选择窗口长度、空洞率或学习率，测试集就不再是独立的最终评价数据；
用电预测涉及公共基础设施，本教学模型不具备实际部署所需的可靠性检查、数据变化监控和安全审查。
发布前检查#
运行结束后，还要确认标准化没有改变最终报告中的用电量单位，测试指标已经还原到原始用电量尺度，并且每个结果表都注明对应的时间范围。若服务器只负责生成文档，应关闭案例的自动执行功能，避免每次构建网页时重复下载数据和训练模型。
综合练习#
手算核大小
[5,3,3]、空洞率[1,2,4]、步幅均为 1 时的感受野，并用代码验证。保持数据划分、随机种子、参数规模和训练更新次数不变，只把
LOOKBACK改为 24、72、168，比较 MAE。增加“过去 24 小时同一时刻”的季节性基线，并解释它与持续性基线各自利用了什么假设。
将模型改为严格的左侧因果填充，输出每个时间位置的预测；写断言证明改变窗口最后一个时点之后的值不会影响较早输出。
报告至少三个随机种子的均值和标准差，并说明空洞卷积的优势是否稳定。