Add export raw data toggle with optimized health check evaluation using vectorized operations and improved timestamp handling for baseline/recent period masking with persistent bad date detection

panel_app/kpi_health_check_panel.py

@@ -1383,6 +1383,11 @@ export_button = pn.widgets.FileDownload(
     button_type="primary",
 )
 
+export_include_raw_data = pn.widgets.Checkbox(
+    name="Include raw KPI data (slow)",
+    value=True,
+)
+
 alert_pack_button = pn.widgets.FileDownload(
     label="Download Alert Pack",
     filename="KPI_Alert_Pack.xlsx",
@@ -3349,6 +3354,12 @@ def run_health_check(event=None) -> None:
 
 
 def _build_export_bytes() -> bytes:
+    include_raw = bool(export_include_raw_data.value)
+    daily_by_rat = (
+        current_daily_by_rat
+        if (include_raw and isinstance(current_daily_by_rat, dict))
+        else None
+    )
     return build_export_bytes(
         datasets_df=(
             datasets_table.value
@@ -3364,9 +3375,7 @@ def _build_export_bytes() -> bytes:
         status_df=(
             current_status_df if isinstance(current_status_df, pd.DataFrame) else None
         ),
-        daily_by_rat=(
-            current_daily_by_rat if isinstance(current_daily_by_rat, dict) else None
-        ),
+        daily_by_rat=daily_by_rat,
         granularity=str(granularity_select.value or "Daily"),
         multirat_summary_df=(
             current_multirat_df
@@ -3617,6 +3626,13 @@ def _on_drilldown_params_change(event=None) -> None:
     _refresh_validation_state()
 
 
+def _on_export_options_change(event=None) -> None:
+    global current_export_bytes
+    if _applying_profile or _loading_datasets:
+        return
+    current_export_bytes = None
+
+
 def _on_granularity_change(event=None) -> None:
     if _applying_profile or _loading_datasets:
         return
@@ -3665,6 +3681,8 @@ recent_days.param.watch(_on_drilldown_params_change, "value")
 rel_threshold_pct.param.watch(_on_drilldown_params_change, "value")
 min_consecutive_days.param.watch(_on_drilldown_params_change, "value")
 
+export_include_raw_data.param.watch(_on_export_options_change, "value")
+
 
 def _on_rules_table_change(event=None) -> None:
     global current_export_bytes, current_alert_pack_bytes
@@ -3930,6 +3948,7 @@ sidebar = pn.Column(
         sizing_mode="stretch_width",
     ),
     pn.Card(
+        export_include_raw_data,
         export_button,
         alert_pack_button,
         title="Export",

process_kpi/kpi_health_check/engine.py

@@ -111,7 +111,7 @@ def evaluate_health_check(
         int(min_consecutive_days) * 24 if is_hourly else int(min_consecutive_days)
     )
 
-    end_dt = _to_timestamp(daily[time_col].max())
+    end_dt = _to_timestamp(pd.to_datetime(daily[time_col], errors="coerce").max())
     if end_dt is None:
         return pd.DataFrame(), pd.DataFrame()
 
@@ -121,6 +121,11 @@ def evaluate_health_check(
 
     rat_rules = rules_df[rules_df["RAT"] == rat].copy()
     kpis = [k for k in rat_rules["KPI"].tolist() if k in daily.columns]
+    rules_by_kpi = {
+        str(r["KPI"]): r
+        for r in rat_rules.to_dict(orient="records")
+        if str(r.get("KPI", ""))
+    }
 
     rows = []
 
@@ -131,9 +136,12 @@ def evaluate_health_check(
             else None
         )
         g_site = g_site.sort_values(time_col)
+        t_all = pd.to_datetime(g_site[time_col], errors="coerce")
+        baseline_mask_all = (t_all >= baseline_start_dt) & (t_all <= baseline_end_dt)
+        recent_mask_all = (t_all >= recent_start_dt) & (t_all <= recent_end_dt)
 
         for kpi in kpis:
-            rule = rat_rules[rat_rules["KPI"] == kpi].iloc[0]
+            rule = rules_by_kpi.get(str(kpi), {})
             direction = str(rule.get("direction", "higher_is_better"))
             policy = str(rule.get("policy", "enforce") or "enforce").strip().lower()
             sla = rule.get("sla", np.nan)
@@ -144,8 +152,9 @@ def evaluate_health_check(
 
             sla_eval = None if policy == "notify" else sla_val
 
-            s = g_site[[time_col, kpi]].dropna(subset=[kpi])
-            if s.empty:
+            vals = pd.to_numeric(g_site[kpi], errors="coerce")
+            has_any = bool(vals.notna().any())
+            if not has_any:
                 rows.append(
                     {
                         "RAT": rat,
@@ -157,29 +166,38 @@ def evaluate_health_check(
                 )
                 continue
 
-            t = pd.to_datetime(s[time_col], errors="coerce")
-            baseline_mask = (t >= baseline_start_dt) & (t <= baseline_end_dt)
-            recent_mask = (t >= recent_start_dt) & (t <= recent_end_dt)
+            baseline_vals = vals.loc[baseline_mask_all]
+            recent_vals = vals.loc[recent_mask_all]
+            t_recent = t_all.loc[recent_vals.index]
 
             baseline = (
-                s.loc[baseline_mask, kpi].median() if baseline_mask.any() else np.nan
+                baseline_vals.median(skipna=True) if baseline_mask_all.any() else np.nan
             )
-            recent = s.loc[recent_mask, kpi].median() if recent_mask.any() else np.nan
-
-            daily_recent = s.loc[recent_mask, [time_col, kpi]].copy()
-            bad_dates = []
-            if not daily_recent.empty:
-                for d, v in zip(
-                    daily_recent[time_col].tolist(), daily_recent[kpi].tolist()
-                ):
-                    if is_bad(
-                        float(v) if pd.notna(v) else None,
-                        float(baseline) if pd.notna(baseline) else None,
-                        direction,
-                        rel_threshold_pct,
-                        sla_eval,
-                    ):
-                        bad_dates.append(d)
+            recent = (
+                recent_vals.median(skipna=True) if recent_mask_all.any() else np.nan
+            )
+
+            bad_dates: list = []
+            if recent_mask_all.any() and recent_vals.notna().any():
+                thr = float(rel_threshold_pct) / 100.0
+                b = float(baseline) if pd.notna(baseline) else None
+                bad_series = pd.Series(False, index=recent_vals.index)
+
+                if b is not None:
+                    if direction == "higher_is_better":
+                        bad_series = bad_series | (recent_vals < (b - abs(b) * thr))
+                    else:
+                        bad_series = bad_series | (recent_vals > (b + abs(b) * thr))
+
+                if sla_eval is not None and pd.notna(sla_eval):
+                    if direction == "higher_is_better":
+                        bad_series = bad_series | (recent_vals < float(sla_eval))
+                    else:
+                        bad_series = bad_series | (recent_vals > float(sla_eval))
+
+                bad_series = bad_series & recent_vals.notna() & t_recent.notna()
+                if bool(bad_series.any()):
+                    bad_dates = t_recent.loc[bad_series].tolist()
 
             max_streak = max_consecutive_periods(bad_dates, step=step)
             persistent = max_streak >= int(min_periods)
@@ -193,16 +211,20 @@ def evaluate_health_check(
             )
 
             is_bad_current = is_bad_recent
-            if not daily_recent.empty:
-                last_row = daily_recent.sort_values(time_col).iloc[-1]
-                last_val = last_row[kpi]
-                is_bad_current = is_bad(
-                    float(last_val) if pd.notna(last_val) else None,
-                    float(baseline) if pd.notna(baseline) else None,
-                    direction,
-                    rel_threshold_pct,
-                    sla_eval,
-                )
+            try:
+                last_mask = recent_mask_all & vals.notna() & t_all.notna()
+                if bool(last_mask.any()):
+                    idx_last = t_all.loc[last_mask].idxmax()
+                    last_val = vals.loc[idx_last]
+                    is_bad_current = is_bad(
+                        float(last_val) if pd.notna(last_val) else None,
+                        float(baseline) if pd.notna(baseline) else None,
+                        direction,
+                        rel_threshold_pct,
+                        sla_eval,
+                    )
+            except Exception:  # noqa: BLE001
+                pass
 
             had_bad_recent = (len(bad_dates) > 0) or bool(is_bad_recent)