Assistance Program Eligibility Taxonomy in Municipal Utility Billing

Assistance programs are where billing automation meets public obligation — and where a single misapplied flag becomes a real hardship for a household or an audit finding for the utility. This guide sits inside the broader Municipal Utility Billing Architecture & Rate Taxonomy and covers one workflow end to end: turning raw eligibility records for low-income households, senior citizens, and medical-baseline customers into deterministic, auditable adjustments on the bill.

# Problem Framing

In a production billing environment the failure mode is rarely “we forgot to apply the discount.” It is subtler and more expensive: a subsidy that keeps applying three months after a medical certification lapsed, a percentage waiver stacked on top of a fixed-charge waiver so the customer is credited twice, or an expired record that silently flips a household back to the standard residential rate mid-cycle with no notice. Each of these is a revenue-leakage or public-harm event, and each traces back to an eligibility model that was treated as a static configuration table instead of a governed, versioned data asset.

The taxonomy is the operational backbone that maps eligibility attributes — income verification status, household composition, program tier, effective and expiration dates, jurisdictional overrides, and meter validation flags — onto concrete billing operations: discount percentages, fee waivers, baseline allowances, and arrears handling. Get the taxonomy right and every downstream engine (rating, reconciliation, arrears routing) inherits a single, defensible source of truth. Get it wrong and the errors compound across every batch cycle.

This workflow assumes the standard billing pipeline is already ingesting meter reads and resolving rate schedules; assistance handling is the layer that adjusts the resolved charge before it is posted. Because eligibility payloads and rate payloads can each be missing or stale, the taxonomy shares its exception path with Fallback Routing for Missing Rate Data and its identity model with Security Boundaries & Role-Based Access Control.

# Prerequisites

Before implementing the code in this guide, pin the following in your billing service:

• Python 3.11+ — required for datetime.UTC, zoneinfo in the standard library, and modern type syntax (list[Decimal], X | None).
• Pydantic v2 (pydantic>=2.6) — the schema layer below uses v2 field_validator / model_validator signatures, which differ from v1.
• decimal.Decimal for all money — never float. Percentage waivers, block-rate splits, and ledger postings must be exact and reproducibly rounded. Configure the context once at process start.
• zoneinfo (standard library) — grace-period expiry and effective-date boundaries are evaluated in the utility’s local civil time, not UTC, so daylight-saving transitions resolve correctly.
• Data assumptions — eligibility records arrive keyed by account_id (format ACC- + 8 digits) with a program tier drawn from an approved municipal code list, and each account has a resolvable service tier from Customer Class & Service Tier Mapping.
• Access permissions — write access to eligibility state is restricted; assistance data carries income-verification and medical indicators governed by Data Governance & Privacy Compliance.

from decimal import Decimal, ROUND_HALF_UP, getcontext

# Set once at process start. 28 digits of precision is ample for utility money;
# quantize explicitly at every boundary rather than relying on the global context.
getcontext().prec = 28

CENTS = Decimal("0.01")

def to_money(value: Decimal) -> Decimal:
    """Round any Decimal to a posted currency amount (banker-safe HALF_UP)."""
    return value.quantize(CENTS, rounding=ROUND_HALF_UP)

# Architecture Overview

The workflow is a short pipeline: a validated eligibility record is resolved against the billing service period to produce a status, the status selects a waiver rule, the rule is applied to the base charge with an explicit cap, and the decision is logged with a time-bounded expiry check. Verification always fails toward the standard rate plus a review flag — never toward an unbounded subsidy.

Figure: Eligibility routing — verified attributes map to a program tier, then to a capped, logged, time-bounded credit.

# Step-by-Step Implementation

# Step 1 — Define the canonical eligibility schema

Enforce strict schema validation at ingestion so malformed records never reach the rate engine. The Pydantic v2 model below validates identifiers, restricts the program tier to approved codes, and rejects any record whose expiration is not strictly after its effective date. This is the same defensive posture used in Schema Validation & Data Quality Checks for meter telemetry, applied here to eligibility payloads.

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


class ProgramTier(str, Enum):
    LIFELINE_BASIC = "LIFELINE_BASIC"        # low-income percentage discount
    SENIOR_DISCOUNT = "SENIOR_DISCOUNT"      # fixed-charge waiver
    MEDICAL_BASELINE = "MEDICAL_BASELINE"    # extra allowance / exemption


class EligibilityRecord(BaseModel):
    account_id: str = Field(pattern=r"^ACC-\d{8}$")
    program_tier: ProgramTier
    effective_date: date
    expiration_date: date
    income_verified: bool
    household_size: int = Field(ge=1, le=20)
    jurisdiction_code: str = Field(min_length=2, max_length=4)
    meter_serial: str = Field(pattern=r"^MET-[A-Z0-9]{10}$")
    # Grace window applied after expiration before reverting to standard rate.
    grace_period_days: int = Field(default=30, ge=0, le=90)

    @model_validator(mode="after")
    def validate_chronology(self) -> "EligibilityRecord":
        if self.expiration_date <= self.effective_date:
            raise ValueError("expiration_date must be strictly after effective_date")
        return self

Validation is the first line of defense against data drift; every downstream engine can now assume it is handling a canonical eligibility state.

# Step 2 — Resolve status against the service period

Eligibility is never a bare boolean at billing time. Given a bill’s service period (in the utility’s civil time zone), a record resolves to ACTIVE, GRACE, EXPIRED, or INELIGIBLE. The grace window keeps a household on its subsidy for a bounded period after expiration instead of reverting mid-cycle — a customer-protection requirement, not a convenience.

from datetime import date, timedelta
from enum import Enum
from zoneinfo import ZoneInfo


class EligibilityStatus(str, Enum):
    ACTIVE = "ACTIVE"
    GRACE = "GRACE"
    EXPIRED = "EXPIRED"
    INELIGIBLE = "INELIGIBLE"


def resolve_status(
    record: EligibilityRecord,
    service_period_end: date,
) -> EligibilityStatus:
    """Resolve an eligibility record against the last day of the service period.

    service_period_end must already be expressed in the utility's local civil
    date (see Step 5 for the tz-aware conversion) so grace math never drifts
    across a DST boundary.
    """
    if not record.income_verified:
        return EligibilityStatus.INELIGIBLE
    if service_period_end < record.effective_date:
        return EligibilityStatus.INELIGIBLE
    if service_period_end <= record.expiration_date:
        return EligibilityStatus.ACTIVE
    grace_end = record.expiration_date + timedelta(days=record.grace_period_days)
    if service_period_end <= grace_end:
        return EligibilityStatus.GRACE
    return EligibilityStatus.EXPIRED

# Step 3 — Apply the waiver with an explicit cap

Once a record resolves to ACTIVE or GRACE, the tier selects a waiver rule. Isolate the base charge first, then apply exactly one adjustment, then cap the credit so it can never exceed the gross charge. This is where precedence matters: the interaction with tiered consumption pricing is governed by Step-Rate vs Block-Rate Structure Design, which requires the base rate to be resolved before any percentage- or dollar-based waiver is applied so subsidies are never double-counted across overlapping brackets.

from decimal import Decimal


def base_charge_from_blocks(
    consumption: Decimal,
    block_thresholds: list[Decimal],
    block_rates: list[Decimal],
) -> Decimal:
    """Sum a block-rate charge before any assistance is applied.

    block_thresholds are cumulative upper bounds; block_rates[i] applies to the
    volume between block_thresholds[i-1] and block_thresholds[i].
    """
    total = Decimal("0")
    remaining = consumption
    prev = Decimal("0")
    for threshold, rate in zip(block_thresholds, block_rates):
        span = min(remaining, threshold - prev)
        if span <= 0:
            break
        total += span * rate
        remaining -= span
        prev = threshold
    if remaining > 0:  # consumption above the top block uses the last rate
        total += remaining * block_rates[-1]
    return total


def apply_assistance(
    base_charge: Decimal,
    tier: ProgramTier,
    *,
    discount_pct: Decimal = Decimal("0"),      # e.g. 0.25 for 25%
    fixed_waiver: Decimal = Decimal("0"),       # dollar amount, e.g. service charge
    baseline_allowance: Decimal = Decimal("0"), # dollar value of exempt volume
) -> Decimal:
    """Return the adjusted charge. Exactly one rule fires per tier; the credit
    is capped so the customer is never credited below zero."""
    if tier is ProgramTier.LIFELINE_BASIC:
        credit = base_charge * discount_pct
    elif tier is ProgramTier.SENIOR_DISCOUNT:
        credit = fixed_waiver
    elif tier is ProgramTier.MEDICAL_BASELINE:
        credit = baseline_allowance
    else:  # pragma: no cover - enum is exhaustive
        credit = Decimal("0")

    credit = min(credit, base_charge)          # never credit past the gross charge
    return to_money(base_charge - credit)

# Step 4 — Route stale or missing payloads to a fallback

Households transition between income brackets, certifications lapse, and funding cycles shift, so eligibility payloads are routinely stale or absent at billing time. An EXPIRED status — or a payload with no resolvable rate schedule — must not silently reverse the rate. Instead the account is billed at a provisional standard rate, marked PENDING_REVIEW, and pushed to the finance reconciliation queue. This handoff follows the same deterministic pattern documented in Routing Low-Income Assistance Flags Automatically.

from dataclasses import dataclass, field
from decimal import Decimal


@dataclass
class BillingDecision:
    account_id: str
    status: str          # ACTIVE | GRACE | EXPIRED | INELIGIBLE
    base_charge: Decimal
    posted_charge: Decimal
    review_flag: bool = False
    notes: list[str] = field(default_factory=list)


def decide(
    record: EligibilityRecord | None,
    base_charge: Decimal,
    service_period_end: date,
    **waiver_kwargs: Decimal,
) -> BillingDecision:
    # No eligibility payload at all -> bill standard, flag for review.
    if record is None:
        return BillingDecision(
            account_id="UNKNOWN",
            status="INELIGIBLE",
            base_charge=to_money(base_charge),
            posted_charge=to_money(base_charge),
            review_flag=True,
            notes=["no eligibility payload; provisional standard rate applied"],
        )

    status = resolve_status(record, service_period_end)
    if status in (EligibilityStatus.ACTIVE, EligibilityStatus.GRACE):
        posted = apply_assistance(base_charge, record.program_tier, **waiver_kwargs)
        note = "grace period active" if status is EligibilityStatus.GRACE else "active"
        return BillingDecision(
            account_id=record.account_id,
            status=status.value,
            base_charge=to_money(base_charge),
            posted_charge=posted,
            review_flag=(status is EligibilityStatus.GRACE),
            notes=[note],
        )

    # EXPIRED or INELIGIBLE -> standard rate, queued for manual review.
    return BillingDecision(
        account_id=record.account_id,
        status=status.value,
        base_charge=to_money(base_charge),
        posted_charge=to_money(base_charge),
        review_flag=True,
        notes=["subsidy lapsed; routed to finance reconciliation queue"],
    )

# Step 5 — Reconcile waivers against jurisdictional tables

Municipal utilities operate across overlapping tax districts, special-assessment zones, and regional authorities, so waived amounts must be reconciled against jurisdictional exemption tables during nightly batch processing to prevent subsidy leakage or over-application. Post the net adjustment to the general ledger tagged to an ASSISTANCE_PROGRAM cost center, and raise a variance alert when cumulative deviation exceeds a fraction of projected revenue.

from decimal import Decimal

VARIANCE_THRESHOLD = Decimal("0.005")  # 0.5% of projected monthly revenue


def reconcile(
    decisions: list[BillingDecision],
    projected_revenue: Decimal,
) -> dict[str, Decimal | bool]:
    total_waived = sum(
        (d.base_charge - d.posted_charge for d in decisions), Decimal("0")
    )
    deviation = (total_waived / projected_revenue) if projected_revenue else Decimal("0")
    return {
        "total_waived": to_money(total_waived),
        "deviation_ratio": deviation,
        "alert": deviation > VARIANCE_THRESHOLD,
    }

# Edge-Case Handling

• DST and grace-window boundaries. Compute the service-period end date from a timezone-aware datetime before calling resolve_status. A grace window that expires on the “spring forward” or “fall back” day must be evaluated in local civil time, or an account can gain or lose a day of subsidy: service_period_end = billing_cutoff_utc.astimezone(ZoneInfo("America/Chicago")).date().
• Leap-year proration. When a medical-baseline allowance is prorated by days in the billing month, divide by the actual day count (calendar.monthrange) so February 2028’s 29 days are not billed as 28. Keep the divisor and dividend as Decimal and quantize only the final credit.
• Retroactive PUC orders. A commission order can re-rate closed periods. Never mutate a posted BillingDecision; re-run the pipeline against the archived record version for the affected periods and post the delta as a separate adjustment so the audit trail shows both the original and corrected charge.
• Expired certification inside the grace window. A GRACE status still applies the subsidy but sets review_flag=True, so the customer keeps the benefit while caseworkers re-verify — the household is never dropped mid-cycle without notice.
• Duplicate or overlapping records. If two records cover the same period, select the one with the latest effective_date; if both are still active, prefer the more specific jurisdiction_code and flag the overlap rather than summing the waivers.
• Null verification on a formerly active account. income_verified=False forces INELIGIBLE regardless of dates — a cleared verification flag must revoke the subsidy on the next cycle, not persist on stale state.

# Verification and Audit Trail

Assistance decisions must be reproducible and tamper-evident. Serialize each BillingDecision to a canonical form (sorted keys, Decimal rendered as fixed strings), hash it with SHA-256, and store the digest alongside the ledger posting. Re-hashing the archived decision during an audit proves the posted charge was never altered after the fact — the non-repudiation property municipal finance audits require.

import hashlib
import json
from decimal import Decimal


def seal_decision(d: BillingDecision) -> str:
    payload = {
        "account_id": d.account_id,
        "status": d.status,
        "base_charge": str(d.base_charge),
        "posted_charge": str(d.posted_charge),
        "review_flag": d.review_flag,
        "notes": d.notes,
    }
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def test_lifeline_discount_is_exact():
    rec = EligibilityRecord(
        account_id="ACC-00010001",
        program_tier=ProgramTier.LIFELINE_BASIC,
        effective_date=date(2026, 1, 1),
        expiration_date=date(2026, 12, 31),
        income_verified=True,
        household_size=4,
        jurisdiction_code="TX07",
        meter_serial="MET-A1B2C3D4E5",
    )
    decision = decide(
        rec,
        base_charge=Decimal("120.00"),
        service_period_end=date(2026, 6, 30),
        discount_pct=Decimal("0.25"),
    )
    assert decision.posted_charge == Decimal("90.00")   # exact, not 89.999...
    assert len(seal_decision(decision)) == 64

Run the full billing simulation against a corpus of historical, hand-verified fixtures before every production deployment, and compare posted charges with Decimal equality — never float tolerance. A single mismatch means a precedence or rounding regression that would otherwise surface as thousands of off-by-a-cent invoices.

# Troubleshooting

# Related Topics

• Up one level: Municipal Utility Billing Architecture & Rate Taxonomy
• Routing Low-Income Assistance Flags Automatically
• Customer Class & Service Tier Mapping
• Step-Rate vs Block-Rate Structure Design
• Fallback Routing for Missing Rate Data
• Data Governance & Privacy Compliance