Surcharge & Fee Application Logic in Municipal Utility Billing

Surcharges and fees are the part of a municipal bill most prone to silent, compounding error. Unlike the base commodity charge, they are dynamic, policy-driven constructs — drought surcharges, franchise fees, storm-water assessments, state utility taxes, and low-income riders — that must be applied deterministically across millions of service accounts. This guide sits under Automated Rate Calculation & Rule Engines and picks up exactly where base pricing ends: after usage has been rated, but before the invoice total is finalized and posted to the ledger.

The core failure this workflow addresses is fee drift — the slow divergence between what a bill charges and what the governing ordinance actually mandates. Drift creeps in when a rule is edited in place instead of versioned, when a seasonal window is expressed as a hardcoded date, when floating-point rounding accumulates across a batch, or when a mid-cycle Public Utilities Commission (PUC) order is applied without point-in-time resolution. The remedy is an engineering discipline built on four pillars: strict schema validation at ingestion, timezone-aware temporal resolution, memory-bounded deterministic evaluation, and an immutable audit trail. When these are combined, billing cycles become predictable, compliance becomes verifiable, and financial drift is caught before it reaches a ratepayer.

# Prerequisites

This workflow targets a modern, reproducible Python stack. Pin the following before implementing anything below:

• Python 3.11+ — required for zoneinfo in the standard library and datetime | None union syntax.
• Pydantic v2 (pydantic>=2.6) — all rule and payload contracts use v2 field_validator / model_validator semantics; v1 will not validate these models.
• decimal.Decimal for all money — never float. Monetary arithmetic uses Decimal with an explicit ROUND_HALF_UP quantization at the charge boundary.
• zoneinfo (not pytz) — seasonal windows and effective dates are resolved in the utility’s local civil time, so every boundary comparison must be timezone-aware.

Data assumptions: base commodity charges have already been computed by the rate engine and are available per account as base_usage_cost (a Decimal); each account carries a service_date, a service_address (for jurisdiction resolution), and a validated customer_class. Surcharge definitions arrive from an authoritative rule repository — the same source of truth used by schema validation & data quality checks upstream. Write access to the rule repository must be governed; see Security Boundaries & Role-Based Access for the dual-authorization pattern that gates fee-schedule edits.

# Architecture Overview

Surcharge application is a staged pipeline, not a single function. Each stage has an explicit exception path so a malformed rule or a null read is diverted rather than allowed to corrupt a batch. The base-rated account enters from the rate engine, the active fee set is resolved for the service date and jurisdiction, each fee is evaluated with decimal-safe math, and every application emits an audit record before the decorated account is handed to reconciliation.

The remaining sections implement each stage in order.

# Step-by-Step Implementation

# Step 1 — Enforce a strict surcharge rule schema at ingestion

Municipal billing data arrives from disparate upstream systems: SCADA meter reads, GIS service boundaries, customer information systems (CIS), and regulatory fee schedules. Without strict typing, a malformed fee definition cascades into incorrect statements, audit failures, and customer disputes. Pydantic v2 enforces structural integrity before any calculation occurs. A surcharge rule must capture the fee identifier, calculation method, rate value, effective-date range, jurisdictional scope, and the PUC reference that authorizes it.

from pydantic import BaseModel, Field, field_validator, model_validator
from datetime import date
from decimal import Decimal
from enum import Enum

class CalculationMethod(str, Enum):
    FLAT = "flat"
    PERCENTAGE = "percentage"
    TIERED = "tiered"

class TierBand(BaseModel):
    # Upper bound is exclusive; the final band uses None to mean "no ceiling".
    upper_bound: Decimal | None = Field(default=None, ge=0)
    rate_value: Decimal = Field(..., ge=0)

class SurchargeRule(BaseModel):
    fee_id: str = Field(..., pattern=r"^FEE-\d{4}$")
    puc_reference: str = Field(..., min_length=5)
    jurisdiction_code: str = Field(..., max_length=4)
    calculation_method: CalculationMethod
    rate_value: Decimal = Field(default=Decimal("0"), ge=0)
    tier_bands: list[TierBand] = Field(default_factory=list)
    effective_start: date
    effective_end: date | None = None
    seasonal_start: tuple[int, int] | None = None  # (month, day), inclusive
    seasonal_end: tuple[int, int] | None = None     # (month, day), inclusive

    @field_validator("effective_end")
    @classmethod
    def validate_date_range(cls, v: date | None, info) -> date | None:
        start = info.data.get("effective_start")
        if v and start and v < start:
            raise ValueError("effective_end must be on or after effective_start")
        return v

    @model_validator(mode="after")
    def validate_method_payload(self) -> "SurchargeRule":
        if self.calculation_method is CalculationMethod.TIERED and not self.tier_bands:
            raise ValueError("tiered rules require at least one tier band")
        if self.calculation_method is not CalculationMethod.TIERED and self.rate_value < 0:
            raise ValueError("rate_value must be non-negative")
        return self

By validating incoming payloads against this contract, non-conforming records are rejected at the ingestion layer instead of corrupting the downstream engine. This validation-first posture is the same discipline that governs schema validation & data quality checks on the meter side — clean, typed input is the precondition for deterministic evaluation.

# Step 2 — Resolve the active fee window (effective dates + seasonal boundaries)

Once validated, each surcharge must be mapped to precise temporal boundaries. Many municipal fees are calendar-dependent: drought surcharges activate during summer, winter heating-assistance riders apply only between November and March, and storm-water assessments may scale with quarterly precipitation windows. Rather than hardcoding these ranges per fee, the rule carries its own seasonal bounds and the resolver evaluates them against a timezone-aware service date. This is the fee-side counterpart to the broader Seasonal Rate Mapping & Calendar Logic that governs base pricing.

from datetime import date, datetime
from zoneinfo import ZoneInfo

UTILITY_TZ = ZoneInfo("America/Chicago")

def _in_seasonal_window(service_date: date, rule: SurchargeRule) -> bool:
    """Handle both normal (May-Sep) and year-wrapping (Nov-Mar) seasons."""
    if rule.seasonal_start is None or rule.seasonal_end is None:
        return True  # no seasonal restriction — always in-window
    (sm, sd), (em, ed) = rule.seasonal_start, rule.seasonal_end
    start = date(service_date.year, sm, sd)
    end = date(service_date.year, em, ed)
    if start <= end:                      # same-year window, e.g. May 1 – Sep 30
        return start <= service_date <= end
    # Year-wrapping window, e.g. Nov 1 – Mar 31
    return service_date >= start or service_date <= end

def resolve_active_fee_window(read_dt: datetime, rule: SurchargeRule) -> bool:
    """True when the rule applies to a read at read_dt in local civil time."""
    local_date = read_dt.astimezone(UTILITY_TZ).date()
    if not (rule.effective_start <= local_date <= (rule.effective_end or date.max)):
        return False
    return _in_seasonal_window(local_date, rule)

Resolving the read in the utility’s local time zone before extracting the calendar date is what keeps a read taken at 11:40 PM on the last day of a season from silently rolling into the next day (and out of the window) under UTC. When a billing cycle closes, the system resolves the active fee set by cross-referencing each read against the municipal calendar so that seasonal transitions produce neither overlapping nor missing charges.

# Step 3 — Evaluate each fee with decimal-safe arithmetic

With the active set known, each fee is computed by method. Flat fees are constant; percentage fees scale the base commodity charge; tiered fees allocate the charge base across bands. Every result is quantized once, at the charge boundary, with an explicit rounding mode — this is the single most important defense against accumulated rounding variance.

from decimal import Decimal, ROUND_HALF_UP

CENTS = Decimal("0.01")

def _money(value: Decimal) -> Decimal:
    return value.quantize(CENTS, rounding=ROUND_HALF_UP)

def compute_charge(rule: SurchargeRule, charge_base: Decimal) -> Decimal:
    """charge_base is the amount the fee applies to (usage cost or usage units)."""
    if rule.calculation_method is CalculationMethod.FLAT:
        return _money(rule.rate_value)

    if rule.calculation_method is CalculationMethod.PERCENTAGE:
        return _money(charge_base * (rule.rate_value / Decimal("100")))

    # TIERED: apply each band's rate only to the slice of base within that band.
    total = Decimal("0")
    lower = Decimal("0")
    for band in rule.tier_bands:
        ceiling = band.upper_bound if band.upper_bound is not None else charge_base
        slice_amount = max(Decimal("0"), min(charge_base, ceiling) - lower)
        total += slice_amount * band.rate_value
        lower = ceiling
        if band.upper_bound is not None and charge_base <= band.upper_bound:
            break
    return _money(total)

Tiered surcharges must slice the charge base across bands rather than dumping the whole amount into the top or bottom band — the latter distorts conservation incentives and can violate PUC guidelines. The band model here mirrors the marginal-slice approach documented in Step-Rate vs Block-Rate Structure Design, applied to fees instead of the commodity rate.

# Step 4 — Apply fees across a memory-bounded batch

Municipal billing cycles routinely process millions of accounts. Loading an entire dataset into memory triggers garbage-collection overhead and out-of-memory failures. Streaming the source in bounded chunks keeps memory flat regardless of dataset size — the same principle behind async batch processing for high-volume reads.

from typing import Iterator, Generator

def stream_account_chunks(
    source: Iterator[dict], chunk_size: int = 5000
) -> Generator[list[dict], None, None]:
    """Yield bounded batches to prevent memory saturation."""
    batch: list[dict] = []
    for account in source:
        batch.append(account)
        if len(batch) >= chunk_size:
            yield batch
            batch = []          # rebind, never .clear() an already-yielded list
    if batch:
        yield batch

def apply_surcharges_batch(
    accounts: list[dict], active_rules: list[SurchargeRule]
) -> list[dict]:
    """Deterministic fee application across one memory-bounded batch."""
    for acct in accounts:
        applied: list[dict] = []
        base_cost: Decimal = acct["base_usage_cost"]
        for rule in active_rules:
            if rule.jurisdiction_code != acct["jurisdiction_code"]:
                continue
            if not resolve_active_fee_window(acct["read_dt"], rule):
                continue
            charge = compute_charge(rule, base_cost)
            applied.append({"fee_id": rule.fee_id, "amount": charge})
        acct["applied_surcharges"] = applied
        acct["surcharge_total"] = sum((f["amount"] for f in applied), Decimal("0"))
    return accounts

Note the subtle correctness fix in stream_account_chunks: after a batch is yielded, the buffer is rebound to a fresh list rather than cleared in place. Calling .clear() on a list that a consumer is still iterating would mutate the data the caller is holding. Decoupling ingestion from evaluation this way gives linear memory scaling and stable throughput during month-end close, when SLA pressure is highest.

# Step 5 — Emit an immutable audit record for every application

Regulatory precision demands verifiable proof of how each charge was derived, not just an accurate total. Every fee application generates a tamper-evident record capturing the input state, the rule version, the computed charge, and a chained hash that links each record to the one before it.

import hashlib
from datetime import datetime, timezone

def audit_record(
    account_id: str, rule: SurchargeRule, charge: Decimal, prev_hash: str
) -> dict:
    ts = datetime.now(timezone.utc).isoformat()
    payload = f"{prev_hash}|{account_id}|{rule.fee_id}|{rule.puc_reference}|{rule.rate_value}|{charge}|{ts}"
    digest = hashlib.sha256(payload.encode()).hexdigest()
    return {
        "account_id": account_id,
        "fee_id": rule.fee_id,
        "rule_version": rule.puc_reference,
        "charge_applied": str(charge),
        "processed_at": ts,
        "prev_hash": prev_hash,
        "integrity_hash": digest,
    }

Chaining prev_hash into each digest makes the log an append-only structure: altering any historical record invalidates every hash after it, which is what an external auditor verifies. PUC mandates typically require retaining these trails for a minimum of seven years. Audit payloads must carry operational metadata only — account IDs, fee codes, timestamps — with personally identifiable information stripped or hashed per Data Governance & Privacy Compliance.

# Edge-Case Handling

Municipal fee data is full of boundary conditions that a naive implementation gets wrong. The scenarios below are the ones that most often produce silent variance:

• Mid-cycle PUC rate change. A commission order that takes effect on the 15th must not retroactively reprice the whole cycle. Because rules are versioned by effective_start and never edited in place, the resolver returns the version in force on each read’s service date; consumption straddling the boundary is split and each segment priced under its own version.
• Year-wrapping seasonal windows. A winter rider running Nov 1 – Mar 31 fails any start <= date <= end test because the start month is numerically greater than the end month. _in_seasonal_window handles the wrap explicitly (date >= start or date <= end).
• DST boundary reads. A read timestamped during the “fall-back” hour is ambiguous in local time. Resolving through zoneinfo with an explicit tz and comparing on the extracted local date (not datetime) keeps a late-night read on the correct billing day.
• Leap day. A fee whose seasonal window includes Feb 29 must not raise in non-leap years. Because windows are expressed as (month, day) and reconstructed against the read’s own year, Feb 28 vs 29 resolves naturally without a hardcoded calendar.
• Null or missing base cost. A percentage fee cannot be computed when the base commodity charge is absent (an unresolved rate or a null AMI read). Rather than defaulting to zero — which silently under-bills — the account is diverted to Fallback Routing for Missing Rate Data for provisional handling and a true-up flag.
• Jurisdiction mismatch. Fees are authorized per jurisdiction. Matching jurisdiction_code from the service address — not from estimated consumption — prevents a franchise fee from one municipality being applied to a neighboring one, a common source of cross-boundary billing errors.

# Verification and Audit Trail

Correctness is confirmed two ways: deterministic replay and reconciliation against a regulatory baseline. Replay proves the engine returns the same output for the same input; reconciliation proves the aggregate matches what the ordinance predicts.

from decimal import Decimal

def verify_batch(accounts: list[dict], tolerance: Decimal = Decimal("0.0005")) -> None:
    # 1. Deterministic replay: recomputing must reproduce the stored total.
    for acct in accounts:
        recomputed = sum((f["amount"] for f in acct["applied_surcharges"]), Decimal("0"))
        assert recomputed == acct["surcharge_total"], acct["account_id"]

    # 2. Drift detection against the expected regulatory baseline.
    billed = sum((a["surcharge_total"] for a in accounts), Decimal("0"))
    baseline = sum((a["expected_surcharge_baseline"] for a in accounts), Decimal("0"))
    if baseline > 0:
        variance = abs(billed - baseline) / baseline
        assert variance <= tolerance, f"drift {variance:.4%} exceeds tolerance"

Run the same suite against a fixture of historical bills whose totals are already reconciled — this catches regressions introduced by a rule or code change before they reach production. When variance exceeds tolerance (commonly ±0.05%), the pipeline should halt downstream posting, isolate the affected batches, and trigger a rule-version rollback rather than posting suspect charges. Transient failures encountered while writing audit records or posting to the ledger should be retried through the patterns in Error Handling & Retry Workflows so a single flaky write does not abort an otherwise-clean cycle.

# Troubleshooting

# Frequently Asked Questions

# Why must surcharge math use decimal.Decimal instead of float?

Binary floating point cannot represent most decimal fractions exactly, so a percentage fee like 3.5% of a base charge introduces a tiny error on every line. Across millions of accounts those errors accumulate into unexplained variance that reconciliation cannot close. decimal.Decimal gives exact decimal arithmetic, and quantizing once at the charge boundary with an explicit rounding mode makes every cent reproducible and auditable.

# How do I apply a fee that changes partway through a billing cycle?

Version the rule by its effective_start date and never edit it in place. When a cycle spans the change, split the consumption at the boundary and price each segment under the rule version in force for that segment. Point-in-time resolution reconstructs the exact fee context as of each service date, so a bill can be reproduced or corrected later without ambiguity.

# What happens when the base commodity charge is missing?

A percentage-based surcharge has nothing to scale, so the engine must not default it to zero — that silently under-bills. The account is diverted to fallback routing, which produces a provisional charge tagged for true-up once the real read or rate resolves, keeping the cycle on schedule without posting a wrong figure.

# How is a year-wrapping seasonal fee handled correctly?

A window like November 1 through March 31 fails a naive start <= date <= end test because the start month is numerically larger than the end month. The resolver detects the wrap and instead checks whether the date is on or after the start or on or before the end, which correctly spans the year boundary.

# How do I prove to an auditor that a historical charge was not altered?

Each audit record chains the previous record’s hash into its own SHA-256 digest, forming an append-only log. Changing any past record changes its hash and therefore every hash after it, so recomputing the chain and comparing to the stored digests immediately reveals tampering. Combined with seven-year retention, this satisfies typical PUC audit requirements.

# Related Topics

• Automated Rate Calculation & Rule Engines — the parent guide; base pricing that this fee layer decorates.
• Seasonal Rate Mapping & Calendar Logic — timezone-aware windows and effective-date proration for base rates.
• Step-Rate vs Block-Rate Structure Design — the marginal-slice tier model reused here for tiered fees.
• Fallback Routing for Missing Rate Data — provisional handling when a fee’s base cost is absent.
• Error Handling & Retry Workflows — resilient posting of audit records and ledger entries.

Up: Automated Rate Calculation & Rule Engines