10 min read2,190 words

Risk-Scoring Utility Assets with Python

A maintenance schedule is only as defensible as the number it sorts on, and for a utility asset register that number is risk — the product of how likely an asset is to fail and how much its failure costs. Ranking by age alone sends crews to old-but-harmless pipe; ranking by customers-served alone ignores the cast-iron main that is one freeze away from a break. Doing the arithmetic by hand, or in a spreadsheet that nobody can reproduce six months later, does not survive a rate-case review or an ISO 55001 audit, where the question is always “show me exactly how this asset earned its priority.” A vectorized Python model over the GIS asset register answers that question the same way every time. This page implements the probability-times-consequence score at the core of condition-based maintenance scheduling, the entry discipline of Asset Lifecycle & Maintenance Automation. The consequence side depends on a clean criticality roll-up from asset hierarchy design for water and electric, and the probability side pairs naturally with the age-driven cycles in automating inspection-cycle generation from asset age.

Environment Prerequisites

A model that runs on malformed inputs produces a plausible ranking that is quietly wrong, which is worse than a crash. Lock the following before scoring:

  • Python runtime: Python 3.11 in an isolated environment created with conda create -n un-risk python=3.11, so numeric behavior stays reproducible across engineers.
  • Dependencies: geopandas>=1.0, pandas>=2.0, numpy>=1.26, and shapely>=2.0, installed via conda install -c conda-forge geopandas pandas numpy shapely. Vectorized column math over the register requires the pandas 2.x and numpy APIs used below.
  • Source register: An asset layer readable by geopandas.read_file (GeoPackage, Shapefile, or an OGR-accessible enterprise connection), carrying at minimum asset_id, install_date, material, criticality, and customers_served. Export from a reconciled, unversioned snapshot so the attributes reflect committed edits.
  • Service-life lookup: A material-to-expected-life table, versioned alongside the code so a change to an assumption is a reviewable commit rather than a silent edit.
  • Criticality provenance: Confirm criticality and customers_served were rolled up from a validated containment model; a consequence weight computed over a fragmented network under-counts and mis-ranks. Enforce topology validation upstream.

Schema-Aware Validation Protocol — Run Before the Score

Most risk-scoring errors originate in the inputs, not the formula. Work this ordered checklist first; the earliest item is the most frequent culprit.

  1. Confirm every scoring column is present and typed. A missing material or a customers_served read as text produces silent coercion to NaN, which propagates through the product and drops assets out of the ranking entirely. Assert the schema before any arithmetic.
  2. Detect defaulted install dates. Import sentinels — a pile-up on 1900-01-01 or the load date — masquerade as real ages. Flag suspiciously dense dates and route them to a documented estimation rule rather than scoring them as literal.
  3. Validate the material domain. Any material outside the service-life lookup must fail loudly, not fall to a default life, because a wrong life rescales the whole probability curve for that class.
  4. Bound the consequence inputs. Negative or zero customers_served, or a criticality class outside the permitted set, indicate a broken roll-up. Clamp and log rather than letting an outlier dominate the top of the ranking.
  5. Decide the inspection-grade policy explicitly. An expired inspection grade is not current condition. Choose — and document — whether a stale grade is ignored (age dominates) or retained, because the choice moves assets between bands.

Minimal Reproducible Implementation

The following model reads the register, validates the scoring attributes, derives a 0-to-1 probability of failure from age against material-specific service life, derives a consequence weight from criticality and customers served, multiplies them into a bounded risk score, and returns the register sorted descending with structured run metadata. It handles missing columns, out-of-domain materials, and future-dated installs as explicit exceptions rather than silent NaN.

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date
import logging

import numpy as np
import pandas as pd
import geopandas as gpd

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

# Expected service life (years) and consequence weight per criticality class.
SERVICE_LIFE = {"DI": 100, "CI": 75, "PVC": 80, "PE": 60, "AC": 50, "STEEL": 65}
CRITICALITY_WEIGHT = {"low": 1.0, "medium": 2.5, "high": 5.0, "critical": 10.0}
REQUIRED_COLS = ("asset_id", "install_date", "material", "criticality", "customers_served")


@dataclass
class RiskResult:
    """Structured output of a scoring run."""
    scored: gpd.GeoDataFrame
    run_date: date
    asset_count: int
    dropped: int
    top_asset: str | None = field(default=None)


def _probability_of_failure(ages: pd.Series, materials: pd.Series) -> pd.Series:
    """Vectorized 0..1 probability of failure from age vs. material service life."""
    lives = materials.map(SERVICE_LIFE)
    if lives.isna().any():
        bad = sorted(materials[lives.isna()].unique())
        raise KeyError(f"unknown material class(es): {bad}")
    # Fraction of service life consumed, clamped to [0, 1]; failure risk rises
    # non-linearly near end of life, so square the consumed fraction.
    consumed = np.clip(ages / lives, 0.0, 1.0)
    return np.round(consumed ** 2, 4)


def _consequence(criticality: pd.Series, customers: pd.Series) -> pd.Series:
    """Consequence weight from criticality class and customers served."""
    weights = criticality.str.lower().map(CRITICALITY_WEIGHT)
    if weights.isna().any():
        bad = sorted(criticality[weights.isna()].unique())
        raise KeyError(f"unknown criticality class(es): {bad}")
    served = customers.clip(lower=0).fillna(0)
    # Log-scale the customer count so a 10x span does not swamp criticality.
    return np.round(weights * (1.0 + np.log1p(served)), 4)


def score_register(register: gpd.GeoDataFrame, as_of: date) -> RiskResult:
    """Compute a probability x consequence risk score for every asset.

    Validates schema and domains, drops rows with unrecoverable inputs,
    and returns a GeoDataFrame sorted by descending risk plus run metadata.
    """
    missing = [c for c in REQUIRED_COLS if c not in register.columns]
    if missing:
        raise KeyError(f"register is missing required columns: {missing}")

    df = register.copy()
    df["install_date"] = pd.to_datetime(df["install_date"], errors="coerce")
    before = len(df)
    df = df[df["install_date"].notna()]  # drop unusable installs, count them
    dropped = before - len(df)
    if dropped:
        logging.warning("dropped %d assets with unparseable install_date", dropped)

    ages = (pd.Timestamp(as_of) - df["install_date"]).dt.days / 365.25
    if (ages < 0).any():
        raise ValueError("one or more install_date values are in the future")

    df["prob_failure"] = _probability_of_failure(ages, df["material"])
    df["consequence"] = _consequence(df["criticality"], df["customers_served"])
    df["risk_score"] = np.round(df["prob_failure"] * df["consequence"] * 100.0, 2)

    df = df.sort_values("risk_score", ascending=False).reset_index(drop=True)
    top = str(df.loc[0, "asset_id"]) if len(df) else None
    logging.info("scored %d assets; highest risk asset=%s", len(df), top)

    return RiskResult(scored=df, run_date=as_of, asset_count=len(df),
                      dropped=dropped, top_asset=top)


if __name__ == "__main__":
    demo = gpd.GeoDataFrame({
        "asset_id": ["MN-101", "MN-102", "LT-9", "SW-4"],
        "install_date": ["1961-04-01", "2018-09-12", "1999-01-01", "1975-06-01"],
        "material": ["CI", "PE", "PVC", "STEEL"],
        "criticality": ["critical", "low", "medium", "high"],
        "customers_served": [4200, 12, 340, 1100],
        "geometry": [None, None, None, None],
    })
    result = score_register(demo, as_of=date(2026, 7, 13))
    print(result.scored[["asset_id", "prob_failure", "consequence", "risk_score"]]
          .to_string(index=False))
    print(f"top risk: {result.top_asset} | dropped: {result.dropped}")

The score is the product of two independent axes, and reading it as a matrix makes the ranking legible to planners who will never open the code: an asset is urgent only when it is both likely to fail and costly to lose.

Risk as probability of failure times consequence, on a three-by-three matrix A three-by-three grid with probability of failure increasing up the vertical axis and consequence increasing along the horizontal axis. The lower-left cells, where both are low, are shaded as monitor. The diagonal middle cells are shaded as scheduled. The upper-right cells, where both probability and consequence are high, are shaded as immediate. The label emphasizes that risk is the product of the two axes, so only the high-high corner is urgent. Probability of failure ↑ Consequence → Scheduled Immediate Immediate Monitor Scheduled Immediate Monitor Monitor Scheduled low medium high high med low

Two modeling choices in the code are deliberate and worth defending in a review. Squaring the consumed-life fraction makes probability rise steeply near end of life rather than linearly, which matches observed break-rate curves for buried pipe. Log-scaling the customer count stops a single large-consumer feeder from swamping the ranking, so criticality class still discriminates among high-count assets. Both are policy inputs — swap the exponent or the scaling and the ranking shifts — which is exactly why they live in named, testable functions rather than buried in one expression.

Production Deployment Pattern

A notebook that scores once is a diagnostic; an enforced, scheduled score is an engineering control. Promote it as follows:

  1. Score against a reconciled snapshot. Read the register from a read-only replica or the DEFAULT version with no active edit session, so the score reflects committed geometry and attributes and never contends with editor locks — the same discipline as any data ingestion pipeline for utility assets.
  2. Pin every assumption in version control. The service-life table, criticality weights, and the probability exponent are commits, not spreadsheet cells, so any change to the ranking basis is reviewable and dated.
  3. Run on a schedule and gate on drift. Invoke score_register nightly from CI. Compare the new top-decile set against the prior run; a large unexplained churn is the signal that an upstream attribute broke, and it should open a ticket rather than silently re-prioritize crews.
  4. Write scores back with backoff. Push asset_id → risk_score, band into the enterprise asset register over its REST endpoint, wrapping the write in a bounded retry so a transient lock produces a retry, not a failed run.
  5. Persist the run metadata. Append the run date, asset count, dropped-row count, model versions, and the top-risk asset to a timestamped, version-controlled log — the chain of custody an ISO 55001 or AWWA G400 review expects.

Conclusion

Risk scoring turns a register full of attributes into a defensible priority order by doing one honest calculation: probability of failure, from age against material service life, times consequence, from criticality and customers served. Implemented as a vectorized GeoDataFrame pass with explicit validation and structured output, it runs the same way every night, drops bad inputs loudly instead of scoring them wrong, and leaves an audit trail. The natural next step is to feed the ranked score into inspection-cycle generation so the highest-risk assets are not only flagged but dated, and to close the loop by writing each completed inspection grade back so tomorrow’s score reflects observed condition.

Related pages