#!/usr/bin/env python3 """ test_quickstart.py 功能： 1. 构造 1 小时窗口，演示实时异常检测（正常 / 异常两种场景） 2. 构造 7 天数据，演示异常模式聚合 3. 输出格式化的LLM文案，方便直接接入大模型 运行方式： python test_quickstart.py """ from __future__ import annotations import sys from pathlib import Path from datetime import datetime, timedelta import json import numpy as np import random import random ROOT_DIR = Path(__file__).parent.resolve() sys.path.insert(0, str(ROOT_DIR)) from wearable_anomaly_detector import WearableAnomalyDetector import importlib.util # 动态导入 utils.formatter，避免相对路径问题 formatter_spec = importlib.util.spec_from_file_location( "formatter", ROOT_DIR / "utils" / "formatter.py" ) formatter_module = importlib.util.module_from_spec(formatter_spec) formatter_spec.loader.exec_module(formatter_module) AnomalyFormatter = formatter_module.AnomalyFormatter FORMATTER = AnomalyFormatter() TEST_WINDOW_FILE = ROOT_DIR / "test_data" / "example_window.json" WINDOW_SIZE = 12 # 12 * 5 分钟 = 1 小时 INTERVAL_MINUTES = 5 def make_point(ts: datetime, device_id: str, hrv: float, hr: float, include_static: bool = True) -> dict: """构造单个数据点""" return { "timestamp": ts.isoformat(), "deviceId": device_id, "features": { "hr": float(hr), "hr_resting": 65.0, "hrv_rmssd": float(hrv), "hrv_sdnn": float(hrv * 1.2), "time_period_primary": "day", "time_period_secondary": "workday", "is_weekend": 0.0, "data_quality": "high", "baseline_hrv_mean": 75.0, "baseline_hrv_std": 5.0, }, "static_features": { "age_group": 2, "sex": 0, "exercise": 1, "coffee": 1, "drinking": 0, "MEQ": 50.0, } if include_static else {}, } def generate_window( device_id: str, start: datetime, base_hrv: float, base_hr: float, anomaly_level: float = 0.0, include_static: bool = True, missing_ratio: float = 0.0, ) -> list: """生成 1 小时窗口数据""" data = [] base_hrv_for_day = max(30, base_hrv - 18 * anomaly_level) base_hr_for_day = min(125, base_hr + 10 * anomaly_level) for i in range(WINDOW_SIZE): noise_hrv = np.random.normal(0, 3) noise_hr = np.random.normal(0, 1.5) decline = -15 * anomaly_level * (i / WINDOW_SIZE) increase = 8 * anomaly_level * (i / WINDOW_SIZE) hrv = max(25, base_hrv_for_day + noise_hrv + decline) hr = min(125, base_hr_for_day + noise_hr + increase) ts = start + timedelta(minutes=INTERVAL_MINUTES * i) point = make_point(ts, device_id, hrv, hr, include_static=include_static) if missing_ratio > 0 and random.random() < missing_ratio: point["features"].pop("hr_resting", None) point["features"].pop("baseline_hrv_mean", None) point["features"].pop("baseline_hrv_std", None) if random.random() < 0.5: point["static_features"] = {} data.append(point) return data def load_window_from_file(path: Path) -> list | None: try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) assert isinstance(data, list) and data, "JSON needs to be a non-empty list" return data except Exception as exc: print(f" ⚠️ 读取 {path.name} 失败: {exc}") return None def demo_from_file(detector: WearableAnomalyDetector) -> None: print("\n" + "=" * 80) print("示例文件推理（test_data/example_window.json）") print("=" * 80) if not TEST_WINDOW_FILE.exists(): print(f" ⚠️ 未找到 {TEST_WINDOW_FILE}, 请确认仓库中存在该文件") return window = load_window_from_file(TEST_WINDOW_FILE) if not window: return avg_hrv = np.nanmean([pt["features"]["hrv_rmssd"] for pt in window]) avg_hr = np.nanmean([pt["features"]["hr"] for pt in window]) print(f" - 数据点数: {len(window)}") print(f" - 平均 HRV: {avg_hrv:.2f} ms, 平均心率: {avg_hr:.1f} bpm") result = detector.detect_realtime(window, update_baseline=False) print( f" -> 是否异常: {'是 ⚠️' if result.get('is_anomaly') else '否'} | " f"分数: {result.get('anomaly_score', 0):.4f} | 阈值: {result.get('threshold', 0):.4f}" ) baseline_info = { "baseline_mean": 76.0, "baseline_std": 5.0, "current_value": avg_hrv, "deviation_pct": (avg_hrv - 76.0) / 76.0 * 100, } llm_text = FORMATTER.format_for_llm(result, baseline_info=baseline_info) print("\n LLM 文本片段（前 350 字符）:") print("-" * 60) print(llm_text[:350]) print("...") print("-" * 60) def demo_realtime(detector: WearableAnomalyDetector) -> None: print("\n" + "=" * 80) print("实时检测示例") print("=" * 80) start = datetime.now() - timedelta(hours=1) normal_window = generate_window("demo_normal", start, base_hrv=76, base_hr=68, anomaly_level=0.0) anomaly_window = generate_window("demo_anomaly", start, base_hrv=74, base_hr=70, anomaly_level=0.7) for title, window in [("正常窗口", normal_window), ("异常窗口", anomaly_window)]: avg_hrv = np.mean([pt["features"]["hrv_rmssd"] for pt in window]) avg_hr = np.mean([pt["features"]["hr"] for pt in window]) print(f"\n[{title}] HRV≈{avg_hrv:.2f} ms, HR≈{avg_hr:.1f} bpm") result = detector.detect_realtime(window, update_baseline=False) print( f" -> 是否异常: {'是 ⚠️' if result.get('is_anomaly') else '否'} | " f"分数: {result.get('anomaly_score', 0):.4f} | 阈值: {result.get('threshold', 0):.4f}" ) def demo_pattern(detector: WearableAnomalyDetector) -> None: print("\n" + "=" * 80) print("7 天异常模式聚合示例") print("=" * 80) base_date = datetime.now() - timedelta(days=7) daily_data = [] anomaly_plan = [0.0, 0.1, 0.3, 1.0, 1.4, 1.8, 1.8] avg_hrv_per_day = [] for day, anomaly_level in enumerate(anomaly_plan): day_start = base_date + timedelta(days=day) window = generate_window( device_id="demo_pattern", start=day_start.replace(hour=8, minute=0, second=0, microsecond=0), base_hrv=75, base_hr=69, anomaly_level=anomaly_level, ) daily_data.append(window) avg_hrv_per_day.append(np.mean([pt["features"]["hrv_rmssd"] for pt in window])) print(" 日均HRV轨迹: " + ", ".join(f"{val:.1f}" for val in avg_hrv_per_day)) result = detector.detect_pattern( daily_data, days=len(daily_data), min_duration_days=2, format_for_llm=True ) pattern = result.get("anomaly_pattern", {}) print( f" -> 是否有模式: {'是' if pattern.get('has_pattern') else '否'} | " f"持续天数: {pattern.get('duration_days', 0)} | 趋势: {pattern.get('trend', '未知')}" ) if "formatted_for_llm" in result: print("\n格式化输出（前 400 字符）:") print("-" * 60) print(result["formatted_for_llm"][:400]) print("...") print("-" * 60) def demo_missing_data(detector: WearableAnomalyDetector) -> None: print("\n" + "=" * 80) print("数据缺失 / 质量下降示例") print("=" * 80) start = datetime.now() - timedelta(hours=1) incomplete_window = generate_window( device_id="demo_missing", start=start, base_hrv=74, base_hr=71, anomaly_level=0.5, include_static=True, missing_ratio=0.4, ) # 模拟传感器丢包：移除 2 个时间点 & 降低数据质量 for idx in (3, 7): incomplete_window[idx]["features"]["data_quality"] = "low" incomplete_window[idx]["features"]["hr"] = float("nan") avg_hrv = np.nanmean([pt["features"].get("hrv_rmssd", np.nan) for pt in incomplete_window]) available_static = sum(bool(pt["static_features"]) for pt in incomplete_window) print(f" - 有效静态特征点数: {available_static}/{len(incomplete_window)}") print(f" - 平均 HRV（忽略缺失）: {avg_hrv:.2f} ms") result = detector.detect_realtime(incomplete_window, update_baseline=False) print( f" -> 是否异常: {'是' if result.get('is_anomaly') else '否'} | " f"分数: {result.get('anomaly_score', 0):.4f} | 阈值: {result.get('threshold', 0):.4f}" ) def main() -> None: model_dir = ROOT_DIR / "checkpoints" / "phase2" / "exp_factor_balanced" detector = WearableAnomalyDetector(model_dir=model_dir, device="cpu") detector.update_threshold(0.50) demo_from_file(detector) demo_realtime(detector) demo_pattern(detector) demo_missing_data(detector) if __name__ == "__main__": main()