本文將簡要介紹三種常見的異常值以及對應的偵測策略。然後將提供基於兩個支援的 API 的範例程式碼：用於開發時間序列異常值檢測管道的 TODS API 和用於使用第三方套件進行實驗的 scikit-learn API。
時間序列異常值偵測旨在識別資料中意外或罕見的實例。作為資料分析最重要的任務之一，異常值偵測在時間序列資料上有多種應用，例如詐欺偵測、故障偵測和網路安全攻擊偵測。例如，雅虎 [1] 和微軟 [2] 已經建立了自己的時間序列異常值檢測服務來監控他們的業務資料並觸發異常值警報。在時間序列資料上，異常值可分為三種情況：逐點異常值、模式（集體）異常值、系統異常值。
TODS [3] 是一個全端機器學習系統，用於對多元時間序列資料進行異常值偵測。 TODS 為建構基於機器學習的異常值偵測系統提供了詳盡的模組，包括：資料處理、時間序列處理、特徵分析、偵測演算法和強化模組。透過這些模組提供的功能包括：通用資料預處理、時間序列資料平滑/轉換、從時域/頻域中提取特徵、各種檢測演算法，以及涉及人類專業知識來校準系統。可以對時間序列資料執行三種常見的異常值偵測場景：逐點偵測（時間點作為異常值）、模式偵測（子序列為異常值）和系統偵測（時間序列集作為異常值）。
當時間序列中存在潛在的系統故障或小故障時，通常會出現逐點異常值。這種異常值存在於全域（與整個時間序列中的資料點相比）或局部（與相鄰點相比）的單一資料點上。全域異常值通常很明顯，檢測全域異常值的常見做法是取得資料集的統計值（例如，最小值/最大值/平均值/標準差）並設定檢測異常點的閾值。局部異常值通常出現在特定上下文中，具有相同值的資料點如果不在特定上下文中顯示，則不會被識別為異常值。偵測局部異常值的常用策略是識別情境（透過季節性趨勢分解、自相關），然後應用統計/機器學習方法（例如 AutoRegression、IsolationForest、OneClassSVM）來偵測異常值。
當資料中存在異常行為時，通常會出現模式異常值。模式異常值是指與其他子序列相比其行為異常的時間序列資料的子序列（連續點）。檢測模式異常值的常見做法，包括不和諧分析（例如，矩陣配置檔 [6]、HotSAX [7]）和子序列聚類 [4]。 Discords 分析利用滑動視窗將時間序列分割成多個子序列，並計算子序列之間的距離（例如，歐幾里德距離）以找到時間序列資料中的不一致。子序列聚類也將子序列分割應用於時間序列數據，並採用子序列作為每個時間點的特徵，其中滑動視窗的大小為特徵的數量。然後，採用無監督機器學習方法，例如聚類（例如，KMeans、PCA）或逐點異常值檢測演算法來檢測模式異常值。
當許多系統之一處於異常狀態時，系統異常值會不斷發生，其中系統被定義為多元時間序列資料。偵測系統異常值的目標是從許多類似的系統中找出處於異常狀態的系統。例如，從具有多條生產線的工廠檢測異常生產線。檢測這種異常值的常用方法是執行逐點和模式異常值檢測以獲得每個時間點/子序列的異常值分數，然後採用整合技術為每個系統產生整體異常值分數以進行比較和檢測。
在建構機器學習管道的開始，需要大量實驗來調整或分析演算法。在 TODS 中，Scikit-learn 類似 API 可用於大多數模組，讓使用者靈活地將單一函數呼叫到實驗腳本中。這是一個呼叫矩陣設定檔的範例，用於使用 UCR 資料集識別模式異常值 [5]。
# !pip install -e git+https://github.com/datamllab/tods.git#egg=tods
import numpy as np
from tods.sk_interface.detection_algorithm.MatrixProfile_skinterface import MatrixProfileSKI
from sklearn.metrics import precision_recall_curve
from sklearn.metrics import accuracy_score
from sklearn.metrics import confusion_matrix
from sklearn.metrics import classification_report
# 数据准备
data = np.loadtxt("./500_UCR_Anomaly_robotDOG1_10000_19280_19360.txt")
X_train = np.expand_dims(data[:10000], axis=1)
X_test = np.expand_dims(data[10000:], axis=1)
transformer = MatrixProfileSKI()
transformer.fit(X_train)
prediction_labels_train = transformer.predict(X_train)
prediction_labels = transformer.predict(X_test)
prediction_score = transformer.predict_score(X_test)
y_true = prediction_labels_train
y_pred = prediction_labels
print('Accuracy Score: ', accuracy_score(y_true, y_pred))
confusion_matrix(y_true, y_pred)
print(classification_report(y_true, y_pred))
結果如下：
Accuracy Score: 0.89
precision recall f1-score support
0 0.90 0.98 0.94 9005
1 0.21 0.04 0.06 995
accuracy 0.89 10000
macro avg 0.55 0.51 0.50 10000
weighted avg 0.83 0.89 0.85 10000
在管道探索的后期阶段，需要在没有开发工作的情况下以可重复的方式管理实验，因为会有更多的超参数和组件组合。在 TODS 中，我们的管道构建和执行 API 允许用户使用单个脚本生成各种可重现的管道。生成的管道将存储为 .json 或 .yml 文件等类型的描述文件，这些文件可以轻松地使用不同的数据集进行复制/执行以及共享给同事。下面的示例利用 TODS API 以 .json 格式建立自动编码器管道，并使用 TODS 后端引擎运行管道以检测雅虎网络入侵数据集中的点异常值 [1]。
管道生成脚本提供如下。虽然它看起来比 Scikit-learn 界面更长，但用户可以轻松地添加带有候选的超参数。
from d3m import index
from d3m.metadata.base import ArgumentType
from d3m.metadata.pipeline import Pipeline, PrimitiveStep
# Creating pipeline
pipeline_description = Pipeline()
pipeline_description.add_input(name='inputs')
# Step 0: dataset_to_dataframe
step_0 = PrimitiveStep(primitive=index.get_primitive('d3m.primitives.tods.data_processing.dataset_to_dataframe'))
step_0.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='inputs.0')
step_0.add_output('produce')
pipeline_description.add_step(step_0)
# Step 1: column_parser
step_1 = PrimitiveStep(primitive=index.get_primitive('d3m.primitives.tods.data_processing.column_parser'))
step_1.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.0.produce')
step_1.add_output('produce')
pipeline_description.add_step(step_1)
# Step 2: extract_columns_by_semantic_types(attributes)
step_2 = PrimitiveStep(primitive=index.get_primitive('d3m.primitives.tods.data_processing.extract_columns_by_semantic_types'))
step_2.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.1.produce')
step_2.add_output('produce')
step_2.add_hyperparameter(name='semantic_types', argument_type=ArgumentType.VALUE,
data=['[https://metadata.datadrivendiscovery.org/types/Attribute](https://link.zhihu.com/?target=https%3A//metadata.datadrivendiscovery.org/types/Attribute)'])
pipeline_description.add_step(step_2)
# Step 3: extract_columns_by_semantic_types(targets)
step_3 = PrimitiveStep(primitive=index.get_primitive('d3m.primitives.tods.data_processing.extract_columns_by_semantic_types'))
step_3.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.0.produce')
step_3.add_output('produce')
step_3.add_hyperparameter(name='semantic_types', argument_type=ArgumentType.VALUE,
data=['[https://metadata.datadrivendiscovery.org/types/TrueTarget](https://link.zhihu.com/?target=https%3A//metadata.datadrivendiscovery.org/types/TrueTarget)'])
pipeline_description.add_step(step_3)
attributes = 'steps.2.produce'
targets = 'steps.3.produce'
# Step 4: processing
step_4 = PrimitiveStep(primitive=index.get_primitive('d3m.primitives.tods.timeseries_processing.transformation.axiswise_scaler'))
step_4.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference=attributes)
step_4.add_output('produce')
pipeline_description.add_step(step_4)
# Step 5: algorithm
step_5 = PrimitiveStep(primitive=index.get_primitive('d3m.primitives.tods.detection_algorithm.pyod_ae'))
step_5.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.4.produce')
step_5.add_output('produce')
pipeline_description.add_step(step_5)
# Step 6: Predictions
step_6 = PrimitiveStep(primitive=index.get_primitive('d3m.primitives.tods.data_processing.construct_predictions'))
step_6.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.5.produce')
step_6.add_argument(name='reference', argument_type=ArgumentType.CONTAINER, data_reference='steps.1.produce')
step_6.add_output('produce')
pipeline_description.add_step(step_6)
# Final Output
pipeline_description.add_output(name='output predictions', data_reference='steps.6.produce')
# Output to json
data = pipeline_description.to_json()
with open('autoencoder_pipeline.json', 'w') as f:
f.write(data)
print(data)
创建管道描述文件后，我们可以按如下方式运行管道描述文件并评估无监督管道：
import sys
import argparse
import os
import pandas as pd
from tods import generate_dataset, load_pipeline, evaluate_pipeline
this_path = os.path.dirname(os.path.abspath(__file__))
table_path = os.path.join(this_path, 'yahoo_sub_5.csv') # file path to the dataset
target_index = 6 # which column is the label
pipeline_path = "./autoencoder_pipeline.json"
metric = "ALL"
# Read data and generate dataset
df = pd.read_csv(table_path)
dataset = generate_dataset(df, target_index)
# Load the default pipeline
pipeline = load_pipeline(pipeline_path)
# Run the pipeline
pipeline_result = evaluate_pipeline(dataset, pipeline, metric)
print(pipeline_result.scores)
metricvalue normalized randomSeed fold
0 F1_MACRO 0.509059 0.509059 00
虽然这个API需要一个脚本来生成管道描述文件，但它提供了灵活的接口来生成多个管道。
除了手动创建管道之外，TODS 还利用 TODS API 提供自动模型发现。自动模型发现的目标旨在根据验证集中的标签信息和给定的计算时间限制搜索最佳管道。
import pandas as pd
from axolotl.backend.simple import SimpleRunner
from tods import generate_dataset, generate_problem
from tods.searcher import BruteForceSearch
table_path = 'yahoo_sub_5.csv'
target_index = 6 # what column is the target
time_limit = 30 # How many seconds you wanna search
metric = 'F1_MACRO' # F1 on label 1
# Read data and generate dataset and problem
df = pd.read_csv(table_path)
dataset = generate_dataset(df, target_index=target_index)
problem_description = generate_problem(dataset, metric)
# Start backend
backend = SimpleRunner(random_seed=0)
# Start search algorithm
search = BruteForceSearch(problem_description=problem_description,
backend=backend)
# Find the best pipeline
best_runtime, best_pipeline_result = search.search_fit(input_data=[dataset], time_limit=time_limit)
best_pipeline = best_runtime.pipeline
best_output = best_pipeline_result.output
# Evaluate the best pipeline
best_scores = search.evaluate(best_pipeline).scores
print('Search History:')
for pipeline_result in search.history:
print('-' * 52)
print('Pipeline id:', pipeline_result.pipeline.id)
print(pipeline_result.scores)
print('Best pipeline:')
print('-' * 52)
print('Pipeline id:', best_pipeline.id)
print('Pipeline json:', best_pipeline.to_json())
print('Output:')
print(best_output)
print('Scores:')
print(best_scores)
管道搜索完成后，用户可以通过管道id访问所有搜索到的管道，并保存任何管道描述文件以供后续使用。
该项目团队正在为该项目积极开发更多功能，包括带有可视化工具的图形用户界面、半监督学习算法和高级管道搜索器。目标是使时间序列数据的异常值检测变得可访问且更容易。我希望你喜欢阅读这篇文章，在接下来的文章中，我将详细介绍在时间序列数据中检测不同类型异常值的常见策略，并介绍 TODS 中具有合成标准的数据合成器。
以上是用於時間序列異常值檢測的全端機器學習系統的詳細內容。更多資訊請關注PHP中文網其他相關文章！
