diff --git a/openspec/changes/ts-as-modality/tasks.md b/openspec/changes/ts-as-modality/tasks.md index 03b5dcf..197d4be 100644 --- a/openspec/changes/ts-as-modality/tasks.md +++ b/openspec/changes/ts-as-modality/tasks.md @@ -37,9 +37,9 @@ ## 5. M5 · 完整评测 -- [ ] 5.1 回答解析 `eval/parse_answer.py`:文本回答→JSON 异常段→逐点分数 `[T]`,失败→全 0+失败标记(验证:多格式可解析、失败有兜底) -- [ ] 5.2 异常检测指标 `eval/ts_metrics.py`:`tsb_uad` 算 VUS-PR(主),自实现 Affiliation-F1(多 λ)/AUC-PR/Point-F1(无PA),PA-F1 单独对照(验证:已知段算出合理值;trivial 全报异常 VUS-PR 不虚高) -- [ ] 5.3 问答评测 `eval/qa_judge.py`:规则解析轨(JSON/数值容差→准确率)+ LLM-judge 轨(GPT-4o 打 0-5 分),不一致样本导出抽检(验证:对 eval 集产出 6 类任务的 {规则准确率,judge 均分,解析成功率}) -- [ ] 5.4 对照基线 `eval/baselines.py`:纯 LLM 数值喂文本 / Time-LLM 风格重映射 / ChatTS(若可复现) / Trivial(Random/Constant/全报异常),全走同一套 parse+metrics(验证:4 类基线产出同口径指标表) -- [ ] 5.5 评测脚本与报告 `scripts/run_eval.sh`:跑 `eval_synth.jsonl`+`eval_real.jsonl`,产 `reports/eval-YYYYMMDD.md`(含每数据集×每任务全部必报指标、标注是否用 PA、含 trivial 对照)(验证:一键产出满足设计规格 §5.4 必报清单的完整报告) +- [x] 5.1 回答解析 `eval/parse_answer.py`:文本回答→JSON 异常段(先 JSON、失败回退正则)→逐点分数 `[T]`,解析失败→全 0+失败标记(验证:13 单测——多格式/嵌套/prose 中 JSON/多 JSON 对象/malformed 回退/空段/单点段/越界裁剪) +- [x] 5.2 异常检测指标 `eval/ts_metrics.py`:`vus_pr`(主,阈值无关,常数分数塌缩为 prevalence 不虚高)、`affiliation_f1`(自实现,距离衰减)、`auc_pr`、`point_f1`(无 PA)、`point_f1_with_pa`(仅对照)(验证:12 单测——完美/部分/逆序/远预测/常数不虚高/随机中等) +- [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 确定性/精确匹配奖励) +- [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)。 +- [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 diff --git a/scripts/run_eval.sh b/scripts/run_eval.sh new file mode 100755 index 0000000..4c8a107 --- /dev/null +++ b/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}" \ + "$@" diff --git a/src/tsmm/eval/baselines.py b/src/tsmm/eval/baselines.py new file mode 100644 index 0000000..0bd1546 --- /dev/null +++ b/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)") diff --git a/src/tsmm/eval/parse_answer.py b/src/tsmm/eval/parse_answer.py new file mode 100644 index 0000000..d7a1fbe --- /dev/null +++ b/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 diff --git a/src/tsmm/eval/qa_judge.py b/src/tsmm/eval/qa_judge.py new file mode 100644 index 0000000..b80b19e --- /dev/null +++ b/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"} diff --git a/src/tsmm/eval/report.py b/src/tsmm/eval/report.py new file mode 100644 index 0000000..7436b70 --- /dev/null +++ b/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() diff --git a/src/tsmm/eval/ts_metrics.py b/src/tsmm/eval/ts_metrics.py new file mode 100644 index 0000000..b0bee69 --- /dev/null +++ b/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 diff --git a/tests/test_baselines.py b/tests/test_baselines.py new file mode 100644 index 0000000..e8d1632 --- /dev/null +++ b/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() diff --git a/tests/test_parse_answer.py b/tests/test_parse_answer.py new file mode 100644 index 0000000..561f616 --- /dev/null +++ b/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 diff --git a/tests/test_qa_judge.py b/tests/test_qa_judge.py new file mode 100644 index 0000000..63d6a03 --- /dev/null +++ b/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 diff --git a/tests/test_ts_metrics.py b/tests/test_ts_metrics.py new file mode 100644 index 0000000..30fc2bd --- /dev/null +++ b/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 == ()