aeon

aeon

概述

aeon 为时间序列机器学习任务提供统一且兼容 scikit-learn 的 API，涵盖分类、回归、聚类、分段、标注、相似性搜索和转换。它遵循与 scikit-learn 相同的 fit(X, y) / predict(X) 模式，其中 X 是形状为 (n_instances, n_channels, n_timepoints) 的 3D NumPy 数组。aeon 汇集了时间序列领域文献中的先进算法，包括用于分类的 ROCKET 及其变体（MiniROCKET、MultiROCKET）、用于聚类的基于 DTW 的 k-means，以及用于分段的 CLASP；同时提供基准测试工具，用于跨数据集比较算法。它是在 2022 年治理分拆后由社区维护的 sktime 分支。

适用场景

• 使用先进算法对 ECG、EEG、加速度计或传感器时间序列进行分类
• 根据时间序列输入预测连续标量目标（例如，根据生命体征波形预测患者病情严重程度）
• 在没有类别标签时，按形状相似性对时间序列进行聚类
• 检测变化点，或将连续记录分割为同质区间
• 从可变长度的时间序列中提取固定长度特征向量，用于后续机器学习
• 在 UCR/UEA 存档上对时间序列算法进行基准测试，并开展可复现的比较
• 需要更大的生态系统或现有代码依赖其 API 时使用 sktime；专注于 DTW 的工作则使用 tslearn

先决条件

• Python 包：aeon、numpy、scikit-learn、matplotlib
• 数据格式：3D NumPy 数组 (n_instances, n_channels, n_timepoints)；单变量数据也可使用 2D 数组 (n_instances, n_timepoints)
• 可选项：numba（ROCKET、DTW 所必需——通过 pip install aeon[all_extras] 安装）

pip install aeon
pip install aeon[all_extras]  # includes numba, statsmodels for full algorithm support

快速入门

import numpy as np
from aeon.classification.convolution_based import RocketClassifier
from aeon.datasets import load_unit_test

# Load a small benchmark dataset
X_train, y_train = load_unit_test(split="train")  # shape: (n, 1, timepoints)
X_test, y_test = load_unit_test(split="test")

print(f"Train: {X_train.shape}, classes: {np.unique(y_train)}")

clf = RocketClassifier(num_kernels=500, random_state=42)
clf.fit(X_train, y_train)
accuracy = clf.score(X_test, y_test)
print(f"ROCKET accuracy: {accuracy:.3f}")

核心 API

模块 1：时间序列分类

涵盖卷积、字典、距离、特征、区间和形状片段系列的 30 多种分类器。

import numpy as np
from aeon.datasets import load_unit_test
from aeon.classification.convolution_based import RocketClassifier, MiniRocketClassifier
from aeon.classification.distance_based import KNeighborsTimeSeriesClassifier
from aeon.classification.feature_based import Catch22Classifier
from sklearn.metrics import accuracy_score

X_train, y_train = load_unit_test(split="train")
X_test, y_test   = load_unit_test(split="test")

classifiers = {
    "ROCKET":    RocketClassifier(num_kernels=1000, random_state=42),
    "MiniROCKET": MiniRocketClassifier(random_state=42),
    "KNN-DTW":   KNeighborsTimeSeriesClassifier(n_neighbors=1, distance="dtw"),
    "Catch22":   Catch22Classifier(random_state=42),
}

for name, clf in classifiers.items():
    clf.fit(X_train, y_train)
    acc = accuracy_score(y_test, clf.predict(X_test))
    print(f"{name:15s}: {acc:.3f}")

# Multivariate classification (multiple channels)
from aeon.classification.convolution_based import MultiRocketMultivariateClassifier

# Synthetic multivariate time series: 100 instances, 3 channels, 50 timepoints
np.random.seed(0)
X_mv_train = np.random.randn(100, 3, 50)
y_mv_train = (X_mv_train[:, 0, :].mean(axis=1) > 0).astype(str)
X_mv_test  = np.random.randn(30, 3, 50)
y_mv_test  = (X_mv_test[:, 0, :].mean(axis=1) > 0).astype(str)

clf_mv = MultiRocketMultivariateClassifier(random_state=42)
clf_mv.fit(X_mv_train, y_mv_train)
print(f"Multivariate accuracy: {clf_mv.score(X_mv_test, y_mv_test):.3f}")

模块 2：时间序列回归

根据时间序列输入预测连续目标。

import numpy as np
from aeon.regression.convolution_based import RocketRegressor
from aeon.regression.distance_based import KNeighborsTimeSeriesRegressor
from sklearn.metrics import mean_squared_error

# Synthetic regression: predict the mean of each series
np.random.seed(42)
X_train = np.random.randn(200, 1, 100)
y_train = X_train[:, 0, :].mean(axis=1) + np.random.randn(200) * 0.1
X_test  = np.random.randn(50, 1, 100)
y_test  = X_test[:, 0, :].mean(axis=1) + np.random.randn(50) * 0.1

reg = RocketRegressor(num_kernels=500, random_state=42)
reg.fit(X_train, y_train)
y_pred = reg.predict(X_test)
mse = mean_squared_error(y_test, y_pred)
print(f"ROCKET regressor MSE: {mse:.4f}")

模块 3：时间序列聚类

在没有类别标签的情况下，按形状相似性对时间序列分组。

import numpy as np
from aeon.clustering.k_means import TimeSeriesKMeans
from aeon.clustering.k_medoids import TimeSeriesKMedoids

# Synthetic clustering dataset: 3 distinct shapes
np.random.seed(0)
n_per_class = 30
class_0 = np.sin(np.linspace(0, 2*np.pi, 50)) + np.random.randn(n_per_class, 1, 50)*0.1
class_1 = np.cos(np.linspace(0, 2*np.pi, 50)) + np.random.randn(n_per_class, 1, 50)*0.1
class_2 = np.linspace(0, 1, 50) + np.random.randn(n_per_class, 1, 50)*0.05
X = np.concatenate([class_0, class_1, class_2], axis=0)

# K-Means with DTW averaging
km = TimeSeriesKMeans(n_clusters=3, metric="dtw", averaging_method="ba",
                       random_state=42, n_init=3, max_iter=50)
labels = km.fit_predict(X)
print(f"Cluster sizes: {np.bincount(labels)}")
print(f"Cluster centers shape: {km.cluster_centers_.shape}")

模块 4：分段与变化点检测

检测连续时间序列中不同状态之间的边界。

import numpy as np
from aeon.segmentation import ClaSPSegmenter, EAggloSegmenter

# Simulate signal with 3 segments
np.random.seed(0)
seg1 = np.random.randn(100) * 0.5 + 0
seg2 = np.random.randn(100) * 0.5 + 3
seg3 = np.random.randn(100) * 0.5 - 2
signal = np.concatenate([seg1, seg2, seg3])

# CLASP: Classification Score Profile segmentation
clasp = ClaSPSegmenter(period_length=10, n_change_points=2)
labels = clasp.fit_predict(signal)
change_points = clasp.change_points_
print(f"Detected change points: {change_points}")
# Expected near indices 100 and 200

模块 5：转换（特征提取）

将时间序列转换为固定长度的特征向量或变换后的序列。

import numpy as np
from aeon.transformations.collection.convolution_based import Rocket, MiniRocket
from aeon.transformations.collection.feature_based import Catch22
from sklearn.linear_model import RidgeClassifierCV
from sklearn.pipeline import Pipeline

np.random.seed(42)
X_train = np.random.randn(100, 1, 50)
y_train = (X_train[:, 0, :].mean(axis=1) > 0).astype(str)
X_test  = np.random.randn(30, 1, 50)

# ROCKET transform → Ridge classifier (the classic ROCKET pipeline)
rocket_pipe = Pipeline([
    ("rocket", Rocket(num_kernels=1000, random_state=42)),
    ("clf", RidgeClassifierCV(alphas=[0.1, 1.0, 10.0])),
])
rocket_pipe.fit(X_train, y_train)
print(f"ROCKET pipeline accuracy: {rocket_pipe.score(X_test, (X_test[:, 0, :].mean(axis=1) > 0).astype(str)):.3f}")

# Catch22: 22 canonical time series features per channel
catch22 = Catch22()
X_features = catch22.fit_transform(X_train)
print(f"Catch22 feature matrix: {X_features.shape}")  # (n_instances, 22)

模块 6：数据集加载与基准测试

from aeon.datasets import load_classification, load_from_tsf_file
from aeon.benchmarking.results_loaders import get_estimator_results_as_array
import numpy as np

# Load UCR/UEA dataset
X_train, y_train = load_classification("BasicMotions", split="train")
X_test, y_test   = load_classification("BasicMotions", split="test")
print(f"BasicMotions: train={X_train.shape}, classes={np.unique(y_train)}")

# Load custom .ts file (UCR format)
# X, y = load_from_tsf_file("my_dataset.ts")

# Compare to published benchmark results
# from aeon.benchmarking import plot_critical_difference_diagram

核心概念

数据格式约定

aeon 使用形状为 (n_instances, n_channels, n_timepoints) 的 3D NumPy 数组：

• n_instances：时间序列的数量（类似 sklearn 中的 n_samples）
• n_channels：每个时间序列的变量数（单变量为 1）
• n_timepoints：每个序列的长度（长度不等时，不同实例的长度可能不同）

Scikit-learn 使用 2D 格式 (n_samples, n_features)。使用 aeon 的转换器进行转换。

ROCKET 系列

ROCKET（随机卷积核变换）会随机生成 10,000 个长度、膨胀率和偏置各异的卷积核，然后将其应用于每个序列，以提取 PPV（正值占比）和最大值特征。随后使用线性模型对这 20,000 个特征进行分类。MiniROCKET 使用数量更少且固定的卷积核，在准确率相近的情况下，速度约为 ROCKET 的 75 倍。MultiROCKET 将其扩展到多变量序列。

常见工作流

工作流 1：交叉验证分类器比较

import numpy as np
from aeon.datasets import load_unit_test
from aeon.classification.convolution_based import RocketClassifier, MiniRocketClassifier
from aeon.classification.feature_based import Catch22Classifier
from sklearn.model_selection import cross_val_score

X, y = load_unit_test()  # Full dataset (train+test merged)

classifiers = {
    "ROCKET":     RocketClassifier(num_kernels=500, random_state=42),
    "MiniROCKET": MiniRocketClassifier(random_state=42),
    "Catch22":    Catch22Classifier(random_state=42),
}

print(f"Dataset: {X.shape}, classes: {np.unique(y)}")
for name, clf in classifiers.items():
    scores = cross_val_score(clf, X, y, cv=5, scoring="accuracy")
    print(f"{name:15s}: {scores.mean():.3f} ± {scores.std():.3f}")

工作流 2：包含预处理和分类的管道

import numpy as np
from aeon.datasets import load_unit_test
from aeon.transformations.collection.convolution_based import Rocket
from aeon.transformations.collection.normalize import TimeSeriesScaler
from sklearn.pipeline import Pipeline
from sklearn.linear_model import RidgeClassifierCV
from sklearn.preprocessing import StandardScaler

X_train, y_train = load_unit_test(split="train")
X_test, y_test   = load_unit_test(split="test")

# Normalize series → ROCKET features → Ridge
pipe = Pipeline([
    ("normalize", TimeSeriesScaler()),
    ("rocket",    Rocket(num_kernels=2000, random_state=42)),
    ("scale",     StandardScaler(with_mean=False)),
    ("clf",       RidgeClassifierCV(alphas=[0.01, 0.1, 1.0, 10.0])),
])

pipe.fit(X_train, y_train)
print(f"Pipeline accuracy: {pipe.score(X_test, y_test):.3f}")

关键参数

参数	模块/类	默认值	范围 / 选项	作用
num_kernels	Rocket, RocketClassifier	10000	500–50000	随机核的数量；数量越多，准确率越高，但速度越慢
n_clusters	TimeSeriesKMeans	8	2–50	K-Means 的簇数
metric	TimeSeriesKMeans, KNN	"dtw"	"dtw", "euclidean", "msm", "twe"	用于相似度计算的距离度量
n_change_points	ClaSPSegmenter	1	1–20	预期检测到的变点数量
period_length	ClaSPSegmenter	auto	5–100	最小分段长度
n_neighbors	KNeighborsTimeSeriesClassifier	1	1–20	KNN 中的 k 值；对于时间序列，k=1 通常效果最好
averaging_method	TimeSeriesKMeans	"ba"	"ba"（重心），"mean"	计算簇原型的方法
random_state	所有随机算法	None	int	用于确保结果可复现的种子

常用配方

配方：将 ROCKET 特征导出到 DataFrame 以用于 AutoML

import numpy as np
import pandas as pd
from aeon.transformations.collection.convolution_based import MiniRocket

np.random.seed(0)
X = np.random.randn(200, 1, 100)  # 200 univariate time series, length 100
y = (X[:, 0, :].mean(axis=1) > 0).astype(int)

transformer = MiniRocket(random_state=42)
X_features = transformer.fit_transform(X)  # shape: (200, 9996)

df = pd.DataFrame(X_features, columns=[f"f{i}" for i in range(X_features.shape[1])])
df["label"] = y
df.to_parquet("rocket_features.parquet", index=False)
print(f"Feature matrix: {df.shape}, saved to rocket_features.parquet")

配方：加载并分类自定义数据集

import numpy as np
from aeon.classification.convolution_based import RocketClassifier

# Custom dataset: time series stored as 3D array
# Shape: (n_instances, n_channels, n_timepoints)
np.random.seed(42)
X_train = np.random.randn(150, 2, 80)  # 150 bivariate series, length 80
y_train = np.random.choice(["class_A", "class_B", "class_C"], 150)
X_test  = np.random.randn(50, 2, 80)
y_test  = np.random.choice(["class_A", "class_B", "class_C"], 50)

clf = RocketClassifier(num_kernels=2000, random_state=42)
clf.fit(X_train, y_train)
print(f"Test accuracy: {clf.score(X_test, y_test):.3f}")

# Save predictions
y_pred = clf.predict(X_test)
print(f"Predictions: {y_pred[:10]}")

故障排除

问题	原因	解决方案
使用 ROCKET 时出现 ImportError: numba	未安装 numba	pip install aeon[all_extras] 或 pip install numba
输入形状错误：期望 3D 数组	传入了 2D (n, t)，而不是 3D (n, 1, t)	重塑形状：X = X[:, np.newaxis, :]
KNN-DTW 在大型数据集上速度很慢	DTW 对每一对样本的计算复杂度为 O(n·m)；全对比较的复杂度为 O(N²)	当 N>1000 时改用 ROCKET；或缩短序列长度
聚类将所有点分配到同一个簇	初始质心过于接近，或度量不合适	设置 n_init=10；尝试其他 metric（"msm" 或 "euclidean"）
load_classification 下载失败	网络超时或数据集名称拼写错误	检查可用数据集：aeon.datasets.list_datasets()；手动下载
ClaSPSegmenter 检测到的变点数量不正确	n_change_points 过高或 period_length 设置不当	将 period_length 设为预期的最小分段长度；使用已知信号进行验证
大型面板数据集出现内存错误	ROCKET 转换需要大小为 n_kernels × n_instances 的矩阵	减少 num_kernels；使用 partial_fit 分批处理

相关技能

• neurokit2 — 生成可输入 aeon 分类器的时间序列（ECG、EDA）
• scikit-learn-machine-learning — 用于处理 aeon 特征的 sklearn 兼容下游模型
• matplotlib-scientific-plotting — 时间序列和分类结果的可视化

参考资料

• aeon 文档 — API 参考、算法目录和教程
• aeon GitHub — 源代码、基准测试和贡献指南
• ROCKET 论文：Dempster 等（2020），DMKD — ROCKET 算法说明和 UCR 基准测试结果
• MiniROCKET 论文：Dempster 等（2021），KDD — 准确率相当的更快变体
• UCR 时间序列档案 — 包含 128 个用于时间序列分类的基准数据集