Upload api/app.py

api/app.py

@@ -0,0 +1,275 @@
+"""
+api/app.py — FastAPI backend for the TS Anomaly Detection Benchmark
+====================================================================
+
+Designed to run on Hugging Face Spaces (Docker, port 7860).
+
+Endpoints:
+  POST /api/run              — start a benchmark job
+  GET  /api/status/{job_id} — poll for progress + results
+  GET  /                    — health check
+"""
+
+import os
+import sys
+import uuid
+import math
+import base64
+import threading
+import io
+from typing import Optional, List
+
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+
+# ── Add benchmark package to Python path ──
+# Allows importing data.synthetic, evaluation.scorer, models.*, etc.
+_BENCH_PATH = os.path.join(os.path.dirname(__file__), "..", "ts-anomaly-benchmark")
+sys.path.insert(0, os.path.abspath(_BENCH_PATH))
+
+# ── App ──────────────────────────────────────────────────────────────
+app = FastAPI(title="TS Anomaly Benchmark API", version="1.0.0")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ── Job store ─────────────────────────────────────────────────────────
+# Simple in-memory store. Fine for a demo — one server instance.
+_jobs: dict = {}
+_run_lock = threading.Lock()  # one benchmark at a time (models are CPU-heavy)
+
+
+# ── Request / Response models ─────────────────────────────────────────
+
+class RunRequest(BaseModel):
+    models: List[str]                          # e.g. ["moment", "isolation_forest"]
+    synthetic_types: List[str] = [
+        "sine_with_spikes",
+        "random_walk_with_shift",
+        "seasonal_with_noise",
+        "ecg_like",
+    ]
+    num_points: int = 512
+    custom_csv: Optional[str] = None           # base64-encoded CSV content
+    custom_value_col: str = "value"
+    custom_label_col: Optional[str] = None
+
+
+# ── Endpoints ─────────────────────────────────────────────────────────
+
+@app.get("/")
+def root():
+    return {"status": "ok", "message": "TS Anomaly Benchmark API"}
+
+
+@app.post("/api/run")
+def start_run(req: RunRequest):
+    """Start a benchmark job. Returns a job_id for polling."""
+    job_id = str(uuid.uuid4())
+    _jobs[job_id] = {
+        "status": "running",
+        "logs": [],
+        "results": None,
+        "error": None,
+    }
+    thread = threading.Thread(
+        target=_execute_job,
+        args=(job_id, req),
+        daemon=True,
+    )
+    thread.start()
+    return {"job_id": job_id}
+
+
+@app.get("/api/status/{job_id}")
+def get_status(job_id: str):
+    """Poll job progress. When status=='done', results are included."""
+    job = _jobs.get(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return {
+        "status":  job["status"],
+        "logs":    job["logs"],
+        "results": job["results"],
+        "error":   job["error"],
+    }
+
+
+# ── Job execution ─────────────────────────────────────────────────────
+
+def _execute_job(job_id: str, req: RunRequest):
+    """Run the full benchmark in a background thread."""
+    job = _jobs[job_id]
+    logs = job["logs"]
+
+    # Serialise concurrent jobs — models are memory-heavy
+    with _run_lock:
+        original_stdout = sys.stdout
+        sys.stdout = _LogCapture(logs, original_stdout)
+        try:
+            config = _build_config(req)
+
+            # Import benchmark modules (torch may be slow to first import)
+            from data.synthetic import generate_all as generate_synthetic
+            from evaluation.scorer import build_models, run_benchmark
+
+            # ── Datasets ──
+            all_datasets = {}
+            if req.synthetic_types:
+                all_datasets.update(generate_synthetic(config["datasets"]["synthetic"]))
+            if req.custom_csv:
+                all_datasets.update(_load_custom_csv(req))
+
+            if not all_datasets:
+                raise ValueError("No datasets loaded.")
+
+            # ── Models ──
+            models = build_models(config["models"])
+            if not models:
+                raise ValueError("No models enabled. Select at least one.")
+
+            # ── Run ──
+            results_df = run_benchmark(models, all_datasets, config["evaluation"])
+
+            job["results"] = _serialise(results_df, all_datasets)
+            job["status"] = "done"
+
+        except Exception as exc:
+            job["status"] = "error"
+            job["error"] = str(exc)
+            logs.append(f"ERROR: {exc}")
+        finally:
+            sys.stdout = original_stdout
+
+
+def _build_config(req: RunRequest) -> dict:
+    enabled = set(req.models)
+    return {
+        "models": {
+            "moment": {
+                "enabled": "moment" in enabled,
+                "pretrained": "AutonLab/MOMENT-1-large",
+                "task": "reconstruction",
+            },
+            "isolation_forest": {
+                "enabled": "isolation_forest" in enabled,
+                "n_estimators": 100,
+                "contamination": "auto",
+                "window_features": True,
+                "feature_window": 20,
+            },
+            "lof": {
+                "enabled": "lof" in enabled,
+                "n_neighbors": 20,
+                "contamination": "auto",
+                "window_features": True,
+                "feature_window": 20,
+            },
+            "moving_window": {
+                "enabled": "moving_window" in enabled,
+                "window_size": 15,
+                "sigma_multiplier": 2.0,
+            },
+        },
+        "datasets": {
+            "synthetic": {
+                "enabled": bool(req.synthetic_types),
+                "num_points": req.num_points,
+                "types": req.synthetic_types,
+                "anomaly_rate": 0.03,
+                "random_seed": 42,
+            }
+        },
+        "evaluation": {"threshold_method": "best_f1"},
+    }
+
+
+def _load_custom_csv(req: RunRequest) -> dict:
+    import pandas as pd
+    import numpy as np
+
+    csv_bytes = base64.b64decode(req.custom_csv)
+    df = pd.read_csv(io.BytesIO(csv_bytes))
+
+    if req.custom_value_col not in df.columns:
+        numeric = df.select_dtypes(include=[np.number]).columns
+        if len(numeric) == 0:
+            raise ValueError("No numeric columns found in uploaded CSV.")
+        value_col = numeric[0]
+    else:
+        value_col = req.custom_value_col
+
+    series = df[value_col].to_numpy(dtype=float)
+    labels = None
+    if req.custom_label_col and req.custom_label_col in df.columns:
+        labels = df[req.custom_label_col].to_numpy(dtype=int)
+
+    valid = ~__import__("numpy").isnan(series)
+    series = series[valid]
+    if labels is not None:
+        labels = labels[valid]
+
+    return {"custom_upload": {"series": series, "labels": labels}}
+
+
+def _serialise(results_df, all_datasets: dict) -> dict:
+    metrics = []
+    for _, row in results_df.iterrows():
+        metrics.append({
+            "model":        row.get("model"),
+            "dataset":      row.get("dataset"),
+            "auc_roc":      _f(row.get("auc_roc")),
+            "auc_pr":       _f(row.get("auc_pr")),
+            "f1":           _f(row.get("f1")),
+            "precision":    _f(row.get("precision")),
+            "recall":       _f(row.get("recall")),
+            "time_seconds": _f(row.get("time_seconds")),
+            "scores":       _arr(row.get("_scores")),
+        })
+
+    series_out = {}
+    for name, data in all_datasets.items():
+        series_out[name] = {
+            "series": [round(float(v), 4) for v in data["series"]],
+            "labels": data["labels"].tolist() if data.get("labels") is not None else None,
+        }
+
+    return {"metrics": metrics, "series": series_out}
+
+
+def _f(v):
+    try:
+        f = float(v)
+        return None if math.isnan(f) else round(f, 4)
+    except Exception:
+        return None
+
+
+def _arr(v):
+    try:
+        if v is None or not hasattr(v, "__len__"):
+            return None
+        return [round(float(x), 4) for x in v]
+    except Exception:
+        return None
+
+
+# ── Stdout capture ────────────────────────────────────────────────────
+
+class _LogCapture:
+    def __init__(self, log_list: list, original):
+        self._log = log_list
+        self._orig = original
+
+    def write(self, text: str):
+        if text and text.strip():
+            self._log.append(text.strip())
+        self._orig.write(text)
+
+    def flush(self):
+        self._orig.flush()