Add RCA group classification, tags, and hints to multi-RAT top anomalies table with helper functions for slug generation, numeric formatting, and metadata construction based on KPI groups, status, baseline/recent values, and multi-RAT impact

process_kpi/kpi_health_check/multi_rat.py

@@ -1,5 +1,74 @@
 import pandas as pd
 
+from process_kpi.kpi_health_check.kpi_groups import classify_kpi
+
+
+def _slug(value: str) -> str:
+    s = str(value or "").strip().lower()
+    out = []
+    prev_underscore = False
+    for ch in s:
+        if ch.isalnum():
+            out.append(ch)
+            prev_underscore = False
+        else:
+            if not prev_underscore:
+                out.append("_")
+                prev_underscore = True
+    return "".join(out).strip("_")
+
+
+def _fmt_num(value) -> str:
+    try:
+        v = pd.to_numeric(value, errors="coerce")
+        if pd.isna(v):
+            return "NA"
+        return f"{float(v):.3g}"
+    except Exception:  # noqa: BLE001
+        return "NA"
+
+
+def _build_rca_tags(row: dict) -> str:
+    tags: list[str] = []
+    group = str(row.get("rca_group") or "Other")
+    tags.append(_slug(group) if group else "other")
+
+    status = str(row.get("status") or "").strip().upper()
+    if status == "PERSISTENT_DEGRADED":
+        tags.append("persistent")
+    elif status == "DEGRADED":
+        tags.append("degraded")
+    elif status:
+        tags.append(_slug(status))
+
+    baseline = row.get("baseline_median")
+    recent = row.get("recent_median")
+    if pd.isna(pd.to_numeric(baseline, errors="coerce")):
+        tags.append("missing_baseline")
+    if pd.isna(pd.to_numeric(recent, errors="coerce")):
+        tags.append("missing_recent")
+
+    impact = pd.to_numeric(row.get("impacted_rats"), errors="coerce")
+    if pd.notna(impact) and float(impact) >= 2:
+        tags.append("multi_rat")
+
+    return ",".join([t for t in tags if t])
+
+
+def _build_rca_hint(row: dict) -> str:
+    group = str(row.get("rca_group") or "Other")
+    kpi = str(row.get("KPI") or "")
+    rat = str(row.get("RAT") or "")
+    status = str(row.get("status") or "")
+    baseline_s = _fmt_num(row.get("baseline_median"))
+    recent_s = _fmt_num(row.get("recent_median"))
+    streak = int(pd.to_numeric(row.get("max_streak_recent"), errors="coerce") or 0)
+    bad = int(pd.to_numeric(row.get("bad_days_recent"), errors="coerce") or 0)
+    return (
+        f"{group} | {rat} | {kpi} | {status} | "
+        f"baseline={baseline_s} recent={recent_s} | streak={streak}d bad={bad}d"
+    )
+
 
 def compute_multirat_views(
     status_df: pd.DataFrame,
@@ -142,6 +211,22 @@ def compute_multirat_views(
         ascending=[False, False, False, False],
     )
 
+    try:
+        top = pd.merge(top, impacted, on="site_code", how="left")
+        top["impacted_rats"] = (
+            pd.to_numeric(top["impacted_rats"], errors="coerce").fillna(0).astype(int)
+        )
+    except Exception:  # noqa: BLE001
+        top["impacted_rats"] = 0
+
+    top["rca_group"] = top["KPI"].apply(classify_kpi)
+    try:
+        top["rca_hint"] = top.apply(lambda r: _build_rca_hint(r.to_dict()), axis=1)
+        top["rca_tags"] = top.apply(lambda r: _build_rca_tags(r.to_dict()), axis=1)
+    except Exception:  # noqa: BLE001
+        top["rca_hint"] = ""
+        top["rca_tags"] = ""
+
     top_cols = [
         c
         for c in [
@@ -151,7 +236,11 @@ def compute_multirat_views(
             "site_code",
             "City",
             "KPI",
+            "rca_group",
+            "rca_tags",
+            "rca_hint",
             "status",
+            "impacted_rats",
             "baseline_median",
             "recent_median",
             "bad_days_recent",