Clean and aggregate factory event data in Pandas

You are given factory event data with possible duplicate events for the same logical event_id. Duplicates are defined as rows that share the same event_id but may have slightly different ts_utc values (up to a couple of seconds apart). The goal is to deduplicate these events in a deterministic way while preserving chronological ordering. Using the events table below: - Treat ts_utc as the canonical timestamp (already in UTC). - If multiple rows have the same event_id, keep only the row with the latest ts_utc. - Return one row per event_id with the columns: event_id, machine_id, ts_utc, event_type, batch_id. Format `ts_utc` as `YYYY-MM-DDTHH24:MI:SSZ`. - Order the result by machine_id ascending, then ts_utc ascending. Write a SQL query to produce this deduplicated, time-ordered list of events.

You are building per-machine, per-hour features for a 4-hour window on a UTC time axis, combining deduplicated production events with machine telemetry. There are three tables (`events`, `telemetry`, `calendar`). For this question only `events` and `telemetry` are needed. Assume the window is the 4 hourly buckets from **2025-05-26 00:00:00 UTC** through **2025-05-26 03:00:00 UTC** (inclusive of both endpoints), so the hour-start timestamps are 00:00, 01:00, 02:00, and 03:00 on 2025-05-26. Produce the feature table as follows: 1. **Deduplicate events.** The `events` table may contain duplicate rows for the same `event_id` (a late re-emission). For each `event_id`, keep only the row with the latest `ts_utc`. 2. **Normalize telemetry to UTC.** `telemetry.ts_local` is a wall-clock timestamp with no zone; the `timezone` column holds the IANA zone name (e.g. `'UTC'`, `'America/Los_Angeles'`). Convert each reading to a UTC instant before bucketing it. 3. **Build a complete hour grid.** Output one row for **every** machine and **every** one of the 4 hour buckets in the window — even hours with no events and no telemetry. The set of machines is the distinct machines that appear in either the deduplicated events or the telemetry. 4. **`throughput` (INT).** Using the deduplicated, time-ordered events per machine, a *completed production cycle* is a `'stop'` event whose immediately preceding event (by `ts_utc`, same machine) is a `'start'`. Count, per machine and hour, how many such completed cycles have their `'stop'` timestamp falling in that hour. Hours with no completed cycle have `throughput = 0`. 5. **`temp_p95`.** The 95th percentile (discrete, `percentile_disc`) of `temperature_c` over the normalized telemetry readings that fall in that machine/hour. NULL when the hour has no telemetry. 6. **`avg_power_kw`.** The average `power_kw` over the normalized telemetry readings in that machine/hour. NULL when the hour has no telemetry. 7. **`power_kw_z24`.** A rolling 24-hour z-score of `avg_power_kw` per machine, ordered by `hour_start_utc`, over the frame `ROWS BETWEEN 23 PRECEDING AND CURRENT ROW` (up to the previous 23 hours plus the current hour). The z-score is `(avg_power_kw - mean) / stddev`, where `mean` and the **sample** standard deviation (`stddev_samp`) are taken over the non-null hourly averages in that frame. If fewer than 2 non-null hourly averages are in the frame, or the sample standard deviation is 0, the z-score is NULL. **Return exactly these columns**, one row per machine and hour: `machine_id`, `hour_start_utc`, `throughput`, `temp_p95`, `avg_power_kw`, `power_kw_z24`. **Sort the result by `machine_id`, then `hour_start_utc` ascending.**

Two source tables capture factory activity over a fixed time window: - **`events`** — machine lifecycle events. Each physical event may be **duplicated** (same `event_id` inserted more than once with a slightly different `ts_utc`); keep only **one** copy per `event_id` (the latest `ts_utc`). `ts_utc` is already in UTC (`TIMESTAMP WITH TIME ZONE`). `event_type` is one of `'start'`, `'stop'`, `'fault'`. - **`telemetry`** — periodic sensor readings (`temperature_c`, `power_kw`) stored in **machine-local time** (`ts_local`, a naive timestamp) together with the IANA `timezone` of that machine. Normalize every reading to UTC with `ts_local AT TIME ZONE timezone`. (The `calendar` table is provided for context only and is not required by this question.) Build a **tidy machine-hour panel** over the window **2025-05-26 00:00:00 UTC through 2025-05-26 05:00:00 UTC**, inclusive of both endpoints — i.e. the **6 hourly buckets** `00:00, 01:00, 02:00, 03:00, 04:00, 05:00` (UTC). The grid must be **complete**: emit **one row per `machine_id` and `hour_start_utc`** even for hours with no events and no telemetry. A machine appears in the grid if it has at least one row in `events` or in `telemetry`. **Step 1 — per machine and hour, compute the raw hourly metrics:** - `throughput` = number of **completed cycles** ending in that hour. A completed cycle is a deduplicated `'stop'` event whose immediately preceding event for the same machine (ordered by `ts_utc`) is a `'start'`. Count a cycle in the hour bucket `[hour_start_utc, hour_start_utc + 1 hour)` that contains its `'stop'` timestamp. - `temp_p95` = the 95th-percentile temperature (`percentile_disc(0.95)`) over telemetry readings whose UTC timestamp falls in that hour bucket; `NULL` if the hour has no telemetry. - `avg_power_kw` = the average `power_kw` over telemetry readings in that hour bucket; `NULL` if none. - `power_kw_z24` = the rolling 24-hour z-score of `avg_power_kw` within each machine, over the window frame `ROWS BETWEEN 23 PRECEDING AND CURRENT ROW` ordered by `hour_start_utc`. Define it as `(avg_power_kw - mean) / stddev_samp` over that frame, and leave it `NULL` when fewer than 2 non-null values are in the frame or the sample standard deviation is 0. **Step 2 — impute missing values:** - `throughput`: replace `NULL` with `0`. - `temp_p95`: replace `NULL` with the **mean of `temp_p95` for that `machine_id` over the whole window** (ignoring `NULL`s) → `temp_p95_imputed`. - `avg_power_kw`: replace `NULL` with the **mean of `avg_power_kw` for that `machine_id` over the whole window** (ignoring `NULL`s) → `avg_power_kw_imputed`. - `power_kw_z24`: replace `NULL` with `0` → `power_kw_z24_imputed`. **Step 3 — anomaly flag:** add a boolean `is_anomaly` that is `TRUE` when `ABS(power_kw_z24_imputed) > 3`, else `FALSE`. **Output:** one row per `machine_id`, `hour_start_utc` with the columns, in this order: `machine_id`, `hour_start_utc`, `throughput`, `temp_p95_imputed`, `avg_power_kw_imputed`, `power_kw_z24_imputed`, `is_anomaly`. Round the three imputed numeric columns to **4 decimal places**. Order the result by `machine_id`, then `hour_start_utc` ascending.