Add artifact.metrics_analysis/src/artifact_metrics_analysis/core.py

artifact.metrics_analysis/src/artifact_metrics_analysis/core.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import logging
+from typing import Dict, Any
+import pandas as pd
+from statistics import mean
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
+
+class MetricResults:
+    """Repräsentation des Ergebnisses der Metrikanalyse inklusive Delta-Berechnung."""
+
+    def __init__(self, retry_tail_p99: float, band_width: float, delta_band_width: float) -> None:
+        assert isinstance(retry_tail_p99, (int, float)), "retry_tail_p99 muss numerisch sein"
+        assert isinstance(band_width, (int, float)), "band_width muss numerisch sein"
+        assert isinstance(delta_band_width, (int, float)), "delta_band_width muss numerisch sein"
+
+        self.retry_tail_p99 = float(retry_tail_p99)
+        self.band_width = float(band_width)
+        self.delta_band_width = float(delta_band_width)
+
+    def to_dict(self) -> Dict[str, float]:
+        """Serialisiert die Ergebnisse als dictionary für JSON-Ausgabe."""
+        return {
+            "retry_tail_p99": self.retry_tail_p99,
+            "band_width": self.band_width,
+            "delta_band_width": self.delta_band_width,
+        }
+
+
+class MetricDataError(Exception):
+    """Custom-Exception für ungültige Metrikdaten."""
+    pass
+
+
+def analyze_metrics(card_data_40: Dict[str, Any], card_data_42: Dict[str, Any]) -> MetricResults:
+    """Vergleicht Metriken von zwei Evidence Cards anhand der Eingabedaten und erstellt eine Ergebnisanalyse."""
+    if not isinstance(card_data_40, dict) or not isinstance(card_data_42, dict):
+        raise MetricDataError("Beide Eingaben müssen Dictionaries sein.")
+
+    required_fields = {"retry_tail_p99", "band_width"}
+    for cid, data in (("card_40", card_data_40), ("card_42", card_data_42)):
+        missing = required_fields - data.keys()
+        if missing:
+            raise MetricDataError(f"Fehlende Felder in {cid}: {missing}")
+        for field in required_fields:
+            if not isinstance(data[field], (int, float)):
+                raise MetricDataError(f"Ungültiger Typ für {field} in {cid}: {type(data[field])}")
+
+    df = pd.DataFrame([
+        {"card": "40", **card_data_40},
+        {"card": "42", **card_data_42}
+    ])
+
+    retry_tail_p99_avg = mean(df["retry_tail_p99"])
+    bw_40 = df.loc[df["card"] == "40", "band_width"].iloc[0]
+    bw_42 = df.loc[df["card"] == "42", "band_width"].iloc[0]
+    delta_band_width = bw_42 - bw_40
+
+    logger.info("Analyse durchgeführt: ΔBandbreite=%.4f", delta_band_width)
+
+    result = MetricResults(
+        retry_tail_p99=retry_tail_p99_avg,
+        band_width=mean([bw_40, bw_42]),
+        delta_band_width=delta_band_width,
+    )
+
+    assert isinstance(result.retry_tail_p99, float)
+    assert isinstance(result.band_width, float)
+    assert isinstance(result.delta_band_width, float)
+
+    return result