Add robustness_check/src/robustness_check/core.py

robustness_check/src/robustness_check/core.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+import statistics
+from typing import List, Dict, Any
+
+
+class RobustnessResult:
+    """Datenstruktur zur Speicherung der Ergebnisse der Robustheitsanalyse."""
+
+    def __init__(self, k_value: int, flip_flops: int, average_decision: float) -> None:
+        if not isinstance(k_value, int) or not isinstance(flip_flops, int) or not isinstance(average_decision, (int, float)):
+            raise TypeError("Invalid types for RobustnessResult initialization.")
+        self.k_value = k_value
+        self.flip_flops = flip_flops
+        self.average_decision = float(average_decision)
+
+    def __repr__(self) -> str:
+        return (
+            f"RobustnessResult(k_value={self.k_value}, "
+            f"flip_flops={self.flip_flops}, average_decision={self.average_decision:.4f})"
+        )
+
+
+
+def check_robustness(runs_data: List[Dict[str, Any]]) -> RobustnessResult:
+    """Analysiert die Robustheit der Gate-v0-Entscheidung anhand von Frozen-Run-Daten."""
+
+    if not isinstance(runs_data, list):
+        raise TypeError("runs_data must be a list of dictionaries.")
+    if not runs_data:
+        raise ValueError("runs_data cannot be empty.")
+
+    for item in runs_data:
+        if not isinstance(item, dict):
+            raise TypeError("Each run data entry must be a dictionary.")
+        for field in ["run_id", "p95_value", "pinned"]:
+            if field not in item:
+                raise ValueError(f"Missing field '{field}' in run data entry.")
+
+    # Beispielhafte k-Wert-Definition für Analysezwecke
+    k_value = 3
+
+    # Entscheidungen basierend auf Pinned-Status (vereinfacht)
+    decisions = [1.0 if run["pinned"] else 0.0 for run in runs_data]
+    avg_decision = statistics.mean(decisions)
+
+    # Flip-Flops zählen (Veränderung zwischen aufeinanderfolgenden Entscheidungen)
+    flip_flops = 0
+    for i in range(1, len(decisions)):
+        if decisions[i] != decisions[i - 1]:
+            flip_flops += 1
+
+    return RobustnessResult(k_value=k_value, flip_flops=flip_flops, average_decision=avg_decision)