Automating Senior and Student Fare Validation Rules

The task is narrow and unforgiving: given a tap event carrying a rider’s age and enrollment status, decide — deterministically, and with a defensible audit trail — whether a senior or student concession applies, then compute the discounted fare. Transit agencies bleed revenue and invite compliance findings when this decision is left to manual overrides at the gate or buried in rigid legacy validators that no analyst can re-derive. This page shows how to encode municipal senior and student subsidy policy as a stateless Python function, one that a revenue analyst can reconcile line-by-line and a mobility-tech developer can ship to production. It is one of the concrete concession workflows inside the Discount Eligibility Engines stage, which itself sits within the broader Fare Rule Validation & Calculation Engines pipeline.

The design goal throughout is externalized policy. Age thresholds, discount rates, and program tiers do not belong hardcoded inside a transaction processor where every subsidy change triggers a firmware patch cycle. They live in a version-controlled configuration object that flows into the evaluator, so a mid-fiscal-year change to the senior age threshold propagates without a full redeployment. Incoming tap payloads are normalized upstream — validated against a strict schema, with timestamps coerced to UTC — following the same discipline documented in Implementing Pydantic Models for AFC Event Streams, before they ever reach the eligibility layer described here.

Eligibility Decision Flow

The evaluator applies a fixed precedence so that dual-eligible riders always resolve the same way. Chronological age is checked first; if the rider does not clear the senior threshold, active university enrollment is checked next; a pending enrollment resolves to the standard fare but is flagged for manual reconciliation rather than silently denied. The diagram below traces that precedence from tap to final fare:

The discounted fare is computed with exact decimal arithmetic and half-up rounding, never floating point, so that a batch of taps reconciled against the clearinghouse never drifts by a stray cent:

$$\text{final\_fare} = \operatorname{round_{HALF\_UP}}\bigl(\text{base\_fare} \times (1 - \text{discount})\bigr)$$

Step-by-Step Implementation

The implementation is a single stateless function, evaluate_eligibility, plus the typed models it consumes and returns. Every monetary field is a Decimal; every branch appends to a machine-readable audit_flags list so the decision can be replayed. The rule_version travels with the result, which is what makes downstream ledger reconciliation across a policy change tractable.

import uuid
from decimal import Decimal, ROUND_HALF_UP
from datetime import datetime, timezone
from enum import Enum
from typing import Optional, List

import structlog
from pydantic import BaseModel, Field, ValidationError, field_validator

logger = structlog.get_logger()


class StudentStatus(str, Enum):
    ACTIVE = "active"
    EXPIRED = "expired"
    PENDING = "pending"


class FareTransaction(BaseModel):
    card_id: str = Field(..., min_length=8, max_length=16, pattern=r"^[A-Z0-9]+$")
    tap_timestamp: datetime
    rider_age: Optional[int] = Field(None, ge=0, le=120)
    student_status: Optional[StudentStatus] = None
    program_tier: str = Field(..., pattern="^(standard|subsidized|university_partner)$")

    @field_validator("tap_timestamp")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        """Normalize all ingress timestamps to UTC."""
        if v.tzinfo is None:
            return v.replace(tzinfo=timezone.utc)
        return v.astimezone(timezone.utc)


class PolicyConfig(BaseModel):
    """Externalized, version-controlled subsidy policy — never hardcoded."""
    senior_age_threshold: int = Field(65, ge=60, le=70)
    student_discount_rate: Decimal = Field(Decimal("0.50"), ge=0, le=1)
    senior_discount_rate: Decimal = Field(Decimal("0.50"), ge=0, le=1)
    base_fare: Decimal = Decimal("2.90")
    rule_version: str = "v2.4.1"


class ValidationResult(BaseModel):
    transaction_id: str
    eligible: bool
    applied_discount: Decimal
    final_fare: Decimal
    rule_version: str
    decision_path: str
    audit_flags: List[str]
    requires_manual_review: bool


class PolicyEvaluationError(Exception):
    """Raised when validation logic encounters unrecoverable state."""


def evaluate_eligibility(tx: FareTransaction, policy: PolicyConfig) -> ValidationResult:
    """
    Stateless senior/student concession evaluator with an explicit audit trail.

    Precedence: senior age first, then active university enrollment. A pending
    enrollment resolves to standard fare but is flagged for reconciliation.
    """
    audit_flags: List[str] = []
    decision_path = "standard_fare"
    eligible = False
    discount = Decimal("0.00")

    try:
        # 1. Senior eligibility — chronological age clears the configured threshold.
        if tx.rider_age is not None and tx.rider_age >= policy.senior_age_threshold:
            eligible = True
            discount = policy.senior_discount_rate
            decision_path = "senior_age_override"
            audit_flags.append(f"age_verified:{tx.rider_age}")

        # 2. Student eligibility — active enrollment on a partner program.
        elif tx.student_status == StudentStatus.ACTIVE and tx.program_tier == "university_partner":
            eligible = True
            discount = policy.student_discount_rate
            decision_path = "student_active_enrollment"
            audit_flags.append("enrollment_verified:active")

        # 3. Pending enrollment — standard fare, but queued for manual review.
        elif tx.student_status == StudentStatus.PENDING:
            audit_flags.append("student_pending_review")

        final_fare = (policy.base_fare * (Decimal("1") - discount)).quantize(
            Decimal("0.01"), rounding=ROUND_HALF_UP
        )
        requires_review = "student_pending_review" in audit_flags or tx.rider_age is None

        return ValidationResult(
            transaction_id=str(uuid.uuid4()),
            eligible=eligible,
            applied_discount=discount,
            final_fare=final_fare,
            rule_version=policy.rule_version,
            decision_path=decision_path,
            audit_flags=audit_flags,
            requires_manual_review=requires_review,
        )

    except ValidationError as ve:
        logger.error("schema_validation_failed", error=str(ve), card_id=tx.card_id)
        raise PolicyEvaluationError(f"Malformed transaction payload: {ve}") from ve
    except (ArithmeticError, TypeError) as exc:
        logger.critical("evaluation_error", error=str(exc), card_id=tx.card_id)
        raise PolicyEvaluationError(
            f"Policy evaluation failed for card {tx.card_id}: {exc}"
        ) from exc

Two implementation details carry disproportionate weight. First, rider_age is None forces requires_manual_review=True even when no discount applied — an absent age is never treated as a silent “not a senior.” Second, the senior branch is checked before the student branch by deliberate policy choice; if your agency reverses that (student discount deeper than senior), reorder the branches rather than mutating the rates, so the decision_path still names exactly which rule fired.

Validation & Test Cases

Every concession path needs a fixture with a stated expected output. The table below fixes the four load-bearing cases against a default PolicyConfig (base fare 2.90, both rates 0.50, threshold 65):

The same cases as runnable assertions — one normal path and one edge/error path per concession:

from decimal import Decimal
import pytest

POLICY = PolicyConfig()


def _tx(**kwargs) -> FareTransaction:
    base = dict(card_id="TRANSIT8842A", tap_timestamp="2026-07-03T14:00:00+00:00",
                program_tier="standard")
    base.update(kwargs)
    return FareTransaction(**base)


def test_senior_discount_applies():
    result = evaluate_eligibility(_tx(rider_age=67), POLICY)
    assert result.decision_path == "senior_age_override"
    assert result.final_fare == Decimal("1.45")
    assert result.requires_manual_review is False


def test_pending_student_flags_for_review():
    tx = _tx(rider_age=19, student_status="pending", program_tier="subsidized")
    result = evaluate_eligibility(tx, POLICY)
    assert result.eligible is False
    assert result.final_fare == Decimal("2.90")
    assert "student_pending_review" in result.audit_flags
    assert result.requires_manual_review is True


def test_missing_age_forces_manual_review():
    result = evaluate_eligibility(_tx(rider_age=None), POLICY)
    assert result.requires_manual_review is True  # absent age is never "not senior"


def test_malformed_card_id_is_rejected_at_schema():
    with pytest.raises(ValidationError):
        _tx(card_id="lower-case!", rider_age=67)  # fails ^[A-Z0-9]+$ before evaluation

Note the last test: a malformed card_id is caught by the Pydantic schema at construction time, before evaluate_eligibility runs. That boundary is intentional — the evaluator assumes it only ever sees structurally valid transactions, so bad payloads fail loudly at ingress rather than producing a plausible-but-wrong fare.

Production Debugging and Reconciliation

Concession logic that passes unit tests still generates revenue discrepancies once real validators, real clocks, and real third-party enrollment APIs enter the picture. Four failure modes account for most reconciliation tickets:

1. Timezone and DST boundary drift. Tap events arrive from gates with misaligned system clocks. The enforce_utc validator normalizes on ingress, but you should still run a nightly job flagging any transaction where the reported local tap time and the stored UTC differ by more than a plausible offset. Python’s datetime documentation is the reference for astimezone() behavior across ambiguous DST windows.
2. Stale enrollment API responses. Student status often depends on an institutional API. When it returns 503 or times out, default to requires_manual_review=True and reuse the last known status — never let None silently collapse to “inactive.” Log API latency alongside each fare decision so upstream degradation is visible before it hits settlement. Validators that must keep deciding while that dependency is unreachable follow Fallback Routing Strategies to cache and replay locally.
3. Schema drift and policy versioning. Funding mandates move age thresholds mid-year. Because rule_version rides on every ValidationResult, you can isolate policy-driven fare deltas by querying the warehouse for v2.4.0 versus v2.4.1. Validate incoming payloads in Pydantic strict mode to catch silent type coercion; see Pydantic’s strict-mode guidance.
4. Validator firmware sync. Offline gates cache policy locally. Push only changed PolicyConfig fields as a delta so edge devices never run a stale senior_age_threshold, and cross-reference decision_path logs between cloud and edge to detect split-brain fare calculations before they compound into settlement shortfalls. The immutable audit trail this evaluator emits is what an auditor replays under the controls described in AFC System Security Boundaries.

Integration Note

This evaluator is one leaf of the Discount Eligibility Engines stage: it consumes a normalized, schema-validated tap and emits a priced, audit-flagged result that the ledger posts. Its expiry-window cousin — deciding when a concession lapses rather than whether it applies today — is handled by threshold work such as Dynamic Peak Pricing Threshold Adjustment Scripts, and a discounted senior tap that later chains into a free transfer is resolved by Calculating Cross-Operator Transfer Windows with Python. Keeping eligibility, thresholds, and transfers as separate stages means a subsidy-policy change never forces a rewrite of transfer logic, and vice versa.

FAQ

Related

• Discount Eligibility Engines — the parent stage this evaluator plugs into
• Dynamic Peak Pricing Threshold Adjustment Scripts — tuning the thresholds and windows that concessions are checked against
• Calculating Cross-Operator Transfer Windows with Python — how a discounted tap chains into a free transfer
• Implementing Pydantic Models for AFC Event Streams — the upstream schema enforcement that feeds clean transactions here
• Fallback Routing Strategies — deciding concessions when the enrollment API or backhaul is unreachable

Part of Discount Eligibility Engines, within Fare Rule Validation & Calculation Engines.