Automating Daily Fare Variance Reports

This page builds the report generator that turns two aggregated totals — expected fare revenue from the calculation engines and settled totals from acquirer files — into the daily variance report ops reads every morning. The task is concrete: group both sides by operator and fare product, compute the signed variance and variance percentage in Decimal, flag any line that breaks a configurable tolerance, and emit both a structured summary and a CSV-ready row set for the finance export. It is the reporting half of Daily Variance Reconciliation, inside Revenue Reconciliation & Settlement, and it assumes the reconciler has already produced per-bucket verdicts.

Report generation flow

The generator is a straight pipeline: join the two aggregates on the bucket key, difference and grade each line, then split the output into a header summary and a flat row list. Nothing branches on external state, which is what makes the report reproducible from the same two inputs. The flow below shows the four stages from paired totals to the two emitted artifacts:

Step 1 — Model the report line

Every row of the report is one bucket. The model carries both raw totals, the signed variance, its percentage, and a grade, plus a serialization method that produces the exact string form the CSV export needs. Money stays Decimal end to end; only the CSV projection turns it into a formatted string, and it does so with a fixed quantization so a re-export is byte-stable.

from __future__ import annotations

import logging
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP

logger = logging.getLogger("transit.variance.report")

PCT = Decimal("0.0001")


@dataclass(frozen=True)
class ReportLine:
    operator_id: str
    fare_product: str
    expected: Decimal      # minor units (cents)
    settled: Decimal       # minor units (cents)
    variance: Decimal      # expected - settled, signed
    variance_pct: Decimal  # fraction of expected
    flagged: bool          # True when beyond tolerance

    def to_csv_row(self) -> dict[str, str]:
        """Project to the flat string form the finance export consumes."""
        return {
            "operator_id": self.operator_id,
            "fare_product": self.fare_product,
            "expected_cents": str(self.expected),
            "settled_cents": str(self.settled),
            "variance_cents": str(self.variance),
            "variance_pct": str((self.variance_pct * 100).quantize(
                Decimal("0.01"), rounding=ROUND_HALF_UP)),
            "flagged": "Y" if self.flagged else "N",
        }

Step 2 — Configure tolerance and compute a line

A single line’s math is a subtraction and a division, but both have to survive the degenerate cases: a bucket with zero expected revenue cannot be divided, and a bucket that settled money against zero expected is always a flag. The tolerance is a two-part rule — an absolute cent floor and a relative fraction — and a line is flagged only when both floors break, so a large-percentage blip on a tiny bucket does not page anyone.

from dataclasses import dataclass


@dataclass(frozen=True)
class ProductTolerance:
    abs_cents: Decimal
    rel_fraction: Decimal   # e.g. Decimal("0.005") for 0.5%


def _variance_pct(variance: Decimal, expected: Decimal) -> Decimal:
    if expected == 0:
        return Decimal("0")
    return (variance / expected).quantize(PCT, rounding=ROUND_HALF_UP)


def build_line(
    operator_id: str,
    fare_product: str,
    expected: Decimal,
    settled: Decimal,
    tol: ProductTolerance,
) -> ReportLine:
    """Difference and grade one bucket into a report line."""
    expected = expected.quantize(Decimal("1"))
    settled = settled.quantize(Decimal("1"))
    variance = expected - settled
    pct = _variance_pct(variance, expected)

    if expected == 0:
        # Settled money against a product that priced nothing is never rounding.
        flagged = variance != 0
    else:
        flagged = abs(variance) > tol.abs_cents and abs(pct) > tol.rel_fraction

    if flagged:
        logger.warning(
            "flagged %s/%s: variance=%s pct=%s",
            operator_id, fare_product, variance, pct,
        )
    return ReportLine(
        operator_id=operator_id,
        fare_product=fare_product,
        expected=expected,
        settled=settled,
        variance=variance,
        variance_pct=pct,
        flagged=flagged,
    )

Step 3 — Generate the full report

The generator ranges over the union of buckets so a product missing from either side still produces a line, sorts for stable output, and folds a summary as it goes. The summary carries the day’s total expected, total settled, net variance, and — the number ops actually watches — the count of flagged lines. Returning both the summary and the row list from one pass keeps the two artifacts guaranteed consistent.

from dataclasses import dataclass, field
from typing import Mapping


@dataclass(frozen=True)
class ReportSummary:
    business_date: str
    total_expected: Decimal
    total_settled: Decimal
    net_variance: Decimal
    line_count: int
    flagged_count: int


@dataclass(frozen=True)
class DailyVarianceReport:
    summary: ReportSummary
    lines: list[ReportLine] = field(default_factory=list)

    def csv_rows(self) -> list[dict[str, str]]:
        return [line.to_csv_row() for line in self.lines]

    def exceptions(self) -> list[ReportLine]:
        return [line for line in self.lines if line.flagged]


def generate_report(
    business_date: str,
    expected: Mapping[tuple[str, str], Decimal],
    settled: Mapping[tuple[str, str], Decimal],
    tolerances: Mapping[tuple[str, str], ProductTolerance],
    default_tol: ProductTolerance,
) -> DailyVarianceReport:
    """Produce the graded report and its CSV rows from two aggregates."""
    lines: list[ReportLine] = []
    total_exp = Decimal("0")
    total_set = Decimal("0")
    flagged = 0

    for key in sorted(set(expected) | set(settled)):
        operator_id, fare_product = key
        exp = expected.get(key, Decimal("0"))
        settle = settled.get(key, Decimal("0"))
        tol = tolerances.get(key, default_tol)
        line = build_line(operator_id, fare_product, exp, settle, tol)
        lines.append(line)
        total_exp += line.expected
        total_set += line.settled
        flagged += int(line.flagged)

    summary = ReportSummary(
        business_date=business_date,
        total_expected=total_exp,
        total_settled=total_set,
        net_variance=total_exp - total_set,
        line_count=len(lines),
        flagged_count=flagged,
    )
    logger.info(
        "report %s: %d lines, %d flagged, net=%s",
        business_date, summary.line_count, flagged, summary.net_variance,
    )
    return DailyVarianceReport(summary=summary, lines=lines)

Validation & Test Cases

The tests pin the two behaviors that matter: a clean day nets to zero and flags nothing, and a single beyond-tolerance product surfaces as an exception without dragging the clean lines with it.

from decimal import Decimal

default_tol = ProductTolerance(abs_cents=Decimal("500"), rel_fraction=Decimal("0.005"))

# Normal case: two products, both within tolerance.
expected = {
    ("MUNI_BUS", "adult_single"): Decimal("1250000"),   # $12,500.00
    ("MUNI_BUS", "senior_single"): Decimal("330000"),   #  $3,300.00
}
settled = {
    ("MUNI_BUS", "adult_single"): Decimal("1249850"),   # $1.50 short, under floor
    ("MUNI_BUS", "senior_single"): Decimal("330000"),   # exact
}
report = generate_report("2026-07-16", expected, settled, {}, default_tol)

assert report.summary.flagged_count == 0
assert report.summary.net_variance == Decimal("150")   # 150 cents net
adult = report.lines[0]
assert adult.variance == Decimal("150")
# 150 / 1_250_000 = 0.00012 -> well under the 0.5% relative floor
assert not adult.flagged

# Edge case: a product settled money the engine never priced -> forced flag.
expected_2 = {("MUNI_BUS", "adult_single"): Decimal("1250000")}
settled_2 = {
    ("MUNI_BUS", "adult_single"): Decimal("1180000"),   # $700 short, both floors broken
    ("REGIONAL_RAIL", "adult_single"): Decimal("4200"), # $42 with zero expected
}
report_2 = generate_report("2026-07-16", expected_2, settled_2, {}, default_tol)

exceptions = report_2.exceptions()
flagged_keys = {(e.operator_id, e.fare_product) for e in exceptions}
assert flagged_keys == {("MUNI_BUS", "adult_single"), ("REGIONAL_RAIL", "adult_single")}
# The zero-expected line reports a 0% variance_pct but is still flagged.
rail = next(e for e in exceptions if e.operator_id == "REGIONAL_RAIL")
assert rail.expected == Decimal("0") and rail.variance == Decimal("-4200")
assert rail.variance_pct == Decimal("0") and rail.flagged

# CSV projection is stable and string-typed for the finance export.
row = adult.to_csv_row()
assert row["variance_cents"] == "150" and row["flagged"] == "N"

The adult line in the normal case is short by 150 cents, but 150 / 1,250,000 is 0.00012 — two orders of magnitude below the 0.5% relative floor — so it stays clean and only surfaces in trend monitoring, not the exception queue. The rail line in the edge case proves the zero-expected rule: its percentage is reported as 0 because there is no denominator, yet the $42 that settled against a product that priced nothing is flagged unconditionally, because no rounding story explains it.

Edge Cases

• Products present on only one side. The union iteration guarantees a line for a product that priced revenue but settled nothing (a dropped batch) and for one that settled money with zero expected (a miscoded product or a stray file). An inner join would silently drop both — the two failures this report exists to catch.
• Sign convention drift. variance = expected - settled means positive is under-collection. Every consumer, dashboard, and export must share that convention; flipping it on one screen turns a shortfall into an apparent surplus and delays the investigation.
• Percentage on a near-zero denominator. A product with two dollars of expected revenue can post a 300% variance from a two-cent rounding difference. The absolute-and-relative rule suppresses it, but the raw percentage is still stored so trend tooling can see it if the pattern recurs.

Integration Note

This generator consumes the aggregates and grading contract defined in the parent Daily Variance Reconciliation topic, which owns the business-date cutoff and the ledger read that produces the expected side. Its immediate upstream sibling is Matching Settlement Files to Tap Ledgers: that row-level match confirms each settlement line is real and complete before the totals it feeds are aggregated here, so a truncated file is caught as a missing row rather than surfacing in this report as a phantom under-collection.

FAQ

Related

• Daily Variance Reconciliation — the parent topic that owns the reconciler and the business-date cutoff.
• Matching Settlement Files to Tap Ledgers — the row-level match that validates settlement lines before they are aggregated into this report.
• Ledger & Audit Trail Design — the append-only source of the expected totals this report differences against.
• Inter-Agency Settlement Clearing — downstream consumer that will not net a day with open exceptions from this report.

Part of Daily Variance Reconciliation, within Revenue Reconciliation & Settlement.