TDgpt 算法扩展开发指南为 TDengine 时序分析平台编写自定义分析与预测算法【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengineTDgpt 是 TDengine 生态中可扩展的高级时序数据分析 Agent本文基于官方开发者指南docs/en/09-ai-and-advanced-analytics/01-tdgpt/06-dev/index.md并对照仓库源码系统讲解如何在 anonode 上开发、注册并调用自定义统计、机器学习与基础模型算法。读完本文你将掌握 TDgpt 的算法加载机制、Python 类开发规范命名、继承、属性初始化、SQL 调用方式以及如何编写一个可直接运行的异常检测与预测算法。一、TDgpt 的扩展模型半动态算法加载TDgpt 的扩展性核心在于anode 的半动态算法加载机制anode 启动时会扫描指定目录将符合约定要求的 Python 文件中的算法注册到平台之后通过 SQL 语句即可直接调用这些算法无需重启 TDengine 服务端taosd。整个注册链路在源码中非常清晰tools/tdgpt/taosanalytics/app.py 在应用初始化时调用loader.register_all_services()tools/tdgpt/taosanalytics/service_registry.py 中register_all_services()依次扫描algo/ad、algo/fc、algo/imputat、algo/correl四个内置算法目录以及algo/custom/ad、algo/custom/fc两个自定义目录同时扫描动态模型目录dynamic_model_dir目录下的 JSON 配置文件可随时增删模型在服务运行期间即会被加载或卸载见 service_registry.py 的sync_dynamic_services这就是半动态的含义。由于 TDgpt 与 TDengine 解耦在 anode 上新增或升级算法不会影响 taosd 本身应用侧只需更新 SQL 语句即可使用新算法。二、添加算法到 TDgpt 的三个步骤按官方指南向 TDgpt 添加一个算法只需三步开发按照 TDgpt 的规范用 Python 编写分析算法部署将源码文件放入 anode 的指定目录并重启 anode 服务注册执行CREATE ANODE语句将 anode 加入 TDengine 集群。之后算法即可通过 SQL 调用。需要说明的是第 3 步仅在 anode 尚未注册到集群时执行一次日常算法升级只需完成第 2 步重启 anode应用侧更新 SQL 即可。2.1 准备开发环境TDgpt 的源码位于本仓库的 tools/tdgpt 目录下核心 Python 包为 tools/tdgpt/taosanalytics。开发环境要求 Python 3.10 及以上见 tools/tdgpt/README.md并依赖 numpy、statsmodels、scikit-learn、Flask 等库安装完成后 taosanode 会创建虚拟环境venv并注册为系统服务taosanoded。2.2 anode 目录结构安装后的 anode 目录结构如下摘自官方文档. ├── bin ├── cfg ├── lib │ └── taosanalytics │ ├── algo │ │ ├── ad │ │ └── fc │ ├── misc │ └── test ├── log - /var/log/taos/taosanode ├── model - /var/lib/taos/taosanode/model └── venv - /var/lib/taos/taosanode/venv目录说明taosanalytics平台源码。algo子目录存放算法ad为异常检测、fc为预测test存放单元与集成测试misc存放其他辅助文件venvPython 虚拟环境model数据集对应的已训练模型cfg配置文件仓库中 tools/tdgpt/taosanalytics/algo/ad 与 tools/tdgpt/taosanalytics/algo/fc 的目录组织与安装布局一一对应内置算法包括异常检测adksigma、grubbs、iqr、lof、shesd、pyod_stat预测fcholtwinters、arima、prophet、theta、ets、ces、chronos、moirai、timemoe、timesfm、gpt。三、算法开发硬性规范决定能否被自动加载anode 采用自动扫描注册因此文件命名、类命名、继承关系必须严格遵守以下约定否则算法会被静默跳过。3.1 存放位置限制异常检测算法的 Python 源码必须放在./taosanalytics/algo/ad预测算法的 Python 源码必须放在./taosanalytics/algo/fc自定义算法可放入algo/custom/ad与algo/custom/fc见 service_registry.py。3.2 类命名规则算法类名必须以下划线开头、以Service结尾。例如_KSigmaService就是 k-sigma 异常检测算法的类名见 tools/tdgpt/taosanalytics/algo/ad/ksigma.py。这条规则在注册源码中得到了验证。在 service_registry.py 中扫描器遍历模块内的所有类并做如下过滤跳过抽象基类AbstractAnomalyDetectionService等跳过不以_开头的类跳过定义在其他模块的类防止误注册 import 进来的第三方类跳过__init__.py、__pycache__及非.py文件。if class_name in ServiceRegistry._base_class_name or ( not class_name.startswith(_) ): continue3.3 类继承规则所有异常检测算法必须继承AbstractAnomalyDetectionService并实现execute方法所有预测算法必须继承AbstractForecastService并实现execute方法。这两个抽象基类定义在 tools/tdgpt/taosanalytics/base.py异常检测与 base.py预测其公共父类是AbstractAnalyticsService再上层是AnalyticsService。AbstractAnomalyDetectionService的关键约定valid_code 1结果列表中等于valid_code的值表示正常点否则视为异常点type anomaly-detection用于平台按类型索引算法基类已实现set_input_list支持一维/二维输入与set_params支持valid_code参数。AbstractForecastService的关键约定内置参数period周期、start_ts起始时间戳、time_step步长、rows预测行数、return_conf是否返回置信区间、conf置信度默认 0.95、precision时间精度默认ms、tz时区set_params强制要求start_ts、time_step、rows三个参数必须存在且time_step、rows必须大于 0见 base.py预测结果以字典形式返回res中第一行为预测时间戳序列后续行为预测值及可选的置信区间上下界。对于基于 StatsForecast 的统计模型还可继承 base.py 中定义的AbstractStatsForecastService只需实现_fit_model并复用其execute即可自动得到预测值、置信区间与 MSE。3.4 类属性初始化每个算法类必须初始化以下两个类属性name算法标识符只能使用小写字母。该标识符会显示在SHOW语句的算法列表中并在 SQL 中通过algoname指定desc算法的基础描述会展示给用户并用于平台列表接口。-- 示例algo 键即取类中定义的 name 值 SELECT COUNT(*) FROM foo ANOMALY_WINDOW(col_name, algoname)四、源码级示例一编写异常检测算法以 k-sigma 为例仓库自带的_KSigmaServicetools/tdgpt/taosanalytics/algo/ad/ksigma.py是规范实现的最小完整范例可作为模板ksigma class definition import numpy as np from taosanalytics.base import AbstractAnomalyDetectionService class _KSigmaService(AbstractAnomalyDetectionService): KSigma algorithm is to check the anomaly data in the input list name ksigma desc the k-sigma algorithm (or 3σ rule) expresses a conventional heuristic that nearly all values are taken to lie within k (usually three) standard deviations of the mean, and thus it is empirically useful to treat 99.7% probability as near certainty _builtins True def __init__(self, k_val3): super().__init__() self.k_val k_val def execute(self): def get_k_sigma_range(vals, k_value): avg np.mean(vals) std np.std(vals) upper avg k_value * std lower avg - k_value * std return [float(lower), float(upper)] if self.input_is_empty(): return [] threshold get_k_sigma_range(self.list, self.k_val) return [-1 if k threshold[0] or k threshold[1] else 1 for k in self.list] def set_params(self, params): super().set_params(params) if k in params: k int(params[k]) if k 1 or k 3: raise ValueError(k value out of range, valid range [1, 3]) self.k_val k def get_params(self): return {k: self.k_val}对照规范逐项检查命名文件为ksigma.py类名_KSigmaService以_开头、以Service结尾满足扫描条件继承继承AbstractAnomalyDetectionService并实现execute属性name ksigma全小写、desc描述了 3σ 准则原理另设_builtins True标记为内置算法参数通过重写set_params支持k参数取值范围 [1, 3]默认 3get_params向平台暴露参数列表。执行语义上execute返回与输入等长的标签列表正常点返回1等于基类valid_code异常点返回-1。上层 tools/tdgpt/taosanalytics/algo/anomaly.py 的do_ad_check会统计非valid_code的点数并通过convert_results_to_windows将异常点转换为异常窗口供 SQL 侧使用。编写同类算法时仿照该结构即可继承基类 → 初始化name/desc→ 在execute中读取self.list并返回标签列表 → 可选重写set_params支持自定义参数。五、源码级示例二编写预测算法以 Holt-Winters 为例预测算法的参照实现是_HoltWintersServicetools/tdgpt/taosanalytics/algo/fc/holtwinters.pyholt winters definition from statsmodels.tsa.holtwinters import ExponentialSmoothing, SimpleExpSmoothing from taosanalytics.algo.forecast import insert_ts_list from taosanalytics.base import AbstractForecastService class _HoltWintersService(AbstractForecastService): Holt winters algorithm is to do the fc in the input list name holtwinters desc forecast algorithm by using exponential smoothing _builtins True def __init__(self): super().__init__() self.trend_option None self.seasonal_option None def set_params(self, params): super().set_params(params) self.trend_option params[trend] if trend in params else None if self.trend_option is not None: if self.trend_option not in (add, mul): raise ValueError(trend parameter can only be mul or add) self.seasonal_option params[seasonal] if seasonal in params else None if self.seasonal_option is not None: if self.seasonal_option not in (add, mul): raise ValueError(seasonal parameter can only be mul or add) def get_params(self): p super().get_params() p.update({trend: self.trend_option, seasonal: self.seasonal_option}) return p def execute(self): if self.list is None or len(self.list) self.period: raise ValueError(number of input data is less than the periods) if self.rows 0: raise ValueError(fc rows is not specified yet) res, mse self.__do_forecast_helper(self.list, self.rows) insert_ts_list(res, self.start_ts, self.time_step, self.rows) return {mse: mse, res: res}预测算法与异常检测算法的关键差异在于execute的返回结构必须返回包含res键的字典其中res的第一行为预测时间戳序列由insert_ts_list基于start_ts与time_step生成后续行为预测值序列及若return_conf开启置信区间上下界。上层 tools/tdgpt/taosanalytics/algo/forecast.py 的do_forecast会执行check_forecast_results校验各序列长度一致并把period与algo写入结果。另外注意当请求的算法名加载失败时do_forecast会自动回退到holtwinters因此 Holt-Winters 是平台默认的兜底预测算法。AbstractStatsForecastService则提供了更省力的路径子类只需实现_fit_model基类的execute会统一完成数据校验、预测、置信区间计算与时间戳生成base.pyarima、prophet、theta、ets等内置算法均基于此模式。六、动态模型免重启加载与卸载除了内置 Python 算法TDgpt 还支持通过JSON 配置文件注册动态模型。配置文件放在动态模型目录默认model_dir/dynamic格式支持模型名.json旧格式或模型名/模型名.json新格式两种布局配置中必须包含algo字段见 service_registry.py 的register_service_from_file。algo字段当前支持以下模型类型见 service_registry.py预测forecastarima、prophet、deepar异常检测anomalyiforest、svm回归regressionlinear_regression、lasso、ridge、elasticnet、svr、polynomial_regression分类classificationlogistic_regression、decision_tree。动态模型与内置算法的差异在于内置算法随代码加载动态模型则从 JSON 配置 序列化的模型文件中加载。平台会为动态模型创建对应的DynamicForecastService/DynamicAnomalyService等包装服务tools/tdgpt/taosanalytics/handlers/dynamic。在服务运行期间get_service每次调用都会触发sync_dynamic_services新增配置文件会被注册配置被删除的模型会被从内存移除孤立的.pkl文件会被清理service_registry.py——这就是前文所述半动态机制的实际落地。需要训练与部署动态模型时可参考 tools/tdgpt/taosanalytics/misc/train_ad_model.py 及 tools/tdgpt/taosanalytics/misc/model_downloader.py。七、在 SQL 中调用自定义算法算法注册成功后即可通过 TDengine SQL 直接调用。官方指南给出了异常检测的调用示例SELECT COUNT(*) FROM foo ANOMALY_WINDOW(col_name, algoname)其中algoname的name就是算法类中定义的name属性。预测算法同理例如使用内置的 Holt-Winters 或时间序列基础模型SELECT forecast(val, algoholtwinters) FROM foo; SELECT forecast(val, algotdtsfm_1) FROM forecast.electricity_demand;forecast支持通过 options 传入预测行数、置信度、时间精度等参数tools/tdgpt/taosanalytics/algo/forecast.py 的add_forecast_params展示了参数映射关系forecast_rows→rows、start→start_ts、every→time_step、conf→ 置信度、return_conf→ 是否返回置信区间、prec→ 精度、tz→ 时区。7.1 anode 的集群注册与管理首次使用时需将 anode 注册进 TDengine 集群相关管理语句见 docs/en/09-ai-and-advanced-analytics/01-tdgpt/03-management.mdCREATE ANODE {node_url}; -- 注册 anode SHOW ANODES; -- 查看已注册 anode SHOW ANODES FULL; -- 查看详情 DROP ANODE {anode_id}; -- 移除 anode UPDATE ANODE {anode_id}; -- 更新 anode 信息 UPDATE ALL ANODES;7.2 通过 RESTful 接口校验算法anode 以 Flask 提供 RESTful 服务tools/tdgpt/taosanalytics/app.py可用以下接口快速验证算法是否注册成功GET /返回 anode 版本号GET /status返回服务状态GET /list返回全部可用算法清单含类型、名称、描述、参数与状态见loader.get_service_list()service_registry.pyGET /models返回可用模型列表。八、服务配置与验证测试8.1 关键配置项anode 的配置文件为taosanode.config.pyLinux 下默认位于/etc/taos/仓库模板见 tools/tdgpt/cfg/taosanode.config.py也可通过环境变量TDGPT_CONF指定路径app.py。常用配置项包括配置项默认值说明bind0.0.0.0:6035服务监听地址与端口workers2worker 进程数建议 2×CPU 核数 1threads由 CPU 核数推导每进程线程数适合模型部署场景timeout/keepalive1200超时与 keep-alive秒log_levelDEBUG日志级别model_dirinstall_dir/model模型存储目录dynamic_model_dirmodel_dir/dynamic动态模型目录运行期热加载draw_resultFalse是否绘制查询结果图输出到img_dirmodels见配置文件时间序列基础模型tdtsfm、timemoe为必需moirai、chronos、timesfm、moment为可选及其端口/端点用于自动推导服务 URLconf.py默认路径定义可在 tools/tdgpt/taosanalytics/conf.py 中查到日志/var/log/taos/taosanode/、模型目录/usr/local/taos/taosanode/model/、动态模型目录model/dynamic。修改配置后重启服务即可生效systemctl restart taosanoded。8.2 使用单元测试验证算法仓库在 tools/tdgpt/tests 与 tools/tdgpt/taosanalytics/test 提供了大量测试用例。例如 tools/tdgpt/tests/unit_test.py 中的test_generate_anomaly_window直接验证了异常检测结果向异常窗口的转换逻辑convert_results_to_windowsforecast_test.py、anomaly_test.py、restful_api_test.py则分别覆盖预测、异常检测与接口层。开发新算法时建议参照这些测试为execute的正确性、set_params的参数校验非法参数应抛出ValueError补充用例并运行pytest确保回归。九、开发建议与常见问题算法未出现在SHOW列表中优先检查类名是否以_开头且以Service结尾、类是否定义在扫描目录的模块内、name是否全小写、文件是否为.py且非__init__.py查看 anode 日志/var/log/taos/taosanode/taosanode.app.log中的load algorithm与failed to register service记录。name冲突注册表不允许重名重复注册会抛出RuntimeErrorservice_registry.py请确保name全局唯一。输入数据校验异常检测的execute应先调用input_is_empty()处理空输入预测算法应在execute中校验数据量是否满足周期要求如 Holt-Winters 要求数据量不小于period。参数校验set_params中所有非法参数都应抛出ValueError平台会捕获并返回错误信息见 tools/tdgpt/taosanalytics/handlers/forecast.py。动态模型优于内置算法升级如果算法只是模型权重更新优先使用动态模型目录热替换 JSON 配置避免重启 anode 影响正在执行的查询。Python 版本兼容注册器会对特定 Python 版本做兼容性过滤例如 Python 3.12 下因 pandas 兼容问题跳过_SHESDService见 service_registry.py自研算法应关注运行环境的 Python 版本。十、总结TDgpt 通过约定优于配置的目录扫描机制将自定义算法的接入成本降到最低把遵循命名、继承与属性规范编写的 Python 文件放入algo/ad或algo/fc目录并重启 anode算法即可通过algoname在 SQL 中调用而动态模型机制则进一步支持运行期免重启的模型热更新。本文给出的 k-sigma 与 Holt-Winters 两个源码范例覆盖了异常检测与预测两大算法类型的完整开发范式读者可直接基于 tools/tdgpt/taosanalytics/algo 中的任一内置实现仿写配合 tools/tdgpt/tests 的测试用例快速验证即可将自有算法无缝接入 TDengine 的时序分析链路。【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengine创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 TDgpt 算法扩展开发指南：为 TDengine 时序分析平台编写自定义分析与预测算法
TDgpt 算法扩展开发指南为 TDengine 时序分析平台编写自定义分析与预测算法【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengineTDgpt 是 TDengine 生态中可扩展的高级时序数据分析 Agent本文基于官方开发者指南docs/en/09-ai-and-advanced-analytics/01-tdgpt/06-dev/index.md并对照仓库源码系统讲解如何在 anonode 上开发、注册并调用自定义统计、机器学习与基础模型算法。读完本文你将掌握 TDgpt 的算法加载机制、Python 类开发规范命名、继承、属性初始化、SQL 调用方式以及如何编写一个可直接运行的异常检测与预测算法。一、TDgpt 的扩展模型半动态算法加载TDgpt 的扩展性核心在于anode 的半动态算法加载机制anode 启动时会扫描指定目录将符合约定要求的 Python 文件中的算法注册到平台之后通过 SQL 语句即可直接调用这些算法无需重启 TDengine 服务端taosd。整个注册链路在源码中非常清晰tools/tdgpt/taosanalytics/app.py 在应用初始化时调用loader.register_all_services()tools/tdgpt/taosanalytics/service_registry.py 中register_all_services()依次扫描algo/ad、algo/fc、algo/imputat、algo/correl四个内置算法目录以及algo/custom/ad、algo/custom/fc两个自定义目录同时扫描动态模型目录dynamic_model_dir目录下的 JSON 配置文件可随时增删模型在服务运行期间即会被加载或卸载见 service_registry.py 的sync_dynamic_services这就是半动态的含义。由于 TDgpt 与 TDengine 解耦在 anode 上新增或升级算法不会影响 taosd 本身应用侧只需更新 SQL 语句即可使用新算法。二、添加算法到 TDgpt 的三个步骤按官方指南向 TDgpt 添加一个算法只需三步开发按照 TDgpt 的规范用 Python 编写分析算法部署将源码文件放入 anode 的指定目录并重启 anode 服务注册执行CREATE ANODE语句将 anode 加入 TDengine 集群。之后算法即可通过 SQL 调用。需要说明的是第 3 步仅在 anode 尚未注册到集群时执行一次日常算法升级只需完成第 2 步重启 anode应用侧更新 SQL 即可。2.1 准备开发环境TDgpt 的源码位于本仓库的 tools/tdgpt 目录下核心 Python 包为 tools/tdgpt/taosanalytics。开发环境要求 Python 3.10 及以上见 tools/tdgpt/README.md并依赖 numpy、statsmodels、scikit-learn、Flask 等库安装完成后 taosanode 会创建虚拟环境venv并注册为系统服务taosanoded。2.2 anode 目录结构安装后的 anode 目录结构如下摘自官方文档. ├── bin ├── cfg ├── lib │ └── taosanalytics │ ├── algo │ │ ├── ad │ │ └── fc │ ├── misc │ └── test ├── log - /var/log/taos/taosanode ├── model - /var/lib/taos/taosanode/model └── venv - /var/lib/taos/taosanode/venv目录说明taosanalytics平台源码。algo子目录存放算法ad为异常检测、fc为预测test存放单元与集成测试misc存放其他辅助文件venvPython 虚拟环境model数据集对应的已训练模型cfg配置文件仓库中 tools/tdgpt/taosanalytics/algo/ad 与 tools/tdgpt/taosanalytics/algo/fc 的目录组织与安装布局一一对应内置算法包括异常检测adksigma、grubbs、iqr、lof、shesd、pyod_stat预测fcholtwinters、arima、prophet、theta、ets、ces、chronos、moirai、timemoe、timesfm、gpt。三、算法开发硬性规范决定能否被自动加载anode 采用自动扫描注册因此文件命名、类命名、继承关系必须严格遵守以下约定否则算法会被静默跳过。3.1 存放位置限制异常检测算法的 Python 源码必须放在./taosanalytics/algo/ad预测算法的 Python 源码必须放在./taosanalytics/algo/fc自定义算法可放入algo/custom/ad与algo/custom/fc见 service_registry.py。3.2 类命名规则算法类名必须以下划线开头、以Service结尾。例如_KSigmaService就是 k-sigma 异常检测算法的类名见 tools/tdgpt/taosanalytics/algo/ad/ksigma.py。这条规则在注册源码中得到了验证。在 service_registry.py 中扫描器遍历模块内的所有类并做如下过滤跳过抽象基类AbstractAnomalyDetectionService等跳过不以_开头的类跳过定义在其他模块的类防止误注册 import 进来的第三方类跳过__init__.py、__pycache__及非.py文件。if class_name in ServiceRegistry._base_class_name or ( not class_name.startswith(_) ): continue3.3 类继承规则所有异常检测算法必须继承AbstractAnomalyDetectionService并实现execute方法所有预测算法必须继承AbstractForecastService并实现execute方法。这两个抽象基类定义在 tools/tdgpt/taosanalytics/base.py异常检测与 base.py预测其公共父类是AbstractAnalyticsService再上层是AnalyticsService。AbstractAnomalyDetectionService的关键约定valid_code 1结果列表中等于valid_code的值表示正常点否则视为异常点type anomaly-detection用于平台按类型索引算法基类已实现set_input_list支持一维/二维输入与set_params支持valid_code参数。AbstractForecastService的关键约定内置参数period周期、start_ts起始时间戳、time_step步长、rows预测行数、return_conf是否返回置信区间、conf置信度默认 0.95、precision时间精度默认ms、tz时区set_params强制要求start_ts、time_step、rows三个参数必须存在且time_step、rows必须大于 0见 base.py预测结果以字典形式返回res中第一行为预测时间戳序列后续行为预测值及可选的置信区间上下界。对于基于 StatsForecast 的统计模型还可继承 base.py 中定义的AbstractStatsForecastService只需实现_fit_model并复用其execute即可自动得到预测值、置信区间与 MSE。3.4 类属性初始化每个算法类必须初始化以下两个类属性name算法标识符只能使用小写字母。该标识符会显示在SHOW语句的算法列表中并在 SQL 中通过algoname指定desc算法的基础描述会展示给用户并用于平台列表接口。-- 示例algo 键即取类中定义的 name 值 SELECT COUNT(*) FROM foo ANOMALY_WINDOW(col_name, algoname)四、源码级示例一编写异常检测算法以 k-sigma 为例仓库自带的_KSigmaServicetools/tdgpt/taosanalytics/algo/ad/ksigma.py是规范实现的最小完整范例可作为模板ksigma class definition import numpy as np from taosanalytics.base import AbstractAnomalyDetectionService class _KSigmaService(AbstractAnomalyDetectionService): KSigma algorithm is to check the anomaly data in the input list name ksigma desc the k-sigma algorithm (or 3σ rule) expresses a conventional heuristic that nearly all values are taken to lie within k (usually three) standard deviations of the mean, and thus it is empirically useful to treat 99.7% probability as near certainty _builtins True def __init__(self, k_val3): super().__init__() self.k_val k_val def execute(self): def get_k_sigma_range(vals, k_value): avg np.mean(vals) std np.std(vals) upper avg k_value * std lower avg - k_value * std return [float(lower), float(upper)] if self.input_is_empty(): return [] threshold get_k_sigma_range(self.list, self.k_val) return [-1 if k threshold[0] or k threshold[1] else 1 for k in self.list] def set_params(self, params): super().set_params(params) if k in params: k int(params[k]) if k 1 or k 3: raise ValueError(k value out of range, valid range [1, 3]) self.k_val k def get_params(self): return {k: self.k_val}对照规范逐项检查命名文件为ksigma.py类名_KSigmaService以_开头、以Service结尾满足扫描条件继承继承AbstractAnomalyDetectionService并实现execute属性name ksigma全小写、desc描述了 3σ 准则原理另设_builtins True标记为内置算法参数通过重写set_params支持k参数取值范围 [1, 3]默认 3get_params向平台暴露参数列表。执行语义上execute返回与输入等长的标签列表正常点返回1等于基类valid_code异常点返回-1。上层 tools/tdgpt/taosanalytics/algo/anomaly.py 的do_ad_check会统计非valid_code的点数并通过convert_results_to_windows将异常点转换为异常窗口供 SQL 侧使用。编写同类算法时仿照该结构即可继承基类 → 初始化name/desc→ 在execute中读取self.list并返回标签列表 → 可选重写set_params支持自定义参数。五、源码级示例二编写预测算法以 Holt-Winters 为例预测算法的参照实现是_HoltWintersServicetools/tdgpt/taosanalytics/algo/fc/holtwinters.pyholt winters definition from statsmodels.tsa.holtwinters import ExponentialSmoothing, SimpleExpSmoothing from taosanalytics.algo.forecast import insert_ts_list from taosanalytics.base import AbstractForecastService class _HoltWintersService(AbstractForecastService): Holt winters algorithm is to do the fc in the input list name holtwinters desc forecast algorithm by using exponential smoothing _builtins True def __init__(self): super().__init__() self.trend_option None self.seasonal_option None def set_params(self, params): super().set_params(params) self.trend_option params[trend] if trend in params else None if self.trend_option is not None: if self.trend_option not in (add, mul): raise ValueError(trend parameter can only be mul or add) self.seasonal_option params[seasonal] if seasonal in params else None if self.seasonal_option is not None: if self.seasonal_option not in (add, mul): raise ValueError(seasonal parameter can only be mul or add) def get_params(self): p super().get_params() p.update({trend: self.trend_option, seasonal: self.seasonal_option}) return p def execute(self): if self.list is None or len(self.list) self.period: raise ValueError(number of input data is less than the periods) if self.rows 0: raise ValueError(fc rows is not specified yet) res, mse self.__do_forecast_helper(self.list, self.rows) insert_ts_list(res, self.start_ts, self.time_step, self.rows) return {mse: mse, res: res}预测算法与异常检测算法的关键差异在于execute的返回结构必须返回包含res键的字典其中res的第一行为预测时间戳序列由insert_ts_list基于start_ts与time_step生成后续行为预测值序列及若return_conf开启置信区间上下界。上层 tools/tdgpt/taosanalytics/algo/forecast.py 的do_forecast会执行check_forecast_results校验各序列长度一致并把period与algo写入结果。另外注意当请求的算法名加载失败时do_forecast会自动回退到holtwinters因此 Holt-Winters 是平台默认的兜底预测算法。AbstractStatsForecastService则提供了更省力的路径子类只需实现_fit_model基类的execute会统一完成数据校验、预测、置信区间计算与时间戳生成base.pyarima、prophet、theta、ets等内置算法均基于此模式。六、动态模型免重启加载与卸载除了内置 Python 算法TDgpt 还支持通过JSON 配置文件注册动态模型。配置文件放在动态模型目录默认model_dir/dynamic格式支持模型名.json旧格式或模型名/模型名.json新格式两种布局配置中必须包含algo字段见 service_registry.py 的register_service_from_file。algo字段当前支持以下模型类型见 service_registry.py预测forecastarima、prophet、deepar异常检测anomalyiforest、svm回归regressionlinear_regression、lasso、ridge、elasticnet、svr、polynomial_regression分类classificationlogistic_regression、decision_tree。动态模型与内置算法的差异在于内置算法随代码加载动态模型则从 JSON 配置 序列化的模型文件中加载。平台会为动态模型创建对应的DynamicForecastService/DynamicAnomalyService等包装服务tools/tdgpt/taosanalytics/handlers/dynamic。在服务运行期间get_service每次调用都会触发sync_dynamic_services新增配置文件会被注册配置被删除的模型会被从内存移除孤立的.pkl文件会被清理service_registry.py——这就是前文所述半动态机制的实际落地。需要训练与部署动态模型时可参考 tools/tdgpt/taosanalytics/misc/train_ad_model.py 及 tools/tdgpt/taosanalytics/misc/model_downloader.py。七、在 SQL 中调用自定义算法算法注册成功后即可通过 TDengine SQL 直接调用。官方指南给出了异常检测的调用示例SELECT COUNT(*) FROM foo ANOMALY_WINDOW(col_name, algoname)其中algoname的name就是算法类中定义的name属性。预测算法同理例如使用内置的 Holt-Winters 或时间序列基础模型SELECT forecast(val, algoholtwinters) FROM foo; SELECT forecast(val, algotdtsfm_1) FROM forecast.electricity_demand;forecast支持通过 options 传入预测行数、置信度、时间精度等参数tools/tdgpt/taosanalytics/algo/forecast.py 的add_forecast_params展示了参数映射关系forecast_rows→rows、start→start_ts、every→time_step、conf→ 置信度、return_conf→ 是否返回置信区间、prec→ 精度、tz→ 时区。7.1 anode 的集群注册与管理首次使用时需将 anode 注册进 TDengine 集群相关管理语句见 docs/en/09-ai-and-advanced-analytics/01-tdgpt/03-management.mdCREATE ANODE {node_url}; -- 注册 anode SHOW ANODES; -- 查看已注册 anode SHOW ANODES FULL; -- 查看详情 DROP ANODE {anode_id}; -- 移除 anode UPDATE ANODE {anode_id}; -- 更新 anode 信息 UPDATE ALL ANODES;7.2 通过 RESTful 接口校验算法anode 以 Flask 提供 RESTful 服务tools/tdgpt/taosanalytics/app.py可用以下接口快速验证算法是否注册成功GET /返回 anode 版本号GET /status返回服务状态GET /list返回全部可用算法清单含类型、名称、描述、参数与状态见loader.get_service_list()service_registry.pyGET /models返回可用模型列表。八、服务配置与验证测试8.1 关键配置项anode 的配置文件为taosanode.config.pyLinux 下默认位于/etc/taos/仓库模板见 tools/tdgpt/cfg/taosanode.config.py也可通过环境变量TDGPT_CONF指定路径app.py。常用配置项包括配置项默认值说明bind0.0.0.0:6035服务监听地址与端口workers2worker 进程数建议 2×CPU 核数 1threads由 CPU 核数推导每进程线程数适合模型部署场景timeout/keepalive1200超时与 keep-alive秒log_levelDEBUG日志级别model_dirinstall_dir/model模型存储目录dynamic_model_dirmodel_dir/dynamic动态模型目录运行期热加载draw_resultFalse是否绘制查询结果图输出到img_dirmodels见配置文件时间序列基础模型tdtsfm、timemoe为必需moirai、chronos、timesfm、moment为可选及其端口/端点用于自动推导服务 URLconf.py默认路径定义可在 tools/tdgpt/taosanalytics/conf.py 中查到日志/var/log/taos/taosanode/、模型目录/usr/local/taos/taosanode/model/、动态模型目录model/dynamic。修改配置后重启服务即可生效systemctl restart taosanoded。8.2 使用单元测试验证算法仓库在 tools/tdgpt/tests 与 tools/tdgpt/taosanalytics/test 提供了大量测试用例。例如 tools/tdgpt/tests/unit_test.py 中的test_generate_anomaly_window直接验证了异常检测结果向异常窗口的转换逻辑convert_results_to_windowsforecast_test.py、anomaly_test.py、restful_api_test.py则分别覆盖预测、异常检测与接口层。开发新算法时建议参照这些测试为execute的正确性、set_params的参数校验非法参数应抛出ValueError补充用例并运行pytest确保回归。九、开发建议与常见问题算法未出现在SHOW列表中优先检查类名是否以_开头且以Service结尾、类是否定义在扫描目录的模块内、name是否全小写、文件是否为.py且非__init__.py查看 anode 日志/var/log/taos/taosanode/taosanode.app.log中的load algorithm与failed to register service记录。name冲突注册表不允许重名重复注册会抛出RuntimeErrorservice_registry.py请确保name全局唯一。输入数据校验异常检测的execute应先调用input_is_empty()处理空输入预测算法应在execute中校验数据量是否满足周期要求如 Holt-Winters 要求数据量不小于period。参数校验set_params中所有非法参数都应抛出ValueError平台会捕获并返回错误信息见 tools/tdgpt/taosanalytics/handlers/forecast.py。动态模型优于内置算法升级如果算法只是模型权重更新优先使用动态模型目录热替换 JSON 配置避免重启 anode 影响正在执行的查询。Python 版本兼容注册器会对特定 Python 版本做兼容性过滤例如 Python 3.12 下因 pandas 兼容问题跳过_SHESDService见 service_registry.py自研算法应关注运行环境的 Python 版本。十、总结TDgpt 通过约定优于配置的目录扫描机制将自定义算法的接入成本降到最低把遵循命名、继承与属性规范编写的 Python 文件放入algo/ad或algo/fc目录并重启 anode算法即可通过algoname在 SQL 中调用而动态模型机制则进一步支持运行期免重启的模型热更新。本文给出的 k-sigma 与 Holt-Winters 两个源码范例覆盖了异常检测与预测两大算法类型的完整开发范式读者可直接基于 tools/tdgpt/taosanalytics/algo 中的任一内置实现仿写配合 tools/tdgpt/tests 的测试用例快速验证即可将自有算法无缝接入 TDengine 的时序分析链路。【免费下载链接】TDengineHigh-performance, scalable time-series database designed for Industrial IoT (IIoT) scenarios项目地址: https://gitcode.com/GitHub_Trending/tde/TDengine创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考