Scoring Violations by Population and Exposure Duration

Two drinking-water violations that breach the identical contaminant threshold are rarely equal in consequence: one that persists for a single afternoon across a hamlet of 40 connections carries far less public-health weight than the same exceedance sustained for a full quarter across a metropolitan service area of two million people. This page sits inside the parent Severity Scoring Models section and covers exactly one half of that calculus — how to fold population served and exposure duration into a defensible, reproducible severity number. The other half, the contaminant-specific health penalty, is developed separately in weighting contaminant health risk in severity scores; here that penalty enters only as a fixed input coefficient so the two components can be composed without double-counting. The score this page produces is what ranks a queue of concurrent events, drives the tiering logic behind public notification workflows, and lets a small compliance team triage the events that matter first.

Prerequisites & Environment Setup

The implementation runs on Python 3.10+ and leans on pydantic for validated event models and pandas for scoring a batch of open violations at once. No SCADA connectivity is needed at this layer — the scorer consumes already-detected violation records emitted upstream by the rule engine, so it is pure, deterministic, and trivially unit-testable. Population figures come from your system inventory (the SDWIS-style facility record that pins population served to each public water system), and the exposure window comes from the detected violation’s start and end timestamps.

python3 -m venv .venv && source .venv/bin/activate
pip install "pydantic==2.7.*" "pandas==2.2.*"
# Optional: only needed if you persist scored events
pip install "sqlalchemy==2.0.*"

One input deserves a caution before any code is written. The exposure duration is only as trustworthy as the timestamps bounding it, and an undetected monitoring lapse can make a chronic exposure look brief. Reconcile the window against the output of the monitoring gap detection algorithms before scoring, so that a silent sensor outage is not misread as a period of compliance.

Step-by-Step Implementation

Step 1 — Model the scoring inputs

Every score derives from three quantities: the population served by the affected system, the duration of the exceedance window, and the health-risk coefficient supplied by the sibling model. Capture them in a validated model so that a negative population or an inverted time window fails loudly at construction rather than silently skewing a ranking.

from datetime import datetime, timezone
from pydantic import BaseModel, Field, field_validator, model_validator


class ViolationEvent(BaseModel):
    event_id: str
    system_id: str
    population_served: int = Field(ge=0)
    exposure_start: datetime
    exposure_end: datetime
    # Health-risk weight w_h in [0, 1] comes from the sibling health-risk model;
    # this page treats it strictly as a fixed input coefficient.
    health_weight: float = Field(ge=0.0, le=1.0)
    contaminant_class: str = Field(pattern="^(acute|chronic)$")

    @field_validator("exposure_start", "exposure_end")
    @classmethod
    def must_be_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("timestamps must be timezone-aware UTC")
        return v.astimezone(timezone.utc)

    @model_validator(mode="after")
    def window_must_be_ordered(self) -> "ViolationEvent":
        if self.exposure_end < self.exposure_start:
            raise ValueError("exposure_end precedes exposure_start")
        return self

    @property
    def duration_days(self) -> float:
        seconds = (self.exposure_end - self.exposure_start).total_seconds()
        return seconds / 86_400.0

Step 2 — Log-scale the population weight

Raw population served spans five orders of magnitude across regulated systems — from a 25-person transient non-community well to a city serving several million. If population entered the score linearly, a single large utility would dwarf every rural system on the board, and the ranking would degenerate into a proxy for system size. Compressing population with a base-10 logarithm keeps the ordering intact while collapsing that dynamic range, so a mega-utility outranks a hamlet by a factor of roughly four or five rather than by five full orders of magnitude. Normalizing against a reference population maps the weight into a stable $[0, 1]$ band:

Here $P$ is the population served and $P_\text{ref}$ is a fixed ceiling (the largest system you intend to score, or a policy constant). The $1 + P$ term keeps the logarithm finite when $P = 0$, which matters for a newly provisioned system whose population record has not yet been populated.

import math

REFERENCE_POPULATION = 3_500_000  # policy ceiling for normalization


def population_weight(population: int, reference: int = REFERENCE_POPULATION) -> float:
    """Log-compressed population weight normalized to [0, 1]."""
    if population < 0:
        raise ValueError("population_served must be non-negative")
    if reference <= 0:
        raise ValueError("reference population must be positive")
    weight = math.log10(1 + population) / math.log10(1 + reference)
    return min(weight, 1.0)  # clamp systems larger than the reference ceiling

A quick sanity read of the curve: 500 people yield about 0.41, 50,000 yield about 0.72, and 3,500,000 yield 1.0. The small system is not erased — it retains roughly 40% of the maximum population weight — which is precisely the behavior that keeps rural exceedances visible in a shared triage queue.

Step 3 — Derive the duration multiplier from the exceedance window

Duration enters as a multiplier rather than an additive term, because time amplifies whatever health and population weight the event already carries. The multiplier starts at 1.0 for an instantaneous exceedance and rises toward a bounded ceiling as the window lengthens, following a saturating (Hill-type) curve so that the marginal severity of each additional day shrinks once the exposure is clearly chronic:

$d$ is the exposure duration in days, $\mu$ is the multiplier ceiling, and $d_{50}$ is the half-saturation duration — the number of days at which the multiplier reaches the midpoint between 1 and $\mu$. Acute contaminants (nitrate, nitrite, E. coli), where a short exposure is already dangerous, use a small $d_{50}$ so the curve climbs fast; chronic contaminants (arsenic, total trihalomethanes), where harm accrues over quarters, use a larger $d_{50}$ that reflects the running-average nature of their standards.

# Contaminant-class duration profiles: (half-saturation days, ceiling mu)
DURATION_PROFILES = {
    "acute": (1.5, 3.0),    # climbs steeply; days matter
    "chronic": (45.0, 2.0),  # accrues over the averaging window
}


def duration_multiplier(duration_days: float, contaminant_class: str) -> float:
    """Saturating multiplier in [1.0, mu] driven by the exceedance window."""
    if duration_days < 0:
        raise ValueError("duration_days must be non-negative")
    try:
        d50, mu = DURATION_PROFILES[contaminant_class]
    except KeyError as exc:
        raise ValueError(f"unknown contaminant_class: {contaminant_class}") from exc
    return 1.0 + (mu - 1.0) * (duration_days / (duration_days + d50))

Step 4 — Compose the severity score

The composite score is the product of the health coefficient, the population weight, and the duration multiplier, scaled onto a 0–100 range for human-readable triage. Multiplication is deliberate: a violation is severe only when the contaminant is harmful and the affected population is meaningful and the exposure has persisted. Any near-zero factor rightly pulls the score down, which is the behavior a compliance officer expects when, say, a trace contaminant briefly clips its limit in a tiny system.

The middle and right factors are what this page owns; $w_h$ is the health penalty carried in from the sibling model. Because the duration multiplier can exceed 1, the raw product may push past 100 for a prolonged exposure in a large system, so the final value is clamped.

def composite_severity(event: "ViolationEvent") -> float:
    """Combine health weight, log-scaled population, and duration into 0-100."""
    p_w = population_weight(event.population_served)
    d_m = duration_multiplier(event.duration_days, event.contaminant_class)
    raw = 100.0 * event.health_weight * p_w * d_m
    return round(min(raw, 100.0), 2)

Scoring an open caseload is then a single vectorized pass, which is how the ranking reaches an operator’s dashboard:

import pandas as pd


def score_caseload(events: list["ViolationEvent"]) -> pd.DataFrame:
    rows = [
        {
            "event_id": e.event_id,
            "system_id": e.system_id,
            "population_served": e.population_served,
            "duration_days": round(e.duration_days, 3),
            "severity": composite_severity(e),
        }
        for e in events
    ]
    return pd.DataFrame(rows).sort_values("severity", ascending=False, ignore_index=True)

Configuration Reference

Treat every constant below as versioned policy, reviewed with your primacy agency, not as an inline literal. Changing REFERENCE_POPULATION or a duration profile reshapes every historical score, so pin them and record the version alongside each stored result.

Scoring parameters

Duration profiles by contaminant class

Worked population weights (with $P_\text{ref} = 3{,}500{,}000$)

Verification & Testing

The scorer is deterministic, so its properties can be asserted exactly. The tests below pin the monotonic behaviors that make the ranking trustworthy: longer exposure never lowers a score, larger population never lowers it, and the log compression keeps a giant system within a small multiple of a tiny one.

import math
from datetime import datetime, timedelta, timezone


def _event(pop, days, cls="chronic", hw=0.8):
    start = datetime(2026, 1, 1, tzinfo=timezone.utc)
    return ViolationEvent(
        event_id="e", system_id="s", population_served=pop,
        exposure_start=start, exposure_end=start + timedelta(days=days),
        health_weight=hw, contaminant_class=cls,
    )


def test_population_weight_is_log_compressed():
    small = population_weight(25)
    large = population_weight(3_500_000)
    assert large == 1.0
    # 5 orders of magnitude in population compress to well under 5x in weight
    assert large / small < 5.0


def test_duration_multiplier_saturates_below_ceiling():
    assert duration_multiplier(0.0, "chronic") == 1.0
    assert duration_multiplier(10_000, "chronic") < 2.0  # never reaches mu


def test_score_is_monotonic_in_duration_and_population():
    base = composite_severity(_event(50_000, 5))
    longer = composite_severity(_event(50_000, 40))
    bigger = composite_severity(_event(500_000, 5))
    assert longer > base
    assert bigger > base


def test_score_is_clamped_to_100():
    assert composite_severity(_event(3_500_000, 3650, cls="acute", hw=1.0)) <= 100.0

Acceptance criteria before wiring the score into triage:

• REFERENCE_POPULATION and both duration profiles are pinned as reviewed policy and versioned with each stored score.
• Population weight is log-compressed: a 3.5M system scores under 5x a 25-person system on the population factor alone.
• Duration multiplier equals exactly 1.0 at zero days and never reaches $\mu$ for any finite window.
• Composite score is monotonically non-decreasing in both population and duration.
• Final score is clamped to the 0–100 range and rounded consistently.
• Exposure windows are reconciled against detected monitoring gaps before scoring.
• Inverted or timezone-naive windows raise at model construction, never score silently.

Troubleshooting & Gotchas

• A large utility dominates the entire queue. If big systems still crowd out every rural exceedance, population is probably entering somewhere as a linear term, or REFERENCE_POPULATION is set well above your true maximum, stretching the denominator. Confirm the weight is the log ratio in Step 2 and set $P_\text{ref}$ to the largest system you actually score.
• Chronic violations score lower than expected. The saturating multiplier is doing its job — a chronic profile with $\mu = 2.0$ can at most double the base. If chronic events genuinely need to outrank acute ones, that priority belongs in the health coefficient $w_h$ from the sibling model, not in the duration curve; resist the urge to inflate $\mu$ to compensate.
• Duration looks far too short for a known long-running problem. The exposure window is bounded by the timestamps you passed, and an undetected reporting lapse truncates it. Cross-check the window against the monitoring gap detection algorithms so a silent outage is not scored as compliant time.
• Scores drift after a routine data refresh. A revised population figure in the system inventory silently rescores every historical event. Snapshot population_served and the parameter version onto each stored score at the moment of computation so past rankings remain reproducible and auditable.
• Ties at the top of the queue. When several events share an identical score, the multiplicative form has landed them on the same contour. Break ties deterministically on raw population, then on duration, so the ordering handed to the public notification workflows is stable across runs.

Related

• Severity Scoring Models — the parent section and the overall scoring contract
• Weighting Contaminant Health Risk in Severity Scores — the sibling model that supplies the $w_h$ coefficient used here
• Public Notification Workflows for SDWA Violations — where the ranked score selects a notification tier
• Violation Detection & Rule Engine Logic — the parent architecture this scorer plugs into