atai-newton-omega-model-data-prep

Newton 数据预处理 — 清理 → 切分 → 特征化

用于时间序列传感器数据的建模前处理流程。三个可组合的构建块接收原始多变量 DataFrame，并返回下游 KNN / Isolation Forest 分类器所需的 (X, y, metadata) 数组。

来源。 这三个引入的脚本源自 Lucas（解决方案工程团队）的工作，并已在真实的数据预处理流程中端到端使用。仓库中的副本是权威版本——如需引入更新版本，请联系 Lucas。

何时使用

• 用户有原始多变量传感器 CSV（1 个或多个传感器、时间戳不规则、存在 NaN 缺失），准备使用 atai-newton-omega-model 为数据生成嵌入，并需要先清理数据。
• 用户询问“我应该删除包含 NaN 的行吗？”、“如何处理数据缺口？”或“时间序列数据怎样划分训练集和测试集比较合适？”
• 用户的分类器表现好得可疑或差得可疑，而你怀疑存在时间泄漏——DataSplitter(mode='oot') 可以解决此问题。
• 用户有逐传感器嵌入（来自 Omega 模型），需要将其整合为单个特征矩阵以供 KNN 使用——这正是 FeaturePreparer 的用途。
• 用户想要以代码实现 atai-newton-omega-model 中介绍的“联合状态”模式。

在以下情况下，请改用外部 omega-1-4-preflight 检查：你需要在开始运行前进行只读的准入检查。Preflight 不会更改数据。本技能会进行更改——按块切分、填补缺失值、降维。两者互为补充：preflight 用于判断数据集是否可挽救；本技能用于清理数据集。

以下情况不要使用本技能：

• 任务涉及视频、图像或文本（请使用 atai-newton-fusion-model）。
• 数据已经非常干净（采样规则、没有 NaN、没有泄漏风险）——此时处理流程不会产生变化，可以直接将 DataFrame 交给下游技能。

三个构建块

1. DataPreprocessor — 诊断 + 感知缺口的分块处理

references/data_preprocessor.py

“块”是一个连续的数据段，其中所有选定的传感器都有数据，并且 NaN 连续缺失段均不超过 gap_threshold_samples。长缺口区域会被拆分为单独的块，因为跨越长缺口填补数据会凭空捏造数据。

分两阶段使用：

from data_preprocessor import DataPreprocessor

prep = DataPreprocessor(
    timestamp_col='timestamp',
    sampling_rate_minutes=1,
    gap_threshold_samples=5,        # NaN runs > 5 samples split into a new block
    imputation_method='linear',     # or 'time', 'ffill', 'bfill', 'spline' + kwargs
)

# Phase 1: diagnose (read-only). Decide gap_threshold from this.
report = prep.diagnose(df_raw, verbose=True, plot=True)

# Phase 2: build. Returns the cleaned df with added 'block_id' + 'imputed' columns.
df_clean = prep.build(df_raw)

diagnose() 返回结构化的 {'timestamp': {...}, 'nulls': {...}} 报告，涵盖：采样率 CV（规则性）、包含 (starts_at, ends_at, size) 的前 N 个缺口、每列的空值计数，以及每列的最长连续空值段——这是选择 gap_threshold_samples 时最重要的数值。设置 plot=True 还会生成差值直方图和缺失情况图。

build() 会在输出中添加两列：

• block_id（int）：该行所属的连续数据块
• imputed（bool）：如果该行中的任一传感器原本为 NaN，则为 True

长缺口区域内的行会被完全丢弃（不进行填补），因为填补会跨越阈值。

重要参数：

• gap_threshold_samples：根据 _diagnose_nulls 的 max_consecutive_null 设置。保守的默认值为 5；对于噪声较大的传感器，可以调高，但切勿高到跨越真实的中断。
• imputation_method：'linear' 是合适的默认值。如果采样并非完全规则，请使用 'time'。仅当传感器类似分类数据、不应进行插值时，才使用 'ffill' / 'bfill'。
• drop_sensors：任何已知不可靠或与客户无关的传感器。应在缺口检测之前删除，这样这些传感器的 NaN 就不会造成不必要的分块。

2. DataSplitter — 按时间外推划分训练集/测试集

references/data_splitter.py

时间序列交叉验证做错了会带来虚假的信心。随机划分会让未来信息泄漏到训练集中；OOT 划分则不会。

from data_splitter import DataSplitter

splitter = DataSplitter(mode='oot', test_size=0.3, timestamp_column='timestamp')
X_train, X_test, y_train, y_test, meta_train, meta_test = splitter.split(X, y, metadata)

默认使用 mode='oot'。仅将 mode='random' 用作合理性检查（“我到底有没有学到任何东西？”）——绝不要将其用作生产划分方式。

对于无监督任务（例如 Isolation Forest），y 可以是 None，此时返回的 y_train / y_test 也会是 None。

3. FeaturePreparer — 联合状态特征化

references/feature_preparer.py

将按传感器、按窗口生成的嵌入数据框转换为单个 (X, y, metadata) 矩阵，其中每一行表示给定窗口中所有传感器的联合状态（拼接后的嵌入——在 Omega 笔记中有时称为“philosophy 2”）。可选进行 L2 归一化、标准化和/或 PCA 降维。

from feature_preparer import FeaturePreparer

preparer = FeaturePreparer(
    normalize='l2',          # or 'standardize', or None
    reduce_dim=50,           # or None to skip PCA
    sensor_order=None,       # None = alphabetical, fixed across train/test
)

# df_emb is the output of an Omega embedding step
# (see atai-newton-omega-model for how to produce it).
X, y, metadata = preparer.prepare(df_emb, label_column='machine_state')

重要规则：

• 训练集和测试集的 sensor_order 必须相同。如果使用默认的字母顺序，就会保持一致；但有任何疑问时，请明确指定。
• 同一窗口的各传感器之间，每窗口标签必须一致。发现不一致时，类会报错。
• reduce_dim 必须小于 D 且小于 N。对于小型数据集，请跳过 PCA。
• 下游使用余弦类距离的 KNN 时，normalize='l2' 是合适的默认值；下游使用度量模型或线性模型时，则使用 'standardize'。

完整工作流

# 1. Diagnose, then clean
prep = DataPreprocessor(timestamp_col='timestamp', sampling_rate_minutes=1, gap_threshold_samples=5)
report = prep.diagnose(df_raw, verbose=True)
df_clean = prep.build(df_raw)

# 2. Embed (see atai-newton-omega-model) — one row per (sensor, window) with an embedding column.
df_emb = embedding_generator.generate(
    df_clean, data_columns=['sensor_a', 'sensor_b'], label_columns=['machine_state'],
)

# 3. Featurize into (X, y, metadata)
preparer = FeaturePreparer(normalize='l2', reduce_dim=50)
X, y, metadata = preparer.prepare(df_emb, label_column='machine_state')

# 4. Split out-of-time
splitter = DataSplitter(mode='oot', test_size=0.3)
X_train, X_test, y_train, y_test, meta_train, meta_test = splitter.split(X, y, metadata)

# 5. Hand to your classifier of choice (KNN / IsolationForest).

与其他技能组合使用

技能	关系
atai-newton-omega-model	同级 + 下游 — Omega 嵌入步骤位于 DataPreprocessor.build() 和 FeaturePreparer.prepare() 之间，随后使用生成的 (X, y, metadata)，针对 /query Omega 嵌入执行逐窗口 KNN。本技能是其入口（清理）和出口（特征化）。
atai-operational-state-monitoring-agent-data-prep	用于托管 OSM Agent 的同级技能 — 要将数据提供给 OSM Agent，而不是本地 KNN？该技能会将带标签的记录转换为 OSM 平台接受的角色文件（精确网格、每种状态一个训练文件、连续评分文件，以及对平台规则的预检）。如有需要，请先在这里清理杂乱的原始数据；FeaturePreparer 在该流程中没有作用（平台会生成嵌入）。
omega-1-4-preflight	上游检查（外部仓库） — 只读静态检查。在使用本技能之前运行，以判断数据集是否值得清理。

常见问题

• 对时间序列进行随机划分。 几乎总是错误的。使用 mode='oot'。例外情况：消融实验中，你明确希望忽略时间顺序，以隔离非时间因素的影响。
• 跨长间隔进行插补。 不要为了“增加数据”而调高 gap_threshold_samples——这等于伪造真实情况。如果间隔长到插值不再可信，应将该区域单独划为一个区块（或直接丢弃）。
• 每窗口标签不一致。 如果 FeaturePreparer.prepare() 因标签不一致而报错，问题出在上游：你在嵌入步骤中传播标签所用的策略（'last' / 'first' / 'mode'）与原始行的标签分配方式不匹配。应在上游解决，而不是关闭检查。
• 训练集和测试集使用不同的 sensor_order。 这会悄悄破坏预测，因为拼接后的联合状态向量顺序发生了变化。请明确指定 sensor_order。
• 在小型数据集上使用 PCA。 PCA 要求样本数和维度都大于组件数。样本数约少于 100 时请跳过。

本地设置

# In a virtualenv (system pythons are often PEP 668 "externally managed"):
python3 -m venv .venv && source .venv/bin/activate
pip install -r skills/atai-newton-omega-model-data-prep/references/requirements.txt
pip install matplotlib  # optional: only for plot=True in diagnose()

无需 Newton API 密钥——本技能只进行本地数据整理。

文件布局

skills/atai-newton-omega-model-data-prep/
├── SKILL.md                              ← this file
├── references/
│   ├── requirements.txt                  ← runtime deps (pip install -r requirements.txt)
│   ├── data_preprocessor.py              ← gap-aware blocking + diagnostics
│   ├── data_splitter.py                  ← OOT / random train/test split
│   └── feature_preparer.py               ← joint-state featurization + PCA
└── tests/
    ├── conftest.py                       ← shared fixtures
    ├── test_data_preprocessor.py
    ├── test_data_splitter.py
    └── test_feature_preparer.py

运行测试

pip install -r skills/atai-newton-omega-model-data-prep/references/requirements.txt pytest
pytest skills/atai-newton-omega-model-data-prep/tests/ -v

CI 会在任何修改 skills/atai-newton-omega-model-data-prep/** 的 PR 上自动运行测试套件（参见 .github/workflows/test-atai-newton-omega-model-data-prep.yml）。