diff --git a/data/models/sample_ad_model b/data/models/sample_ad_model new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_detection.py b/tests/test_detection.py new file mode 100644 index 0000000..94b797a --- /dev/null +++ b/tests/test_detection.py @@ -0,0 +1,79 @@ +"""detection 模块测试(mock TDengine execute 返回预设窗口)。""" + +from unittest.mock import MagicMock +import pytest +from ts_anomaly_td.detection import run_anomaly_window, detect_all_algos, ALL_ALGOS + + +def make_mock_conn(rows=None): + """构造 mock TDConnection,execute 返回预设行。""" + conn = MagicMock() + conn.execute.return_value = rows or [] + return conn + + +def test_run_anomaly_window_returns_windows(): + """正常返回两个异常窗口。""" + conn = make_mock_conn([ + (1000000, 2000000), + (3000000, 4000000), + ]) + windows = run_anomaly_window(conn, "finance_001", "ksigma") + assert windows == [(1000000, 2000000), (3000000, 4000000)] + + +def test_run_anomaly_window_empty(): + """无异常窗口时返回空列表。""" + conn = make_mock_conn([]) + windows = run_anomaly_window(conn, "finance_001", "iqr") + assert windows == [] + + +def test_run_anomaly_window_invalid_algo(): + """不支持的算法抛 ValueError。""" + conn = make_mock_conn() + with pytest.raises(ValueError, match="不支持的算法"): + run_anomaly_window(conn, "test", "nonexistent") + + +def test_run_anomaly_window_sql_error(): + """SQL 执行失败传播异常。""" + conn = make_mock_conn() + conn.execute.side_effect = Exception("TDengine error") + + with pytest.raises(Exception, match="TDengine error"): + run_anomaly_window(conn, "test", "ksigma") + + +def test_detect_all_algos_independent_fault_tolerance(): + """单个算法失败不影响其他算法。""" + conn = MagicMock() + + def side_effect(sql): + if "iqr" in sql: + raise Exception("iqr failed") + return [(1000, 2000)] + + conn.execute.side_effect = side_effect + + results = detect_all_algos(conn, "test") + + # 所有 6 个算法都在结果中 + assert set(results.keys()) == set(ALL_ALGOS) + + # iqr 记录了 error + assert results["iqr"]["error"] == "iqr failed" + assert results["iqr"]["windows"] == [] + + # 其他算法正常 + for algo in ALL_ALGOS: + if algo != "iqr": + assert results[algo]["error"] is None + assert len(results[algo]["windows"]) == 1 + + +def test_detect_all_algos_custom_algo_list(): + """指定算法子集。""" + conn = make_mock_conn([(1000, 2000)]) + results = detect_all_algos(conn, "test", algos=["ksigma", "iqr"]) + assert set(results.keys()) == {"ksigma", "iqr"} diff --git a/ts_anomaly_td/detection.py b/ts_anomaly_td/detection.py new file mode 100644 index 0000000..4275677 --- /dev/null +++ b/ts_anomaly_td/detection.py @@ -0,0 +1,77 @@ +"""异常检测模块:ANOMALY_WINDOW × 6 算法多路跑 + 容错。""" + +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +ALL_ALGOS = ["ksigma", "iqr", "grubbs", "shesd", "lof", "sample_ad_model"] + + +def run_anomaly_window(conn, stable: str, algo: str) -> list[tuple[int, int]]: + """执行单算法 ANOMALY_WINDOW。 + + Args: + conn: TDConnection 实例 + stable: supertable 名称(不含 ds_ 前缀) + algo: 算法名 + + Returns: + [(start_ts_ms, end_ts_ms), ...] 异常窗口列表 + + Raises: + ValueError: 算法不支持 + Exception: SQL 执行失败(透传) + """ + if algo not in ALL_ALGOS: + raise ValueError(f"不支持的算法: {algo},可选: {ALL_ALGOS}") + + sql = ( + f"SELECT _WSTART, _WEND " + f"FROM (ANOMALY_WINDOW(value, \"algo={algo}\") " + f"FROM ds_{stable})" + ) + + rows = conn.execute(sql) + + windows = [] + for row in rows: + if row[0] is not None and row[1] is not None: + windows.append((int(row[0]), int(row[1]))) + return windows + + +def detect_all_algos(conn, stable: str, algos: list[str] | None = None) -> dict[str, Any]: + """依次执行所有指定算法,独立容错。 + + Args: + conn: TDConnection 实例 + stable: supertable 名称 + algos: 算法列表,默认 ALL_ALGOS + + Returns: + {algo_name: {"windows": [...], "error": None | str}, ...} + """ + algos = algos or ALL_ALGOS + results: dict[str, Any] = {} + + for algo in algos: + logger.info("running ANOMALY_WINDOW algo=%s stable=%s", algo, stable) + try: + if algo == "sample_ad_model": + # 检查模型文件 + import os + model_path = os.path.join("data", "models", "sample_ad_model") + if not os.path.exists(model_path): + results[algo] = {"windows": [], "error": "model file not found"} + logger.warning("sample_ad_model skip: model file not found at %s", model_path) + continue + + windows = run_anomaly_window(conn, stable, algo) + results[algo] = {"windows": windows, "error": None} + logger.info("algo=%s: %d windows found", algo, len(windows)) + except Exception as e: + results[algo] = {"windows": [], "error": str(e)} + logger.warning("algo=%s failed (non-fatal): %s", algo, e) + + return results diff --git a/ts_anomaly_td/forecast.py b/ts_anomaly_td/forecast.py new file mode 100644 index 0000000..4ad9de3 --- /dev/null +++ b/ts_anomaly_td/forecast.py @@ -0,0 +1,108 @@ +"""预测式异常模块:FORECAST + 区间外点判定。""" + +import logging +from typing import TypedDict + +logger = logging.getLogger(__name__) + + +class ForecastPoint(TypedDict): + ts: int + value: float + _flow: float + _fhigh: float + is_anomaly: bool + + +def run_forecast( + conn, + stable: str, + algo: str = "holtwinters", + rows: int = 10, + conf: int = 95, +) -> list[tuple[int, float, float]]: + """执行 FORECAST 查询。 + + Args: + conn: TDConnection 实例 + stable: supertable 名称 + algo: 预测算法,默认 holtwinters + rows: 预测行数 + conf: 置信度百分比 + + Returns: + [(frowts_ms, flow, fhigh), ...] + + Raises: + RuntimeError: SQL 执行失败 + """ + sql = ( + f"SELECT _FROWTS, _FLOW, _FHIGH " + f"FROM (FORECAST(value, \"algo={algo},rows={rows},conf={conf}\") " + f"FROM ds_{stable})" + ) + try: + raw = conn.execute(sql) + except Exception as e: + raise RuntimeError(f"FORECAST 执行失败: {e}") from e + + results = [] + for row in raw: + frowts = int(row[0]) if row[0] is not None else 0 + flow = float(row[1]) if row[1] is not None else 0.0 + fhigh = float(row[2]) if row[2] is not None else 0.0 + results.append((frowts, flow, fhigh)) + return results + + +def forecast_anomaly( + conn, stable: str, algo: str = "holtwinters", rows: int = 10, conf: int = 95 +) -> list[ForecastPoint]: + """FORECAST 预测并判定区间外异常。 + + 步骤: + 1. 执行 FORECAST,获取 [(_flow, _fhigh)] 区间 + 2. 查询预测区间对应的原始数据 + 3. 比较判定:value < _flow or value > _fhigh → is_anomaly + + Args: + conn: TDConnection 实例 + stable: supertable 名称 + algo: 预测算法 + rows: 预测行数 + conf: 置信度百分比 + + Returns: + ForecastPoint 列表(含 is_anomaly 标记) + """ + forecasts = run_forecast(conn, stable, algo, rows, conf) + if not forecasts: + return [] + + frowtses = [f[0] for f in forecasts] + start_ts = min(frowtses) + end_ts = max(frowtses) + + sql = f"SELECT ts, value FROM s_{stable} WHERE ts >= {start_ts} AND ts <= {end_ts}" + raw = conn.execute(sql) + + fmap = {f[0]: (f[1], f[2]) for f in forecasts} + results: list[ForecastPoint] = [] + for row in raw: + ts = int(row[0]) + value = float(row[1]) + if ts in fmap: + flow, fhigh = fmap[ts] + is_anom = value < flow or value > fhigh + else: + flow, fhigh = 0.0, 0.0 + is_anom = False + results.append(ForecastPoint( + ts=ts, value=value, _flow=flow, _fhigh=fhigh, is_anomaly=is_anom + )) + + logger.info( + "forecast_anomaly: %d points, %d anomalies", + len(results), sum(1 for r in results if r["is_anomaly"]), + ) + return results