KDnuggets
使用 Python Itertools 进行时间序列特征工程
8.5Score
TL;DR · AI 摘要
使用 Python 的 itertools 模块构建时间序列特征，提供灵活的迭代方法。
核心要点
- 文章介绍了如何利用 itertools 构建七类时间序列特征。
- 通过 islice 实现滞后特征，支持自定义偏移量。
- 代码示例展示了如何生成和应用这些特征到实际数据集。
结构提纲
按章节快速跳转。
- §引言
介绍时间序列特征工程与表格数据的不同之处，并指出 itertools 的适用性。
生成一个包含温度、湿度和功率读数的传感器数据集用于后续分析。
使用 itertools 的 islice 方法创建不同时间间隔的滞后特征。
思维导图
用一张图看清主题之间的关系。
查看大纲文本（无障碍 / 无 JS 友好）
- Time-Series Feature Engineering with Python Itertools
- Introduction
- Time Series vs Tabular Data
- itertools Applicability
- Sample Dataset Creation
- Sensor Readings Generation
- Data Structure Overview
- Lag Features Generation
- islice Method Usage
- Custom Offset Implementation
金句 / Highlights
值得收藏与分享的关键句。
时间序列特征工程与表格数据遵循不同的规则。
滞后特征是最基本的时间序列特征：变量在固定步数前的值。
你可以在 GitHub 上获取代码：https://github.com/balapriyac/data-science-tutorials/tree/main/time-series-feature-engineering
#Python#时间序列#itertools
打开原文#引言
时间序列特征工程并不遵循与表格数据相同的规则。观测值不是独立的，行顺序不是偶然的，最有用的特征很少是单个读数。您必须识别跨时间的模式，如变化率、滞后比较、与滚动基线的偏差等。
构建滞后、滑动窗口和跨分辨率的分组，本质上都是有序序列的迭代问题。[Python 的 itertools 模块](https://docs.python.org/3/library/itertools.html) 非常适合这类工作。它并不能替代像 .rolling() 这样的高级 [pandas](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.rolling.html) 抽象，但它为您提供了更低层次的构建块，以便精确构建您需要的特征，并完全控制逻辑。
在本文中，您将使用 itertools 构建七类时间序列特征。您还将把每种特征应用到示例数据集中。
[您可以在 GitHub 上获取代码](https://github.com/balapriyac/data-science-tutorials/tree/main/time-series-feature-engineering)。
#创建示例数据集
在我们开始构建特征之前，让我们创建一个示例传感器数据集，以便在整篇文章中使用。
code
import numpy as np
import pandas as pd
import itertools
np.random.seed(42)
periods = 168 # one week of hourly readings
index = pd.date_range(start="2024-03-01", periods=periods, freq="h")
hours = np.arange(periods)
# Temperature (°C): daily cycle + gradual drift + noise
temp_base = 3.5
temp_daily = 1.2 * np.sin(2 * np.pi * hours / 24)
temp_drift = 0.003 * hours
temp_noise = np.random.normal(0, 0.3, periods)
temperature = temp_base + temp_daily + temp_drift + temp_noise
# Humidity (%): inverse relationship with temperature + noise
humidity = 78 - 2.1 * (temperature - temp_base) + np.random.normal(0, 1.2, periods)
# Power draw (kW): peaks during business hours, higher on weekdays
day_of_week = index.dayofweek
business_hours = ((index.hour >= 8) & (index.hour <= 18)).astype(int)
weekend_factor = np.where(day_of_week >= 5, 0.6, 1.0)
power = (
42.0
+ 18.0 * business_hours * weekend_factor
+ np.random.normal(0, 2.1, periods)
)
df = pd.DataFrame({
"temperature_c": np.round(temperature, 3),
"humidity_pct": np.round(humidity, 2),
"power_kw": np.round(power, 2),
}, index=index)
df.index.name = "timestamp"
print(df.head(8))
print(f"\nShape: {df.shape}")输出：
code
temperature_c humidity_pct power_kw
timestamp
2024-03-01 00:00:00 3.649 77.39 40.27
2024-03-01 01:00:00 3.772 76.52 41.33
2024-03-01 02:00:00 4.300 75.25 42.87
2024-03-01 03:00:00 4.814 74.26 40.82
2024-03-01 04:00:00 4.481 75.85 40.27
2024-03-01 05:00:00 4.604 76.09 42.51
2024-03-01 06:00:00 5.192 74.78 42.51
2024-03-01 07:00:00 4.910 76.03 40.94
Shape: (168, 3)我们现在有三个传感器通道的 168 个每小时读数。现在让我们构建特征。
#1. 使用 `islice` 生成滞后特征
滞后特征是最基本的时间序列特征：变量在过去固定步数处的值。例如，1 步前、6 步前或 24 步前的值可以分别捕捉不同的模式，如短期波动、周期性周期内行为以及长期趋势或季节性。
让我们使用 islice 为示例数据集构建滞后特征：
code
sensor_readings = df["temperature_c"].tolist()
lag_offsets = [1, 6, 12, 24]
lag_features = {}
for lag in lag_offsets:
lagged = list(itertools.islice(sensor_readings, 0, len(sensor_readings) - lag))
# Pad the beginning with None to preserve index alignment
lag_features[f"temp_lag_{lag}h"] = [None] * lag + lagged
lag_df = pd.DataFrame(lag_features, index=df.index)
lag_df["temperature_c"] = df["temperature_c"]
print(lag_df.iloc[24:30])输出：
code
temp_lag_1h temp_lag_6h temp_lag_12h temp_lag_24h \
timestamp
2024-03-02 00:00:00 2.831 2.082 3.609 3.649
2024-03-02 01:00:00 3.409 1.974 2.654 3.772
2024-03-02 02:00:00 3.919 2.960 2.425 4.300
2024-03-02 03:00:00 3.833 2.647 2.528 4.814
2024-03-02 04:00:00 4.542 2.986 2.205 4.481
2024-03-02 05:00:00 4.443 2.831 2.486 4.604
temperature_c
timestamp
2024-03-02 00:00:00 3.409
2024-03-02 01:00:00 3.919
2024-03-02 02:00:00 3.833
2024-03-02 03:00:00 4.542
2024-03-02 04:00:00 4.443
2024-03-02 05:00:00 4.659islice(sensor_readings, 0, len - lag) 提取了向后移动 lag 步的序列，而无需创建完整列表的副本。前面的 None 填充使每个滞后特征与原始索引保持对齐。这在您稍后为模型训练删除 NaN 时很重要。
#2. 使用 `islice` 和 `accumulate` 构建滚动窗口特征
单个滞后值告诉您传感器在过去某个时间点的读数。而滚动统计量则告诉您传感器在一段时间窗口内的行为，这通常更有用。
code
readings = df["temperature_c"].tolist()
window_size = 6 # 6-hour rolling window
rolling_features = []python
for i in range(len(readings)):
if i < window_size:
rolling_features.append({
"rolling_mean_6h": None,
"rolling_std_6h": None,
"rolling_min_6h": None,
"rolling_max_6h": None,
})
continue
window = list(itertools.islice(readings, i - window_size, i))
# 使用 accumulate 计算运行总和以获取均值
running_sum = list(itertools.accumulate(window))
window_mean = running_sum[-1] / window_size
window_mean_sq = sum(x**2 for x in window) / window_size
rolling_features.append({
"rolling_mean_6h": round(window_mean, 4),
"rolling_std_6h": round((window_mean_sq - window_mean**2) ** 0.5, 4),
"rolling_min_6h": round(min(window), 4),
"rolling_max_6h": round(max(window), 4),
})
roll_df = pd.DataFrame(rolling_features, index=df.index)
roll_df["temperature_c"] = df["temperature_c"]
print(roll_df.iloc[6:12])输出：
code
rolling_mean_6h rolling_std_6h rolling_min_6h \
timestamp
2024-03-01 06:00:00 4.2700 0.4256 3.649
2024-03-01 07:00:00 4.5272 0.4386 3.772
2024-03-01 08:00:00 4.7168 0.2929 4.300
2024-03-01 09:00:00 4.7372 0.2662 4.422
2024-03-01 10:00:00 4.6912 0.2728 4.422
2024-03-01 11:00:00 4.6095 0.3769 3.991
rolling_max_6h temperature_c
timestamp
2024-03-01 06:00:00 4.814 5.192
2024-03-01 07:00:00 5.192 4.910
2024-03-01 08:00:00 5.192 4.422
2024-03-01 09:00:00 5.192 4.538
2024-03-01 10:00:00 5.192 3.991
2024-03-01 11:00:00 5.192 3.704这里的 accumulate 调用计算窗口的运行总和，这样我们就能在一次遍历中获得总值 — running_sum[-1] — 而无需单独调用 sum()。对于以流式方式处理的大型数据集，避免对相同数据进行冗余遍历是高效的。
#3. 使用 `product` 创建季节性交互特征
许多时间序列表现出分层季节性，其中多个时间周期相互影响 — 例如一天中的时间、一周中的天数以及更广泛的操作或周期时段。结合这些维度的交互特征可以捕捉到单独时间组件可能忽略的模式。
现在让我们使用 product 构建交互特征：
code
hours_of_day = list(range(24))
day_types = ["weekday", "weekend"]
operational_shifts = ["off_peak", "on_peak"] # on_peak: 08:00–18:00
# 为所有组合构建完整的查找网格
season_grid = list(itertools.product(hours_of_day, day_types, operational_shifts))
season_df = pd.DataFrame(season_grid, columns=["hour", "day_type", "shift"])
# 模拟每个组合的预期基准温度
np.random.seed(14)
season_df["baseline_temp_c"] = np.round(
3.5
+ 0.8 * np.sin(2 * np.pi * season_df["hour"] / 24)
+ np.where(season_df["day_type"] == "weekend", 0.3, 0.0)
+ np.where(season_df["shift"] == "on_peak", 0.5, 0.0)
+ np.random.normal(0, 0.1, len(season_df)),
3
)
print(season_df[season_df["hour"].isin([0, 8, 14, 20])].head(16).to_string(index=False))
print(f"\nTotal grid combinations: {len(season_df)}")输出：
code
hour day_type shift baseline_temp_c
0 weekday off_peak 3.655
0 weekday on_peak 4.008
0 weekend off_peak 3.817
0 weekend on_peak 4.293
8 weekday off_peak 4.325
8 weekday on_peak 4.601
8 weekend off_peak 4.446
8 weekend on_peak 4.978
14 weekday off_peak 3.370
14 weekday on_peak 3.628
14 weekend off_peak 3.279
14 weekend on_peak 3.959
20 weekday off_peak 2.726
20 weekday on_peak 3.256
20 weekend off_peak 3.056
20 weekend on_peak 3.530
Total grid combinations: 96这个网格作为 baseline_temp_c 特征合并回主数据集中的每一行 — 为每个读数提供一个上下文感知的预期值。与该基准的偏差 temperature_c - baseline_temp_c 随后成为一个有用的异常检测特征。
#4. 使用 `tee` 提取滑动窗口统计量
有时你需要同时通过多个统计视角处理相同的序列 — 均值、方差、变化率 — 而无需多次遍历它。itertools.tee 从单个源创建独立的迭代器，这正是你所需要的。
code
def sliding_window_stats(series, window_size):
"""使用 tee 计算滑动窗口上的均值、范围和变化率"""
results = []
it = iter(series)
window = list(itertools.islice(it, window_size))
if len(window) < window_size:
return results
results.append({
"window_mean": round(sum(window) / window_size, 4),
"window_range": round(max(window) - min(window), 4),
"rate_of_change": round(window[-1] - window[0], 4),
})
for next_val in it:
window = window[1:] + [next_val]
# tee 为同一窗口创建两个独立的迭代器
iter_a, iter_b = itertools.tee(iter(window))
values_a = list(iter_a)
values_b = list(iter_b)
mean_val = sum(values_a) / window_size
results.append({
"window_mean": round(mean_val, 4),
"window_range": round(max(values_b) - min(values_b), 4),
"rate_of_change": round(window[-1] - window[0], 4),
})
return results
power_readings = df["power_kw"].tolist()
stats = sliding_window_stats(power_readings, window_size=8)
stats_df = pd.DataFrame(stats, index=df.index[7:])
stats_df["power_kw"] = df["power_kw"].iloc[7:].values
print(stats_df.iloc[0:8])输出：
code
窗口均值 窗口范围 变化率 功率千瓦
时间戳
2024-03-01 07:00:00 41.4400 2.60 0.67 40.94
2024-03-01 08:00:00 43.7825 18.74 17.68 59.01
2024-03-01 09:00:00 46.1775 20.22 17.62 60.49
2024-03-01 10:00:00 47.9387 20.22 16.14 56.96
2024-03-01 11:00:00 49.9663 20.22 16.77 57.04
2024-03-01 12:00:00 52.2437 19.55 15.98 58.49
2024-03-01 13:00:00 54.3738 19.55 17.04 59.55
2024-03-01 14:00:00 56.6412 19.71 19.71 60.65如示例所示，tee 函数允许您将同一个窗口迭代器传递给两个独立的下游计算，无需手动回退或复制列表。
#5. 使用 `chain` 组合多分辨率时序特征
有用的时序特征通常同时来自多个时间分辨率：原始小时读数、6小时滚动均值、24小时滚动均值，以及像一天中小时数这样的日历特征。这些特征通常位于不同的数组中，需要整合成一个清晰的特征列表。以下是使用 chain 组合这些特征的方法：
code
humidity = df["humidity_pct"].tolist()
def rolling_means(series, window):
means = []
for i in range(len(series)):
if i < window:
means.append(None)
else:
w = list(itertools.islice(series, i - window, i))
means.append(round(sum(w) / window, 3))
return means
rolling_6h = rolling_means(humidity, 6)
rolling_24h = rolling_means(humidity, 24)
hour_of_day = df.index.hour.tolist()
is_business_hour = [1 if 8 <= h <= 18 else 0 for h in hour_of_day]
# chain 从逻辑分组子列表中组装特征名称列表
feature_names = list(itertools.chain(
["humidity_raw"],
["humidity_roll_6h", "humidity_roll_24h"],
["hour_of_day", "is_business_hour"],
))
multi_res_df = pd.DataFrame({
name: vals for name, vals in zip(
feature_names,
[humidity, rolling_6h, rolling_24h, hour_of_day, is_business_hour]
)
}, index=df.index)
print(multi_res_df.iloc[24:30])输出：
code
原始湿度 6小时滚动湿度 24小时滚动湿度 \
时间戳
2024-03-02 00:00:00 78.45 79.622 78.055
2024-03-02 01:00:00 75.63 79.105 78.100
2024-03-02 02:00:00 77.51 78.190 78.062
2024-03-02 03:00:00 76.27 78.088 78.157
2024-03-02 04:00:00 74.96 77.805 78.240
2024-03-02 05:00:00 75.75 77.208 78.203
一天中小时 是否营业时间
时间戳
2024-03-02 00:00:00 0 0
2024-03-02 01:00:00 1 0
2024-03-02 02:00:00 2 0
2024-03-02 03:00:00 3 0
2024-03-02 04:00:00 4 0
2024-03-02 05:00:00 5 0这里的 chain 函数从逻辑分组的子列表（原始传感器数据、滚动聚合数据、日历特征）中组装特征名称列表。随着特征集在更多传感器通道和更多分辨率上的扩展，chain 能保持这种组装的易读性和易扩展性。
#6. 使用 `combinations` 计算成对时序相关性
在多传感器场景中，变量间随时间变化的关系通常包含有价值的信号，这些信号是单独测量无法捕捉的。例如，两个传感器同时出现上升趋势可能揭示出新兴状态或交互作用，而这些在单独分析每个序列时并不明显。
引入反映这些联合动态的特征可以提升模型检测细微模式和依赖关系的能力。让我们使用 combinations 构建成对相关性：
code
sensor_cols = ["temperature_c", "humidity_pct", "power_kw"]
window_size = 12
pairwise_features = {}
for col_a, col_b in itertools.combinations(sensor_cols, 2):
feature_name = f"corr_{col_a[:4]}_{col_b[:4]}_12h"
correlations = []
series_a = df[col_a].tolist()
series_b = df[col_b].tolist()
for i in range(len(series_a)):
if i < window_size:
correlations.append(None)
continue
win_a = list(itertools.islice(series_a, i - window_size, i))
win_b = list(itertools.islice(series_b, i - window_size, i))
mean_a = sum(win_a) / window_size
mean_b = sum(win_b) / window_size
cov = sum((a - mean_a) * (b - mean_b) for a, b in zip(win_a, win_b)) / window_size
std_a = (sum((a - mean_a)**2 for a in win_a) / window_size) ** 0.5
std_b = (sum((b - mean_b)**2 for b in win_b) / window_size) ** 0.5
corr = round(cov / (std_a * std_b), 4) if std_a > 0 and std_b > 0 else None
correlations.append(corr)
pairwise_features[feature_name] = correlations
corr_df = pd.DataFrame(pairwise_features, index=df.index)
print(corr_df.iloc[12:18])输出：
code
温度湿度12h相关性 温度功率12h相关性 \
时间戳
2024-03-01 12:00:00 -0.6700 -0.2281
2024-03-01 13:00:00 -0.7208 -0.4960
2024-03-01 14:00:00 -0.7442 -0.6669
2024-03-01 15:00:00 -0.7678 -0.7076
2024-03-01 16:00:00 -0.8116 -0.7265
2024-03-01 17:00:00 -0.8368 -0.7482
湿度功率12h相关性
时间戳
2024-03-01 12:00:00 0.5380
2024-03-01 13:00:00 0.6614
2024-03-01 14:00:00 0.7202
2024-03-01 15:00:00 0.7311
2024-03-01 16:00:00 0.7233
2024-03-01 17:00:00 0.7219#7. 使用 `accumulate` 累积运行基线
特定数值的重要性取决于它在序列中出现的时间点。关键在于该数值与动态基线之间的差异——即该时间点之前的运行均值。使用诸如 accumulate 这样的增量计算方法，您可以高效地计算运行均值，而无需存储完整历史数据。
code
readings = df["temperature_c"].tolist()
running_sums = list(itertools.accumulate(readings))
running_counts = list(itertools.accumulate([1] * len(readings)))
running_means = [
round(s / c, 4)
for s, c in zip(running_sums, running_counts)
]
# 运行最大值——当前观测到的最高温度，适用于阈值突破追踪
running_max = list(itertools.accumulate(readings, func=max))
deviation_from_baseline = [
round(r - m, 4)
for r, m in zip(readings, running_means)
]
baseline_df = pd.DataFrame({
"temperature_c": readings,
"running_mean": running_means,
"running_max": running_max,
"deviation_from_baseline": deviation_from_baseline,
}, index=df.index)
print(baseline_df.iloc[20:28])输出结果：
code
temperature_c running_mean running_max \
timestamp
2024-03-01 20:00:00 2.960 3.5857 5.192
2024-03-01 21:00:00 2.647 3.5430 5.192
2024-03-01 22:00:00 2.986 3.5188 5.192
2024-03-01 23:00:00 2.831 3.4902 5.192
2024-03-02 00:00:00 3.409 3.4869 5.192
2024-03-02 01:00:00 3.919 3.5035 5.192
2024-03-02 02:00:00 3.833 3.5157 5.192
2024-03-02 03:00:00 4.542 3.5524 5.192
deviation_from_baseline
timestamp
2024-03-01 20:00:00 -0.6257
2024-03-01 21:00:00 -0.8960
2024-03-01 22:00:00 -0.5328
2024-03-01 23:00:00 -0.6592
2024-03-02 00:00:00 -0.0779
2024-03-02 01:00:00 0.4155
2024-03-02 02:00:00 0.3173
2024-03-02 03:00:00 0.9896#总结
时间序列特征工程的核心在于描述*上下文*——这个信号相对于我们预期行为的表现轨迹。本文介绍的每个函数都是将这个问题形式化为模型可学习特征的独特方式。
以下是本文涵盖模式的总结：
| itertools 函数 | 时间序列特征 | 示例 | | --- | --- | --- | | islice | 滞后特征 | 1小时/6小时/24小时前的温度 | | islice + accumulate | 滚动窗口统计 | 6小时均值/标准差/最小值/最大值 | | product | 季节性交互网格 | 小时×日期类型×班次基线 | | tee | 并行窗口统计 | 均值+极差+变化率 | | chain | 多分辨率特征组合 | 原始特征+滚动特征+日历特征 | | combinations | 跨传感器配对相关性 | 温度-湿度/温度-功率滚动相关性 | | accumulate | 运行基线+偏差 | 相对于历史均值的漂移检测 |
由于 itertools 工作在迭代器层级，所有这些模式都能优雅地组合成流式处理管道。祝您特征工程愉快！
[](https://twitter.com/balawc27)Bala Priya C**** 是来自印度的开发者和技术作家。她热衷于在数学、编程、数据科学和内容创作的交叉领域工作。她的兴趣专长包括 DevOps、数据科学和自然语言处理。她热爱阅读、写作、编程和咖啡！目前她正通过撰写教程、操作指南、观点文章等方式，持续学习并与开发者社区分享知识。Bala 还擅长创作引人入胜的资源概览和编程教程。