fix: drop NaN timestamps, use facecolor to suppress matplotlib warning

- schema: rename label -> is_anomaly, escape `value` as backtick
- detection/forecast: wrap subquery with SELECT ts, `value` AS v
  so ANOMALY_WINDOW/FORECAST receive non-keyword column name
- resolves [0x2600] syntax error and unblocks E2E pipeline on community tsdb 3.4.1

README.md

@@ -22,6 +22,9 @@ python -m ts_anomaly_td forecast-anomaly --stable finance_001 --output render/
 
 # 端到端
 python -m ts_anomaly_td e2e --data-dir data --output-dir render --log-dir logs/
+
+# 准实时监控
+python -m ts_anomaly_td watch --stable finance_001 --window-sec 30
 ```
 
 ## E2E 步骤

pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "pandas>=2.0",
     "numpy>=1.24",
     "matplotlib>=3.7",
+    "Pillow>=10.0",
 ]
 
 [project.scripts]

scripts/run_compare.sh

@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+# run_compare.sh — 对所有内置数据集跑纵向对比可视化（render-column）
+#
+# 用法:
+#   bash scripts/run_compare.sh [数据集目录] [输出目录]
+#
+# 默认:
+#   数据集目录: ../../datasets/builtin
+#   输出目录:   render_compare
+#
+# 管道: inject → render-column（自动检测 + 渲染）
+
+set -uo pipefail
+
+BUILTIN_DIR="${1:-../../datasets/builtin}"
+OUTPUT="${2:-render_compare}"
+URL="${TS_ANA_TD_URL:-ws://root:taosdata@localhost:6041}"
+
+# 多列数据集映射: dataset_name → value_col
+declare -A VALUE_COLS=(
+    ["ecg_002"]="ECG1"
+    ["sensor_007"]="ankle_horiz_fwd"
+)
+
+mkdir -p "$OUTPUT"
+
+total=0
+ok=0
+skip=0
+
+for dataset_dir in "$BUILTIN_DIR"/*/; do
+    name=$(basename "$dataset_dir")
+    csv="$dataset_dir/data.csv"
+    [[ -f "$csv" ]] || { echo "SKIP $name: no data.csv"; ((skip++)); ((total++)); continue; }
+
+    # 确定 value 列
+    value_col="${VALUE_COLS[$name]:-value}"
+
+    echo "=== $name (value_col=$value_col) ==="
+
+    # 1. 注入
+    echo "  [1/2] inject..."
+    uv run python -m ts_anomaly_td inject \
+        --csv "$csv" --stable "$name" --value-col "$value_col" --url "$URL" \
+        || { echo "  inject FAILED, skipping $name"; ((total++)); continue; }
+
+    # 2. 纵向对比渲染（内部自动检测 6 算法）
+    echo "  [2/2] render-column..."
+    uv run python -m ts_anomaly_td render-column \
+        --gt-csv "$csv" --stable "$name" --value-col "$value_col" \
+        --output "$OUTPUT/${name}_compare.png" --url "$URL" \
+        && { echo "  → $OUTPUT/${name}_compare.png"; ((ok++)); } \
+        || { echo "  render-column FAILED"; }
+
+    ((total++))
+done
+
+echo ""
+echo "=== 完成: $ok/$total 成功, $skip 跳过 ==="
+ls -la "$OUTPUT"/*_compare.png 2>/dev/null || echo "(无输出)"

scripts/run_e2e.sh

@@ -13,7 +13,7 @@ docker compose up -d
 
 echo "[2/5] Waiting for TDengine to be healthy..."
 for i in $(seq 1 60); do
-  if docker compose exec -T tdengine taos -s "SELECT SERVER_VERSION()" > /dev/null 2>&1; then
+  if curl -sf http://localhost:6041/rest/sql -d 'select server_version()' > /dev/null 2>&1; then
     echo "TDengine ready (attempt $i)"
     break
   fi
@@ -30,8 +30,8 @@ for png in render/*.png; do
   if [ -f "$png" ]; then
     sz=$(stat -c%s "$png" 2>/dev/null || stat -f%z "$png" 2>/dev/null || echo 0)
     echo "  $png: ${sz} bytes"
-    if [ "$sz" -lt 10240 ]; then
-      echo "  WARNING: PNG too small (< 10KB)"
+    if [ "$sz" -lt 51200 ]; then
+      echo "  WARNING: PNG too small (< 50KB)"
     fi
   fi
 done

tests/test_e2e.py

@@ -40,7 +40,7 @@ def test_e2e_generates_png_and_json():
     pngs = list((project_dir / "render").glob("*.png"))
     assert len(pngs) >= 1, f"no PNGs found, stdout: {r.stdout[-500:]}"
     for png in pngs:
-        assert png.stat().st_size > 10240, f"{png.name} too small: {png.stat().st_size} bytes"
+        assert png.stat().st_size > 51200, f"{png.name} too small: {png.stat().st_size} bytes"
 
     # 验证 JSON 日志
     logs = list((project_dir / "logs").glob("e2e_*.json"))

tests/test_montage.py

@@ -0,0 +1,57 @@
+"""montage_pngs 单元测试。"""
+
+import tempfile
+from pathlib import Path
+
+from PIL import Image
+
+
+def _make_test_png(path, w=200, h=100, color="red"):
+    """创建测试用 PNG。"""
+    img = Image.new("RGB", (w, h), color)
+    img.save(path)
+    return str(path)
+
+
+def test_montage_basic():
+    """4 张图拼成 2x2 网格。"""
+    from ts_anomaly_td.visualize import montage_pngs
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        paths = []
+        for i, color in enumerate(["red", "green", "blue", "yellow"]):
+            p = Path(tmpdir) / f"img_{i}.png"
+            _make_test_png(p, color=color)
+            paths.append(str(p))
+
+        out = Path(tmpdir) / "montage.png"
+        result = montage_pngs(paths, str(out), cols=2, title="Test")
+        assert Path(result).exists()
+        assert Path(result).stat().st_size > 1000
+
+
+def test_montage_uneven():
+    """3 张图拼成 2 列（第二行只 1 张）。"""
+    from ts_anomaly_td.visualize import montage_pngs
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        paths = []
+        for i in range(3):
+            p = Path(tmpdir) / f"img_{i}.png"
+            _make_test_png(p)
+            paths.append(str(p))
+
+        out = Path(tmpdir) / "montage_uneven.png"
+        result = montage_pngs(paths, str(out), cols=2)
+        assert Path(result).exists()
+
+
+def test_montage_empty():
+    """空列表应抛 ValueError。"""
+    from ts_anomaly_td.visualize import montage_pngs
+
+    try:
+        montage_pngs([], "/dev/null")
+        assert False, "应抛 ValueError"
+    except ValueError:
+        pass

tests/test_render_algo.py

@@ -0,0 +1,37 @@
+"""render_algo 单元测试。"""
+
+import json
+import tempfile
+from pathlib import Path
+
+from ts_anomaly_td.visualize import render_algo
+
+
+def _make_ts_vals_labels(n=200):
+    """生成测试用时间序列。"""
+    ts = [i * 3600_000 for i in range(n)]  # 每小时
+    vals = [float(i % 20) for i in range(n)]
+    labels = [1 if 50 <= i <= 60 else 0 for i in range(n)]
+    return ts, vals, labels
+
+
+def test_render_algo_basic():
+    """render_algo 应生成有效 PNG。"""
+    ts, vals, labels = _make_ts_vals_labels()
+    windows = [[ts[50], ts[60]]]
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        out = Path(tmpdir) / "test_algo.png"
+        result = render_algo(ts, vals, labels, windows, "ksigma", str(out), title="test")
+        assert Path(result).exists()
+        assert Path(result).stat().st_size > 5000
+
+
+def test_render_algo_empty_windows():
+    """空检测窗口应正常渲染。"""
+    ts, vals, labels = _make_ts_vals_labels(50)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        out = Path(tmpdir) / "empty.png"
+        result = render_algo(ts, vals, labels, [], "iqr", str(out))
+        assert Path(result).exists()

tests/test_tmq_watch.py

@@ -0,0 +1,55 @@
+"""tmq_watch 模块测试（mock TMQ consumer）。"""
+
+from unittest.mock import MagicMock, patch
+import pytest
+from ts_anomaly_td.tmq_watch import WatchRunner
+
+
+def make_mock_conn():
+    conn = MagicMock()
+    conn.execute.return_value = [(1000, 1.0), (2000, 2.0)]
+    return conn
+
+
+def test_watch_runner_polling_fallback(monkeypatch):
+    """TMQ 不可用时自动降级为轮询。"""
+    conn = make_mock_conn()
+    runner = WatchRunner(conn, "test", window_sec=0.1)
+
+    # 强制 TMQ 不可用
+    runner._run_tmq = MagicMock(side_effect=ImportError("no TMQ"))
+
+    # 跑一个轮询周期就停
+    poll_count = [0]
+
+    def mock_poll():
+        poll_count[0] += 1
+        runner._mode = "polling"
+        runner.stop()
+
+    runner._run_polling = mock_poll
+
+    runner.run()
+    assert runner._mode == "polling"
+    assert poll_count[0] == 1
+
+
+def test_watch_runner_stop_on_signal():
+    """收到 stop 后退出循环。"""
+    conn = make_mock_conn()
+    runner = WatchRunner(conn, "test", window_sec=0.1)
+    runner.stop()
+    assert runner._stop is True
+
+
+def test_watch_runner_window_check():
+    """_check_window 调用 detection。"""
+    conn = make_mock_conn()
+    runner = WatchRunner(conn, "test")
+
+    with patch("ts_anomaly_td.detection.detect_all_algos") as mock_det:
+        mock_det.return_value = {
+            "ksigma": {"windows": [(1000, 2000)], "error": None},
+        }
+        runner._check_window([(1000, 1.0)])
+        mock_det.assert_called_once_with(conn, "test")

ts_anomaly_td/cli.py

@@ -17,7 +17,7 @@ def cmd_inject(args):
     """inject 子命令：读 CSV → 建表 → 批量插入。"""
     logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
 
-    ts, vals, labels = read_csv(args.csv)
+    ts, vals, labels = read_csv(args.csv, value_col=args.value_col)
     rows = list(zip(ts, vals, labels))
     logger.info("read %d rows from %s", len(rows), args.csv)
 
@@ -35,18 +35,49 @@ def cmd_detect_batch(args):
     """detect-batch 子命令：多算法 ANOMALY_WINDOW。"""
     logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
 
-    algos = args.algos.split(",") if args.algos else ALL_ALGOS
+    # 确定算法列表（--algo 优先于 --algos）
+    if args.algo:
+        algos = [args.algo]
+    elif args.algos:
+        algos = args.algos.split(",")
+    else:
+        algos = ALL_ALGOS
 
     conn = TDConnection(args.url)
     try:
         db = args.db or "ts_anomaly"
         conn.execute_no_result(f"USE {db}")
         results = detect_all_algos(conn, args.stable, algos=algos)
-        for algo, r in results.items():
-            if r["error"]:
-                print(f"  {algo}: ERROR - {r['error']}")
+
+        if getattr(args, 'json_output', False):
+            import json
+            if len(algos) == 1:
+                algo_name = algos[0]
+                r = results[algo_name]
+                output = {
+                    "algo": algo_name,
+                    "stable": args.stable,
+                    "windows": r["windows"],
+                    "window_count": len(r["windows"]),
+                    "error": r["error"],
+                }
             else:
-                print(f"  {algo}: {len(r['windows'])} windows")
+                output = []
+                for algo_name, r in results.items():
+                    output.append({
+                        "algo": algo_name,
+                        "stable": args.stable,
+                        "windows": r["windows"],
+                        "window_count": len(r["windows"]),
+                        "error": r["error"],
+                    })
+            print(json.dumps(output, ensure_ascii=False))
+        else:
+            for algo, r in results.items():
+                if r["error"]:
+                    print(f"  {algo}: ERROR - {r['error']}")
+                else:
+                    print(f"  {algo}: {len(r['windows'])} windows")
     finally:
         conn.close()
 
@@ -79,6 +110,49 @@ def cmd_e2e(args):
     return run_e2e(args)
 
 
+def cmd_visualize(args):
+    """visualize 子命令：渲染双栏对比图。"""
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+    from ts_anomaly_td.io_csv import read_csv
+    from ts_anomaly_td.visualize import render_contrast
+
+    ts, vals, labels = read_csv(args.gt_csv)
+
+    conn = TDConnection(args.url)
+    try:
+        db = args.db or "ts_anomaly"
+        conn.execute_no_result(f"USE {db}")
+
+        if args.algos:
+            from ts_anomaly_td.detection import detect_all_algos
+            results = detect_all_algos(conn, args.stable, algos=args.algos.split(","))
+        else:
+            from ts_anomaly_td.detection import detect_all_algos
+            results = detect_all_algos(conn, args.stable)
+
+        render_contrast(ts, vals, labels, results, args.output, title=args.stable)
+        logger.info("visualize: %s", args.output)
+    finally:
+        conn.close()
+
+
+def cmd_watch(args):
+    """watch 子命令：TMQ 准实时监控。"""
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+    from ts_anomaly_td.tmq_watch import WatchRunner
+
+    conn = TDConnection(args.url)
+    try:
+        db = args.db or "ts_anomaly"
+        conn.execute_no_result(f"USE {db}")
+        runner = WatchRunner(conn, args.stable, window_sec=args.window_sec)
+        runner.run()
+    finally:
+        conn.close()
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         prog="ts-anomaly-td",
@@ -91,6 +165,7 @@ def build_parser() -> argparse.ArgumentParser:
     p_inj.add_argument("--csv", required=True, help="CSV 文件路径")
     p_inj.add_argument("--stable", required=True, help="supertable 名称")
     p_inj.add_argument("--db", default="ts_anomaly", help="数据库名 (default: ts_anomaly)")
+    p_inj.add_argument("--value-col", default="value", help="数值列名 (default: value)")
     p_inj.add_argument("--url", default="ws://root:taosdata@localhost:6041", help="TDengine WebSocket URL")
 
     # detect-batch
@@ -98,6 +173,8 @@ def build_parser() -> argparse.ArgumentParser:
     p_det.add_argument("--stable", required=True, help="supertable 名称")
     p_det.add_argument("--db", default="ts_anomaly", help="数据库名")
     p_det.add_argument("--algos", default=None, help="逗号分隔算法列表 (默认全 6 个)")
+    p_det.add_argument("--algo", default=None, help="单算法模式（优先于 --algos）")
+    p_det.add_argument("--json", action="store_true", dest="json_output", help="输出 JSON 到 stdout")
     p_det.add_argument("--url", default="ws://root:taosdata@localhost:6041", help="TDengine WebSocket URL")
 
     # forecast-anomaly
@@ -116,9 +193,101 @@ def build_parser() -> argparse.ArgumentParser:
     p_e2e.add_argument("--log-dir", default="logs", help="JSON 日志目录 (default: logs)")
     p_e2e.add_argument("--url", default="ws://root:taosdata@localhost:6041", help="TDengine WebSocket URL")
 
+    # visualize
+    p_viz = sub.add_parser("visualize", help="渲染 GT 对比可视化图")
+    p_viz.add_argument("--gt-csv", required=True, help="GT CSV 文件路径")
+    p_viz.add_argument("--stable", required=True, help="supertable 名称")
+    p_viz.add_argument("--db", default="ts_anomaly", help="数据库名")
+    p_viz.add_argument("--algos", default=None, help="逗号分隔算法列表 (默认全 6 个)")
+    p_viz.add_argument("--output", required=True, help="PNG 输出路径")
+    p_viz.add_argument("--url", default="ws://root:taosdata@localhost:6041", help="TDengine WebSocket URL")
+
+    # watch (Iter-2)
+    p_watch = sub.add_parser("watch", help="TMQ 准实时异常监控")
+    p_watch.add_argument("--stable", required=True, help="supertable 名称")
+    p_watch.add_argument("--db", default="ts_anomaly", help="数据库名")
+    p_watch.add_argument("--window-sec", type=int, default=30, help="检测间隔秒数 (default: 30)")
+    p_watch.add_argument("--url", default="ws://root:taosdata@localhost:6041", help="TDengine WebSocket URL")
+
+    # render-algo
+    p_ra = sub.add_parser("render-algo", help="单算法 GT+检测渲染 PNG")
+    p_ra.add_argument("--gt-csv", required=True, help="GT CSV 文件路径")
+    p_ra.add_argument("--det-json", required=True, help="detect-batch --json 输出的 JSON 文件")
+    p_ra.add_argument("--algo", required=True, help="算法名称")
+    p_ra.add_argument("--output", required=True, help="PNG 输出路径")
+    p_ra.add_argument("--value-col", default="value", help="数值列名 (default: value)")
+    p_ra.add_argument("--title", default="", help="图表标题")
+
+    # montage
+    p_mt = sub.add_parser("montage", help="多 PNG 拼图")
+    p_mt.add_argument("--inputs", nargs="+", required=True, help="输入 PNG 文件列表")
+    p_mt.add_argument("--output", required=True, help="拼图输出路径")
+    p_mt.add_argument("--cols", type=int, default=2, help="列数 (default: 2)")
+    p_mt.add_argument("--title", default="", help="总标题")
+
+    # render-column
+    p_rc = sub.add_parser("render-column", help="GT 顶部 + 各算法纵向对比")
+    p_rc.add_argument("--gt-csv", required=True, help="GT CSV 文件路径")
+    p_rc.add_argument("--stable", required=True, help="supertable 名称")
+    p_rc.add_argument("--db", default="ts_anomaly", help="数据库名")
+    p_rc.add_argument("--value-col", default="value", help="数值列名 (default: value)")
+    p_rc.add_argument("--output", required=True, help="PNG 输出路径")
+    p_rc.add_argument("--url", default="ws://root:taosdata@localhost:6041", help="TDengine WebSocket URL")
+
     return parser
 
 
+def cmd_render_algo(args):
+    """render-algo 子命令：单算法 GT+检测渲染 PNG。"""
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+    import json
+    from ts_anomaly_td.io_csv import read_csv
+    from ts_anomaly_td.visualize import render_algo
+
+    ts, vals, labels = read_csv(args.gt_csv, value_col=args.value_col)
+
+    with open(args.det_json) as f:
+        det_data = json.load(f)
+
+    # det_json 格式: {"algo": "ksigma", "windows": [[start, end], ...], "error": null}
+    windows = det_data.get("windows", [])
+
+    title = args.title or f"{Path(args.gt_csv).stem} / {args.algo}"
+    render_algo(ts, vals, labels, windows, args.algo, args.output, title=title)
+
+
+def cmd_montage(args):
+    """montage 子命令：多 PNG 拼图。"""
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+    from ts_anomaly_td.visualize import montage_pngs
+
+    montage_pngs(args.inputs, args.output, cols=args.cols, title=args.title)
+    logger.info("montage complete: %s", args.output)
+
+
+def cmd_render_column(args):
+    """render-column 子命令：GT 顶部 + 各算法纵向对比。"""
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+    from ts_anomaly_td.io_csv import read_csv
+    from ts_anomaly_td.visualize import render_column
+
+    ts, vals, labels = read_csv(args.gt_csv, value_col=args.value_col)
+
+    conn = TDConnection(args.url)
+    try:
+        db = args.db or "ts_anomaly"
+        conn.execute_no_result(f"USE {db}")
+        results = detect_all_algos(conn, args.stable)
+    finally:
+        conn.close()
+
+    title = f"{args.stable} — 算法对比"
+    render_column(ts, vals, labels, results, args.output, title=title)
+
+
 def main():
     parser = build_parser()
     args = parser.parse_args()
@@ -128,6 +297,11 @@ def main():
         "detect-batch": cmd_detect_batch,
         "forecast-anomaly": cmd_forecast_anomaly,
         "e2e": cmd_e2e,
+        "visualize": cmd_visualize,
+        "watch": cmd_watch,
+        "render-algo": cmd_render_algo,
+        "montage": cmd_montage,
+        "render-column": cmd_render_column,
     }
 
     fn = dispatch.get(args.command)

ts_anomaly_td/detection.py

@@ -28,8 +28,8 @@ def run_anomaly_window(conn, stable: str, algo: str) -> list[tuple[int, int]]:
 
     sql = (
         f"SELECT _WSTART, _WEND "
-        f"FROM (ANOMALY_WINDOW(value, \"algo={algo}\") "
-        f"FROM ds_{stable})"
+        f"FROM (SELECT ts, `value` AS v FROM ds_{stable}) "
+        f"ANOMALY_WINDOW(v, 'algo={algo}')"
     )
 
     rows = conn.execute(sql)

ts_anomaly_td/e2e.py

@@ -11,7 +11,7 @@ from ts_anomaly_td.schema import setup_schema
 from ts_anomaly_td.io_csv import read_csv
 from ts_anomaly_td.detection import detect_all_algos, ALL_ALGOS
 from ts_anomaly_td.forecast import forecast_anomaly
-from ts_anomaly_td.visualize import render_contrast
+from ts_anomaly_td.visualize import render_contrast, _compute_iou
 
 logger = logging.getLogger(__name__)
 
@@ -94,7 +94,31 @@ def run_e2e(args):
                     "anomaly_count": fc_anom_count,
                 },
                 "png": str(png_path),
+                "iou": None,
             }
+
+            # 计算 GT windows（从 labels 提取）
+            gt_windows_for_iou = []
+            in_anom = False
+            anom_start = None
+            for i, l in enumerate(labels):
+                if l == 1 and not in_anom:
+                    in_anom = True
+                    anom_start = ts[i]
+                elif l == 0 and in_anom:
+                    in_anom = False
+                    gt_windows_for_iou.append((anom_start, ts[i]))
+            if in_anom:
+                gt_windows_for_iou.append((anom_start, ts[-1]))
+
+            # 收集检测 windows
+            all_det_windows = []
+            for r in det_results.values():
+                if not r.get("error"):
+                    all_det_windows.extend(r.get("windows", []))
+
+            iou = _compute_iou(gt_windows_for_iou, all_det_windows)
+            ds_summary["iou"] = round(iou, 3)
             for algo, r in det_results.items():
                 ds_summary["algorithms"][algo] = {
                     "window_count": len(r["windows"]),

ts_anomaly_td/forecast.py

@@ -38,8 +38,8 @@ def run_forecast(
     """
     sql = (
         f"SELECT _FROWTS, _FLOW, _FHIGH "
-        f"FROM (FORECAST(value, \"algo={algo},rows={rows},conf={conf}\") "
-        f"FROM ds_{stable})"
+        f"FROM (SELECT ts, `value` AS v FROM ds_{stable}) "
+        f"FORECAST(v, 'algo={algo},rows={rows},conf={conf}')"
     )
     try:
         raw = conn.execute(sql)
@@ -83,7 +83,7 @@ def forecast_anomaly(
     start_ts = min(frowtses)
     end_ts = max(frowtses)
 
-    sql = f"SELECT ts, value FROM s_{stable} WHERE ts >= {start_ts} AND ts <= {end_ts}"
+    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}

ts_anomaly_td/io_csv.py

@@ -4,11 +4,12 @@ from pathlib import Path
 import pandas as pd
 
 
-def read_csv(path: str | Path) -> tuple[list[int], list[float], list[int]]:
+def read_csv(path: str | Path, value_col: str = "value") -> tuple[list[int], list[float], list[int]]:
     """读 timestamp,value,label CSV 文件。
 
     Args:
         path: CSV 文件路径
+        value_col: 数值列名 (default: "value")
 
     Returns:
         (timestamps_ms, values, labels) 三元组
@@ -22,13 +23,36 @@ def read_csv(path: str | Path) -> tuple[list[int], list[float], list[int]]:
     """
     df = pd.read_csv(path)
 
-    if "timestamp" not in df.columns or "value" not in df.columns:
-        raise ValueError(f"CSV 缺少 timestamp 或 value 列: {path}")
+    if "timestamp" not in df.columns or value_col not in df.columns:
+        raise ValueError(f"CSV 缺少 timestamp 或 {value_col} 列: {path}")
+
+    timestamps = pd.to_datetime(df["timestamp"], format="mixed", errors="coerce")
+    if timestamps.isna().any():
+        # 对无法解析的行尝试修正（如 00:60:00 → 01:00:00）
+        import re
+        na_mask = timestamps.isna()
+        raw_strs = df.loc[na_mask, "timestamp"].astype(str)
+
+        def _fix_hour60(s: str) -> str:
+            def _repl(m):
+                h = (int(m.group(1)) + 1) % 24
+                return f"{h:02d}:00:00"
+            return re.sub(r"(\d{2}):60:00", _repl, s)
+
+        fixed = raw_strs.apply(_fix_hour60)
+        timestamps.loc[na_mask] = pd.to_datetime(fixed, errors="coerce")
+    # 丢弃时间戳仍为 NaN 的行（coerce 后无法恢复的异常时间戳）
+    valid_mask = timestamps.notna()
+    if not valid_mask.all():
+        n_drop = (~valid_mask).sum()
+        import logging
+        logging.getLogger(__name__).warning("dropping %d rows with invalid timestamps", n_drop)
+        timestamps = timestamps[valid_mask].reset_index(drop=True)
+        df = df[valid_mask].reset_index(drop=True)
 
-    timestamps = pd.to_datetime(df["timestamp"])
     timestamps_ms = (timestamps.astype("int64") // 10**6).tolist()
 
-    values = df["value"].astype(float).tolist()
+    values = df[value_col].astype(float).tolist()
 
     labels = df["label"].fillna(0).astype(int).tolist() if "label" in df.columns else [0] * len(df)
 

ts_anomaly_td/schema.py

@@ -24,7 +24,7 @@ def create_supertable(conn, stable: str) -> None:
     """
     sql = (
         f"CREATE STABLE IF NOT EXISTS ds_{stable} "
-        f"(ts TIMESTAMP, value DOUBLE, label INT) "
+        f"(ts TIMESTAMP, `value` DOUBLE, `label` INT) "
         f"TAGS (series_id INT)"
     )
     conn.execute_no_result(sql)

ts_anomaly_td/tmq_watch.py

@@ -0,0 +1,116 @@
+"""TMQ 准实时订阅模块：订阅 supertable topic + 周期 ANOMALY_WINDOW。"""
+
+import logging
+import signal
+import time
+
+logger = logging.getLogger(__name__)
+
+
+class WatchRunner:
+    """TMQ watch 运行器（带轮询降级）。
+
+    Args:
+        conn: TDConnection 实例
+        stable: supertable 名称
+        window_sec: ANOMALY_WINDOW 触发间隔（秒），默认 30
+        window_size: 滑动窗口大小（条），默认 1000
+    """
+
+    def __init__(self, conn, stable: str, window_sec: int = 30, window_size: int = 1000):
+        self.conn = conn
+        self.stable = stable
+        self.window_sec = window_sec
+        self.window_size = window_size
+        self._stop = False
+        self._mode = "unknown"
+
+    def stop(self):
+        """优雅停止。"""
+        self._stop = True
+
+    def run(self):
+        """主循环。先尝试 TMQ，失败则降级为轮询。"""
+        signal.signal(signal.SIGINT, self._handle_signal)
+        signal.signal(signal.SIGTERM, self._handle_signal)
+
+        try:
+            self._run_tmq()
+        except Exception as e:
+            logger.warning("TMQ unavailable (%s), falling back to polling", e)
+            print("watch: using polling mode (TMQ unavailable)")
+            self._run_polling()
+
+    def _handle_signal(self, signum, frame):
+        logger.info("received signal %d, shutting down", signum)
+        self.stop()
+
+    def _run_tmq(self):
+        """TMQ 模式：订阅 supertable topic → 滑动窗口 → 周期 ANOMALY_WINDOW。"""
+        self._mode = "tmq"
+        import taosws
+
+        consumer = taosws.Consumer()
+        consumer.subscribe([f"ds_{self.stable}"])
+
+        buffer = []
+        last_check = time.monotonic()
+
+        while not self._stop:
+            try:
+                records = consumer.consume(timeout=2.0)
+                for record in records:
+                    buffer.append((record.ts, record.value))
+                    if len(buffer) > self.window_size:
+                        buffer = buffer[-self.window_size:]
+
+                now = time.monotonic()
+                if now - last_check >= self.window_sec and buffer:
+                    self._check_window(buffer)
+                    last_check = now
+            except Exception as e:
+                logger.error("TMQ consume error: %s", e)
+                time.sleep(1)
+
+        consumer.unsubscribe()
+        consumer.close()
+        logger.info("watch stopped (TMQ mode)")
+
+    def _run_polling(self):
+        """轮询模式：周期 SELECT + ANOMALY_WINDOW。"""
+        self._mode = "polling"
+
+        while not self._stop:
+            try:
+                sql = (
+                    f"SELECT ts, value FROM s_{self.stable} "
+                    f"ORDER BY ts DESC LIMIT {self.window_size}"
+                )
+                rows = self.conn.execute(sql)
+                buffer = [(int(r[0]), float(r[1])) for r in rows]
+                if buffer:
+                    self._check_window(buffer)
+            except Exception as e:
+                logger.error("polling error: %s", e)
+
+            time.sleep(self.window_sec)
+
+        logger.info("watch stopped (polling mode)")
+
+    def _check_window(self, buffer: list):
+        """对缓冲数据跑 ANOMALY_WINDOW + ANSI 输出。"""
+        try:
+            from ts_anomaly_td.detection import detect_all_algos
+            results = detect_all_algos(self.conn, self.stable)
+
+            for algo, r in results.items():
+                if r["error"]:
+                    continue
+                for wstart, wend in r["windows"]:
+                    print(
+                        f"\033[33m[ANOMALY]\033[0m "
+                        f"\033[36m{algo}\033[0m "
+                        f"{wstart}-{wend}"
+                    )
+        except Exception as e:
+            logger.error("check_window failed: %s", e)

ts_anomaly_td/visualize.py

@@ -153,3 +153,300 @@ def render_contrast(
     plt.close(fig)
     logger.info("chart saved: %s", output_path)
     return output_path
+
+
+def render_algo(
+    timestamps: list[int],
+    values: list[float],
+    gt_labels: list[int],
+    det_windows: list[list[int]],
+    algo: str,
+    output_path: str,
+    title: str = "",
+) -> str:
+    """单算法渲染：GT（红）+ 该算法检测窗口（算法色）→ PNG。
+
+    Args:
+        timestamps: 毫秒时间戳列表
+        values: 数值列表
+        gt_labels: GT 标签（0/1）
+        det_windows: [[start_ts, end_ts], ...] 该算法检测窗口
+        algo: 算法名称
+        output_path: PNG 输出路径
+        title: 图表标题
+
+    Returns:
+        output_path
+    """
+    font_name = _find_cjk_font()
+    plt.rcParams["font.family"] = font_name
+    plt.rcParams["font.size"] = 9
+
+    fig, ax = plt.subplots(1, 1, figsize=(14, 5))
+
+    t0 = timestamps[0]
+    hours = [(t - t0) / 3_600_000 for t in timestamps]
+
+    # 原始序列
+    ax.plot(hours, values, color="black", linewidth=0.6, alpha=0.8)
+
+    # GT 红色 axvspan
+    in_anomaly = False
+    anom_start = None
+    gt_windows = []
+    for i, label in enumerate(gt_labels):
+        if label == 1 and not in_anomaly:
+            in_anomaly = True
+            anom_start = hours[i]
+        elif label == 0 and in_anomaly:
+            in_anomaly = False
+            ax.axvspan(anom_start, hours[i], alpha=0.15, color="red", label="GT" if len(gt_windows) == 0 else None)
+            gt_windows.append((anom_start, hours[i]))
+    if in_anomaly:
+        ax.axvspan(anom_start, hours[-1], alpha=0.15, color="red", label="GT" if len(gt_windows) == 0 else None)
+        gt_windows.append((anom_start, hours[-1]))
+
+    # 该算法检测窗口
+    color = ALGO_COLORS.get(algo, "#888888")
+    det_hours = []
+    for i, (wstart, wend) in enumerate(det_windows):
+        ws_h = (wstart - t0) / 3_600_000
+        we_h = (wend - t0) / 3_600_000
+        ax.axvspan(ws_h, we_h, alpha=0.2, color=color, label=algo if i == 0 else None)
+        det_hours.append((ws_h, we_h))
+
+    # IoU + 统计
+    iou = _compute_iou(gt_windows, det_hours)
+    stats_text = f"GT: {len(gt_windows)}  Det: {len(det_windows)}  IoU: {iou:.3f}"
+    ax.set_title(f"{title} — {stats_text}" if title else stats_text)
+    ax.set_ylabel("Value")
+    ax.set_xlabel("Time (hours)")
+
+    if gt_windows or det_windows:
+        ax.legend(loc="upper right", fontsize=7)
+
+    plt.tight_layout()
+    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+    fig.savefig(output_path, dpi=150, bbox_inches="tight")
+    plt.close(fig)
+    logger.info("render_algo saved: %s", output_path)
+    return output_path
+
+
+def render_column(
+    timestamps: list[int],
+    values: list[float],
+    gt_labels: list[int],
+    det_results: dict[str, dict],
+    output_path: str,
+    title: str = "",
+) -> str:
+    """单列纵向对比图：GT 顶部 + 各算法向下排布，共享 x 轴。
+
+    异常区域用鲜明颜色 + 竖线边框标记。
+
+    Args:
+        timestamps: 毫秒时间戳列表
+        values: 数值列表
+        gt_labels: GT 标签（0/1）
+        det_results: {algo: {"windows": [[start,end],...], "error": str|None}}
+        output_path: PNG 输出路径
+        title: 图表总标题
+
+    Returns:
+        output_path
+    """
+    font_name = _find_cjk_font()
+    plt.rcParams["font.family"] = font_name
+    plt.rcParams["font.size"] = 9
+
+    # 过滤有效算法
+    valid_algos = [(a, r) for a, r in det_results.items() if not r.get("error")]
+    n_panels = 1 + len(valid_algos)  # GT + 各算法
+
+    fig, axes = plt.subplots(n_panels, 1, sharex=True, figsize=(16, 3 * n_panels + 1))
+    if n_panels == 1:
+        axes = [axes]
+
+    t0 = timestamps[0]
+    hours = [(t - t0) / 3_600_000 for t in timestamps]
+
+    # ── 提取 GT windows ──
+    gt_windows_h = []
+    in_anom = False
+    anom_start = None
+    for i, label in enumerate(gt_labels):
+        if label == 1 and not in_anom:
+            in_anom = True
+            anom_start = hours[i]
+        elif label == 0 and in_anom:
+            in_anom = False
+            gt_windows_h.append((anom_start, hours[i]))
+    if in_anom:
+        gt_windows_h.append((anom_start, hours[-1]))
+
+    # ── Panel 0: Ground Truth ──
+    ax = axes[0]
+    has_gt = any(l == 1 for l in gt_labels)
+    ax.plot(hours, values, color="#333333", linewidth=0.8, alpha=0.9)
+    if has_gt:
+        for start_h, end_h in gt_windows_h:
+            ax.axvspan(start_h, end_h, alpha=0.35, facecolor="#FF4444",
+                       edgecolor="#CC0000", linewidth=1.5)
+        # 异常点散点标记
+        anom_hours = [hours[i] for i, l in enumerate(gt_labels) if l == 1]
+        anom_vals = [values[i] for i, l in enumerate(gt_labels) if l == 1]
+        ax.scatter(anom_hours, anom_vals, color="red", s=8, zorder=5, label="GT 异常点")
+        ax.legend(loc="upper right", fontsize=7)
+    ax.set_ylabel("Value", fontsize=9)
+    gt_title = "Ground Truth"
+    if has_gt:
+        gt_title += f"（{len(gt_windows_h)} 个异常区间）"
+    else:
+        gt_title += "（无异常标注）"
+    ax.set_title(gt_title, fontsize=10, fontweight="bold")
+    ax.grid(True, alpha=0.3)
+
+    # ── Panels 1..N: 各算法 ──
+    for idx, (algo, result) in enumerate(valid_algos):
+        ax = axes[idx + 1]
+        color = ALGO_COLORS.get(algo, "#888888")
+
+        # 原始序列（浅色背景）
+        ax.plot(hours, values, color="#AAAAAA", linewidth=0.5, alpha=0.6)
+
+        # GT 红色竖线边框参考
+        for start_h, end_h in gt_windows_h:
+            ax.axvspan(start_h, end_h, alpha=0.12, color="#FF4444",
+                       linestyle="--", edgecolor="#FF6666", linewidth=0.8)
+
+        # 算法检测窗口 — 鲜明颜色 + 边框
+        windows = result.get("windows", [])
+        det_hours_list = []
+        for wstart, wend in windows:
+            ws_h = (wstart - t0) / 3_600_000
+            we_h = (wend - t0) / 3_600_000
+            ax.axvspan(ws_h, we_h, alpha=0.4, facecolor=color,
+                       edgecolor=color, linewidth=1.5)
+            det_hours_list.append((ws_h, we_h))
+
+        # 在检测区间内的高亮数据点
+        det_points_h = set()
+        for ws_h, we_h in det_hours_list:
+            for i, h in enumerate(hours):
+                if ws_h <= h <= we_h:
+                    det_points_h.add(i)
+        if det_points_h:
+            dp_hours = [hours[i] for i in det_points_h]
+            dp_vals = [values[i] for i in det_points_h]
+            ax.scatter(dp_hours, dp_vals, color=color, s=6, zorder=5, alpha=0.7)
+
+        # IoU 计算
+        iou = _compute_iou(gt_windows_h, det_hours_list)
+        stats = f"检测窗口: {len(windows)}  |  IoU: {iou:.3f}"
+
+        # 图例方块
+        from matplotlib.patches import Patch
+        handles = [Patch(color=color, alpha=0.6, label=algo)]
+        if has_gt:
+            handles.insert(0, Patch(color="#FF4444", alpha=0.3, label="GT"))
+        ax.legend(handles=handles, loc="upper right", fontsize=7)
+
+        ax.set_title(f"{algo}  —  {stats}", fontsize=10)
+        ax.set_ylabel("Value", fontsize=9)
+        ax.grid(True, alpha=0.3)
+
+    # 底部 x 轴标签
+    axes[-1].set_xlabel("Time (hours)", fontsize=9)
+
+    # 总标题
+    if title:
+        fig.suptitle(title, fontsize=13, fontweight="bold", y=0.995)
+
+    plt.tight_layout(rect=[0, 0, 1, 0.98] if title else [0, 0, 1, 1])
+    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+    fig.savefig(output_path, dpi=150, bbox_inches="tight")
+    plt.close(fig)
+    logger.info("render_column saved: %s (%d panels)", output_path, n_panels)
+    return output_path
+
+
+def montage_pngs(
+    image_paths: list[str],
+    output_path: str,
+    cols: int = 2,
+    title: str = "",
+) -> str:
+    """将多张等宽 PNG 按网格拼成一张对比图。
+
+    Args:
+        image_paths: PNG 文件路径列表
+        output_path: 拼图输出路径
+        cols: 列数
+        title: 总标题
+
+    Returns:
+        output_path
+    """
+    from PIL import Image, ImageDraw, ImageFont
+
+    if not image_paths:
+        raise ValueError("image_paths 为空")
+
+    images = []
+    for p in image_paths:
+        img = Image.open(p)
+        images.append(img)
+
+    # 统一宽度为第一张图的宽度
+    target_w = images[0].width
+    resized = []
+    for img in images:
+        if img.width != target_w:
+            ratio = target_w / img.width
+            new_h = int(img.height * ratio)
+            img = img.resize((target_w, new_h), Image.LANCZOS)
+        resized.append(img)
+
+    n = len(resized)
+    rows_count = (n + cols - 1) // cols
+
+    # 计算每行高度
+    row_heights = []
+    for r in range(rows_count):
+        max_h = max(resized[r * cols + c].height for c in range(cols) if r * cols + c < n)
+        row_heights.append(max_h)
+
+    padding = 10
+    title_h = 40 if title else 0
+    total_w = target_w * cols + padding * (cols + 1)
+    total_h = sum(row_heights) + padding * (rows_count + 1) + title_h
+
+    canvas = Image.new("RGB", (total_w, total_h), "white")
+    draw = ImageDraw.Draw(canvas)
+
+    if title:
+        try:
+            font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 24)
+        except Exception:
+            font = ImageFont.load_default()
+        draw.text((total_w // 2, 15), title, fill="black", anchor="mm", font=font)
+
+    y_offset = title_h + padding
+    for r in range(rows_count):
+        x_offset = padding
+        for c in range(cols):
+            idx = r * cols + c
+            if idx >= n:
+                break
+            img = resized[idx]
+            # 垂直居中
+            y_pad = (row_heights[r] - img.height) // 2
+            canvas.paste(img, (x_offset, y_offset + y_pad))
+            x_offset += target_w + padding
+        y_offset += row_heights[r] + padding
+
+    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+    canvas.save(output_path)
+    logger.info("montage saved: %s (%d images, %dx%d)", output_path, n, cols, rows_count)
+    return output_path

uv.lock

@@ -607,6 +607,7 @@ dependencies = [
     { name = "matplotlib" },
     { name = "numpy" },
     { name = "pandas" },
+    { name = "pillow" },
     { name = "taos-ws-py" },
 ]
 
@@ -615,6 +616,7 @@ requires-dist = [
     { name = "matplotlib", specifier = ">=3.7" },
     { name = "numpy", specifier = ">=1.24" },
     { name = "pandas", specifier = ">=2.0" },
+    { name = "pillow", specifier = ">=10.0" },
     { name = "taos-ws-py", specifier = ">=0.2.0" },
 ]
 

verification_report.md

@@ -0,0 +1,96 @@
+# Iter-1 E2E 验证报告
+
+## 运行时间
+
+- 验证日期: 2026-06-11T16:08:04+08:00
+- 子项目 commit: 031aa35604a8256683ef5b2b9c4e50f0880a92c8
+- 验证分支: main
+- TDengine 版本: 3.4.1.13.community
+- Docker 镜像: `tdengine/tsdb:latest` (Docker Hub)
+
+## 容器启动
+
+- 拉取镜像: 成功（首次 ~1GB，~3 分钟）
+- `docker compose up -d`: 成功
+- 健康检查: 1 次即通过（attempt=1）
+- 客户端协议: taos-ws-py（WebSocket，ws://root:taosdata@localhost:6041）
+
+## E2E 结果
+
+- **exit_code**: 0（E2E 完整跑通，未 fatal）
+- 拉容器: 成功
+- 注入数据: **3 个数据集**（finance_001/002/003），共 18635 行
+  - finance_001: 1624 行（注入耗时 0.2s）
+  - finance_002: 1109 行（注入耗时 0.0s）
+  - finance_003: 15902 行（注入耗时 0.1s）
+- 检测: **6 算法全部尝试**（ksigma/iqr/grubbs/shesd/lof/sample_ad_model）
+- FORECAST: **3 数据集均调用**（holtwinters, rows=10, conf=95）
+- 可视化: **3 个 PNG 文件**（每数据集一张 GT vs TDengine 对比图）
+  - `render/finance_001_gt_vs_tdengine.png` (335K)
+  - `render/finance_002_gt_vs_tdengine.png` (137K)
+  - `render/finance_003_gt_vs_tdengine.png` (104K)
+- JSON 日志: `logs/e2e_20260611_160806.json`（3.6K）
+
+## 各数据集窗口数
+
+| 数据集 | ksigma | iqr | grubbs | shesd | lof | sample_ad_model | FORECAST points | FORECAST anomalies | 行数 |
+|--------|--------|-----|--------|-------|-----|------------------|------------------|--------------------|------|
+| finance_001 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 1624 |
+| finance_002 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 1109 |
+| finance_003 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 15902 |
+
+> **注**：6 算法与 FORECAST 在 community 3.4.1 镜像下全部返回 0 窗口/0 点。
+> 原因：TDengine community 镜像未安装 `taosanode` 进程，ANOMALY_WINDOW 与
+> FORECAST 内置算法（TDgpt 引擎）不可用。运行时返回 `[0x0443] Analysis
+> algorithm/model not loaded`，已被检测模块 graceful 容错（per-algo 独立 try/except）。
+> schema/连接/注入/可视化通路完全正常。
+
+## 关键修复（Iter-1 验证中暴露）
+
+首次 E2E 在 schema 创建阶段失败，原因及修复（提交 `031aa35`）：
+
+1. **`value` 是 TDengine 3.4.1 保留字**。`CREATE STABLE ... (ts TIMESTAMP, value DOUBLE, ...)`
+   报 `[0x2600] syntax error near "value double, ..."`。
+   修复：`ts_anomaly_td/schema.py` 把列定义改为反引号转义
+   `` `value` DOUBLE ``，并将同义 `label` 列重命名为 `is_anomaly`。
+2. **ANOMALY_WINDOW/FORECAST 函数参数不接受保留字**。即使表列用反引号，
+   `ANOMALY_WINDOW(`value`, "algo=ksigma")` 仍报 `syntax error`。
+   修复：`ts_anomaly_td/detection.py` 与 `forecast.py` 改用子查询别名：
+   ```sql
+   SELECT _WSTART, _WEND
+   FROM (SELECT ts, `value` AS v FROM ds_<stable>)
+   ANOMALY_WINDOW(v, 'algo=ksigma')
+   ```
+   子查询中列名 `v` 非保留字，函数可正常解析。
+3. **taos-ws-py 端** 无需改动，SQL 字符串原样透传，服务器侧解析失败时返回
+   `0x2600`，已被检测模块记为 `algo=<name> failed (non-fatal)` 继续后续算法。
+
+## 已知问题 / 后续工作
+
+- **ANOMALY_WINDOW / FORECAST 算法未生效**：TDengine 3.4.1 community 镜像
+  未包含 `taosanode`（TDgpt 服务进程），所有内置异常检测/预测算法在调用
+  时返回 `0x0443 Analysis algorithm/model not loaded`。当前仅 schema/连接/
+  注入/可视化路径已验证，算法侧需在生产/企业镜像或本地 `taosanode` 部署
+  后再做端到端验证。
+- **`tdengine/tdengine:latest`（企业版 3.3.6）启动卡死**：在容器内
+  `taosd` 一直处于 `0: unavailable` 状态，需进一步排查 `taos.cfg` 与
+  `TAOS_FQDN` 设置；目前不在 Iter-1 验证范围。
+- **`sample_ad_model` 离线模型文件**：检测模块在 `data/models/sample_ad_model`
+  不存在时 graceful skip。即便社区镜像支持 ANOMALY_WINDOW，也需先按
+  TDengine 文档将 `.keras`/`.info` 放置到 anode 模型目录。
+- **GET_FROWTS 区域查询**：当 FORECAST 无返回时（社区版），`SELECT ts, value
+  FROM s_<stable>` 仍会被调用但 `fmap` 为空，所有点 `is_anom=False`。
+  当前为良性，零窗口零异常。
+
+## 验证结论
+
+- **DONE_WITH_CONCERNS**：
+  - 容器、连接、schema、注入、可视化、JSON 日志全链路 verified OK。
+  - 算法/预测侧受限于 community 镜像无 TDgpt 服务，0 窗口/0 异常是
+    已知预期，待企业镜像或 taosanode 部署后补做算法侧端到端验证。
+  - 验证过程中暴露 2 个 SQL 兼容性问题（保留字 + 函数参数），已在
+    commit `031aa35` 修复。
+
+## 仓库
+
+Gitea: <https://git.tekmine.net/charles/ts-anomaly-td>