第5章 时间序列：从连续记录中发现变化

本章技术主题： 时间序列学习——利用过去的连续记录预测下一时点。
本章技术实验： 把连续环境记录切成时间窗口，比较规则、统计基线和时序模型。
家庭项目： “建立家庭空气与用能观察站”第一阶段——观察、异常与预测。
跨章位置： 本章保存数据质量和预测证据，第11章只在虚拟环境中研究舒适—节能行动策略。
技术主线： 时间记录 → 数据质量 → 历史窗口 → 基线与模型 → 预测误差 → 异常、漂移与使用判断。

学习目标

学完本章，你将能够：

1. 说明时间戳、采样间隔、时间窗口和按时间划分数据的意义；
2. 比较固定规则、上一时点、移动平均和小型时序模型分别能解决什么问题；
3. 运行一个使用一维卷积网络预测下一时点数值的完整Keras/TensorFlow项目；
4. 用缺失、突跳、缓慢漂移和陌生时段测试预测与异常标记；
5. 根据同一测试时段的误差、成本和维护要求，判断是否值得采用时序模型。

项目导入

温度、湿度、空气质量、土壤湿度和用电功率等数据都带有时间顺序。某个读数本身可能很普通，但它相对于前一小时突然升高，或者连续几周缓慢变化，就可能值得复核。处理这类数据时，不能把所有行随意打乱，因为模型在预测某一时刻时只能使用当时已经发生的历史，不能偷看未来。

本章建立一个家庭空气与用能观察站，以固定间隔记录的环境或用能CSV为输入。课堂默认使用教师提供的模拟数据、脱敏记录或预先采集的回放数据，不要求上传真实家庭作息与精细用电明细。项目先检查时间戳、缺失、重复、断点和长期不变，再把连续读数切成历史窗口，用窗口中的过去数据预测下一时点；同时保留固定阈值、上一时点和移动平均等透明基线。只有模型在同一封存测试时段稳定优于基线，才有理由考虑增加它。

本章可以沿用第2章的阳台环境字段，但不再把每行当作可随意打乱的独立样本，而是保留时间顺序，用过去窗口预测下一时点。字段相似不代表学习问题相同：结构化分类回答“这一行属于什么”，时间序列回答“接下来可能怎样变化”。

作品只用于学习趋势、预测和异常复核，不提供健康、消防、电气安全或设备故障结论，不自动断电、开窗或控制真实设备。异常分数表示“与当前模型或近期历史不一致”，不等于已经发现危险和原因。

本章成果

• 一个完成家庭环境或用能数据检查、时间窗口构造、基线比较和时序模型训练的程序；
• 一张“连续记录—历史窗口—下一时点—预测误差”的原理图；
• 一份按时间划分的验证与测试报告；
• 一份缺失、漂移、突跳和故障数据的测试记录；
• 一张说明采用规则、统计基线、时序模型或暂不用AI的判断卡。

考点提示

时间序列的核心不是“画一条折线”，而是时间先后关系。训练、验证和测试必须按时间划分；归一化参数只能从训练时段计算；预测某一点的窗口不能包含该点之后的数据。上一时点和移动平均是重要基线。模型误差较小，不代表模型知道变化原因；预测残差较大，也只表示需要复核。

第一节 把环境记录变成时间窗口

一、先检查时间，再讨论模型

输入文件environment.csv至少包含timestamp和一个待分析数值字段，例如power_w。每一行表示一个时点。开始训练前要回答：采样间隔是否基本固定；时间是否重复或逆序；是否有空值和非数字；是否存在长断点；传感器是否长时间保持完全相同的读数。

假设每10分钟记录一次，用过去12个点预测下一个点，那么一个输入窗口覆盖过去120分钟。第一个窗口使用第1—12个读数，目标是第13个读数；下一个窗口使用第2—13个读数，目标是第14个读数。窗口长度相同，但相邻窗口会共享大部分历史，这是时间序列的正常结构。

缺失时点不能悄悄填零。零可能是一个真实读数，也可能制造突跳。首版程序遇到明显断点时停止，要求学生检查设备日志或明确选择插值方法。任何补值都应写入处理记录，原始CSV保持只读。

二、按时间划分训练、验证和测试

普通表格常见随机划分，但时间序列应保持先后顺序。例如，最早70%用于训练，中间15%用于验证，最后15%作为测试。这样模拟了真实使用：用过去建立模型，再处理未来。

训练集用于调整模型参数；验证集用于选择窗口、网络规模和训练轮数；测试集只在方案确定后形成最终证据。如果反复查看测试结果并据此改模型，测试集就参与了开发，不再是独立证据。季节变化明显时，还应保留跨季节或跨设备的额外测试。

划分边界要以“目标时点”判断，而不是只看窗口从哪里开始。验证集的第一个目标可以使用训练末尾已经发生的历史，这符合真实预测；但它不能使用目标之后的记录。测试集同理，只允许使用测试目标之前已经可见的窗口。若把所有窗口先随机打乱，再按比例切分，相邻且高度重叠的窗口可能分散到训练和测试两边，使误差看起来不真实地偏小。

数值归一化也要遵守时间边界。训练均值和标准差只能由训练时段计算，再用于验证和测试。如果先用全体数据计算均值，未来信息就以统计量的形式泄漏给模型。

三、与AI协同建立时间窗口工程

本章的应用中的AI是小型一维卷积时序模型：它接收过去一段时间的数值序列，输出下一时点的预测。开发与学习AI帮助学生解释数据形状、生成完整程序、分析训练曲线和限定修改；学生负责确认字段含义、采样条件、基线、错误后果和采用范围。

项目文件如下：

ch05-timeseries-project/
├─ input/environment.csv
├─ input/config.json
├─ src/forecast.py
├─ tests/cases.json
├─ output/predictions.csv、summary.json、test_plot.png、model.keras
└─ evidence/dialogue.md、change.diff、regression.md、decision-card.md

config.json示例：

{
  "field": "power_w",
  "window": 12,
  "fixed_threshold": 1200,
  "z_threshold": 3.0,
  "epochs": 40
}

向开发与学习AI说明任务时，应明确“只能用目标时点之前的数据”“按时间划分”“归一化只使用训练时段”“必须同时报告基线和模型”。如果生成的代码先打乱全部窗口再划分，或用测试误差选择训练轮数，应要求修正。

第二节 比较规则、统计基线与时序模型

一、四种方法回答不同问题

固定阈值回答“当前值是否越过人工设定的观察线”。它清楚、稳定，却不会自动适应季节和使用习惯。上一时点基线直接把当前值作为下一时点预测，适合变化缓慢的数据；如果复杂模型不能超过它，就没有证明新增复杂度有价值。

移动平均用最近若干点的平均值预测下一点。窗口短，响应较快但容易受噪声影响；窗口长，曲线平稳却可能延迟。Z分数比较当前点与此前窗口的均值和波动，适合标记相对近期历史较少见的点，但不能说明危险和原因。

时序模型从大量历史窗口中调整参数，尝试学习重复周期、局部趋势和变化组合。本章使用一维卷积提取窗口中的局部模式，再输出下一时点数值。模型是否有用，要用同一测试时段的平均绝对误差与基线比较，而不是只看训练损失是否下降。

二、完整核心程序

下面的src/forecast.py完成数据质量检查、按时间划分、训练集归一化、历史窗口构造、两个预测基线、一维卷积模型、测试误差、残差异常和漂移提示。运行前安装TensorFlow、NumPy和Matplotlib。

from pathlib import Path
from datetime import datetime
import csv
import json

import matplotlib.pyplot as plt
import numpy as np
import tensorflow as tf

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "output"
OUT.mkdir(exist_ok=True)
tf.keras.utils.set_random_seed(42)

config = json.loads(
    (ROOT / "input" / "config.json").read_text(encoding="utf-8")
)
FIELD = config["field"]
WINDOW = int(config["window"])
FIXED_THRESHOLD = float(config["fixed_threshold"])
Z_THRESHOLD = float(config["z_threshold"])
EPOCHS = int(config.get("epochs", 40))

with (ROOT / "input" / "environment.csv").open(
    encoding="utf-8-sig", newline=""
) as file:
    rows = list(csv.DictReader(file))

if not rows or "timestamp" not in rows[0] or FIELD not in rows[0]:
    raise ValueError(f"CSV必须包含timestamp和{FIELD}")

try:
    times = [datetime.fromisoformat(row["timestamp"]) for row in rows]
    values = np.asarray([float(row[FIELD]) for row in rows], dtype=np.float32)
except ValueError as error:
    raise ValueError(f"时间或数值格式错误：{error}") from error

if len(values) < max(80, WINDOW * 5):
    raise ValueError("连续记录过少，无法完成按时间划分和窗口学习")
if not np.all(np.isfinite(values)):
    raise ValueError("数据包含空值、无穷值或非数字")

intervals = np.asarray(
    [(times[i] - times[i - 1]).total_seconds() for i in range(1, len(times))]
)
if np.any(intervals <= 0):
    raise ValueError("时间戳重复或逆序")
median_interval = float(np.median(intervals))
if np.any(intervals > 1.5 * median_interval):
    raise ValueError("发现明显采样断点；请核查缺失时点，不要自动填零")

train_end = int(len(values) * 0.70)
valid_end = int(len(values) * 0.85)
if train_end <= WINDOW or valid_end <= train_end:
    raise ValueError("数据划分后没有足够窗口")

train_mean = float(values[:train_end].mean())
train_std = float(values[:train_end].std())
if train_std < 1e-8:
    raise ValueError("训练时段几乎没有变化，不能建立有效预测模型")
normalized = (values - train_mean) / train_std


def make_windows(series, window):
    features, targets, target_indices = [], [], []
    for target_index in range(window, len(series)):
        features.append(series[target_index - window : target_index])
        targets.append(series[target_index])
        target_indices.append(target_index)
    return (
        np.asarray(features, dtype=np.float32)[..., np.newaxis],
        np.asarray(targets, dtype=np.float32),
        np.asarray(target_indices),
    )


all_x, all_y, all_index = make_windows(normalized, WINDOW)
train_mask = all_index < train_end
valid_mask = (all_index >= train_end) & (all_index < valid_end)
test_mask = all_index >= valid_end
x_train, y_train = all_x[train_mask], all_y[train_mask]
x_valid, y_valid = all_x[valid_mask], all_y[valid_mask]
x_test, y_test = all_x[test_mask], all_y[test_mask]
test_index = all_index[test_mask]

model = tf.keras.Sequential(
    [
        tf.keras.layers.Input((WINDOW, 1)),
        tf.keras.layers.Conv1D(16, 3, activation="relu"),
        tf.keras.layers.Conv1D(8, 3, activation="relu"),
        tf.keras.layers.GlobalAveragePooling1D(),
        tf.keras.layers.Dense(8, activation="relu"),
        tf.keras.layers.Dense(1),
    ]
)
model.compile(optimizer="adam", loss="mse", metrics=["mae"])
early_stop = tf.keras.callbacks.EarlyStopping(
    monitor="val_loss", patience=5, restore_best_weights=True
)
model.fit(
    x_train,
    y_train,
    validation_data=(x_valid, y_valid),
    epochs=EPOCHS,
    batch_size=16,
    callbacks=[early_stop],
    verbose=2,
)
model.save(OUT / "model.keras")


def denormalize(series):
    return np.asarray(series) * train_std + train_mean


model_test = denormalize(model.predict(x_test, verbose=0).squeeze())
actual_test = values[test_index]
previous_test = values[test_index - 1]
moving_test = np.asarray(
    [values[index - WINDOW : index].mean() for index in test_index]
)

model_valid = denormalize(model.predict(x_valid, verbose=0).squeeze())
valid_index = all_index[valid_mask]
valid_actual = values[valid_index]
valid_residual = np.abs(valid_actual - model_valid)
residual_limit = float(valid_residual.mean() + 3 * valid_residual.std())
test_residual = np.abs(actual_test - model_test)


def previous_z(index):
    history = values[index - WINDOW : index]
    std = float(history.std())
    if std < 1e-8:
        return 0.0 if values[index] == history.mean() else float("inf")
    return float((values[index] - history.mean()) / std)


records = []
for position, index in enumerate(test_index):
    z_score = previous_z(int(index))
    records.append(
        {
            "timestamp": times[int(index)].isoformat(),
            "actual": float(actual_test[position]),
            "previous_baseline": float(previous_test[position]),
            "moving_baseline": float(moving_test[position]),
            "model_prediction": float(model_test[position]),
            "absolute_residual": float(test_residual[position]),
            "fixed_flag": bool(actual_test[position] > FIXED_THRESHOLD),
            "z_score": z_score,
            "z_flag": bool(abs(z_score) > Z_THRESHOLD),
            "prediction_flag": bool(test_residual[position] > residual_limit),
        }
    )

with (OUT / "predictions.csv").open(
    "w", encoding="utf-8-sig", newline=""
) as file:
    writer = csv.DictWriter(file, fieldnames=list(records[0]))
    writer.writeheader()
    writer.writerows(records)


def mae(actual, predicted):
    return float(np.mean(np.abs(actual - predicted)))


test_mean = float(actual_test.mean())
drift_score = abs(test_mean - train_mean) / train_std
summary = {
    "field": FIELD,
    "records": len(values),
    "window": WINDOW,
    "train_targets": int(train_mask.sum()),
    "valid_targets": int(valid_mask.sum()),
    "test_targets": int(test_mask.sum()),
    "previous_mae": mae(actual_test, previous_test),
    "moving_mae": mae(actual_test, moving_test),
    "model_mae": mae(actual_test, model_test),
    "prediction_residual_limit_from_validation": residual_limit,
    "prediction_flag_count": int(np.sum(test_residual > residual_limit)),
    "drift_score_in_train_standard_deviations": float(drift_score),
    "notice": "异常与漂移只表示需要复核，不表示危险或原因。",
}
(OUT / "summary.json").write_text(
    json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8"
)

plt.figure(figsize=(9, 4))
test_times = [times[int(index)] for index in test_index]
plt.plot(test_times, actual_test, label="actual")
plt.plot(test_times, previous_test, label="previous baseline", alpha=0.7)
plt.plot(test_times, model_test, label="Conv1D prediction", alpha=0.8)
plt.axhline(FIXED_THRESHOLD, color="orange", linestyle="--", label="fixed threshold")
plt.legend()
plt.tight_layout()
plt.savefig(OUT / "test_plot.png", dpi=150)
plt.close()
print(json.dumps(summary, ensure_ascii=False, indent=2))

这段程序没有把测试时段用于训练，也没有用全体数据计算归一化参数。模型预测、上一时点和移动平均都恢复为原始单位后再计算平均绝对误差，因此可以直接比较。若模型没有明显超过简单基线，应保留基线或继续收集数据，而不是只展示神经网络训练成功。

prediction_flag表示模型预测与实际值的残差超过验证时段形成的界限；z_flag表示当前值相对此前窗口较少见；fixed_flag表示超过人工设定的观察线。三者回答的问题不同，必须分别保留。

本章只做“一步预测”，即每次根据真实的最近窗口预测紧接着的一个时点。连续预测多个未来时点是另一项更困难的任务：后续窗口可能包含模型自己的前一步预测，误差会逐步积累。不能把一步预测的测试结果直接解释为“可以可靠预测未来一天”。如需多步预测，应另行设计输出长度、滚动方式和多步误差测试。

三、比较结果而不是比较“技术高级程度”

评价表至少列出上一时点MAE、移动平均MAE、模型MAE、异常标记数、运行时间和维护要求。若模型只比上一时点改善很少，却需要更多依赖、训练和持续监测，简单基线可能更合适。若数据存在稳定周期，模型在多个封存时段都明显改善，并且错误后果可控，才有进一步采用价值。

只改变窗口长度也可以形成对照实验。例如把window从6改为18，保持数据划分、模型结构、训练轮数上限和测试时段不变。窗口6看到较短历史，窗口18看到较长历史；结果变化不能简单解释为“越长越好”，还要结合采样间隔、周期长度、误差和响应速度。

第三节 测试缺失、漂移、异常与预测边界

一、用固定样例暴露四类问题

突跳既可能是真实事件，也可能是传感器或通信异常；缓慢漂移既可能来自季节，也可能来自设备老化。程序只能指出数值与历史或预测不一致，不能自动确定原因。应回查设备日志、现场记录或参考仪表。

漂移分数比较测试均值与训练均值的距离，只是一个初步提示。均值相近也可能存在其他分布变化；均值不同也可能是合理季节变化。模型投入使用后，要按固定周期保存新数据、重跑基线和测试，并明确谁决定是否重新训练。

二、根据泄漏或边界错误限定修改

一种常见错误是构造目标时把当前点放进输入窗口，或者计算当前点Z分数时把当前点加入自己的历史。发现这类问题后，应让开发与学习AI只修改窗口切片，并保持模型、测试时段和阈值不变：

- history = values[index - WINDOW : index + 1]
+ history = values[index - WINDOW : index]

修改后重新运行正常、突跳、缓慢漂移、恒定读数和缺失时点。突跳标记可能变化，但正常样例不应大量增加误报，缺失时点仍应停止。学生要检查代码差异，确认AI没有顺便改动测试数据、数据划分和阈值。

三、形成预测边界和采用判断

判断是否采用时序模型，应同时检查：数据是否足够连续并覆盖实际变化；测试是否严格晚于训练；模型是否稳定超过简单基线；误报、漏报和延迟分别造成什么后果；本地算力、数据隐私和维护成本是否可承担；漂移后由谁复核和重训。

合理结论可以是“采用固定规则与移动平均，暂不采用模型”，也可以是“模型仅用于生成待复核预测，不参与设备控制”。即使采用模型，固定规则和数据质量检查仍应保留为解释基线和故障退路。语言模型可以依据summary.json协助整理周报，但不得把异常分数改写成未经证实的原因。

本章小结

时间序列的关键信息存在于先后顺序中。连续记录先经过时间戳、间隔、缺失和传感器状态检查，再被切成“过去窗口—下一时点”样本。训练、验证和测试按时间划分，归一化只使用训练时段，才能避免未来信息泄漏。

固定阈值、上一时点、移动平均、Z分数和一维卷积模型回答不同问题。模型必须在同一测试时段与基线比较；预测残差和漂移只说明需要复核，不能证明危险和原因。应用中的AI负责从历史窗口产生预测，开发与学习AI协助编程和分析，人负责数据条件、错误后果与采用决定。

关键术语

• 时间序列：按时间先后排列的一组连续记录。
• 采样间隔：相邻两个记录时点之间的时间距离。
• 时间窗口：用于预测下一时点的一段连续历史。
• 时间划分：按照先后顺序形成训练、验证和测试时段。
• 预测基线：用于判断复杂模型是否真正改善任务的简单预测方法。
• 预测残差：实际值与预测值之间的差。
• 数据漂移：季节、设备或行为变化使新数据分布偏离训练数据。
• 一维卷积：沿时间方向提取局部序列模式的神经网络运算。
• 数据泄漏：训练或开发过程使用了本应不可见的未来或测试信息。

目标测试

一、单项选择题

1. 时间序列通常按时间顺序划分训练和测试，主要是为了（　）。
   A．让图表更美观　B．模拟用过去预测未来并避免泄漏　C．增加字段　D．删除异常
2. 如果复杂模型在测试时段没有超过“直接使用上一时点”的基线，较合理的做法是（　）。
   A．隐藏基线　B．仍宣称模型更先进　C．保留简单方案并检查数据和任务　D．把测试数据加入训练
3. 当前值的预测残差很大，最准确的解释是（　）。
   A．已经证明存在危险　B．当前值与模型预测不一致，需要复核　C．模型一定损坏　D．已经找到变化原因
4. 归一化所用均值和标准差应首先来自（　）。
   A．全部数据　B．测试时段　C．训练时段　D．未来数据

二、判断并改错

5. 为了提高模型精度，可以把测试时段打乱后加入训练集，再继续报告原测试结果。请判断并改正。
6. 一维卷积模型的测试误差较小，就说明它已经知道读数变化的真实原因。请判断并改正。

三、简答与操作题

7. 说明“过去12个点预测下一个点”怎样形成输入和目标，并指出哪种窗口构造会造成未来信息泄漏。
8. 比较固定阈值、上一时点、移动平均和一维卷积模型各自回答的问题及一个局限。
9. 设计缺失、突跳、缓慢漂移和传感器停滞四类测试，分别写出期望行为。
10. 根据数据、基线误差、模型误差、隐私、成本和维护条件，说明你会在什么情况下采用时序模型，什么情况下暂不用AI。

答案编号：A1-5-01—A1-5-10。完整答案和评价要点统一放入书后“目标测试参考答案”。