预测算法

输入约定

execute 是预测算法处理的核心方法。框架调用该方法之前，在对象属性参数 self.list 中已经设置完毕用于预测的历史时间序列数据。

输出约定及父类属性说明

execute 方法执行完成后的返回一个如下字典对象， 预测返回结果如下：

1. return {
2. "mse": mse, # 预测算法的拟合数据最小均方误差(minimum squared error)
3. "res": res  # 结果数组 [时间戳数组, 预测结果数组, 预测结果执行区间下界数组，预测结果执行区间上界数组]
4. }

预测算法的父类 AbstractForecastService 包含的对象属性如下：

示例代码

下面我们开发一个示例预测算法，对于任何输入的时间序列数据，固定返回值 1 作为预测结果。

1. import numpy as np
2. from service import AbstractForecastService

4. # 算法实现类名称 需要以下划线 "_" 开始，并以 Service 结束
5. class _MyForecastService(AbstractForecastService):
6. """ 定义类，从 AbstractForecastService 继承并实现其定义的抽象方法 execute  """

8. # 定义算法调用关键词，全小写ASCII码
9. name = 'myfc'

11. # 该算法的描述信息(建议添加)
12. desc = """return the forecast time series data"""

14. def __init__(self):
15. """类初始化方法"""
16. super().__init__()

18. def execute(self):
19. """ 算法逻辑的核心实现"""
20. res = []

22. """这个预测算法固定返回 1 作为预测值，预测值的数量是用户通过 self.fc_rows 指定"""
23. ts_list = [self.start_ts + i * self.time_step for i in range(self.fc_rows)]
24. res.app(ts_list)  #  设置预测结果时间戳列

26. """生成全部为 1 的预测结果 """
27. res_list = [1] * self.fc_rows
28. res.append(res_list)

30. """检查用户输入，是否要求返回预测置信区间上下界"""
31. if self.return_conf:
32. """对于没有计算预测置信区间上下界的算法，直接返回预测值作为上下界即可"""
33. bound_list = [1] * self.fc_rows
34. res.append(bound_list) #  预测结果置信区间下界
35. res.append(bound_list) #  预测结果执行区间上界

37. """返回结果"""
38. return { "res": res, "mse": 0}

41. def set_params(self, params):
42. """该算法无需任何输入参数，直接重载父类该函数，不处理算法参数设置逻辑"""
43. pass

将该文件保存在 ./taosanalytics/algo/fc/ 目录下，然后重启 taosanode 服务。在 TDengine 命令行接口中执行 SHOW ANODES FULL 能够看到新加入的算法。应用就可以通过 SQL 语句调用该预测算法。

1. --- 对 col 列进行异常检测，通过指定 algo 参数为 myfc 来调用新添加的预测类
2. SELECT  _flow, _fhigh, _frowts, FORECAST(col_name, "algo=myfc")
3. FROM foo;

如果是第一次启动该 Anode, 请按照 TDgpt 安装部署 里的步骤先将该 Anode 添加到 TDengine 系统中。

单元测试

在测试目录taosanalytics/test中的 forecast_test.py 中增加单元测试用例或添加新的测试文件。单元测试依赖 Python Unit test 包。

1. def test_myfc(self):
2. """ 测试 myfc 类 """
3. s = loader.get_service("myfc")

5. # 设置用于预测分析的数据
6. s.set_input_list(self.get_input_list())
7. # 检查预测结果应该全部为 1
8. r = s.set_params(
9. {"fc_rows": 10, "start_ts": 171000000, "time_step": 86400 * 30, "start_p": 0}
10. )
11. r = s.execute()

13. expected_list = [1] * 10
14. self.assertEqlist(r["res"][0], expected_list)