Reducing false alarms with multi-sensor correlation in Python

import pandas as pd
import numpy as np
import logging
from typing import Tuple

logger = logging.getLogger(__name__)

def align_sensor_streams(
    df: pd.DataFrame,
    freq: str = "1min",
    max_gap_minutes: int = 5,
) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """
    Normalizes asynchronous sensor telemetry into a zone-aligned time series.
    Returns the aligned DataFrame and an audit log of dropped/imputed records.
    """
    if df.empty:
        raise ValueError("Input telemetry DataFrame is empty.")

    df = df.copy()
    df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
    df = df.set_index("timestamp").sort_index()

    # Validate required columns exist
    required_cols = {"location_zone", "sensor_id", "temperature_c"}
    if not required_cols.issubset(df.columns):
        raise KeyError(f"Missing required columns: {required_cols - set(df.columns)}")

    # Pivot to wide format: index=timestamp, columns=MultiIndex(zone, sensor_id)
    pivot = df.pivot_table(
        index="timestamp",
        columns=["location_zone", "sensor_id"],
        values="temperature_c",
        aggfunc="mean",
    )

    # Resample to target frequency
    pivot = pivot.resample(freq).mean()

    # Strict forward-fill with explicit gap limit to prevent synthetic data generation.
    # Limit is derived from the grid step so that callers can use any pandas freq alias
    # (e.g. "30s", "1min", "5min") without re-deriving the math.
    step = pd.Timedelta(freq)
    if step <= pd.Timedelta(0):
        raise ValueError(f"freq must resolve to a positive Timedelta, got {freq!r}")
    ffill_limit = max(1, int(pd.Timedelta(minutes=max_gap_minutes) // step))
    aligned = pivot.ffill(limit=ffill_limit)

    # Generate compliance audit log. unrecoverable_gaps counts only cells that
    # remained NaN *after* the bounded ffill — those are the audit-visible gaps.
    audit_log = pd.DataFrame({
        "timestamp": pivot.index,
        "total_sensors": pivot.shape[1],
        "sensors_with_data": pivot.notna().sum(axis=1).to_numpy(),
        "imputed_records": (pivot.isna() & aligned.notna()).sum(axis=1).to_numpy(),
        "unrecoverable_gaps": (pivot.isna() & aligned.isna()).sum(axis=1).to_numpy(),
    })

    logger.info(
        "Aligned %d time steps. Max gap tolerance: %dm (ffill limit: %d periods).",
        len(aligned), max_gap_minutes, ffill_limit,
    )
    return aligned, audit_log

def evaluate_consensus(
    aligned_df: pd.DataFrame,
    window_size: str = "15min",
    z_threshold: float = 2.5,
    min_corroborating_sensors: int = 2
) -> pd.DataFrame:
    """
    Applies spatial-temporal consensus logic to flag validated excursions.
    """
    # Calculate rolling zone statistics. min_periods=2 so the first row, where
    # std() is undefined, does not silently produce False flags.
    rolling_mean = aligned_df.rolling(window=window_size, min_periods=2).mean()
    rolling_std = aligned_df.rolling(window=window_size, min_periods=2).std()

    # Compute z-scores per sensor, handling division by zero
    z_scores = (aligned_df - rolling_mean) / rolling_std.replace(0, np.nan)

    # Identify sensors exceeding threshold
    exceeds_threshold = z_scores.abs() > z_threshold

    # Count corroborating sensors per zone. pandas 2.1+ removed the axis=1
    # kwarg from DataFrame.groupby, so we transpose, group by the zone level
    # of the (zone, sensor_id) MultiIndex, sum, and transpose back.
    zone_corroboration = exceeds_threshold.T.groupby(level="location_zone").sum().T

    # Flag timestamps where minimum corroboration is met
    consensus_flags = zone_corroboration >= min_corroborating_sensors

    # Broadcast zone-level consensus back to every (zone, sensor_id) column
    broadcast = consensus_flags.reindex(
        columns=aligned_df.columns, level="location_zone"
    ).fillna(False)
    alert_matrix = pd.DataFrame(
        np.where(broadcast, 1, 0),
        index=aligned_df.index,
        columns=aligned_df.columns,
    )

    return alert_matrix

def classify_and_route(
    aligned_df: pd.DataFrame,
    alert_matrix: pd.DataFrame,
    static_limits: Tuple[float, float] = (2.0, 8.0)
) -> pd.DataFrame:
    """
    Classifies events into: 'VALID_EXCURSION', 'SENSOR_ARTIFACT', or 'NORMAL'.
    Prepares structured output for QA routing.
    """
    lower, upper = static_limits
    status = pd.DataFrame("NORMAL", index=aligned_df.index, columns=aligned_df.columns)
    
    # Mark static threshold breaches
    static_breach = (aligned_df < lower) | (aligned_df > upper)
    status[static_breach] = "THRESHOLD_BREACH"
    
    # Override with consensus validation
    consensus_active = alert_matrix == 1
    status[consensus_active] = "VALID_EXCURSION"
    
    # Quarantine single-sensor static breaches without consensus
    artifact_mask = static_breach & ~consensus_active
    status[artifact_mask] = "SENSOR_ARTIFACT"
    
    return status