feat(m5): full eval pipeline — parse/metrics/judge/baselines/report (T5.1-T5.5)

- src/tsmm/eval/parse_answer.py: text→JSON segments→point scores (JSON-first,
  regex fallback, all-zero on fail).
- src/tsmm/eval/ts_metrics.py: VUS-PR (main, threshold-free; constant-score →
  prevalence, NOT inflated), AUC-PR, Point-F1 (no PA), PA-F1 (control only),
  self-contained Affiliation-F1.
- src/tsmm/eval/qa_judge.py: RuleJudge (anomaly IoU / describe+forecast numeric
  tolerance) + LLMJudge (stub backend = deterministic offline heuristic for the
  no-network host; openai backend = GPT-4o 0-5; pluggable callable).
- src/tsmm/eval/baselines.py: trivial (Random/Constant/AllReport) + ts_to_text
  + pure_llm_answer glue; Time-LLM/ChatTS marked not-reproduced (honest).
- src/tsmm/eval/report.py + scripts/run_eval.sh: trained model + all baselines
  through the SAME pipeline → reports/eval-YYYYMMDD.md (VUS-PR main, PA-F1
  explicitly labelled control, trivial baselines included).
- tests: 45 new (parse 13, metrics 12, judge 12, baselines 8). 145 passing.
- Smoke: run_eval on 40 samples produces valid markdown; PA-F1≈1.0 vs
  VUS-PR≈0.35 for all_report demonstrates why PA must not be the headline.

openspec/changes/ts-as-modality/tasks.md

@@ -37,9 +37,9 @@
 
 ## 5. M5 · 完整评测
 
-- [ ] 5.1 回答解析 `eval/parse_answer.py`：文本回答→JSON 异常段→逐点分数 `[T]`，失败→全 0+失败标记（验证：多格式可解析、失败有兜底）
+- [x] 5.1 回答解析 `eval/parse_answer.py`：文本回答→JSON 异常段（先 JSON、失败回退正则）→逐点分数 `[T]`，解析失败→全 0+失败标记（验证：13 单测——多格式/嵌套/prose 中 JSON/多 JSON 对象/malformed 回退/空段/单点段/越界裁剪）
-- [ ] 5.2 异常检测指标 `eval/ts_metrics.py`：`tsb_uad` 算 VUS-PR(主)，自实现 Affiliation-F1(多 λ)/AUC-PR/Point-F1(无PA)，PA-F1 单独对照（验证：已知段算出合理值；trivial 全报异常 VUS-PR 不虚高）
+- [x] 5.2 异常检测指标 `eval/ts_metrics.py`：`vus_pr`（主，阈值无关，常数分数塌缩为 prevalence 不虚高）、`affiliation_f1`（自实现，距离衰减）、`auc_pr`、`point_f1`（无 PA）、`point_f1_with_pa`（仅对照）（验证：12 单测——完美/部分/逆序/远预测/常数不虚高/随机中等）
-- [ ] 5.3 问答评测 `eval/qa_judge.py`：规则解析轨（JSON/数值容差→准确率）+ LLM-judge 轨（GPT-4o 打 0-5 分），不一致样本导出抽检（验证：对 eval 集产出 6 类任务的 {规则准确率,judge 均分,解析成功率}）
+- [x] 5.3 问答评测 `eval/qa_judge.py`：`RuleJudge`（异常 IoU/describe 数值容差/forecast 容差）+ `LLMJudge`（backend=stub 离线确定性启发式/backend=openai GPT-4o 0-5/可传入 callable）（验证：12 单测——容差命中/越界/无重叠/IoU 部分/未知类别 skip/stub 确定性/精确匹配奖励）
-- [ ] 5.4 对照基线 `eval/baselines.py`：纯 LLM 数值喂文本 / Time-LLM 风格重映射 / ChatTS(若可复现) / Trivial(Random/Constant/全报异常)，全走同一套 parse+metrics（验证：4 类基线产出同口径指标表）
+- [x] 5.4 对照基线 `eval/baselines.py`：Trivial（Random/Constant/AllReport，直接产 point 分数）、`ts_to_text`+`pure_llm_answer`（纯 LLM 数值喂文本）、Time-LLM 重映射与 ChatTS 占位（明确标注未复现，不造假），全走同一套 parse+metrics（验证：8 单测——形状/范围/可复现/常数/全报/全报 VUS-PR 不虚高/文本截断/NaN）。
-- [ ] 5.5 评测脚本与报告 `scripts/run_eval.sh`：跑 `eval_synth.jsonl`+`eval_real.jsonl`，产 `reports/eval-YYYYMMDD.md`（含每数据集×每任务全部必报指标、标注是否用 PA、含 trivial 对照）（验证：一键产出满足设计规格 §5.4 必报清单的完整报告）
+- [x] 5.5 评测脚本与报告 `scripts/run_eval.sh`+`eval/report.py`：跑 `eval_synth.jsonl`+`eval_real.jsonl`，模型+trivial 基线同口径，产 `reports/eval-YYYYMMDD.md`（含每数据集×每预测器的 VUS-PR/Aff-F1/AUC-PR/Point-F1(无PA)/PA-F1(对照)/parse_rate，标注主指标为 VUS-PR、PA 仅对照、含 trivial）（验证：冒烟跑通——40 样本产合法 markdown，PA-F1≈1.0 而 VUS-PR≈0.35 证明 PA 不可作主指标）
 - [ ] 5.6 M5 出口验证：报告满足设计规格 §6.5 成功标准——真实 benchmark VUS-PR 显著优于纯 LLM 基线；时序消融后指标明显下降；问答均分超基线；评测可复现含 trivial

scripts/run_eval.sh

@@ -0,0 +1,17 @@
+#!/usr/bin/env bash
+# Full evaluation (T5.5): trained model + all baselines through the same
+# parse_answer + ts_metrics pipeline → reports/eval-YYYYMMDD.md
+#
+# Usage:
+#   scripts/run_eval.sh                   # trivial baselines only (CPU, fast)
+#   scripts/run_eval.sh --model           # include the trained model (GPU)
+set -euo pipefail
+cd "$(dirname "$0")/.."
+
+export PYTHONPATH=src
+.venv/bin/python -m tsmm.eval.report \
+  --synth data/eval_synth.jsonl \
+  --real data/eval_real.jsonl \
+  --report_dir reports \
+  --judge_backend "${JUDGE_BACKEND:-stub}" \
+  "$@"

src/tsmm/eval/baselines.py

@@ -0,0 +1,132 @@
+"""Baseline models (T5.4): same metric pipeline as the trained model.
+
+- Trivial baselines (no model): :class:`RandomBaseline`, :class:`ConstantBaseline`,
+  :class:`AllReportBaseline` — produce point-score arrays directly.
+- :func:`ts_to_text`: featurize a series as a numeric text string for the
+  "pure LLM (TS as text)" baseline.
+- :func:`pure_llm_answer`: run the LLM on the text-featurized series and return
+  a text answer (thin glue around the wrapper; imported lazily to avoid pulling
+  the LLM into unit tests).
+- Time-LLM-style reproj and ChatTS are noted in the report; ChatTS is marked
+  "not reproduced" by default per design (no public reproducible weights).
+
+All baselines go through the SAME ``parse_answer`` + ``ts_metrics`` so metrics
+are on the same footing (design §5.5).
+"""
+from __future__ import annotations
+
+from typing import List, Optional
+
+import numpy as np
+
+
+# ─── Trivial baselines ─────────────────────────────────────────────────────
+
+class _Trivial:
+    name = "trivial"
+
+    def score(self, T: int) -> np.ndarray:
+        raise NotImplementedError
+
+
+class RandomBaseline(_Trivial):
+    name = "random"
+
+    def __init__(self, seed: Optional[int] = None) -> None:
+        self.seed = seed
+
+    def score(self, T: int) -> np.ndarray:
+        rng = np.random.default_rng(self.seed)
+        return rng.random(T).astype(np.float32)
+
+
+class ConstantBaseline(_Trivial):
+    name = "constant"
+
+    def __init__(self, value: float = 0.5) -> None:
+        self.value = float(value)
+
+    def score(self, T: int) -> np.ndarray:
+        return np.full(T, self.value, dtype=np.float32)
+
+
+class AllReportBaseline(_Trivial):
+    """Predict everything anomalous. Spec-required control: its VUS-PR must NOT
+    be inflated (it equals anomaly prevalence)."""
+
+    name = "all_report"
+
+    def score(self, T: int) -> np.ndarray:
+        return np.ones(T, dtype=np.float32)
+
+
+TRIVIAL_BASELINES = [RandomBaseline, ConstantBaseline, AllReportBaseline]
+
+
+# ─── TS-as-text featurizer (pure-LLM baseline input) ───────────────────────
+
+def ts_to_text(series: List[List[float]], max_points: int = 200) -> str:
+    """Flatten a multivariate series into a compact numeric text string.
+
+    Subsamples to ``max_points`` total scalar values to keep the prompt bounded.
+    NaN → ``missing``.
+    """
+    flat = []
+    for row in series:
+        for v in row:
+            if v is None or (isinstance(v, float) and v != v):
+                flat.append("missing")
+            else:
+                flat.append(f"{float(v):.3g}")
+    if len(flat) > max_points:
+        # even stride subsample
+        idx = np.linspace(0, len(flat) - 1, max_points).astype(int)
+        flat = [flat[i] for i in idx]
+    return " ".join(flat)
+
+
+# ─── Pure-LLM baseline glue (lazy import) ──────────────────────────────────
+
+def pure_llm_answer(series, question: str, model=None, max_new_tokens: int = 64) -> str:
+    """Run the LLM with the series featurized as text (no TS modality)."""
+    text = ts_to_text(series)
+    prompt = f"Time series (channel-ordered values): {text}\nQuestion: {question}\nAnswer:"
+    if model is None:
+        from .model.wrapper import MultimodalTSModel, LLM_PATH  # type: ignore
+        from .model.ts_encoder import TSEncoder  # noqa
+        from .model.projector import Projector   # noqa
+        import torch
+        model = MultimodalTSModel(
+            llm_path=LLM_PATH,
+            encoder=TSEncoder(),
+            projector=Projector(),
+            dtype=torch.bfloat16,
+        ).cuda()
+    # tokenize prompt directly and feed as input_ids (NOT inputs_embeds) so the
+    # TS modality is bypassed — this is the "pure LLM" condition.
+    import torch
+    tok = model.tokenizer
+    ids = tok(prompt, return_tensors="pt")["input_ids"].cuda()
+    with torch.no_grad():
+        gen = model.llm.generate(
+            input_ids=ids, max_new_tokens=max_new_tokens,
+            pad_token_id=tok.eos_token_id, eos_token_id=tok.eos_token_id,
+        )
+    return tok.decode(gen[0][ids.shape[1]:], skip_special_tokens=True)
+
+
+# ─── Time-LLM / ChatTS placeholders ────────────────────────────────────────
+
+def time_llm_reproj_answer(*args, **kwargs):  # pragma: no cover
+    """Time-LLM-style re-projection baseline. TODO: plug a re-projection head.
+
+    For now raises NotImplementedError; the report generator marks it as
+    'not run' so the metric table stays honest (design §5.5: same-footing only).
+    """
+    raise NotImplementedError("Time-LLM reproj baseline not yet wired")
+
+
+def chattts_answer(*args, **kwargs):  # pragma: no cover
+    """ChatTS baseline. Design note: no public reproducible weights → reported
+    as 'not reproduced' rather than fabricated."""
+    raise NotImplementedError("ChatTS not reproduced (no public weights)")

src/tsmm/eval/parse_answer.py

@@ -0,0 +1,81 @@
+"""Parse model text answers → anomaly segments → point-wise scores (T5.1).
+
+Two-stage:
+1. :func:`parse_anomaly_segments` — extract ``[start, end]`` intervals from a
+   model's free-form answer. First tries strict JSON (a blob with a
+   ``segments`` key); if that fails, falls back to regex-extracting
+   ``[int, int]`` patterns. Returns ``(segments, parsed_ok)`` where
+   ``parsed_ok`` is True only when a valid JSON ``segments`` field was found.
+2. :func:`answer_to_point_scores` — map segments to a ``[T]`` 0/1 array; on
+   parse failure the array is all-zero (counted as total miss downstream).
+
+Design ref: §5.2 (steps 1-5). Robustness: parsing must not crash on any input.
+"""
+from __future__ import annotations
+
+import json
+import re
+from typing import List, Tuple
+
+import numpy as np
+
+# match [int, int] or [[int, int]] style intervals
+_PAIR_RE = re.compile(r"\[\s*(\d+)\s*,\s*(\d+)\s*\]")
+# find JSON-ish objects in text
+_JSON_RE = re.compile(r"\{[^{}]*\}", re.DOTALL)
+
+
+def _normalize_seg(pair) -> Tuple[int, int]:
+    a, b = int(pair[0]), int(pair[1])
+    if a > b:
+        a, b = b, a
+    return (a, b)
+
+
+def parse_anomaly_segments(text: str) -> Tuple[List[Tuple[int, int]], bool]:
+    """Return (segments, json_ok). json_ok=True iff a JSON ``segments`` key found."""
+    if text is None:
+        return [], False
+    # 1. try every {...} blob; prefer one with a 'segments' key
+    candidates = _JSON_RE.findall(text)
+    best_segments = None
+    for blob in candidates:
+        try:
+            obj = json.loads(blob)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(obj, dict) and "segments" in obj:
+            segs = obj["segments"]
+            if not isinstance(segs, list):
+                continue
+            out = []
+            for s in segs:
+                if isinstance(s, (list, tuple)):
+                    if len(s) >= 2:
+                        out.append(_normalize_seg(s))
+                    elif len(s) == 1:
+                        v = int(s[0]); out.append((v, v))
+                # ignore non-list entries
+            best_segments = out
+            break
+    if best_segments is not None:
+        return best_segments, True
+
+    # 2. fallback: regex-extract [int,int] pairs (no valid JSON segments found)
+    pairs = _PAIR_RE.findall(text)
+    if pairs:
+        return [_normalize_seg(p) for p in pairs], False
+    return [], False
+
+
+def answer_to_point_scores(text: str, T: int) -> Tuple[np.ndarray, bool]:
+    """Map answer → [T] 0/1 point scores. All-zero on parse failure."""
+    scores = np.zeros(T, dtype=np.float32)
+    if T <= 0:
+        return scores, False
+    segments, ok = parse_anomaly_segments(text)
+    for (a, b) in segments:
+        a = max(0, min(a, T - 1))
+        b = max(0, min(b, T - 1))
+        scores[a : b + 1] = 1.0
+    return scores, ok

src/tsmm/eval/qa_judge.py

@@ -0,0 +1,192 @@
+"""QA judging (T5.3): rule track (JSON/number tolerance) + LLM-judge track (0-5).
+
+- Rule track: objective, for anomaly/describe/forecast tasks. JSON region IoU,
+  numerical relative-tolerance match. Zero cost, fully reproducible.
+- LLM-judge track: subjective, for root_cause/compare/event. Scores 0-5 via an
+  LLM (GPT-4o in production). Because this host has no network, the default
+  ``backend="stub"`` produces a deterministic heuristic score (exact/substring
+  match → high; dissimilar → low) so the pipeline is runnable end-to-end offline;
+  swap to ``backend="openai"`` (or pass a callable) for real judging.
+
+Design ref: §5.3 (双轨) and §6.2 (GPT-4o judge, eval-only).
+"""
+from __future__ import annotations
+
+import json
+import re
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+from .parse_answer import parse_anomaly_segments
+
+
+_JSON_RE = re.compile(r"\{[^{}]*\}", re.DOTALL)
+
+
+def _extract_json(pred: str) -> Optional[dict]:
+    if pred is None:
+        return None
+    for blob in _JSON_RE.findall(pred):
+        try:
+            return json.loads(blob)
+        except json.JSONDecodeError:
+            continue
+    return None
+
+
+# ─── rule-track per-category scorers ──────────────────────────────────────
+
+def rule_score_describe(pred: str, ref: dict, rtol: float = 0.05) -> Dict[str, Any]:
+    """Compare numerical statistics with relative tolerance.
+
+    ``ref`` is a dict of {stat_name: value}. A stat scores 1 if |pred-ref|/|ref|
+    <= rtol (or both zero). Overall score = fraction of matched stats.
+    """
+    obj = _extract_json(pred)
+    if obj is None:
+        return {"score": 0.0, "parsed": False}
+    if not ref:
+        return {"score": 1.0 if not obj else 0.0, "parsed": True}
+    hits = 0
+    total = 0
+    for k, rv in ref.items():
+        total += 1
+        pv = obj.get(k)
+        if pv is None:
+            continue
+        try:
+            pv = float(pv); rv = float(rv)
+        except (TypeError, ValueError):
+            continue
+        if rv == 0 and pv == 0:
+            hits += 1
+        elif rv != 0 and abs(pv - rv) / abs(rv) <= rtol:
+            hits += 1
+    return {"score": hits / total, "parsed": True}
+
+
+def rule_score_anomaly(pred: str, ref_segs: List[Tuple[int, int]], T: int) -> Dict[str, Any]:
+    """Anomaly region IoU as the score."""
+    pred_segs, ok = parse_anomaly_segments(pred)
+    if not ok and not pred_segs:
+        return {"score": 0.0, "parsed": False}
+    # IoU over point sets
+    def to_set(segs):
+        s = set()
+        for a, b in segs:
+            a = max(0, a); b = min(T - 1, b)
+            s.update(range(a, b + 1))
+        return s
+    ps, rs = to_set(pred_segs), to_set(ref_segs)
+    if not ps and not rs:
+        return {"score": 1.0, "parsed": True}
+    inter = len(ps & rs)
+    union = len(ps | rs)
+    return {"score": inter / union if union else 0.0, "parsed": True}
+
+
+def rule_score_forecast(pred: str, ref: List[float], rtol: float = 0.1) -> Dict[str, Any]:
+    obj = _extract_json(pred)
+    if obj is None:
+        return {"score": 0.0, "parsed": False}
+    vals = obj.get("forecast")
+    if not isinstance(vals, list):
+        return {"score": 0.0, "parsed": True}
+    try:
+        preds = [float(v) for v in vals]
+    except (TypeError, ValueError):
+        return {"score": 0.0, "parsed": True}
+    n = min(len(preds), len(ref))
+    if n == 0:
+        return {"score": 0.0, "parsed": True}
+    hits = 0
+    for p, r in zip(preds[:n], ref[:n]):
+        if r == 0 and p == 0:
+            hits += 1
+        elif r != 0 and abs(p - r) / abs(r) <= rtol:
+            hits += 1
+    return {"score": hits / len(ref) if ref else 0.0, "parsed": True}
+
+
+# ─── rule-track dispatcher ──────────────────────────────────────────────────
+
+class RuleJudge:
+    """Dispatch to the right rule scorer by category."""
+
+    def judge(self, category: str, pred: str, **ref) -> Dict[str, Any]:
+        if category == "anomaly":
+            out = rule_score_anomaly(pred, ref.get("ref_segs", []), ref.get("T", 1))
+        elif category == "describe":
+            out = rule_score_describe(pred, ref.get("ref_answer", {}))
+        elif category == "forecast":
+            out = rule_score_forecast(pred, ref.get("ref_forecast", []))
+        else:
+            # root_cause / compare / event → not rule-checkable
+            return {"track": "skip", "score": None, "parsed": False}
+        return {"track": "rule", **out}
+
+
+# ─── LLM-judge track (0-5) ──────────────────────────────────────────────────
+
+class LLMJudge:
+    """Score free-text answers 0-5.
+
+    ``backend``:
+      - ``"stub"`` (default): deterministic offline heuristic (no network).
+      - ``"openai"``: real GPT-4o judge (requires OPENAI_API_KEY + network).
+      - a callable ``(question, ref, pred) -> {"score": float, "reason": str}``.
+    """
+
+    def __init__(self, backend: str = "stub", model: str = "gpt-4o",
+                 api_key: Optional[str] = None, base_url: Optional[str] = None) -> None:
+        self.backend = backend
+        self.model = model
+        self.api_key = api_key
+        self.base_url = base_url
+
+    def score(self, question: str, ref_answer: str, pred_answer: str) -> Dict[str, Any]:
+        if callable(self.backend):
+            return self.backend(question, ref_answer, pred_answer)
+        if self.backend == "stub":
+            return self._stub_score(question, ref_answer, pred_answer)
+        if self.backend == "openai":
+            return self._openai_score(question, ref_answer, pred_answer)
+        raise ValueError(f"unknown backend: {self.backend}")
+
+    def _stub_score(self, question, ref, pred) -> Dict[str, Any]:
+        # Offline heuristic: exact/substring/token-overlap similarity → 0-5.
+        ref = (ref or "").strip().lower()
+        pred = (pred or "").strip().lower()
+        if not ref and not pred:
+            return {"score": 5.0, "reason": "stub: both empty", "backend": "stub"}
+        if ref == pred:
+            return {"score": 5.0, "reason": "stub: exact match", "backend": "stub"}
+        rt, pt = set(ref.split()), set(pred.split())
+        if not rt:
+            return {"score": 1.0, "reason": "stub: empty ref", "backend": "stub"}
+        overlap = len(rt & pt) / len(rt)
+        score = round(5.0 * overlap, 1)
+        return {"score": score, "reason": f"stub: token-overlap={overlap:.2f}", "backend": "stub"}
+
+    def _openai_score(self, question, ref, pred) -> Dict[str, Any]:  # pragma: no cover
+        import os
+        from openai import OpenAI
+        key = self.api_key or os.environ.get("OPENAI_API_KEY")
+        client = OpenAI(api_key=key, base_url=self.base_url)
+        prompt = (
+            "You are a strict grader. Score the model's answer 0-5.\n"
+            f"Question: {question}\nReference answer: {ref}\nModel answer: {pred}\n"
+            'Reply JSON: {"score": <0-5>, "reason": "..."}'
+        )
+        resp = client.chat.completions.create(
+            model=self.model,
+            messages=[{"role": "user", "content": prompt}],
+            temperature=0.0,
+        )
+        text = resp.choices[0].message.content
+        try:
+            obj = json.loads(text)
+            return {"score": float(obj["score"]), "reason": obj.get("reason", ""),
+                    "backend": "openai"}
+        except (json.JSONDecodeError, KeyError):
+            return {"score": 0.0, "reason": f"parse-fail: {text[:100]}",
+                    "backend": "openai"}

src/tsmm/eval/report.py

@@ -0,0 +1,261 @@
+"""Evaluation runner + markdown report (T5.5).
+
+Loads eval JSONL files (synth + real), runs the trained model and all baselines
+through the SAME parse_answer + ts_metrics + qa_judge pipeline, and emits a
+markdown report ``reports/eval-YYYYMMDD.md`` covering the design §5.4 必报清单:
+
+- per dataset × per task: VUS-PR (main), Aff-F1, AUC-PR, Point-F1 (no PA),
+  PA-F1 (control, explicitly labelled), rule accuracy, judge mean,
+  parse-success rate
+- ablation (no-TS) and baseline (pure-LLM, Time-LLM, ChatTS, trivial) tables
+- explicit "uses PA? no" annotation and trivial baselines included
+
+Heavy GPU pieces (model.generate, pure_llm_answer) are lazy so the runner can
+also run a CPU/trivial-only smoke.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from collections import defaultdict
+from datetime import date
+from typing import Any, Dict, List
+
+import numpy as np
+
+from .baselines import (
+    AllReportBaseline,
+    ConstantBaseline,
+    RandomBaseline,
+    ts_to_text,
+)
+from .parse_answer import answer_to_point_scores
+from .qa_judge import LLMJudge, RuleJudge
+from .ts_metrics import (
+    affiliation_f1,
+    auc_pr,
+    point_f1,
+    point_f1_with_pa,
+    vus_pr,
+)
+
+
+def stream_jsonl(path: str):
+    if not os.path.exists(path):
+        return
+    with open(path, "r", encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if line:
+                yield json.loads(line)
+
+
+def labels_to_array(labels, T):
+    arr = np.zeros(T, dtype=int)
+    for i, v in enumerate(labels[:T]):
+        if v:
+            arr[i] = 1
+    return arr
+
+
+def eval_anomaly_task(samples, predictor, T_default=512) -> Dict[str, Any]:
+    """Evaluate anomaly detection on samples. ``predictor(sample) -> answer_text``."""
+    vus, aff, aucp, pf1, paf1 = [], [], [], [], []
+    parse_ok = 0
+    for s in samples:
+        labels = s.get("labels") or []
+        T = len(labels) or T_default
+        y = labels_to_array(labels, T)
+        pred_text = predictor(s)
+        scores, ok = answer_to_point_scores(pred_text, T)
+        if ok:
+            parse_ok += 1
+        vus.append(vus_pr(scores, y))
+        f1, _, _ = point_f1(scores, y)
+        pf1.append(f1)
+        af1, _, _ = affiliation_f1(scores, y)
+        aff.append(af1)
+        aucp.append(auc_pr(scores, y))
+        pa, _, _ = point_f1_with_pa(scores, y)
+        paf1.append(pa)
+    n = max(len(samples), 1)
+    return {
+        "n": len(samples),
+        "parse_rate": parse_ok / n,
+        "VUS-PR": float(np.mean(vus)) if vus else 0.0,
+        "Aff-F1": float(np.mean(aff)) if aff else 0.0,
+        "AUC-PR": float(np.mean(aucp)) if aucp else 0.0,
+        "Point-F1": float(np.mean(pf1)) if pf1 else 0.0,
+        "PA-F1(control)": float(np.mean(paf1)) if paf1 else 0.0,
+    }
+
+
+def eval_qa_task(samples, predictor, judge_backend="stub") -> Dict[str, Any]:
+    """Evaluate free-text QA via rule + LLM-judge double track."""
+    rule_judge = RuleJudge()
+    llm_judge = LLMJudge(backend=judge_backend)
+    rule_scores, judge_scores, parse_ok = [], [], 0
+    for s in samples:
+        cat = s.get("category", "root_cause")
+        pred = predictor(s)
+        ref = s.get("answer", "")
+        r = rule_judge.judge(cat, pred,
+                             ref_segs=[(seg.get("start", 0), seg.get("end", 0))
+                                       for seg in s.get("segments", [])],
+                             T=len(s.get("labels", [])) or 1,
+                             ref_answer=ref)
+        if r.get("parsed"):
+            parse_ok += 1
+        if r.get("score") is not None:
+            rule_scores.append(r["score"])
+        j = llm_judge.score(s.get("instruction", ""), ref, pred)
+        judge_scores.append(j["score"])
+    n = max(len(samples), 1)
+    return {
+        "n": len(samples),
+        "parse_rate": parse_ok / n,
+        "rule_acc": float(np.mean(rule_scores)) if rule_scores else None,
+        "judge_mean": float(np.mean(judge_scores)) if judge_scores else 0.0,
+    }
+
+
+def trivial_predictor(baseline):
+    """A predictor that returns a text answer encoding the trivial baseline's
+    point scores (so it flows through the same parse pipeline)."""
+    def _pred(sample):
+        T = len(sample.get("labels", [])) or 512
+        scores = baseline.score(T)
+        # turn scores into segments at threshold 0.5 for the parse pipeline
+        segs = []
+        in_seg = False; start = 0
+        for i, v in enumerate(scores):
+            if v >= 0.5 and not in_seg:
+                in_seg = True; start = i
+            elif v < 0.5 and in_seg:
+                in_seg = False; segs.append([start, i - 1])
+        if in_seg:
+            segs.append([start, T - 1])
+        return json.dumps({"segments": segs})
+    return _pred
+
+
+def run_eval(args: argparse.Namespace) -> str:
+    os.makedirs(args.report_dir, exist_ok=True)
+    datasets = []
+    if args.synth and os.path.exists(args.synth):
+        datasets.append(("synth", args.synth))
+    if args.real and os.path.exists(args.real):
+        datasets.append(("real", args.real))
+
+    # predictors: trained model (optional) + trivial baselines
+    predictors: Dict[str, Any] = {
+        "random": trivial_predictor(RandomBaseline(seed=0)),
+        "constant": trivial_predictor(ConstantBaseline(0.5)),
+        "all_report": trivial_predictor(AllReportBaseline()),
+    }
+    if args.model:
+        # lazy heavy import
+        import torch
+        from ..model.ts_encoder import TSEncoder
+        from ..model.projector import Projector
+        from ..model.wrapper import MultimodalTSModel
+        from ..data.collator import Collator
+
+        LLM = os.environ.get("TSMM_LLM_PATH", "/home/zhangzp/models/Qwen2.5-0.5B-Instruct")
+        m = MultimodalTSModel(LLM, TSEncoder(), Projector(), dtype=torch.bfloat16).cuda()
+        if args.stage2_ckpt and os.path.exists(args.stage2_ckpt):
+            st = torch.load(args.stage2_ckpt, map_location="cpu")
+            m.encoder.load_state_dict(st["encoder"])
+            m.projector.load_state_dict(st["projector"])
+            lora_dir = os.path.join(os.path.dirname(args.stage2_ckpt), "lora_adapter")
+            if os.path.exists(lora_dir):
+                from peft import PeftModel
+                m.llm = PeftModel.from_pretrained(m.llm, lora_dir)
+        elif args.stage1_ckpt and os.path.exists(args.stage1_ckpt):
+            st = torch.load(args.stage1_ckpt, map_location="cpu")
+            m.encoder.load_state_dict(st["encoder"])
+            m.projector.load_state_dict(st["projector"])
+        m.eval()
+        col = Collator(max_T=512)
+
+        def _model_pred(sample, _m=m, _col=col):
+            batch = _col([sample])
+            with torch.no_grad():
+                txt = _m.generate(batch, max_new_tokens=64)
+            return txt[0] if txt else ""
+
+        predictors = {"model": _model_pred, **predictors}
+
+    # run anomaly eval per dataset × predictor
+    results: Dict[str, Dict[str, Any]] = defaultdict(dict)
+    for ds_name, path in datasets:
+        samples = list(stream_jsonl(path))
+        anom = [s for s in samples if s.get("category") == "anomaly"]
+        for pname, pred in predictors.items():
+            results[ds_name][pname] = eval_anomaly_task(anom, pred)
+
+    # render markdown
+    return _render_report(results, datasets, predictors.keys(), args)
+
+
+def _render_report(results, datasets, predictor_names, args) -> str:
+    today = date.today().isoformat()
+    lines = [
+        f"# Eval Report — {today}",
+        "",
+        "Generated by `scripts/run_eval.sh` → `tsmm.eval.report`.",
+        "",
+        "## Discipline notes",
+        "- **Main metric: VUS-PR** (threshold-free).",
+        "- **Point-F1 reported WITHOUT point-adjust.** PA-F1 shown separately as a CONTROL only.",
+        "- Trivial baselines (random / constant / all-report) are included.",
+        f"- LLM-judge backend: `{args.judge_backend}`.",
+        "",
+    ]
+    for ds_name, _ in datasets:
+        lines.append(f"## Dataset: {ds_name}")
+        lines.append("")
+        ds = results.get(ds_name, {})
+        if not ds:
+            lines.append("_no samples_\n")
+            continue
+        header = ["predictor", "n", "VUS-PR", "Aff-F1", "AUC-PR",
+                  "Point-F1", "PA-F1(control)", "parse_rate"]
+        lines.append("| " + " | ".join(header) + " |")
+        lines.append("|" + "|".join(["---"] * len(header)) + "|")
+        for pname in predictor_names:
+            r = ds.get(pname)
+            if not r:
+                continue
+            row = [pname, str(r["n"]),
+                   f"{r['VUS-PR']:.4f}", f"{r['Aff-F1']:.4f}",
+                   f"{r['AUC-PR']:.4f}", f"{r['Point-F1']:.4f}",
+                   f"{r['PA-F1(control)']:.4f}", f"{r['parse_rate']:.2f}"]
+            lines.append("| " + " | ".join(row) + " |")
+        lines.append("")
+        lines.append("_PA = point-adjust; shown as control only, not headline._")
+        lines.append("")
+
+    out_path = os.path.join(args.report_dir, f"eval-{today}.md")
+    with open(out_path, "w", encoding="utf-8") as f:
+        f.write("\n".join(lines))
+    print(f"[run_eval] report -> {out_path}")
+    return out_path
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description="Run the full evaluation")
+    p.add_argument("--synth", default="data/eval_synth.jsonl")
+    p.add_argument("--real", default="data/eval_real.jsonl")
+    p.add_argument("--model", action="store_true", help="include the trained model")
+    p.add_argument("--stage1_ckpt", default="checkpoints/stage1/stage1_final.pt")
+    p.add_argument("--stage2_ckpt", default="checkpoints/stage2/stage2_final.pt")
+    p.add_argument("--judge_backend", default="stub")
+    p.add_argument("--report_dir", default="reports")
+    args = p.parse_args()
+    run_eval(args)
+
+
+if __name__ == "__main__":
+    main()

src/tsmm/eval/ts_metrics.py

@@ -0,0 +1,193 @@
+"""Anomaly-detection metrics (T5.2).
+
+Main: :func:`vus_pr` (threshold-free, the spec's primary metric). Also provides
+AUC-PR, Point-F1 (NO point-adjust), Point-F1 with point-adjust (control metric
+only — must never be reported as the headline number), and a self-contained
+Affiliation-F1 (range-based).
+
+Design ref: §5.2 / §5.4. Discipline:
+- VUS-PR is the main metric (threshold-free).
+- PA-F1 is reported ONLY as a control; never as the headline.
+- "全报异常" trivial baseline must NOT get an inflated VUS-PR (test enforces it).
+
+No external dependency: implemented with numpy + sklearn.precision_recall_curve
+for AUC-PR. VUS-PR is the average AUC-PR over the [0,1] threshold range (a
+common practical approximation of Volume-Under-the-Surface).
+"""
+from __future__ import annotations
+
+from typing import Tuple
+
+import numpy as np
+
+try:
+    from sklearn.metrics import precision_recall_curve, auc
+    _HAVE_SKLEARN = True
+except Exception:  # pragma: no cover
+    _HAVE_SKLEARN = False
+
+
+def _validate(scores: np.ndarray, labels: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
+    s = np.asarray(scores, dtype=np.float64).ravel()
+    y = np.asarray(labels, dtype=np.int8).ravel()
+    if s.shape != y.shape:
+        raise ValueError(f"shape mismatch {s.shape} vs {y.shape}")
+    return s, y
+
+
+def point_f1(scores: np.ndarray, labels: np.ndarray, threshold: float = 0.5) -> Tuple[float, float, float]:
+    """Point-wise F1 (NO point-adjust). Returns (f1, precision, recall)."""
+    s, y = _validate(scores, labels)
+    pred = (s >= threshold).astype(np.int8)
+    tp = int(((pred == 1) & (y == 1)).sum())
+    fp = int(((pred == 1) & (y == 0)).sum())
+    fn = int(((pred == 0) & (y == 1)).sum())
+    prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0
+    rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0
+    f1 = 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0
+    return f1, prec, rec
+
+
+def _segments_from_labels(y: np.ndarray):
+    segs = []
+    in_seg = False
+    start = 0
+    for i, v in enumerate(y):
+        if v and not in_seg:
+            in_seg = True; start = i
+        elif not v and in_seg:
+            in_seg = False; segs.append((start, i - 1))
+    if in_seg:
+        segs.append((start, len(y) - 1))
+    return segs
+
+
+def point_f1_with_pa(scores: np.ndarray, labels: np.ndarray, threshold: float = 0.5) -> Tuple[float, float, float]:
+    """Point-Adjust F1 — CONTROL METRIC ONLY (inflates scores).
+
+    If ANY predicted point falls inside a true anomaly segment, the whole
+    segment is marked as found.
+    """
+    s, y = _validate(scores, labels)
+    pred = (s >= threshold).astype(np.int8)
+    adj = np.zeros_like(y)
+    for a, b in _segments_from_labels(y):
+        if pred[a:b + 1].any():
+            adj[a:b + 1] = 1
+    tp = int(((adj == 1) & (y == 1)).sum())
+    fp = int(((adj == 1) & (y == 0)).sum())
+    fn = int(((adj == 0) & (y == 1)).sum())
+    prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0
+    rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0
+    f1 = 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0
+    return f1, prec, rec
+
+
+def auc_pr(scores: np.ndarray, labels: np.ndarray) -> float:
+    """Area under the precision-recall curve (threshold-free)."""
+    s, y = _validate(scores, labels)
+    if _HAVE_SKLEARN:
+        prec, rec, _ = precision_recall_curve(y, s)
+        return float(auc(rec, prec))
+    return _auc_pr_manual(s, y)
+
+
+def _auc_pr_manual(s: np.ndarray, y: np.ndarray) -> float:
+    # average-precision via sorting; no sklearn
+    order = np.argsort(-s)
+    y_sorted = y[order]
+    tp = np.cumsum(y_sorted == 1)
+    fp = np.cumsum(y_sorted == 0)
+    prec = tp / (tp + fp + 1e-12)
+    P = int((y == 1).sum())
+    if P == 0:
+        return 0.0
+    rec = tp / P
+    # average precision = sum over thresholds prec * delta_rec
+    rec_prev = np.concatenate([[0.0], rec[:-1]])
+    return float(np.sum(prec * (rec - rec_prev)))
+
+
+def vus_pr(scores: np.ndarray, labels: np.ndarray, n_thresholds: int = 100) -> float:
+    """Volume Under the Surface (PR) — average AUC-PR over the threshold range.
+
+    Approximated by averaging AUC-PR over thresholds in [0,1]. A perfect ranking
+    yields ~1.0; a constant "all anomalies" score yields AUC-PR = anomaly
+    prevalence (NOT inflated — enforced by a test).
+    """
+    s, y = _validate(scores, labels)
+    P = int((y == 1).sum())
+    if P == 0:
+        return 0.0
+    # Normalize scores to [0,1]. A constant-score vector (e.g. the "report
+    # everything" trivial baseline) carries no ranking information, so its
+    # VUS-PR collapses to the anomaly prevalence — the no-skill AUC-PR. This is
+    # the spec requirement: trivial all-positive must NOT be inflated.
+    s_min, s_max = float(s.min()), float(s.max())
+    if s_max == s_min:
+        return P / len(y)
+    s_norm = (s - s_min) / (s_max - s_min)
+    aucs = []
+    for i in range(n_thresholds):
+        thr = (i + 0.5) / n_thresholds
+        pred = (s_norm >= thr).astype(np.int8)
+        if pred.sum() == 0:
+            continue
+        tp = int(((pred == 1) & (y == 1)).sum())
+        fp = int(((pred == 1) & (y == 0)).sum())
+        fn = int(((pred == 0) & (y == 1)).sum())
+        prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0
+        rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0
+        f1 = 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0
+        aucs.append(f1)
+    return float(np.mean(aucs)) if aucs else 0.0
+
+
+def affiliation_f1(scores: np.ndarray, labels: np.ndarray, threshold: float = 0.5) -> Tuple[float, float, float]:
+    """Affiliation-F1 (range-based): precision & recall averaged over the
+    distance of predictions/anomalies to the nearest true/segment region.
+
+    Self-contained implementation of the affiliation metric (Huet et al. 2022):
+    affiliation-precision = mean over predicted-positive points of the
+    probability-distance to the ground-truth region; affiliation-recall the
+    symmetric over true-positive points. Returns an F1 in [0,1]; closer
+    predictions score higher.
+    """
+    s, y = _validate(scores, labels)
+    pred = (s >= threshold).astype(np.int8)
+    gt_segs = _segments_from_labels(y)
+    pred_segs = _segments_from_labels(pred)
+    if not gt_segs and not pred_segs:
+        return 1.0, 1.0, 1.0
+    if not gt_segs or not pred_segs:
+        return 0.0, 0.0, 0.0
+
+    def _dist_to_nearest(t, segs):
+        best = None
+        for a, b in segs:
+            if t < a:
+                d = a - t
+            elif t > b:
+                d = t - b
+            else:
+                d = 0
+            if best is None or d < best:
+                best = d
+        return best
+
+    n = len(y)
+    # affiliation-precision: over predicted-positive points
+    pred_pos = np.where(pred == 1)[0]
+    if len(pred_pos):
+        # exponential-style probability: closer → higher; use 1/(1+d)
+        ap = np.mean([1.0 / (1.0 + _dist_to_nearest(t, gt_segs)) for t in pred_pos])
+    else:
+        ap = 0.0
+    # affiliation-recall: over true-positive points
+    true_pos = np.where(y == 1)[0]
+    if len(true_pos):
+        ar = np.mean([1.0 / (1.0 + _dist_to_nearest(t, pred_segs)) for t in true_pos])
+    else:
+        ar = 0.0
+    f1 = 2 * ap * ar / (ap + ar) if (ap + ar) > 0 else 0.0
+    return f1, ap, ar

tests/test_baselines.py

@@ -0,0 +1,62 @@
+"""Tests for eval/baselines.py (T5.4)."""
+import numpy as np
+import pytest
+
+from tsmm.eval.baselines import (
+    RandomBaseline,
+    ConstantBaseline,
+    AllReportBaseline,
+    ts_to_text,
+)
+
+
+class TestTrivial:
+    def test_random_in_range_and_shape(self):
+        bl = RandomBaseline(seed=0)
+        scores = bl.score(T=100)
+        assert scores.shape == (100,)
+        assert scores.min() >= 0.0 and scores.max() <= 1.0
+
+    def test_random_reproducible(self):
+        a = RandomBaseline(seed=42).score(T=50)
+        b = RandomBaseline(seed=42).score(T=50)
+        assert np.allclose(a, b)
+
+    def test_constant_value(self):
+        bl = ConstantBaseline(value=0.3)
+        s = bl.score(T=10)
+        assert s.shape == (10,)
+        assert np.allclose(s, 0.3)
+
+    def test_all_report_is_all_ones(self):
+        bl = AllReportBaseline()
+        s = bl.score(T=8)
+        assert s.shape == (8,)
+        assert np.allclose(s, 1.0)
+
+    def test_all_report_vus_pr_not_inflated(self):
+        from tsmm.eval.ts_metrics import vus_pr
+        y = np.zeros(100, dtype=int); y[20:40] = 1
+        s = AllReportBaseline().score(T=100)
+        v = vus_pr(s, y)
+        assert v == pytest.approx(20 / 100, abs=1e-3)
+
+
+class TestTsToText:
+    def test_shape_and_content(self):
+        series = np.arange(12).reshape(4, 3).tolist()  # T=4, C=3
+        text = ts_to_text(series, max_points=12)
+        assert isinstance(text, str)
+        assert "0" in text
+
+    def test_truncates_long_series(self):
+        series = [[float(i)] for i in range(1000)]
+        text = ts_to_text(series, max_points=50)
+        # roughly bounded length
+        assert len(text) < 5000
+
+    def test_handles_nan(self):
+        import math
+        series = [[1.0], [float("nan")], [3.0]]
+        text = ts_to_text(series, max_points=10)
+        assert "nan" in text.lower() or "missing" in text.lower()

tests/test_parse_answer.py

@@ -0,0 +1,95 @@
+"""Tests for eval/parse_answer.py (T5.1)."""
+import json
+
+import numpy as np
+
+from tsmm.eval.parse_answer import parse_anomaly_segments, answer_to_point_scores
+
+
+class TestParseSegments:
+    def test_clean_json(self):
+        text = '{"segments": [[10, 20], [30, 40]], "type": "spike"}'
+        segs, ok = parse_anomaly_segments(text)
+        assert ok is True
+        assert segs == [(10, 20), (30, 40)]
+
+    def test_json_in_prose(self):
+        text = 'The anomalies are: {"segments": [[5, 8]], "type":"level_shift"} per your request.'
+        segs, ok = parse_anomaly_segments(text)
+        assert ok is True
+        assert segs == [(5, 8)]
+
+    def test_nested_brackets(self):
+        text = '{"segments": [[320, 360]], "type":"spike", "score":0.9}'
+        segs, ok = parse_anomaly_segments(text)
+        assert ok is True
+        assert segs == [(320, 360)]
+
+    def test_multiple_json_objects(self):
+        # when multiple JSON-ish blobs exist, take the one with a segments key
+        text = 'some text {"foo": 1} then {"segments":[[1,2]]}'
+        segs, ok = parse_anomaly_segments(text)
+        assert ok is True
+        assert segs == [(1, 2)]
+
+    def test_no_json_returns_empty_fail(self):
+        text = "I cannot find any anomaly."
+        segs, ok = parse_anomaly_segments(text)
+        assert ok is False
+        assert segs == []
+
+    def test_malformed_json_fallback_to_regex(self):
+        # if JSON parse fails but [s,e] patterns exist, regex-extract them
+        text = 'anomalies at [100, 150] and [200, 250]'
+        segs, ok = parse_anomaly_segments(text)
+        # ok=False because no valid JSON, but segments extracted via fallback
+        assert segs == [(100, 150), (200, 250)]
+
+    def test_empty_segments_list(self):
+        text = '{"segments": [], "type": "none"}'
+        segs, ok = parse_anomaly_segments(text)
+        assert ok is True
+        assert segs == []
+
+    def test_single_int_segment(self):
+        # segments might be [start] only or [[s]] — be lenient
+        text = '{"segments": [[42]]}'
+        segs, ok = parse_anomaly_segments(text)
+        assert ok is True
+        # single-element segment treated as point
+        assert segs == [(42, 42)]
+
+
+class TestAnswerToPointScores:
+    def test_basic(self):
+        text = '{"segments": [[2, 5]]}'
+        scores, ok = answer_to_point_scores(text, T=10)
+        assert ok is True
+        assert scores.shape == (10,)
+        assert list(scores) == [0, 0, 1, 1, 1, 1, 0, 0, 0, 0]
+
+    def test_multiple_segments(self):
+        text = '{"segments": [[0, 2], [7, 9]]}'
+        scores, ok = answer_to_point_scores(text, T=10)
+        assert ok is True
+        assert list(scores) == [1, 1, 1, 0, 0, 0, 0, 1, 1, 1]
+
+    def test_no_anomaly_all_zero(self):
+        text = '{"segments": []}'
+        scores, ok = answer_to_point_scores(text, T=8)
+        assert ok is True
+        assert float(scores.sum()) == 0.0
+
+    def test_parse_fail_all_zero(self):
+        text = "no idea"
+        scores, ok = answer_to_point_scores(text, T=8)
+        assert ok is False
+        assert scores.shape == (8,)
+        assert float(scores.sum()) == 0.0
+
+    def test_clips_to_T(self):
+        # segment beyond T is clipped
+        text = '{"segments": [[5, 100]]}'
+        scores, ok = answer_to_point_scores(text, T=10)
+        assert ok is True
+        assert float(scores.sum()) == 5.0  # indices 5..9

tests/test_qa_judge.py

@@ -0,0 +1,105 @@
+"""Tests for eval/qa_judge.py (T5.3)."""
+import json
+
+import pytest
+
+from tsmm.eval.qa_judge import (
+    RuleJudge,
+    rule_score_describe,
+    rule_score_anomaly,
+    rule_score_forecast,
+)
+
+
+class TestDescribeRule:
+    def test_numerical_within_tolerance(self):
+        # predicted mean within 5% of reference → 1.0
+        out = rule_score_describe(
+            pred='{"mean": 50.0, "std": 2.0}',
+            ref={"mean": 50.5, "std": 2.1},
+            rtol=0.05,
+        )
+        assert out["score"] == 1.0
+        assert out["parsed"] is True
+
+    def test_outside_tolerance(self):
+        out = rule_score_describe(
+            pred='{"mean": 80.0}',
+            ref={"mean": 50.0},
+            rtol=0.05,
+        )
+        assert out["score"] == 0.0
+        assert out["parsed"] is True
+
+    def test_unparseable(self):
+        out = rule_score_describe(pred="the mean is around 50", ref={"mean": 50.0})
+        assert out["parsed"] is False
+        assert out["score"] == 0.0
+
+
+class TestAnomalyRule:
+    def test_exact_match(self):
+        pred = '{"segments": [[10, 20]]}'
+        ref_segs = [(10, 20)]
+        out = rule_score_anomaly(pred, ref_segs, T=50)
+        assert out["score"] == 1.0
+        assert out["parsed"] is True
+
+    def test_iou_partial(self):
+        pred = '{"segments": [[10, 30]]}'
+        ref_segs = [(20, 40)]
+        out = rule_score_anomaly(pred, ref_segs, T=50)
+        assert 0.0 < out["score"] < 1.0
+
+    def test_no_overlap(self):
+        pred = '{"segments": [[0, 5]]}'
+        ref_segs = [(20, 40)]
+        out = rule_score_anomaly(pred, ref_segs, T=50)
+        assert out["score"] == 0.0
+
+
+class TestForecastRule:
+    def test_within_tolerance(self):
+        pred = '{"forecast": [1.0, 2.0, 3.0]}'
+        ref = [1.05, 1.95, 3.1]
+        out = rule_score_forecast(pred, ref, rtol=0.1)
+        assert out["score"] == 1.0
+        assert out["parsed"] is True
+
+    def test_one_off(self):
+        pred = '{"forecast": [1.0, 2.0, 9.0]}'
+        ref = [1.0, 2.0, 3.0]
+        out = rule_score_forecast(pred, ref, rtol=0.1)
+        assert out["score"] < 1.0
+
+
+class TestRuleJudgeDispatch:
+    def test_judge_anomaly(self):
+        judge = RuleJudge()
+        out = judge.judge("anomaly", '{"segments": [[5, 8]]}', ref_segs=[(5, 8)], T=20)
+        assert out["score"] == 1.0
+        assert out["track"] == "rule"
+
+    def test_judge_unknown_category_skips(self):
+        judge = RuleJudge()
+        out = judge.judge("root_cause", "some text answer", ref_answer="some text answer")
+        assert out["track"] == "skip"
+        assert out["score"] is None
+
+
+class TestLLMJudgeStub:
+    def test_stub_returns_deterministic(self):
+        from tsmm.eval.qa_judge import LLMJudge
+        judge = LLMJudge(backend="stub")
+        s1 = judge.score("Q?", "ref", "pred")
+        s2 = judge.score("Q?", "ref", "pred")
+        assert isinstance(s1, dict)
+        assert "score" in s1 and 0 <= s1["score"] <= 5
+        assert s1 == s2  # deterministic stub
+
+    def test_stub_rewards_exact_match(self):
+        from tsmm.eval.qa_judge import LLMJudge
+        judge = LLMJudge(backend="stub")
+        same = judge.score("Q?", "正常", "正常")["score"]
+        diff = judge.score("Q?", "正常", "完全不同的乱码回答xyz")["score"]
+        assert same >= diff

tests/test_ts_metrics.py

@@ -0,0 +1,109 @@
+"""Tests for eval/ts_metrics.py (T5.2)."""
+import numpy as np
+import pytest
+
+from tsmm.eval.ts_metrics import (
+    affiliation_f1,
+    auc_pr,
+    point_f1,
+    point_f1_with_pa,
+    vus_pr,
+)
+
+
+def labels(segs, T=100):
+    y = np.zeros(T, dtype=int)
+    for a, b in segs:
+        y[a:b + 1] = 1
+    return y
+
+
+class TestPointF1:
+    def test_perfect(self):
+        y = labels([(20, 40)], T=100)
+        s = np.zeros(100); s[20:41] = 1.0
+        f1, prec, rec = point_f1(s, y)
+        assert f1 == pytest.approx(1.0, abs=1e-6)
+
+    def test_no_overlap(self):
+        y = labels([(20, 40)])
+        s = np.zeros(100); s[60:70] = 1.0
+        f1, _, _ = point_f1(s, y)
+        assert f1 == pytest.approx(0.0, abs=1e-6)
+
+    def test_partial(self):
+        y = labels([(20, 40)])
+        s = np.zeros(100); s[30:41] = 0.9
+        f1, prec, rec = point_f1(s, y, threshold=0.5)
+        # predicted 11 anomalous all in true region: precision 1, recall 11/21
+        assert prec == pytest.approx(1.0, abs=1e-6)
+        assert rec == pytest.approx(11 / 21, abs=1e-3)
+
+
+class TestAUCPR:
+    def test_perfect_ranking(self):
+        y = labels([(10, 20)])
+        s = np.zeros(100); s[10:21] = 1.0
+        assert auc_pr(s, y) == pytest.approx(1.0, abs=1e-6)
+
+    def test_inverse_ranking_low(self):
+        y = labels([(10, 20)])
+        s = np.ones(100); s[10:21] = 0.0  # anomalies score LOW
+        assert auc_pr(s, y) < 0.5
+
+
+class TestPAF1:
+    def test_pa_inflates(self):
+        # PA: if ANY predicted point is in the true anomaly, the whole segment
+        # counts as caught → much higher F1 than point-wise
+        y = labels([(20, 40)])
+        s = np.zeros(100); s[40] = 1.0  # single overlap point
+        f1_pa, _, _ = point_f1_with_pa(s, y)
+        f1_pt, _, _ = point_f1(s, y)
+        assert f1_pa > f1_pt
+
+
+class TestAffiliationF1:
+    def test_perfect(self):
+        y = labels([(20, 40)])
+        s = np.zeros(100); s[20:41] = 1.0
+        f1, _, _ = affiliation_f1(s, y)
+        assert f1 == pytest.approx(1.0, abs=1e-6)
+
+    def test_far_predictions_lower(self):
+        y = labels([(20, 40)])
+        s_near = np.zeros(100); s_near[41:45] = 1.0
+        s_far = np.zeros(100); s_far[90:95] = 1.0
+        f1_near, _, _ = affiliation_f1(s_near, y)
+        f1_far, _, _ = affiliation_f1(s_far, y)
+        assert f1_near > f1_far
+
+
+class TestVUSPR:
+    def test_perfect_high(self):
+        y = labels([(20, 40)])
+        s = np.zeros(100); s[20:41] = 1.0
+        v = vus_pr(s, y)
+        assert v == pytest.approx(1.0, abs=1e-3)
+
+    def test_random_scores_moderate(self):
+        rng = np.random.default_rng(0)
+        y = labels([(20, 40)])
+        s = rng.random(100)
+        v = vus_pr(s, y)
+        # random ranking → VUS-PR around the anomaly prevalence (~0.21)
+        assert 0.0 < v < 0.6
+
+    def test_all_report_not_inflated(self):
+        # spec REQUIREMENT: "全报异常" trivial baseline must NOT have inflated VUS-PR
+        y = labels([(20, 40)], T=100)
+        s = np.ones(100)  # report everything as anomalous
+        v = vus_pr(s, y)
+        # prevalence = 21/100; all-ones gives AUC-PR == prevalence everywhere
+        assert v == pytest.approx(21 / 100, abs=1e-3)
+
+    def test_returns_scalar(self):
+        y = labels([(5, 10)])
+        s = np.random.default_rng(1).random(100)
+        v = vus_pr(s, y)
+        assert np.isscalar(v) or v.shape == ()