Skip to main content
Markdown

Reconcile Invoices and Settlements

Match billing-system invoices with payment-processor settlements by invoice ID, classify missing payments and amount differences, and export the result as CSV and Parquet. The workflow preserves every input ID so downstream consumers can account for matched items as well as exceptions.

Uses: Data SDK. Services: none. Difficulty: introductory.

Codon compiles the matching and classification loops to native code; the Data SDK handles CSV parsing and Parquet output. Integer minor units keep this example's single-currency amounts exact. Do not represent money as binary floating point.

Run the Reconciliation​

The program creates two tiny input files so the complete recipe runs without a download. In a real job, replace those two fixture writes with the upstream file delivery step. The one-row-per-invoice contract is enforced: duplicates are an error, not an implicit choice of which payment to keep.

reconcile.codon
import exaloop.data as data

with open("invoices.csv", "w") as stream:
    stream.write("invoice_id,amount_cents\nINV-1,12500\nINV-2,7800\nINV-3,4500\n")
with open("settlements.csv", "w") as stream:
    stream.write("invoice_id,amount_cents\nINV-1,12500\nINV-2,7500\nINV-4,2200\n")

def amounts(path: str) -> dict[str, int]:
    with data.read_csv(path) as source:
        with source["invoice_id"] as keys, source["amount_cents"] as values:
            identifiers = keys.to_list(str)
            cents = values.to_list(int)
    result: dict[str, int] = {}
    for identifier, amount in zip(identifiers, cents):
        if identifier in result:
            raise ValueError("duplicate invoice: " + identifier)
        if amount < 0:
            raise ValueError("negative amount: " + identifier)
        result[identifier] = amount
    return result

invoices = amounts("invoices.csv")
settlements = amounts("settlements.csv")
identifiers = sorted(set(invoices) | set(settlements))
expected = []
received = []
classifications: list[str] = []

for identifier in identifiers:
    expected.append(invoices.get(identifier))
    received.append(settlements.get(identifier))
    if identifier not in invoices:
        classification = "unexpected"
    elif identifier not in settlements:
        classification = "missing"
    elif invoices[identifier] != settlements[identifier]:
        classification = "amount_mismatch"
    else:
        classification = "matched"
    classifications.append(classification)
    print(identifier, classification)

assert classifications == ["matched", "amount_mismatch", "missing", "unexpected"]
with data.table({"invoice_id": data.array(identifiers),
                 "expected_cents": data.array(expected),
                 "received_cents": data.array(received),
                 "status": data.array(classifications)}) as report:
    report.write("reconciliation.parquet")
    report.write("reconciliation.csv")

with data.read_parquet("reconciliation.parquet") as report:
    with report["received_cents"] as column:
        assert column.to_list(Optional[int]) == received
Terminal
exaloop --with data run -release reconcile.codon

Expected output:

Output
INV-1 matched
INV-2 amount_mismatch
INV-3 missing
INV-4 unexpected

The report includes matched rows as well as exceptions so another system can account for every input ID. A missing amount stays null in Parquet, rather than becoming a zero-dollar settlement. CSV is a convenient review copy; Parquet is the typed handoff for downstream processing.

Adapt the Business Contract​

For multiple currencies, key the reconciliation by both invoice ID and currency; never net unrelated currencies. Split payments require an explicit aggregation rule before matching. Refunds need a signed-amount policy rather than simply removing the negative-value check. Include source batch IDs and processing dates in a production report so exceptions can be traced back to their input files.

CSV inference is suitable for this controlled fixture. Ingesting arbitrary exports requires schema checks, mandatory-column validation, and a policy for missing or malformed values. The native conversion rejects null amounts here. For decimal text and declared scale, see the Avro decimal helpers.

Both tables and the key dictionaries are held in memory. For files beyond that budget, partition both inputs by the same stable business key, or perform the join in DuckDB and keep only the custom classification in Codon. Write production reports to a staging path and publish them only after all checks pass; this example is not an atomic multi-file publishing protocol.