Add configurable z-score severity thresholds

README.md

@@ -150,7 +150,7 @@ app.add_middleware(ObyflowASGIMiddleware, service="checkout-api", store=handle.s
 | Median/MAD ("robust") baselining | Yes | No |
 | Rolling time-windowed buckets | Yes | No |
 | Deployment-aware bucketing | Yes | No |
-| Configurable z-score threshold | Yes | No (fixed thresholds in `classify_severity`) |
+| Configurable z-score threshold | Yes | Yes |
 | ML-based detection (IsolationForest) | No | Yes (`detect_ml_anomalies`, `obyflow-python[analysis]`) |
 
 The Python SDK's `analysis/` module is a separate, Python-only convenience toolkit, not a port of `packages/core/src/anomaly/baseline.ts`, which the CLI's `investigate`/`ask`/`incident` commands use internally. `detect_ml_anomalies` (IsolationForest-based) is Python-exclusive, with no TypeScript/core equivalent.

python/obyflow-python/obyflow/analysis/anomaly.py

@@ -7,7 +7,7 @@
 
 from __future__ import annotations
 
-from typing import List, TypedDict
+from typing import TypedDict
 
 from ..events import Event
 from .stats import (
@@ -33,19 +33,22 @@ class MLAnomalyResult(TypedDict):
 DEFAULT_RANDOM_STATE = 0
 
 
-def _feature_vector(event: Event) -> List[float]:
+def _feature_vector(event: Event) -> list[float]:
     duration = event.duration_ms if event.duration_ms is not None else 0.0
     is_error = 1.0 if event.severity in ("error", "critical") else 0.0
     return [duration, is_error]
 
 
 def detect_ml_anomalies(
-    events: List[Event],
+    events: list[Event],
     service: str,
     contamination: float = DEFAULT_CONTAMINATION,
     min_samples: int = DEFAULT_MIN_SAMPLES,
     random_state: int = DEFAULT_RANDOM_STATE,
-) -> List[MLAnomalyResult]:
+    low_threshold: float = 1.0,
+    medium_threshold: float = 2.0,
+    high_threshold: float = 3.0
+) -> list[MLAnomalyResult]:
     try:
         from sklearn.ensemble import IsolationForest
     except ImportError as exc:
@@ -66,7 +69,7 @@ def detect_ml_anomalies(
 
     baseline: BaselineStats = compute_baseline_stats([float(s) for s in raw_scores])
 
-    results: List[MLAnomalyResult] = []
+    results: list[MLAnomalyResult] = []
     for event, score, prediction in zip(service_events, raw_scores, predictions):
         z_score = z_score_of(float(score), baseline)
         results.append(
@@ -75,7 +78,9 @@ def detect_ml_anomalies(
                 "service": service,
                 "anomaly_score": float(score),
                 "z_score": z_score,
-                "severity": classify_severity(z_score),
+                "severity": classify_severity(
+                    z_score, low_threshold, medium_threshold, high_threshold
+                ),
                 "is_anomalous": bool(prediction == -1),
             }
         )

python/obyflow-python/obyflow/analysis/stats.py

@@ -9,7 +9,7 @@
 
 from __future__ import annotations
 
-from typing import List, Optional, TypedDict
+from typing import TypedDict
 
 
 class BaselineStats(TypedDict):
@@ -23,21 +23,21 @@ class BaselineStats(TypedDict):
 ZERO_STDDEV_Z_SCORE = 10.0
 
 
-def mean(values: List[float]) -> float:
+def mean(values: list[float]) -> float:
     if not values:
         return 0.0
     return sum(values) / len(values)
 
 
-def stddev(values: List[float], mean_value: Optional[float] = None) -> float:
+def stddev(values: list[float], mean_value: float | None = None) -> float:
     if not values:
         return 0.0
     m = mean_value if mean_value is not None else mean(values)
     variance = sum((v - m) ** 2 for v in values) / len(values)
     return variance**0.5
 
 
-def compute_baseline_stats(values: List[float]) -> BaselineStats:
+def compute_baseline_stats(values: list[float]) -> BaselineStats:
     m = mean(values)
     return {"mean": m, "stddev": stddev(values, m), "count": len(values)}
 
@@ -50,12 +50,17 @@ def z_score_of(value: float, baseline: BaselineStats) -> float:
     return (value - baseline["mean"]) / baseline["stddev"]
 
 
-def classify_severity(z_score: float) -> DeviationSeverity:
+def classify_severity(
+    z_score: float,
+    low_threshold: float = 1.0,
+    medium_threshold: float = 2.0,
+    high_threshold: float = 3.0
+) -> DeviationSeverity:
     abs_z = abs(z_score)
-    if abs_z < 1:
+    if abs_z < low_threshold:
         return "none"
-    if abs_z < 2:
+    if abs_z < medium_threshold:
         return "low"
-    if abs_z < 3:
+    if abs_z < high_threshold:
         return "medium"
     return "high"

python/obyflow-python/tests/test_analysis_anomaly.py

@@ -58,3 +58,18 @@ def test_detect_ml_anomalies_result_shape():
         assert isinstance(item["z_score"], float)
         assert item["severity"] in ("none", "low", "medium", "high")
         assert isinstance(item["is_anomalous"], bool)
+
+
+def test_detect_ml_anomalies_with_custom_thresholds():
+    events = [_make_event(f"e{i}", 100.0 + i) for i in range(20)]
+    events.append(_make_event("spike", 50000.0))
+    result = detect_ml_anomalies(
+        events,
+        "checkout",
+        min_samples=5,
+        low_threshold=0.5,
+        medium_threshold=1.0,
+        high_threshold=1.5
+    )
+    spike_result = next(r for r in result if r["event_id"] == "spike")
+    assert spike_result["severity"] == "high"

python/obyflow-python/tests/test_analysis_stats.py

@@ -45,3 +45,11 @@ def test_classify_severity_tiers():
     assert classify_severity(2.5) == "medium"
     assert classify_severity(4.0) == "high"
     assert classify_severity(-4.0) == "high"
+
+
+def test_classify_severity_custom_thresholds():
+    assert classify_severity(1.5, low_threshold=2.0) == "none"
+    assert classify_severity(2.5, low_threshold= 2.0, medium_threshold=3.0) == "low"
+    assert classify_severity(3.5, medium_threshold=3.0, high_threshold=5.0) == "medium"
+    assert classify_severity(5.5, high_threshold=5.0) == "high"
+