Time-Series Alignment Strategies for Water Utility Compliance

Time-series alignment is the transformation layer that turns raw, unevenly sampled SCADA telemetry into the regularized, defensible record that regulatory reporting depends on. Readings arrive from field devices on no shared clock: polling latency varies, PLC scan times drift, historians write asynchronously, and daylight-saving transitions duplicate or erase an hour twice a year. Left unaligned, that jitter silently corrupts rolling averages, shifts exceedance flags, and breaks the continuous-monitoring assumptions the Safe Drinking Water Act (SDWA) builds on. This section, part of the parent SCADA Data Ingestion & Time-Series Sync domain, covers the end-to-end alignment workflow — UTC anchoring, rule-based resampling, gap governance, and immutable lineage — for the Python automation builders who own the ingestion service, the environmental compliance teams who sign the resulting reports, and the municipal operations engineers who maintain the field devices. It assumes decoded, quality-flagged values already flow in from Modbus TCP Parsing Workflows and OPC UA Data Extraction, and it hands aligned series forward to the compliance engines that judge them.

Regulatory / Protocol Foundation

Alignment is not a convenience — it is a compliance control point, because the rules that govern water-quality monitoring are written in the language of time. The EPA Safe Drinking Water Act (SDWA) framework and its implementing regulations in 40 CFR Part 141 express limits as averages over fixed windows: a running annual average for total trihalomethanes, a monthly average for many disinfection-byproduct parameters, and locational running annual averages under the Stage 2 Disinfectants and Disinfection Byproducts Rule. A running annual average of a parameter sampled irregularly is only meaningful once every sample sits on a known, uniform time base. The specific numeric limits those averages are compared against are resolved downstream against the SDWA MCL Reference Mapping, and the cadence each parameter must be sampled at is generated by Monitoring Frequency Scheduling — both of which assume the timestamps reaching them are already normalized.

Two temporal constraints drive every design decision in this section:

• A single reference frame. Local wall-clock time is ambiguous by construction — during a fall-back transition an hour repeats, and during spring-forward an hour never exists. Averaging across either boundary double-counts or drops samples inside a compliance window. Every timestamp is therefore converted to UTC the instant it enters the pipeline, and edge-device clocks are disciplined against NIST-traceable time synchronization standards so the reference frame is auditable end to end.
• Continuous-monitoring evidence. A lapse in the telemetry stream is legally a monitoring gap, so alignment must distinguish a genuine reading from an interpolated one and from an unrecoverable void. The output series carries that distinction as metadata, never silently smoothing over a gap that a reviewer is entitled to see.

Architecture & Design Decisions

The alignment service is a pure, deterministic transform: the same raw input under the same configuration must always produce the same aligned output, because a regulatory report has to be reconstructible years after the fact. Three design decisions shape the rest of this section.

Raw records stay immutable. Alignment never overwrites a source reading. The upstream decoders — Modbus TCP Parsing Workflows and OPC UA Data Extraction — write append-only raw records carrying device_id, raw_value, acquisition_timestamp, and a quality_flag. Alignment reads those and emits a new, separate aligned series. Because EPA reporting requires that actual measured values remain producible, keeping the raw layer untouched means a fill rule can be corrected and the whole history re-aligned without any return trip to the field.

Resampling method is configuration, not a global default. Different parameters demand different aggregation, and choosing wrong either hides an excursion or invents one. The method is pinned per tag in a rule table rather than assumed:

Time-weighted averaging for irregular samples. A plain arithmetic mean over-weights whichever interval happened to be polled most densely. For irregular telemetry the compliance-correct aggregate weights each reading by the span it represents:

where $x_i$ is the reading at time $t_i$. Using the time-weighted form keeps a five-minute chlorine value from counting the same as a value that stood for an hour. The decoded, aligned records then flow to the Violation Detection Rule Engine Logic for exceedance evaluation, and high-fan-out backfills are handed to Async Batch Processing Setup so a slow historical re-alignment cannot starve the live poll loop.

Phase-by-Phase Implementation

The workflow is built in four phases, each producing an artifact the next depends on: a UTC-anchored series, a resampled grid, a gap-classified grid, and an audit-ready lineage record.

Phase 1 — UTC anchoring and DST resolution

Field PLCs commonly log in the facility’s local time zone with no offset recorded, so the first phase localizes each naive timestamp against the facility’s IANA zone and converts to UTC. The two boundary cases — an ambiguous fall-back hour and a nonexistent spring-forward hour — must be handled explicitly rather than left to a library default, because a silently mis-resolved boundary shifts every subsequent window.

Implementation steps:

1. Parse heterogeneous timestamp formats in a single pass, coercing unparseable values to NaT and logging their row indices rather than dropping them silently.
2. Localize naive timestamps to the facility’s IANA zone, flagging fall-back ambiguity for review and shifting spring-forward gaps to the next valid wall-clock time.
3. Convert to timezone-aware UTC and sort ascending so all downstream windowing is monotonic.

import logging

import pandas as pd

logger = logging.getLogger("alignment.anchor")


def anchor_to_utc(
    frame: pd.DataFrame,
    tz: str,
    ts_col: str = "timestamp",
) -> pd.DataFrame:
    """Localize facility-local timestamps and anchor every reading to UTC.

    `tz` is an IANA key such as 'America/New_York'. Fall-back ambiguity is
    isolated for manual review; spring-forward gaps shift to the next valid time.
    """
    out = frame.copy()
    naive = pd.to_datetime(out[ts_col], format="mixed", errors="coerce")

    localized = naive.dt.tz_localize(
        tz, ambiguous="NaT", nonexistent="shift_forward"
    )
    ambiguous = localized.isna() & naive.notna()
    if ambiguous.any():
        logger.warning(
            "DST fall-back ambiguity in %d rows; flagged for review",
            int(ambiguous.sum()),
        )
        out.loc[ambiguous, "align_flag"] = "DST_AMBIGUOUS_REVIEW"

    out["ts_utc"] = localized.dt.tz_convert("UTC")
    return out.dropna(subset=["ts_utc"]).sort_values("ts_utc")

Phase 2 — Rule-based resampling to a fixed grid

With every reading anchored to UTC, the pipeline collapses disparate native cadences onto the fixed grid the reporting window requires. The aggregation method is looked up per parameter from the rule table rather than hard-coded, so a volatile turbidity signal is summarized by its peak while a slow chlorine residual is summarized by its mean.

Implementation steps:

1. Set the UTC timestamp as a monotonic index and select the resample method for the parameter from the rule table.
2. Resample to the target frequency, labelling each bucket by its window start so a value can never be attributed to the wrong reporting interval.
3. Preserve the per-bucket sample count so a partially populated window is distinguishable from a fully sampled one downstream.

import pandas as pd

RESAMPLE_METHOD = {
    "turbidity_ntu": "max",
    "chlorine_residual_mg_l": "mean",
    "pump_state": "ffill",
}


def resample_to_grid(
    frame: pd.DataFrame,
    value_col: str,
    freq: str = "1h",
) -> pd.DataFrame:
    """Collapse an irregular UTC series onto a fixed grid by parameter rule."""
    method = RESAMPLE_METHOD.get(value_col, "mean")
    indexed = frame.set_index("ts_utc").sort_index()
    grouper = indexed[value_col].resample(freq, label="left", closed="left")

    if method == "ffill":
        grid = grouper.last().ffill()
    else:
        grid = getattr(grouper, method)()

    counts = grouper.count().rename("sample_count")
    out = pd.concat([grid.rename(value_col), counts], axis=1)
    return out.reset_index().rename(columns={"ts_utc": "window_start"})

Phase 3 — Gap classification and bounded fill

Resampling produces empty buckets wherever telemetry was missing, and how those buckets are filled is the most consequential alignment decision. A short outage in a step-valued signal is safely forward-filled; a long void must be marked as a genuine gap so the compliance engine treats it as missing data rather than a real low reading. Fill is therefore bounded by an explicit tolerance, and anything beyond it is annotated, never fabricated.

Implementation steps:

1. Classify each empty bucket by run length: within tolerance it is eligible for interpolation or forward-fill, beyond tolerance it is a genuine gap.
2. Fill only within tolerance, tagging every filled bucket with the method used so a reviewer can see exactly which values are derived.
3. Route unrecoverable voids to a compliance exception queue rather than emitting a fabricated value into the reporting stream.

import pandas as pd


def classify_and_fill(
    grid: pd.DataFrame,
    value_col: str,
    max_fill_buckets: int = 2,
) -> pd.DataFrame:
    """Fill short gaps within tolerance; annotate longer gaps as genuine voids."""
    out = grid.copy()
    missing = out[value_col].isna()

    # Run length of each consecutive missing streak.
    streak_id = (~missing).cumsum()
    run_len = missing.groupby(streak_id).transform("sum")

    fillable = missing & (run_len <= max_fill_buckets)
    genuine_gap = missing & (run_len > max_fill_buckets)

    out.loc[fillable, "lineage_flag"] = "INTERPOLATED"
    out.loc[genuine_gap, "lineage_flag"] = "GAP"
    out.loc[~missing, "lineage_flag"] = "MEASURED"

    filled = out[value_col].interpolate(method="linear", limit=max_fill_buckets)
    out[value_col] = filled.where(~genuine_gap)  # never fill a genuine gap
    return out

Phase 4 — Lineage capture and audit archival

The final phase records how every aligned value was produced. Each row is stamped with its resample method, fill status, and a checksum, and the series is verified monotonic before it is written to write-once storage. This is what lets a utility reconstruct any historical report during a state or federal audit without manual reconciliation.

Implementation steps:

1. Verify the aligned series is strictly monotonic in UTC and halt the run for reconciliation if it is not.
2. Compute a per-row checksum over the value, window, and lineage flag so any later tampering is detectable.
3. Write raw, resampled, and flagged records to append-only storage for the statutory retention period — typically three to ten years depending on state primacy.

import hashlib

import pandas as pd


def stamp_lineage(frame: pd.DataFrame, value_col: str) -> pd.DataFrame:
    """Verify monotonicity and attach a tamper-evident checksum to each row."""
    if not frame["window_start"].is_monotonic_increasing:
        raise ValueError("Non-monotonic UTC grid; halting for reconciliation")

    out = frame.copy()

    def _checksum(row: pd.Series) -> str:
        payload = f"{row['window_start']}|{row[value_col]}|{row['lineage_flag']}"
        return hashlib.sha256(payload.encode()).hexdigest()[:16]

    out["row_checksum"] = out.apply(_checksum, axis=1)
    return out

Validation, Quality Flags & Edge Cases

Every aligned bucket carries a lineage flag, and downstream averaging engines consume that flag rather than re-deriving trust. Keeping the vocabulary consistent from the raw wire through the report is what lets a compliance calculation exclude a derived or missing value defensibly.

Several edge cases recur in field deployments and each needs a deterministic guard. DST boundaries are the quietest offender: archiving local wall-clock time duplicates a fall-back hour and erases a spring-forward hour, silently double-counting or dropping samples inside a compliance window — which is exactly why Phase 1 anchors to UTC before anything else touches the data. Partial windows at the start of a shift or after an outage must not be averaged as though complete; the sample_count column preserved in Phase 2 is what lets a running annual average built on three of four required quarters be recognized as not-yet-compliant rather than reported as final. Leap seconds and clock step-corrections can momentarily make a series non-monotonic; the Phase 4 monotonicity check halts rather than emitting a report on a scrambled timeline. Backfill collisions occur when a historian replays late-arriving data for a window already aligned; the immutable raw layer plus deterministic re-alignment means the window is simply recomputed, never patched in place. The dedicated worked example in Aligning Irregular SCADA Timestamps to UTC carries these guards through a complete chlorine-residual pipeline.

Deployment & Integration Patterns

Alignment runs as a small, stateless transform stage between the decoders and the compliance engines rather than as a monolith. Each instance consumes decoded records from a message broker (MQTT at the edge, or Kafka for the enterprise stream), applies the four phases, and publishes the aligned series so ingestion, alignment, and evaluation each scale independently.

Deployment guidance:

• Containerize with a read-only root filesystem. The stage needs no local writes beyond a bounded spool directory; enforcing a read-only filesystem shrinks the attack surface on an OT-adjacent host, consistent with the Security Boundary Design for the domain.
• Pin the rule table as versioned configuration. The per-parameter resample methods and fill tolerances are checked into version control and referenced by version in every lineage record, so an audit can reproduce the exact alignment logic that produced a historical report.
• Make every stage idempotent. Re-running alignment over the same raw window must yield byte-identical output; this lets a failed batch be retried safely and lets Async Batch Processing Setup fan out large historical re-alignments without risk of divergent results.
• Fail closed. If a monotonicity or tolerance check fails, the stage halts the affected window and raises an alert rather than publishing an unverified aggregate downstream to the Monitoring Gap Detection Algorithms.

Production Validation Checklist

• Every timestamp anchored to timezone-aware UTC at ingest, with DST fall-back and spring-forward transitions covered by tests
• Resample method for each parameter pinned in a versioned rule table, not hard-coded
• Fill tolerance (max_fill) documented per parameter in the facility Quality Assurance Project Plan (QAPP)
• Interpolated buckets flagged distinctly from measured buckets in every exported series
• Genuine gaps routed to the compliance exception queue, never forward-filled past tolerance
• Partial-window detection via preserved sample_count; incomplete averaging windows never reported as final
• Aligned series verified strictly monotonic in UTC before archival
• Raw, resampled, and flagged records written to append-only storage for the statutory retention period
• Row-level checksums generated for tamper-evident audit trails

Failure Modes & Gotchas

The single most consequential alignment failure is a gap too short to trigger an exception but long enough to skew the average it feeds. A 90-minute telemetry outage on a turbidity sensor falls within many utilities’ forward-fill tolerance, yet it can shift a four-hour average by several tenths of an NTU if the readings just before the outage were elevated. The value parses, passes range gates, and looks healthy on every dashboard — the distortion only surfaces when an auditor reconstructs the raw timeline. Because a plausible filled value carries no error of its own, tolerance thresholds alone cannot catch it. The guard is to log not just gap presence and duration but the statistical impact of each fill on the compliance quantity it feeds: record the pre-fill and post-fill window average alongside the INTERPOLATED flag, and surface any fill that moves the reported value by more than a configured fraction of the limit for manual verification before it reaches a regulatory submission. Re-run that check whenever the fill tolerance or resample method changes, because a widened tolerance can quietly convert yesterday’s GAP into today’s silently interpolated value.

Related

• SCADA Data Ingestion & Time-Series Sync — parent domain and shared ingestion contract
• Modbus TCP Parsing Workflows — decoded, quality-flagged values that feed alignment
• OPC UA Data Extraction — subscription-based acquisition and native status mapping
• Async Batch Processing Setup — non-blocking backfill and high-fan-out re-alignment
• Aligning Irregular SCADA Timestamps to UTC — worked end-to-end UTC normalization example