Add hourly granularity support with period-based time column handling, window calculations using hour intervals, and adaptive date axis formatting for hourly vs daily data across drill-down plots, site views, heatmaps, and correlation analysis

panel_app/kpi_health_check_drilldown_plots.py

@@ -20,6 +20,7 @@ def build_drilldown_plot(
     recent_days_n: int = 7,
     rel_threshold_pct: float = 10.0,
     normalization: str = "None",
+    granularity: str = "Daily",
 ) -> go.Figure | None:
     """
     Builds the drill-down trend plot with native Plotly annotations.
@@ -32,12 +33,16 @@ def build_drilldown_plot(
     if not valid_kpis:
         return None
 
-    # Sort by date
-    # Sort by date
-    plot_df = df.sort_values("date_only").copy()
+    g = str(granularity or "Daily").strip().lower()
+    is_hourly = g.startswith("hour") or g.startswith("h")
+    time_col = (
+        "period_start" if (is_hourly and "period_start" in df.columns) else "date_only"
+    )
+
+    plot_df = df.sort_values(time_col).copy()
 
     try:
-        plot_df["date_only"] = pd.to_datetime(plot_df["date_only"])
+        plot_df[time_col] = pd.to_datetime(plot_df[time_col])
     except Exception:
         pass
 
@@ -69,7 +74,7 @@ def build_drilldown_plot(
 
     for kpi in valid_kpis:
         # Data preparation
-        x_data = plot_df["date_only"]
+        x_data = plot_df[time_col]
         y_data = pd.to_numeric(plot_df[kpi], errors="coerce")
         if do_norm:
             if norm_mode == "Min-Max":
@@ -163,12 +168,17 @@ def build_drilldown_plot(
                             except Exception:
                                 sla_eval = None
 
-            end_dt = pd.to_datetime(plot_df["date_only"]).max()
-            rs = end_dt - timedelta(days=max(int(recent_days_n), 1) - 1)
-            be = rs - timedelta(days=1)
-            bs = be - timedelta(days=max(int(baseline_days_n), 1) - 1)
+            end_dt = pd.to_datetime(plot_df[time_col]).max()
+            if is_hourly:
+                rs = end_dt - timedelta(hours=max(int(recent_days_n), 1) * 24 - 1)
+                be = rs - timedelta(hours=1)
+                bs = be - timedelta(hours=max(int(baseline_days_n), 1) * 24 - 1)
+            else:
+                rs = end_dt - timedelta(days=max(int(recent_days_n), 1) - 1)
+                be = rs - timedelta(days=1)
+                bs = be - timedelta(days=max(int(baseline_days_n), 1) - 1)
 
-            dts = pd.to_datetime(plot_df["date_only"])
+            dts = pd.to_datetime(plot_df[time_col])
             baseline_mask = (dts >= bs) & (dts <= be)
             recent_mask = (dts >= rs) & (dts <= end_dt)
             baseline_val = pd.to_numeric(
@@ -211,7 +221,7 @@ def build_drilldown_plot(
                     hover_txt.append(f"OUTSIDE WINDOW ({main_kpi})")
             fig.add_trace(
                 go.Scatter(
-                    x=plot_df["date_only"],
+                    x=plot_df[time_col],
                     y=[0] * len(plot_df),
                     mode="markers",
                     marker=dict(symbol="square", size=10, color=colors),
@@ -249,7 +259,7 @@ def build_drilldown_plot(
                 if idx_bad:
                     fig.add_trace(
                         go.Scatter(
-                            x=[plot_df["date_only"].iloc[i] for i in idx_bad],
+                            x=[plot_df[time_col].iloc[i] for i in idx_bad],
                             y=[y_main.iloc[i] for i in idx_bad],
                             mode="markers",
                             marker=dict(size=10, color=bad_color, symbol="circle"),
@@ -267,7 +277,7 @@ def build_drilldown_plot(
     if not plot_df.empty and not highlight_bad_days:
         fig.add_trace(
             go.Scatter(
-                x=plot_df["date_only"],
+                x=plot_df[time_col],
                 y=[0] * len(plot_df),
                 mode="markers",
                 opacity=0,
@@ -289,10 +299,10 @@ def build_drilldown_plot(
     try:
         force_all_dates = False
         try:
-            x_min = pd.to_datetime(plot_df["date_only"]).min()
-            x_max = pd.to_datetime(plot_df["date_only"]).max()
+            x_min = pd.to_datetime(plot_df[time_col]).min()
+            x_max = pd.to_datetime(plot_df[time_col]).max()
             span_days = int((x_max - x_min).days) + 1
-            n_dates = int(pd.to_datetime(plot_df["date_only"]).nunique())
+            n_dates = int(pd.to_datetime(plot_df[time_col]).nunique())
             force_all_dates = (span_days <= 200) and (n_dates <= 200)
         except Exception:
             force_all_dates = False
@@ -310,7 +320,7 @@ def build_drilldown_plot(
             col=1,
         )
 
-        if force_all_dates:
+        if force_all_dates and not is_hourly:
             fig.update_xaxes(
                 tickmode="linear",
                 dtick=86400000,
@@ -332,7 +342,10 @@ def build_drilldown_plot(
                 ticklen=6,
                 showgrid=True,
                 tickformatstops=[
-                    {"dtickrange": [None, 86400000], "value": "%d-%b\n%Y"},
+                    {
+                        "dtickrange": [None, 86400000],
+                        "value": "%d-%b\n%H:%M" if is_hourly else "%d-%b\n%Y",
+                    },
                     {"dtickrange": [86400000, 7 * 86400000], "value": "%d-%b"},
                     {"dtickrange": [7 * 86400000, "M1"], "value": "%d-%b"},
                     {"dtickrange": ["M1", "M12"], "value": "%b\n%Y"},

panel_app/kpi_health_check_panel.py

@@ -23,7 +23,9 @@ from process_kpi.kpi_health_check.engine import (
     evaluate_health_check,
     is_bad,
     max_consecutive_days,
+    max_consecutive_periods,
     window_bounds,
+    window_bounds_period,
 )
 from process_kpi.kpi_health_check.export import build_export_bytes
 from process_kpi.kpi_health_check.io import read_bytes_to_df
@@ -31,6 +33,7 @@ from process_kpi.kpi_health_check.kpi_groups import filter_kpis, get_kpis_by_gro
 from process_kpi.kpi_health_check.multi_rat import compute_multirat_views
 from process_kpi.kpi_health_check.normalization import (
     build_daily_kpi,
+    build_period_kpi,
     infer_date_col,
     infer_id_col,
 )
@@ -195,6 +198,7 @@ def _drilldown_cache_key(site_code: int, rat: str, kpi: str) -> tuple:
         int(site_code),
         str(rat or ""),
         str(kpi or ""),
+        str(granularity_select.value or "Daily"),
         compare_kpis_key,
         norm_key,
         ar_key,
@@ -375,6 +379,9 @@ complaint_sites_file = pn.widgets.FileInput(
 only_complaint_sites = pn.widgets.Checkbox(name="Only complaint sites", value=False)
 
 analysis_range = pn.widgets.DateRangePicker(name="Analysis date range (optional)")
+granularity_select = pn.widgets.RadioButtonGroup(
+    name="Granularity", options=["Daily", "Hourly"], value="Daily"
+)
 baseline_days = pn.widgets.IntInput(name="Baseline window (days)", value=30)
 recent_days = pn.widgets.IntInput(name="Recent window (days)", value=7)
 rel_threshold_pct = pn.widgets.FloatInput(
@@ -1485,7 +1492,7 @@ def _validate_inputs() -> tuple[list[str], list[str]]:
     return errors, warnings
 
 
-def _refresh_validation_state(event=None) -> None:
+def _refresh_validation_state() -> None:
     errors, warnings = _validate_inputs()
 
     msgs: list[str] = []
@@ -1499,15 +1506,7 @@ def _refresh_validation_state(event=None) -> None:
     else:
         validation_pane.object = ""
         validation_pane.alert_type = "success"
-        validation_pane.visible = False
-
-    has_data = bool(current_daily_by_rat)
-    has_results = (
-        isinstance(current_status_df, pd.DataFrame)
-        and not current_status_df.empty
-        and isinstance(current_summary_df, pd.DataFrame)
-        and not current_summary_df.empty
-    )
+        validation_pane.visible = bool(errors or warnings)
 
     try:
         run_button.disabled = bool(errors) or not has_data
@@ -1610,7 +1609,16 @@ def _update_kpi_options() -> None:
     kpis = [
         c
         for c in df.columns
-        if c not in {"site_code", "date_only", "Longitude", "Latitude", "City", "RAT"}
+        if c
+        not in {
+            "site_code",
+            "date_only",
+            "period_start",
+            "Longitude",
+            "Latitude",
+            "City",
+            "RAT",
+        }
     ]
     kpis = sorted([str(c) for c in kpis])
 
@@ -1744,7 +1752,15 @@ def _update_site_view(event=None) -> None:
             c
             for c in d.columns
             if c
-            not in {"site_code", "date_only", "Longitude", "Latitude", "City", "RAT"}
+            not in {
+                "site_code",
+                "date_only",
+                "period_start",
+                "Longitude",
+                "Latitude",
+                "City",
+                "RAT",
+            }
         ]
         candidate_kpis = sorted([str(c) for c in candidate_kpis])
         if not candidate_kpis:
@@ -1757,7 +1773,12 @@ def _update_site_view(event=None) -> None:
         new_kpi = candidate_kpis[0]
         _set_widget_value(kpi_select, new_kpi)
         kpi = new_kpi
-    s = d[d["site_code"] == int(code_int)].copy().sort_values("date_only")
+    g = str(granularity_select.value or "Daily").strip().lower()
+    is_hourly = g.startswith("hour") or g.startswith("h")
+    time_col = (
+        "period_start" if (is_hourly and "period_start" in d.columns) else "date_only"
+    )
+    s = d[d["site_code"] == int(code_int)].copy().sort_values(time_col)
     if s.empty:
         trend_plot_pane.object = None
         heatmap_plot_pane.object = None
@@ -1831,6 +1852,7 @@ def _update_site_view(event=None) -> None:
         recent_days_n=int(_coerce_int(recent_days.value) or 7),
         rel_threshold_pct=float(_coerce_float(rel_threshold_pct.value) or 10.0),
         normalization=str(kpi_compare_norm.value or "None"),
+        granularity=str(granularity_select.value or "Daily"),
     )
     trend_plot_pane.object = fig
     kpis_for_heatmap = []
@@ -1883,13 +1905,25 @@ def _update_site_view(event=None) -> None:
                 w = str(corr_window_select.value or "")
                 if w.startswith("Recent"):
                     df_corr = d[
-                        (d["date_only"] >= recent_start)
-                        & (d["date_only"] <= recent_end)
+                        (
+                            pd.to_datetime(d[time_col], errors="coerce")
+                            >= pd.to_datetime(recent_start)
+                        )
+                        & (
+                            pd.to_datetime(d[time_col], errors="coerce")
+                            <= pd.to_datetime(recent_end)
+                        )
                     ].copy()
                 elif w.startswith("Baseline"):
                     df_corr = d[
-                        (d["date_only"] >= baseline_start)
-                        & (d["date_only"] <= baseline_end)
+                        (
+                            pd.to_datetime(d[time_col], errors="coerce")
+                            >= pd.to_datetime(baseline_start)
+                        )
+                        & (
+                            pd.to_datetime(d[time_col], errors="coerce")
+                            <= pd.to_datetime(baseline_end)
+                        )
                     ].copy()
         except Exception:  # noqa: BLE001
             df_corr = d
@@ -2071,18 +2105,52 @@ def _infer_rule_row(rules_df: pd.DataFrame, rat: str, kpi: str) -> dict:
 
 def _compute_site_windows(
     daily_filtered: pd.DataFrame,
-) -> tuple[date, date, date, date] | None:
+) -> (
+    tuple[
+        date | pd.Timestamp,
+        date | pd.Timestamp,
+        date | pd.Timestamp,
+        date | pd.Timestamp,
+    ]
+    | None
+):
     if daily_filtered is None or daily_filtered.empty:
         return None
     rd = _coerce_int(recent_days.value)
     bd = _coerce_int(baseline_days.value)
     if rd is None or rd < 1 or bd is None or bd < 1:
         return None
+    g = str(granularity_select.value or "Daily").strip().lower()
+    is_hourly = g.startswith("hour") or g.startswith("h")
+    time_col = (
+        "period_start"
+        if (is_hourly and "period_start" in daily_filtered.columns)
+        else "date_only"
+    )
+
+    if is_hourly and time_col == "period_start":
+        try:
+            end_dt = pd.to_datetime(daily_filtered[time_col], errors="coerce").max()
+            if pd.isna(end_dt):
+                return None
+            end_dt = pd.Timestamp(end_dt).floor("h")
+        except Exception:  # noqa: BLE001
+            return None
+
+        recent_periods = int(rd) * 24
+        baseline_periods = int(bd) * 24
+        step = timedelta(hours=1)
+
+        recent_start, recent_end = window_bounds_period(end_dt, recent_periods, step)
+        baseline_end = recent_start - step
+        baseline_start, _ = window_bounds_period(baseline_end, baseline_periods, step)
+        return baseline_start, baseline_end, recent_start, recent_end
+
     try:
-        end_raw = max(daily_filtered["date_only"])
-        end_date = (
-            end_raw if isinstance(end_raw, date) else pd.to_datetime(end_raw).date()
-        )
+        end_ts = pd.to_datetime(daily_filtered["date_only"], errors="coerce").max()
+        if pd.isna(end_ts):
+            return None
+        end_date = end_ts.date()
     except Exception:  # noqa: BLE001
         return None
     recent_start, recent_end = window_bounds(end_date, int(rd))
@@ -2105,16 +2173,25 @@ def _build_site_heatmap(
         return None
     baseline_start, baseline_end, recent_start, recent_end = windows
 
+    g = str(granularity_select.value or "Daily").strip().lower()
+    is_hourly = g.startswith("hour") or g.startswith("h")
+    time_col = (
+        "period_start"
+        if (is_hourly and "period_start" in daily_filtered.columns)
+        else "date_only"
+    )
+
     site_daily = daily_filtered[daily_filtered["site_code"] == int(site_code)].copy()
     if site_daily.empty:
         return None
-    site_daily = site_daily.sort_values("date_only")
+    site_daily = site_daily.sort_values(time_col)
 
     dates = []
     cur = recent_start
+    step = timedelta(hours=1) if is_hourly else timedelta(days=1)
     while cur <= recent_end:
         dates.append(cur)
-        cur = cur + timedelta(days=1)
+        cur = cur + step
 
     z = []
     hover = []
@@ -2133,9 +2210,10 @@ def _build_site_heatmap(
 
         sla_eval = None if policy == "notify" else sla_val
 
-        s = site_daily[["date_only", kpi]].dropna(subset=[kpi])
-        baseline_mask = (s["date_only"] >= baseline_start) & (
-            s["date_only"] <= baseline_end
+        s = site_daily[[time_col, kpi]].dropna(subset=[kpi])
+        t = pd.to_datetime(s[time_col], errors="coerce")
+        baseline_mask = (t >= pd.to_datetime(baseline_start)) & (
+            t <= pd.to_datetime(baseline_end)
         )
         baseline = s.loc[baseline_mask, kpi].median() if baseline_mask.any() else np.nan
         baseline_val = float(baseline) if pd.notna(baseline) else None
@@ -2143,7 +2221,11 @@ def _build_site_heatmap(
         row_z = []
         row_h = []
         for d in dates:
-            v_series = site_daily.loc[site_daily["date_only"] == d, kpi]
+            v_series = site_daily.loc[
+                pd.to_datetime(site_daily[time_col], errors="coerce")
+                == pd.to_datetime(d),
+                kpi,
+            ]
             v = v_series.iloc[0] if not v_series.empty else np.nan
             if v is None or (isinstance(v, float) and np.isnan(v)):
                 row_z.append(None)
@@ -2186,7 +2268,7 @@ def _build_site_heatmap(
     fig.update_layout(
         template="plotly_white",
         title=f"{rat} - Site {int(site_code)} - Recent window heatmap",
-        xaxis_title="date",
+        xaxis_title="period",
         yaxis_title="KPI",
         height=420,
         margin=dict(l=40, r=20, t=60, b=40),
@@ -2212,15 +2294,26 @@ def _build_baseline_recent_hist(
         return None
     baseline_start, baseline_end, recent_start, recent_end = windows
 
+    g = str(granularity_select.value or "Daily").strip().lower()
+    is_hourly = g.startswith("hour") or g.startswith("h")
+    time_col = (
+        "period_start"
+        if (is_hourly and "period_start" in daily_filtered.columns)
+        else "date_only"
+    )
+
     site_daily = daily_filtered[daily_filtered["site_code"] == int(site_code)].copy()
     if site_daily.empty:
         return None
 
-    s = site_daily[["date_only", kpi]].dropna(subset=[kpi])
-    baseline_mask = (s["date_only"] >= baseline_start) & (
-        s["date_only"] <= baseline_end
+    s = site_daily[[time_col, kpi]].dropna(subset=[kpi])
+    t = pd.to_datetime(s[time_col], errors="coerce")
+    baseline_mask = (t >= pd.to_datetime(baseline_start)) & (
+        t <= pd.to_datetime(baseline_end)
+    )
+    recent_mask = (t >= pd.to_datetime(recent_start)) & (
+        t <= pd.to_datetime(recent_end)
     )
-    recent_mask = (s["date_only"] >= recent_start) & (s["date_only"] <= recent_end)
 
     baseline_vals = (
         pd.to_numeric(s.loc[baseline_mask, kpi], errors="coerce").dropna().astype(float)
@@ -2276,7 +2369,16 @@ def _build_corr_heatmap(
     cols = [
         c
         for c in cols
-        if c not in {"site_code", "date_only", "Longitude", "Latitude", "City", "RAT"}
+        if c
+        not in {
+            "site_code",
+            "date_only",
+            "period_start",
+            "Longitude",
+            "Latitude",
+            "City",
+            "RAT",
+        }
     ]
     cols = list(dict.fromkeys(cols))
     if len(cols) < 2:
@@ -2600,6 +2702,8 @@ def _refresh_profiles(event=None) -> None:
 def _current_profile_config() -> dict:
     cfg: dict = {}
 
+    cfg["granularity"] = str(granularity_select.value or "Daily")
+
     cfg["analysis_range"] = (
         [
             (
@@ -2653,6 +2757,13 @@ def _apply_profile_config(cfg: dict) -> None:
     _applying_profile = True
 
     try:
+        try:
+            g = str(cfg.get("granularity", "") or "").strip()
+            if g and g in list(granularity_select.options):
+                granularity_select.value = g
+        except Exception:  # noqa: BLE001
+            pass
+
         try:
             ar = cfg.get("analysis_range", [None, None])
             if isinstance(ar, (list, tuple)) and len(ar) == 2 and ar[0] and ar[1]:
@@ -2979,10 +3090,22 @@ def load_datasets(event=None) -> None:
             except Exception:  # noqa: BLE001
                 id_col = None
 
-            daily, kpi_cols = build_daily_kpi(df_raw, rat)
+            daily, kpi_cols = build_period_kpi(df_raw, rat, granularity_select.value)
             current_daily_by_rat[rat] = daily
 
             d = _filtered_daily(daily)
+            periods_n = None
+            try:
+                if (
+                    isinstance(d, pd.DataFrame)
+                    and not d.empty
+                    and "period_start" in d.columns
+                ):
+                    periods_n = int(
+                        pd.to_datetime(d["period_start"], errors="coerce").nunique()
+                    )
+            except Exception:  # noqa: BLE001
+                periods_n = None
             rows.append(
                 {
                     "RAT": rat,
@@ -2992,6 +3115,11 @@ def load_datasets(event=None) -> None:
                     "id_col": id_col,
                     "sites": int(d["site_code"].nunique()),
                     "days": int(d["date_only"].nunique()),
+                    "periods": (
+                        int(periods_n)
+                        if periods_n is not None
+                        else int(d["date_only"].nunique())
+                    ),
                     "kpis": int(len(kpi_cols)),
                 }
             )
@@ -3110,6 +3238,7 @@ def run_health_check(event=None) -> None:
                 int(rd),
                 float(thr),
                 int(mcd),
+                granularity=str(granularity_select.value or "Daily"),
             )
             if not status_df.empty:
                 all_status.append(status_df)
@@ -3230,6 +3359,7 @@ def _build_export_bytes() -> bytes:
         daily_by_rat=(
             current_daily_by_rat if isinstance(current_daily_by_rat, dict) else None
         ),
+        granularity=str(granularity_select.value or "Daily"),
         multirat_summary_df=(
             current_multirat_df
             if isinstance(current_multirat_df, pd.DataFrame)
@@ -3273,6 +3403,7 @@ def _export_callback() -> io.BytesIO:
 
 def _build_alert_pack_bytes() -> bytes:
     params = {
+        "granularity": str(granularity_select.value or "Daily"),
         "baseline_days": baseline_days.value,
         "recent_days": recent_days.value,
         "rel_threshold_pct": rel_threshold_pct.value,
@@ -3478,6 +3609,27 @@ def _on_drilldown_params_change(event=None) -> None:
     _refresh_validation_state()
 
 
+def _on_granularity_change(event=None) -> None:
+    if _applying_profile or _loading_datasets:
+        return
+    _invalidate_drilldown_cache(data_changed=True, healthcheck_changed=True)
+    _refresh_validation_state()
+    try:
+        has_any = bool(
+            (file_2g and file_2g.value)
+            or (file_3g and file_3g.value)
+            or (file_lte and file_lte.value)
+            or (file_twamp and file_twamp.value)
+        )
+    except Exception:  # noqa: BLE001
+        has_any = False
+    if has_any:
+        try:
+            load_datasets()
+        except Exception:  # noqa: BLE001
+            pass
+
+
 rat_select.param.watch(_on_rat_change, "value")
 kpi_group_select.param.watch(
     _on_rat_change, "value"
@@ -3498,6 +3650,7 @@ map_status_filter.param.watch(lambda e: _refresh_map_view(), "value")
 map_auto_fit.param.watch(lambda e: _refresh_map_view(), "value")
 
 analysis_range.param.watch(_on_drilldown_params_change, "value")
+granularity_select.param.watch(_on_granularity_change, "value")
 baseline_days.param.watch(_on_drilldown_params_change, "value")
 recent_days.param.watch(_on_drilldown_params_change, "value")
 rel_threshold_pct.param.watch(_on_drilldown_params_change, "value")
@@ -3586,7 +3739,12 @@ def _build_drilldown_export_bytes() -> bytes:
     if d is None or d.empty:
         return b""
 
-    s = d[d["site_code"] == int(code_int)].copy().sort_values("date_only")
+    g = str(granularity_select.value or "Daily").strip().lower()
+    is_hourly = g.startswith("hour") or g.startswith("h")
+    time_col = (
+        "period_start" if (is_hourly and "period_start" in d.columns) else "date_only"
+    )
+    s = d[d["site_code"] == int(code_int)].copy().sort_values(time_col)
     if s.empty:
         return b""
 
@@ -3598,9 +3756,9 @@ def _build_drilldown_export_bytes() -> bytes:
             selected_kpis = [str(kpi_select.value)] + selected_kpis
 
     selected_kpis = [k for k in selected_kpis if k in d.columns]
-    base_cols = ["date_only"]
+    base_cols = [time_col]
     daily_cols = base_cols + selected_kpis
-    daily_out = s[daily_cols].copy() if selected_kpis else s[["date_only"]].copy()
+    daily_out = s[daily_cols].copy() if selected_kpis else s[[time_col]].copy()
 
     rules_df = (
         rules_table.value
@@ -3623,15 +3781,23 @@ def _build_drilldown_export_bytes() -> bytes:
             except Exception:  # noqa: BLE001
                 sla_val = None
 
-            sk = s[["date_only", k]].copy()
+            sk = s[[time_col, k]].copy()
             sk[k] = pd.to_numeric(sk[k], errors="coerce")
             sk = sk.dropna(subset=[k])
 
-            baseline_mask = (sk["date_only"] >= baseline_start) & (
-                sk["date_only"] <= baseline_end
+            baseline_mask = (
+                pd.to_datetime(sk[time_col], errors="coerce")
+                >= pd.to_datetime(baseline_start)
+            ) & (
+                pd.to_datetime(sk[time_col], errors="coerce")
+                <= pd.to_datetime(baseline_end)
             )
-            recent_mask = (sk["date_only"] >= recent_start) & (
-                sk["date_only"] <= recent_end
+            recent_mask = (
+                pd.to_datetime(sk[time_col], errors="coerce")
+                >= pd.to_datetime(recent_start)
+            ) & (
+                pd.to_datetime(sk[time_col], errors="coerce")
+                <= pd.to_datetime(recent_end)
             )
 
             baseline_med = (
@@ -3644,8 +3810,8 @@ def _build_drilldown_export_bytes() -> bytes:
             )
 
             bad_flags = []
-            recent_vals = sk.loc[recent_mask, ["date_only", k]].sort_values("date_only")
-            bad_dates: list[date] = []
+            recent_vals = sk.loc[recent_mask, [time_col, k]].sort_values(time_col)
+            bad_dates = []
             for _, r in recent_vals.iterrows():
                 v = r.get(k)
                 is_bad_day = bool(
@@ -3660,13 +3826,9 @@ def _build_drilldown_export_bytes() -> bytes:
                 bad_flags.append(is_bad_day)
                 if is_bad_day:
                     try:
-                        d0 = r.get("date_only")
+                        d0 = r.get(time_col)
                         if d0 is not None:
-                            bad_dates.append(
-                                d0
-                                if isinstance(d0, date)
-                                else pd.to_datetime(d0).date()
-                            )
+                            bad_dates.append(pd.to_datetime(d0, errors="coerce"))
                     except Exception:  # noqa: BLE001
                         pass
 
@@ -3680,7 +3842,16 @@ def _build_drilldown_export_bytes() -> bytes:
                     "baseline_median": baseline_med,
                     "recent_median": recent_med,
                     "bad_days_recent": int(sum(bad_flags)),
-                    "max_streak_recent": int(max_consecutive_days(bad_dates)),
+                    "max_streak_recent": int(
+                        max_consecutive_periods(
+                            bad_dates,
+                            step=(
+                                timedelta(hours=1) if is_hourly else timedelta(days=1)
+                            ),
+                        )
+                        if bad_dates
+                        else 0
+                    ),
                 }
             )
         summary_out = pd.DataFrame(rows)
@@ -3692,14 +3863,16 @@ def _build_drilldown_export_bytes() -> bytes:
     )
     if not status_df.empty:
         status_out = status_df[
-            (status_df["site_code"] == int(code)) & (status_df["RAT"] == str(rat))
+            (status_df["site_code"] == int(code_int)) & (status_df["RAT"] == str(rat))
         ].copy()
     else:
         status_out = pd.DataFrame()
 
+    g2 = str(granularity_select.value or "Daily").strip().lower()
+    data_sheet = "Hourly" if (g2.startswith("hour") or g2.startswith("h")) else "Daily"
     return write_dfs_to_excel(
         [summary_out, daily_out, status_out],
-        ["Summary", "Daily", "KPI_Status"],
+        ["Summary", data_sheet, "KPI_Status"],
         index=False,
     )
 
@@ -3732,6 +3905,7 @@ sidebar = pn.Column(
     sidebar,
     pn.Card(
         analysis_range,
+        granularity_select,
         pn.Row(baseline_days, recent_days, sizing_mode="stretch_width"),
         rel_threshold_pct,
         min_consecutive_days,
@@ -3757,7 +3931,6 @@ sidebar = pn.Column(
         min_criticality,
         min_anomaly_score,
         city_filter,
-        only_complaint_sites,
         top_rat_filter,
         top_status_filter,
         title="Filters",

process_kpi/kpi_health_check/engine.py

@@ -1,9 +1,34 @@
-from datetime import date, timedelta
+from datetime import date, datetime, timedelta
 
 import numpy as np
 import pandas as pd
 
 
+def _to_timestamp(value) -> pd.Timestamp | None:
+    if value is None:
+        return None
+    if isinstance(value, pd.Timestamp):
+        return value
+    if isinstance(value, datetime):
+        return pd.Timestamp(value)
+    if isinstance(value, date):
+        return pd.Timestamp(value)
+    try:
+        v = pd.to_datetime(value, errors="coerce")
+        return v if pd.notna(v) else None
+    except Exception:  # noqa: BLE001
+        return None
+
+
+def window_bounds_period(
+    end_dt: pd.Timestamp,
+    periods: int,
+    step: timedelta,
+) -> tuple[pd.Timestamp, pd.Timestamp]:
+    start = end_dt - step * (int(periods) - 1)
+    return start, end_dt
+
+
 def window_bounds(end_date: date, days: int) -> tuple[date, date]:
     start = end_date - timedelta(days=days - 1)
     return start, end_date
@@ -34,14 +59,18 @@ def is_bad(
     return bad or (value > baseline + abs(baseline) * thr)
 
 
-def max_consecutive_days(dates: list[date]) -> int:
-    if not dates:
+def max_consecutive_periods(values: list, step: timedelta) -> int:
+    if not values:
         return 0
-    dates_sorted = sorted(set(dates))
+    ts = [_to_timestamp(v) for v in values]
+    ts2 = [t for t in ts if t is not None]
+    if not ts2:
+        return 0
+    ts_sorted = sorted(set(ts2))
     streak = 1
     best = 1
-    for prev, cur in zip(dates_sorted, dates_sorted[1:]):
-        if cur == prev + timedelta(days=1):
+    for prev, cur in zip(ts_sorted, ts_sorted[1:]):
+        if cur == prev + step:
             streak += 1
         else:
             streak = 1
@@ -50,6 +79,10 @@ def max_consecutive_days(dates: list[date]) -> int:
     return best
 
 
+def max_consecutive_days(dates: list[date]) -> int:
+    return max_consecutive_periods(dates, step=timedelta(days=1))
+
+
 def evaluate_health_check(
     daily: pd.DataFrame,
     rat: str,
@@ -58,14 +91,33 @@ def evaluate_health_check(
     recent_days_n: int,
     rel_threshold_pct: float,
     min_consecutive_days: int,
+    granularity: str = "Daily",
 ) -> tuple[pd.DataFrame, pd.DataFrame]:
     if daily.empty:
         return pd.DataFrame(), pd.DataFrame()
 
-    end_date = max(daily["date_only"])
-    recent_start, recent_end = window_bounds(end_date, int(recent_days_n))
-    baseline_end = recent_start - timedelta(days=1)
-    baseline_start = baseline_end - timedelta(days=int(baseline_days_n) - 1)
+    g = str(granularity or "Daily").strip().lower()
+    is_hourly = g.startswith("hour") or g.startswith("h")
+    time_col = (
+        "period_start"
+        if (is_hourly and "period_start" in daily.columns)
+        else "date_only"
+    )
+
+    step = timedelta(hours=1) if is_hourly else timedelta(days=1)
+    baseline_periods = int(baseline_days_n) * 24 if is_hourly else int(baseline_days_n)
+    recent_periods = int(recent_days_n) * 24 if is_hourly else int(recent_days_n)
+    min_periods = (
+        int(min_consecutive_days) * 24 if is_hourly else int(min_consecutive_days)
+    )
+
+    end_dt = _to_timestamp(daily[time_col].max())
+    if end_dt is None:
+        return pd.DataFrame(), pd.DataFrame()
+
+    recent_start_dt, recent_end_dt = window_bounds_period(end_dt, recent_periods, step)
+    baseline_end_dt = recent_start_dt - step
+    baseline_start_dt, _ = window_bounds_period(baseline_end_dt, baseline_periods, step)
 
     rat_rules = rules_df[rules_df["RAT"] == rat].copy()
     kpis = [k for k in rat_rules["KPI"].tolist() if k in daily.columns]
@@ -78,7 +130,7 @@ def evaluate_health_check(
             if ("City" in g_site.columns and g_site["City"].notna().any())
             else None
         )
-        g_site = g_site.sort_values("date_only")
+        g_site = g_site.sort_values(time_col)
 
         for kpi in kpis:
             rule = rat_rules[rat_rules["KPI"] == kpi].iloc[0]
@@ -92,7 +144,7 @@ def evaluate_health_check(
 
             sla_eval = None if policy == "notify" else sla_val
 
-            s = g_site[["date_only", kpi]].dropna(subset=[kpi])
+            s = g_site[[time_col, kpi]].dropna(subset=[kpi])
             if s.empty:
                 rows.append(
                     {
@@ -105,23 +157,20 @@ def evaluate_health_check(
                 )
                 continue
 
-            baseline_mask = (s["date_only"] >= baseline_start) & (
-                s["date_only"] <= baseline_end
-            )
-            recent_mask = (s["date_only"] >= recent_start) & (
-                s["date_only"] <= recent_end
-            )
+            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 = (
                 s.loc[baseline_mask, kpi].median() if baseline_mask.any() else np.nan
             )
             recent = s.loc[recent_mask, kpi].median() if recent_mask.any() else np.nan
 
-            daily_recent = s.loc[recent_mask, ["date_only", kpi]].copy()
+            daily_recent = s.loc[recent_mask, [time_col, kpi]].copy()
             bad_dates = []
             if not daily_recent.empty:
                 for d, v in zip(
-                    daily_recent["date_only"].tolist(), daily_recent[kpi].tolist()
+                    daily_recent[time_col].tolist(), daily_recent[kpi].tolist()
                 ):
                     if is_bad(
                         float(v) if pd.notna(v) else None,
@@ -132,8 +181,8 @@ def evaluate_health_check(
                     ):
                         bad_dates.append(d)
 
-            max_streak = max_consecutive_days(bad_dates)
-            persistent = max_streak >= int(min_consecutive_days)
+            max_streak = max_consecutive_periods(bad_dates, step=step)
+            persistent = max_streak >= int(min_periods)
 
             is_bad_recent = is_bad(
                 float(recent) if pd.notna(recent) else None,
@@ -145,7 +194,7 @@ def evaluate_health_check(
 
             is_bad_current = is_bad_recent
             if not daily_recent.empty:
-                last_row = daily_recent.sort_values("date_only").iloc[-1]
+                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,

process_kpi/kpi_health_check/export.py

@@ -9,6 +9,7 @@ def build_export_bytes(
     summary_df: pd.DataFrame | None,
     status_df: pd.DataFrame | None,
     daily_by_rat: dict[str, pd.DataFrame] | None = None,
+    granularity: str = "Daily",
     multirat_summary_df: pd.DataFrame | None = None,
     top_anomalies_df: pd.DataFrame | None = None,
     complaint_multirat_df: pd.DataFrame | None = None,
@@ -32,10 +33,12 @@ def build_export_bytes(
 
     max_data_rows = 1048575
     if daily_by_rat and isinstance(daily_by_rat, dict):
+        g = str(granularity or "Daily").strip().lower()
+        prefix = "Hourly" if (g.startswith("hour") or g.startswith("h")) else "Daily"
         for rat, df in daily_by_rat.items():
             if not isinstance(df, pd.DataFrame):
                 continue
-            base = f"Daily_All_{str(rat)}"
+            base = f"{prefix}_All_{str(rat)}"
             if len(df) <= max_data_rows:
                 dfs.append(df)
                 sheet_names.append(base[:31])

process_kpi/kpi_health_check/normalization.py

@@ -228,20 +228,29 @@ def load_physical_db() -> pd.DataFrame:
     return physical_db[keep].drop_duplicates("code")
 
 
-def build_daily_kpi(df_raw: pd.DataFrame, rat: str) -> tuple[pd.DataFrame, list[str]]:
+def build_period_kpi(
+    df_raw: pd.DataFrame,
+    rat: str,
+    granularity: str = "Daily",
+) -> tuple[pd.DataFrame, list[str]]:
     df = df_raw.copy()
     date_col = infer_date_col(df)
     id_col = infer_id_col(df, rat)
 
     df["date"] = parse_datetime(df[date_col])
     df = df.dropna(subset=["date"])
-    df["date_only"] = df["date"].dt.date
+
+    g = str(granularity or "Daily").strip().lower()
+    if g.startswith("hour") or g.startswith("h"):
+        df["period_start"] = df["date"].dt.floor("H")
+    else:
+        df["period_start"] = df["date"].dt.floor("D")
 
     df["site_code"] = df[id_col].apply(extract_site_code)
     df = df.dropna(subset=["site_code"])
     df["site_code"] = df["site_code"].astype(int)
 
-    meta = {date_col, id_col, "date", "date_only", "site_code"}
+    meta = {date_col, id_col, "date", "site_code", "period_start"}
     meta = meta.union(non_kpi_identifier_cols(df, rat))
     candidate_cols = [c for c in df.columns if c not in meta]
 
@@ -259,22 +268,25 @@ def build_daily_kpi(df_raw: pd.DataFrame, rat: str) -> tuple[pd.DataFrame, list[
 
     base = pd.concat(
         [
-            df[["site_code", "date_only"]].reset_index(drop=True),
+            df[["site_code", "period_start"]].reset_index(drop=True),
             numeric_df[kpi_cols].reset_index(drop=True),
         ],
         axis=1,
     )
 
     agg_dict = {k: infer_agg(k) for k in kpi_cols}
-    daily = base.groupby(["site_code", "date_only"], as_index=False).agg(agg_dict)
+    out = base.groupby(["site_code", "period_start"], as_index=False).agg(agg_dict)
+    out["date_only"] = pd.to_datetime(out["period_start"]).dt.date
 
     physical = load_physical_db()
     if not physical.empty:
-        daily = pd.merge(
-            daily, physical, left_on="site_code", right_on="code", how="left"
-        )
-        daily = daily.drop(columns=[c for c in ["code"] if c in daily.columns])
+        out = pd.merge(out, physical, left_on="site_code", right_on="code", how="left")
+        out = out.drop(columns=[c for c in ["code"] if c in out.columns])
 
-    daily["RAT"] = rat
+    out["RAT"] = rat
 
-    return daily, kpi_cols
+    return out, kpi_cols
+
+
+def build_daily_kpi(df_raw: pd.DataFrame, rat: str) -> tuple[pd.DataFrame, list[str]]:
+    return build_period_kpi(df_raw, rat, granularity="Daily")