Reconciliation, audit trails and reproducibility

The original sixteen-order pipeline loses six rows, but its named rules initially explain only four. Reconciliation records rows, keys and totals at each stage; a row audit trail explains specific exclusions and changes. This lesson keeps that original pipeline and the real-month source-overlap analysis, then compares count accounting, identity, run logs and the declared fingerprint policy used for repeat runs.

Each stage accounts for the rows it changes

Reconciliation proves that the numbers in a final table can be traced back to their source. For every stage of a pipeline you count the rows, the distinct keys and the total of an additive column such as revenue, and every difference between two stages is explained by a named rule with a count. An audit trail keeps the rows each rule removed and the values each step changed, with the step and the reason, so that any single row can be accounted for. Reproducibility means that the same inputs, code and versions give the same outputs, and a run log records which inputs, code and versions a run used. Together they answer the three questions every report meets sooner or later: do the numbers add up, where did this row go, and would we get the same answer again?

The habit is small: after every step, record the row count, the number of distinct keys and the total, and have each rule count the rows it removes. A step’s unexplained deltaunexplained delta: A signed stage change remaining after the declared rule's accounted change is subtracted under a fixed reconciliation convention. is its dropped rows minus the rows its rule removed. That delta should be zero under the stated removal contract. Anything else is unexplained, and an unexplained difference is a bug until someone explains it. Here a week of 16 orders goes through four steps, chained with DataFrame.pipe:

"""A week of orders through four steps, with a reconciliation table that explains every row that leaves."""
import pandas as pd

orders = pd.read_csv("orders.csv")
stores = pd.read_csv("stores.csv")


def drop_repeats(df):
    repeat = df.duplicated()
    return df[~repeat], int(repeat.sum())  # the rows a rule removes, counted by the rule itself


def exclude_cancelled(df):
    cancelled = df["status"].eq("cancelled")
    return df[~cancelled], int(cancelled.sum())


def add_region(df):
    return df.merge(stores, on="store"), 0  # adds a column; no rule removes rows here


def keep_sales(df):
    returned = df["status"].eq("returned")
    return df[~returned], int(returned.sum())


def stage(df, name, by_rule=0):
    return {"stage": name, "rows": len(df), "orders": df["order_id"].nunique(), "amount": df["amount"].sum(), "by rule": by_rule}


table = [stage(orders, "source")]
df = orders
for step in [drop_repeats, exclude_cancelled, add_region, keep_sales]:
    df, by_rule = df.pipe(step)
    table.append(stage(df, step.__name__, by_rule))

recon = pd.DataFrame(table)
recon["dropped"] = (-recon["rows"].diff()).fillna(0).astype(int)
recon["unexplained"] = recon["dropped"] - recon["by rule"]
print(recon.to_string(index=False))
lost = recon[recon["unexplained"] != 0]
for row in lost.itertuples():
    print(f"\n{row.stage} lost {row.unexplained} rows that no rule removed")

            stage  rows  orders  amount  by rule  dropped  unexplained
           source    16      15 17040.0        0        0            0
     drop_repeats    15      15 15540.0        1        1            0
exclude_cancelled    13      13 14100.0        2        2            0
       add_region    11      11 12160.0        0        2            2
       keep_sales    10      10 11520.0        1        1            0

add_region lost 2 rows that no rule removed

order_id,store,status,amount
A-1001,S1,delivered,1250.00
A-1002,S2,delivered,480.00
A-1003,S1,cancelled,990.00
A-1004,S3,delivered,2100.00
A-1005,S4,delivered,760.00
A-1006,S2,delivered,1500.00
A-1006,S2,delivered,1500.00
A-1007,S3,returned,640.00
A-1008,S1,delivered,330.00
A-1009,S4,delivered,1180.00
A-1010,S2,cancelled,450.00
A-1011,S3,delivered,870.00
A-1012,S1,delivered,2400.00
A-1013,S2,delivered,560.00
A-1014,S3,delivered,1320.00
A-1015,S1,delivered,710.00

store,region
S1,West
S2,South
S3,North

Six rows leave between the source and the final table, and the rules explain four of them: one exact repeat, two cancelled orders and one return. The join that adds each store’s region was not meant to remove anything, yet it lost two rows, because store S4 is missing from the stores table and an inner join drops orders without a partner. The two orders were worth 1,940, so the week’s sales would have been reported as 11,520 instead of 13,460, and every other line of the report would have looked right. The table also shows what a correct join looks like: the same rows, the same number of orders and the same total before and after.

Use the original 16 rows and two lost join rows to locate the unexplained stage.

Counts, row history and repeatability provide different evidence

A reconciliation table says how many rows each step removed; row audit trailsrow audit trail: Retained row-level evidence of exclusions or changed values, with the responsible stage and reason under a stated identity and retention policy. say which rows and why. Audit trails keep the removed rows with the step and the reason, and every changed value with its old and new value, so that a question about one order becomes a lookup. Reproducibility is what lets anyone, including you in six months, re-run the analysis and get the same numbers. It needs the same input files, proven by their hashes, the same package versions, fixed random seeds and a log of the run. The UK government’s analysis profession makes reproducible analytical pipelines its default way of working and lists the minimum practices.

The parts below apply all three to real data. A full reconciliation of one month of the UCI Online Retail II data shows how two overlapping source sheets can inflate revenue. The count invariant, rows in equals rows out plus rows in the trail, detects the demonstrated unaccounted drop. Exact source-row accounting also needs row identity or an equivalent multiset check. A run-and-compare check re-runs the pipeline from a clean start and compares the outputs.

Keep counts, row identities and repeatability evidence separate when reading the original pipeline.

The exercise reports unexplained losses and additions

Preserve the original input rows while returning stage summaries and signed differences. An added row makes the dropped count negative; clamping it to zero hides the change. Locate the first nonzero unexplained delta, including a negative one. Keep the row and key definitions unchanged and compare the summary with the untouched source frame.

def missing(value):
    """A cell read from a CSV is missing when it is empty or holds only spaces."""
    return value is None or str(value).strip() == ""


def stage_stats(rows, key, amount):
    # The number of rows, the number of distinct non-missing keys, and the total of the non-missing amounts
    # (the amounts are text: convert them with float), rounded to 2 decimal places.
    return {"rows": 0, "keys": 0, "total": 0}


def reconcile(rows, steps, key, amount):
    # One dictionary for the source, then one per step: its stage_stats, the rows it dropped, and the dropped rows
    # that its rule does not explain. Each step is a pair (name, function); function(rows) returns
    # (rows_after, rows_removed_by_its_rule).
    return [{"stage": "source", **stage_stats(rows, key, amount), "dropped": 0, "unexplained": 0}]


def first_unexplained(table):
    # The name of the first stage whose "unexplained" is not 0, or None.
    return None

from recon import first_unexplained, reconcile, stage_stats

ORDERS = [
    {"order_id": "B-1", "store": "S1", "status": "delivered", "amount": "1250.00"},
    {"order_id": "B-2", "store": "S2", "status": "delivered", "amount": "480.50"},
    {"order_id": "B-2", "store": "S2", "status": "delivered", "amount": "480.50"},
    {"order_id": "B-3", "store": "S5", "status": "delivered", "amount": "760.00"},
    {"order_id": "B-4", "store": "S1", "status": "cancelled", "amount": "990.00"},
    {"order_id": "B-5", "store": "S3", "status": "delivered", "amount": ""},
    {"order_id": "", "store": "S3", "status": "delivered", "amount": "100.00"},
    {"order_id": "B-6", "store": "S5", "status": "delivered", "amount": "1180.25"},
]
REGIONS = {"S1": "West", "S2": "South", "S3": "North"}


def drop_repeats(rows):
    seen, out = set(), []
    for row in rows:
        marker = tuple(sorted(row.items()))
        if marker not in seen:
            out.append(row)
        seen.add(marker)
    return out, len(rows) - len(out)


def exclude_cancelled(rows):
    out = [r for r in rows if r["status"] != "cancelled"]
    return out, len(rows) - len(out)


def add_region(rows):
    """Looks up each store's region: an inner join, so stores without a region disappear."""
    return [{**r, "region": REGIONS[r["store"]]} for r in rows if r["store"] in REGIONS], 0


def one_row_per_item(rows):
    """A fan-out: every order of S1 is split into two items, so rows are added."""
    out = []
    for r in rows:
        out.extend([r, {**r, "amount": "0"}] if r["store"] == "S1" else [r])
    return out, 0


def test_stage_stats():
    """counts rows, distinct non-missing keys and the total of the non-missing amounts"""
    assert stage_stats(ORDERS, "order_id", "amount") == {"rows": 8, "keys": 6, "total": 5241.25}
    assert stage_stats([], "order_id", "amount") == {"rows": 0, "keys": 0, "total": 0}


def test_table_rows_and_columns():
    """has one row for the source and one for each step, in order"""
    table = reconcile(ORDERS, [("drop_repeats", drop_repeats), ("exclude_cancelled", exclude_cancelled)], "order_id", "amount")
    assert [r["stage"] for r in table] == ["source", "drop_repeats", "exclude_cancelled"]
    assert table[0] == {"stage": "source", "rows": 8, "keys": 6, "total": 5241.25, "dropped": 0, "unexplained": 0}
    assert table[1] == {"stage": "drop_repeats", "rows": 7, "keys": 6, "total": 4760.75, "dropped": 1, "unexplained": 0}
    assert table[2] == {"stage": "exclude_cancelled", "rows": 6, "keys": 5, "total": 3770.75, "dropped": 1, "unexplained": 0}


def test_finds_the_join_that_loses_rows():
    """flags the step whose dropped rows no rule explains"""
    steps = [("drop_repeats", drop_repeats), ("exclude_cancelled", exclude_cancelled), ("add_region", add_region)]
    table = reconcile(ORDERS, steps, "order_id", "amount")
    assert table[3] == {"stage": "add_region", "rows": 4, "keys": 3, "total": 1830.5, "dropped": 2, "unexplained": 2}
    assert first_unexplained(table) == "add_region"


def test_rows_added_are_unexplained_too():
    """a step that adds rows has a negative drop, which no rule explains either"""
    table = reconcile(ORDERS, [("one_row_per_item", one_row_per_item)], "order_id", "amount")
    assert table[1]["rows"] == 10 and table[1]["dropped"] == -2 and table[1]["unexplained"] == -2
    assert first_unexplained(table) == "one_row_per_item"


def test_clean_pipeline():
    """returns None when every change is explained, and keeps the input rows unchanged"""
    before = [dict(r) for r in ORDERS]
    table = reconcile(ORDERS, [("drop_repeats", drop_repeats)], "order_id", "amount")
    assert first_unexplained(table) is None
    assert ORDERS == before

Retain the original five tests, including the negative-drop fan-out and unchanged input rows.

One real month of Online Retail II, reconciled row by row Pro: opens with a Pro, Premium or Ultimate pass

Audit trails: every removed row and changed value, with its reason Pro: opens with a Pro, Premium or Ultimate pass

Reproducible runs: input hashes, versions, seeds and a second run that must agree Pro: opens with a Pro, Premium or Ultimate pass

Practise the original quality contracts

Interview questions

Key takeaways

• Record rows, distinct keys and the total after every step, and let every rule count what it removes; any difference without a rule is unexplained.
• A step that only adds columns, such as a lookup join, must leave rows, keys and totals unchanged.
• An audit trail keeps removed rows and changed values with the step and reason, so one row’s history is a lookup.
• Reproducibility needs the same inputs, versions and seeds, recorded in a run log next to the report.