ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

TDgpt Anomaly Detection Algorithm Development Guide: Building Custom AD Services for TDengine

TDgpt Anomaly Detection Algorithm Development Guide: Building Custom AD Services for TDengine TDgpt Anomaly Detection Algorithm Development Guide: Building Custom AD Services for TDengine【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengineTDgpt 是 TDengine 内置的 AI 分析服务anode它以ANOMALY_WINDOW等 SQL 函数的形式为时序数据提供异常检测、预测、填补与相关性分析能力。本文以官方开发文档为主体结合仓库源码完整讲解如何基于AbstractAnomalyDetectionService抽象基类编写一个可被taosanode动态加载、可通过 SQL 调用的自定义异常检测算法从框架的输入/输出契约、PyOD 第三方检测器适配到算法类骨架、部署注册、SHOW ANODES FULL验证与单元测试一文打通写算法 → 部署 → SQL 调用的完整闭环。一、异常检测算法在 TDgpt 中的定位在 TDengine 集群中taosanodeTDgpt 分析节点独立于taosd运行负责承载各类机器学习与统计分析算法。其核心交互方式是TDengine SQL 前端将查询拆解为请求通过 HTTP 将时序数据以 JSON 载荷发给 anodeanode 中的 Flask 服务完成参数校验、白噪声检查后把数据交给对应的算法服务执行最后将结果以异常窗口[起始时间, 结束时间]的形式返回给 SQL 层。从源码结构看所有异常检测算法的实现都存放在 tools/tdgpt/taosanalytics/algo/ad/ 目录下仓库自带的算法包括算法名实现文件原理概述grubbsgrubbs.pyGrubbs 检验面向近似正态分布数据检测单个离群点iqriqr.py基于四分位距Inter-Quartile Range判定异常ksigmaksigma.pyk-sigma / 3σ 法则默认k3loflof.py基于密度的局部离群因子Local Outlier Factorecod/hbos/copod/iforest/pcapyod_stat.py基于 PyOD 库包装的统计与集成检测器shesdshesd.py面向长周期序列的季节性混合 ESD 检验这些算法统一继承自AbstractAnomalyDetectionService定义于 tools/tdgpt/taosanalytics/base.py并由 service_registry.py 中的ServiceRegistry单例在启动时扫描注册。自定义算法与内置算法走的是同一条注册链路唯一的区别是内置算法标记了_builtins True。二、框架契约输入与输出限制编写任何异常检测算法的前提是理解框架对算法实例的输入注入方式与对返回值的要求。这两条契约由AbstractAnomalyDetectionService统一保证。2.1 输入self.list与self.input_data_listsexecute是异常检测算法的核心方法。框架在调用execute之前会先把用于异常检测的历史时序数据配置到成员变量self.list中通过set_input_list(input_list, ts_list)注入ts_list为对应时间戳。自v3.4.1.0起所有继承AbstractAnomalyDetectionService的异常检测类新增了一个 Python 成员变量self.input_data_lists。它是一个嵌套数组array of arrays用于存储多列输入数据的所有列。为保持向后兼容框架仍将self.list保留为第一输入列而self.input_data_lists则包含每一个输入列单列逻辑或需要向后兼容的实现中使用self.list算法需要访问多列数据时使用self.input_data_lists。set_input_list在 base.py 中的实现揭示了这一规范的底层逻辑当传入的input_list是二维数组首元素为 list时先校验各列长度一致再将完整数据存入self.input_data_lists、把第一列放入self.list当传入一维数组时self.list即原数组self.input_data_lists被规范化为[input_list]。因此无论输入一列还是多列self.input_data_lists始终是列表的列表算法实现可以放心按二维结构处理。此外基类还提供了input_is_empty()辅助方法判断self.list是否为空或 None内置算法在execute开头均会先做此检查例如 iqr.py 中的if self.input_is_empty(): return []自定义算法建议同样遵循。2.2 输出-1表示异常valid_code表示正常execute方法返回一个与self.list等长的数组数组中的-1表示该位置的数据点为异常。例如对序列[2, 2, 2, 2, 100]假设 100 是异常点则方法应返回[1, 1, 1, 1, -1]。这里的正常值编码并非硬编码的1而是通过基类的valid_code属性管理默认1。set_params支持通过参数valid_code覆盖该值base.py 中的实现为self.valid_code int(params[valid_code])。这一点对 PyOD 适配尤其重要见下一节。下游消费方对返回数组的处理可以在 algo/anomaly.py 中看到do_ad_check依次完成获取服务、注入输入、设置参数、执行算法随后统计异常点数量并调用convert_results_to_windows把逐点标记数组 时间戳转换为连续的异常窗口[skey, ekey]与对应的 mask 列表——这也是 SQL 层最终展示的结果形态。2.3 第三方检测器如 PyOD的标签映射对于 PyOD 这类第三方检测器需要把self.input_data_lists转换成形状为(n_samples, n_features)的特征矩阵样本数 × 特征数后喂给检测器。注意两点差异返回长度PyOD 的fit_predict返回的是每个样本一个标签的数组长度与样本数一致标签语义PyOD 用0表示正常点、1表示离群点而 TDgpt 的框架约定是非valid_code即异常。因此适配层必须把 PyOD 的标签映射为 TDgpt 结果编码正常点使用self.valid_code离群点使用-1。仓库中 lof.py 展示了完整做法先用np.asarray(self.input_data_lists, dtypefloat).T把列数组转置为(n_samples, n_features)的特征矩阵再调用LocalOutlierFactor.fit_predict最后res.tolist()直接返回——因为 sklearn 的 LOF 恰好输出1正常/-1异常与 TDgpt 编码天然一致。而 PyOD 系列算法如 pyod_stat.py 中的 ECOD/HBOS/COPOD/IForest/PCA则统一走 tool/pyod_util.py 中的fit_predict工具函数完成0/1→valid_code/-1的映射可复用于你自己的 PyOD 适配代码。三、编写你的第一个自定义异常检测算法官方文档给出一个最小可运行示例把时间序列的最后一个数据点判定为异常。完整代码如下直接引自官方开发文档from taosanalytics.base import AbstractAnomalyDetectionService # Algorithm files must start with an underscore (_) and end with Service. class _MyAnomalyDetectionService(AbstractAnomalyDetectionService): Define a class inheriting from AbstractAnomalyDetectionService and implementing the abstract method of that class. # Name the algorithm using only lowercase ASCII characters. name myad # Include a description of the algorithm (recommended) desc return the last value as the anomaly data def __init__(self): Method to initialize the class super().__init__() def execute(self): Implementation of algorithm logic Create an array with length len(self.list) whose results are all 1, then set the final value in the array to -1 to indicate an anomaly res [1] * len(self.list) res[-1] -1 Return results return res def set_params(self, params): This algorithm does not take any parameters, so this logic is not included. return super().set_params(params)对照仓库源码这段示例需要满足的隐式约束有类命名类名必须以_开头、以Service结尾如_MyAnomalyDetectionService。ServiceRegistry._register_services_in_dirservice_registry.py在扫描时会过滤掉不以_开头的类同时跳过从其他模块导入的类只注册本模块内定义的算法类name属性仅使用小写 ASCII 字符这是SHOW ANODES FULL与 SQL 中algoxxx的取值依据注册时通过algo_cls.name作为键写入服务表见 service_registry.pydesc属性算法描述可选但推荐会被get_typed_services采集成服务列表展示execute必须返回与self.list等长的数组元素为valid_code默认 1或-1set_params接收parse_options解析出的参数字典。不需要参数时直接透传super().set_params(params)即可。基类的set_params会做类型校验非 dict 抛出ValueError并负责解析valid_code参数base.py。parse_options的参数解析规则见 util.py以逗号分隔的keyvalue字符串例如 SQL 中的algoksigma,k2会被解析为{algo: ksigma, k: 2}。参考内置 ksigma.py 的做法可以在set_params中读取并校验自定义参数如k的合法区间为[1, 3]并配合get_params返回当前参数快照。四、部署与注册让算法对 SQL 可见开发完成后将算法文件保存到 anode 安装目录下的./lib/taosanalytics/algo/ad/目录对应仓库源码中的 tools/tdgpt/taosanalytics/algo/ad/然后重启taosanode服务。注册链路如下taosanode启动时调用ServiceRegistry.register_all_services()service_registry.py其中通过_register_services_in_dir(..., taosanalytics.algo.ad., algo/ad/, True)扫描异常检测算法目录requiredTrue表示该目录必须存在否则快速失败以暴露部署问题对每个.py文件动态importlib.import_module遍历模块内以_开头且在本地定义的类实例化后以algo_cls.name注册进服务表除内置目录外还会加载algo/custom/ad/下的自定义扩展目录requiredFalse目录不存在时静默跳过以及动态模型配置目录中的 JSON 模型。在 TDengine CLI 中执行SHOW ANODES FULL即可看到新注册的算法。确认可见后应用即可通过 SQL 调用--- Detect anomalies in the col column using the newly added myad algorithm SELECT COUNT(*) FROM foo ANOMALY_WINDOW(col, algomyad)SQL 侧的完整执行路径可以结合源码确认handlers/anomaly.py中的handle_anomalyhandlers/anomaly.py先做载荷校验数据行数须在 1040000 之间、时间戳列与数据列提取、可选的白噪声检查随后algo缺省时默认ksigma最终委托do_ad_check完成算法执行与异常窗口转换。如果此前从未启动过 anode请参照 安装指南 先将 anode 加入你的 TDengine 集群。五、单元测试验证算法的正确性框架使用 Python 标准库unittest模块进行测试。可以为自定义算法在taosanalytics/test目录下的anomaly_test.py文件中追加测试用例也可以新建独立的测试文件。官方文档给出的test_myad测试用例def test_myad(self): Test the _IqrService class s loader.get_service(myad) # Configure the data to test s.set_input_list(AnomalyDetectionTest.input_list, None) r s.execute() # The final value is an anomaly self.assertEqual(r[-1], -1) self.assertEqual(len(r), len(AnomalyDetectionTest.input_list))这段测试的关键点loader.get_service(myad)从全局ServiceRegistry单例service_registry.py按名称取回算法实例——测试前需确保服务已注册测试框架会先执行注册流程set_input_list注入待测数据ts_list传None即可该算法不依赖时间戳断言r[-1] -1验证最后一个点为异常的算法逻辑len(r) len(input_list)验证输出与输入等长的框架契约注意get_service返回的是copy.copy的实例副本service_registry.py因此测试间互不干扰。六、内置算法参数速查自定义实现的参照系在为自定义算法设计set_params参数时可参考仓库内置算法的参数风格均在set_params中经super().set_params(params)校验后按需解析算法参数说明依据源码ksigmak标准差倍数合法区间[1, 3]默认 3ksigma.pygrubbsalpha显著性水平取值(0, 1)内部转换为1 - alphagrubbs.pylofneighbors、algorithm近邻数与 sklearn 算法选择lof.pyecod/copodcontamination异常比例经validate_contamination校验pyod_stat.pyhboscontamination、n_binsn_bins为直方图分箱数须 ≥ 2iforestcontamination、n_estimators、random_state树的数量与随机种子均须为非负整数pcacontamination、n_components、standardizationn_components支持正整数或(0,1)内的方差保留比例这些实现同时展示了良好实践参数值校验失败时抛出带明确信息的ValueErrorexecute开头先做input_is_empty()防护多列场景统一从self.input_data_lists构造(n_samples, n_features)特征矩阵。七、开发要点小结契约先行execute返回值必须与self.list等长-1表异常、self.valid_code默认 1表正常多列访问v3.4.1.0起用self.input_data_lists承载全部输入列单列向后兼容走self.listPyOD 适配特征矩阵用(n_samples, n_features)形状标签映射为0→valid_code、1→-1可直接复用 tool/pyod_util.py 的fit_predict命名规范文件与类名以_开头、类名以Service结尾、name用小写 ASCII否则无法通过 service_registry.py 的扫描注册部署验证保存到 anode 的lib/taosanalytics/algo/ad/后重启taosanodeSHOW ANODES FULL确认可见再经ANOMALY_WINDOW(col, algomyad)由 SQL 调用测试兜底用unittest在anomaly_test.py中覆盖输出等长、异常点标记正确两类核心断言。按照以上步骤你可以在不修改 TDgpt 框架任何代码的前提下为 TDengine 的时序分析能力接入任意自定义异常检测算法并以标准的 SQL 语法在应用侧直接消费检测结果。【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengine创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表