feat: ground truth generation — auto severity, CSV upload, string parse — 122/122

Author: marjatmm-sec

src/benchmark/ground_truth.py

@@ -0,0 +1,108 @@
+"""
+Qute — benchmark/ground_truth.py
+Ground truth label generation for benchmark runs.
+
+Three modes:
+  auto      — derive labels from event_severity (>= threshold = anomaly)
+  upload    — parse from uploaded CSV or comma-separated string
+  db        — load from stored ground truth associated with a dataset
+
+Auto mode is the default and requires no user input.
+It is a proxy label, not a verified ground truth — clearly communicated to users.
+"""
+
+from typing import Optional
+
+
+# ── Default severity threshold for anomaly classification ─────────
+DEFAULT_ANOMALY_THRESHOLD = 50   # event_severity >= 50 = anomaly (1)
+
+
+def from_severity(
+    events:    list[dict],
+    threshold: int = DEFAULT_ANOMALY_THRESHOLD,
+) -> list[int]:
+    """
+    Derive binary ground truth labels from event severity.
+    severity >= threshold -> 1 (anomaly)
+    severity <  threshold -> 0 (benign)
+
+    This is a proxy label suitable for benchmarking but not
+    a substitute for verified ground truth from a labelled dataset.
+    """
+    labels = []
+    for e in events:
+        sev = e.get("event_severity")
+        if sev is None:
+            labels.append(0)
+        else:
+            labels.append(1 if int(sev) >= threshold else 0)
+    return labels
+
+
+def from_string(
+    gt_string: str,
+    expected_length: Optional[int] = None,
+) -> tuple[list[int], Optional[str]]:
+    """
+    Parse ground truth from a comma-separated string of 0s and 1s.
+    Returns (labels, error_message). error_message is None on success.
+    """
+    if not gt_string.strip():
+        return [], "Ground truth string is empty."
+
+    try:
+        labels = [int(x.strip()) for x in gt_string.split(",")]
+    except ValueError:
+        return [], "Ground truth must contain only 0s and 1s separated by commas."
+
+    if not all(v in (0, 1) for v in labels):
+        return [], "All ground truth values must be 0 or 1."
+
+    if expected_length is not None and len(labels) != expected_length:
+        return [], (
+            f"Ground truth has {len(labels)} values but "
+            f"{expected_length} events were loaded. They must match."
+        )
+
+    return labels, None
+
+
+def from_csv_upload(
+    content: str,
+    expected_length: Optional[int] = None,
+) -> tuple[list[int], Optional[str]]:
+    """
+    Parse ground truth from CSV file content.
+    Accepts single column CSV or comma-separated values.
+    """
+    # Strip header if present
+    lines = [l.strip() for l in content.strip().splitlines() if l.strip()]
+    if not lines:
+        return [], "CSV file is empty."
+
+    # Try single column format (one value per line)
+    if len(lines) > 1 and all(l in ("0", "1") for l in lines):
+        labels = [int(l) for l in lines]
+        if expected_length and len(labels) != expected_length:
+            return [], (
+                f"CSV has {len(labels)} rows but "
+                f"{expected_length} events were loaded."
+            )
+        return labels, None
+
+    # Try comma-separated on one line
+    return from_string(",".join(lines), expected_length)
+
+
+def summary(labels: list[int]) -> dict:
+    """Return a summary of ground truth labels."""
+    total    = len(labels)
+    anomalies = sum(labels)
+    benign   = total - anomalies
+    return {
+        "total":          total,
+        "anomalies":      anomalies,
+        "benign":         benign,
+        "anomaly_rate":   round(anomalies / total, 3) if total > 0 else 0.0,
+    }

tests/benchmark/test_ground_truth.py

@@ -0,0 +1,106 @@
+"""Tests for ground truth label generation."""
+
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src"))
+
+from benchmark.ground_truth import (
+    from_severity, from_string, from_csv_upload, summary,
+    DEFAULT_ANOMALY_THRESHOLD
+)
+
+
+def test_from_severity_basic():
+    events = [
+        {"event_severity": 75},
+        {"event_severity": 25},
+        {"event_severity": 50},
+        {"event_severity": 49},
+    ]
+    labels = from_severity(events)
+    assert labels == [1, 0, 1, 0]
+
+
+def test_from_severity_none_value():
+    events = [{"event_severity": None}, {"event_severity": 80}]
+    labels = from_severity(events)
+    assert labels == [0, 1]
+
+
+def test_from_severity_custom_threshold():
+    events = [{"event_severity": 60}, {"event_severity": 40}]
+    labels = from_severity(events, threshold=70)
+    assert labels == [0, 0]
+    labels = from_severity(events, threshold=30)
+    assert labels == [1, 1]
+
+
+def test_from_string_valid():
+    labels, err = from_string("1,0,1,1,0")
+    assert err is None
+    assert labels == [1, 0, 1, 1, 0]
+
+
+def test_from_string_with_spaces():
+    labels, err = from_string("1, 0, 1, 0")
+    assert err is None
+    assert labels == [1, 0, 1, 0]
+
+
+def test_from_string_invalid():
+    labels, err = from_string("1,0,2,0")
+    assert err is not None
+    assert labels == []
+
+
+def test_from_string_empty():
+    labels, err = from_string("")
+    assert err is not None
+    assert labels == []
+
+
+def test_from_string_length_mismatch():
+    labels, err = from_string("1,0,1", expected_length=5)
+    assert err is not None
+    assert "3" in err
+    assert "5" in err
+
+
+def test_from_string_length_match():
+    labels, err = from_string("1,0,1", expected_length=3)
+    assert err is None
+    assert len(labels) == 3
+
+
+def test_from_csv_single_column():
+    csv = "1\n0\n1\n0\n1"
+    labels, err = from_csv_upload(csv)
+    assert err is None
+    assert labels == [1, 0, 1, 0, 1]
+
+
+def test_from_csv_comma_separated():
+    csv = "1,0,1,0"
+    labels, err = from_csv_upload(csv)
+    assert err is None
+    assert labels == [1, 0, 1, 0]
+
+
+def test_from_csv_empty():
+    labels, err = from_csv_upload("")
+    assert err is not None
+
+
+def test_summary():
+    labels = [1, 0, 1, 0, 1, 0, 0, 0]
+    s = summary(labels)
+    assert s["total"]     == 8
+    assert s["anomalies"] == 3
+    assert s["benign"]    == 5
+    assert s["anomaly_rate"] == 0.375
+
+
+def test_summary_empty():
+    s = summary([])
+    assert s["total"]       == 0
+    assert s["anomaly_rate"] == 0.0