Skip to content

API reference

Auto-generated from docstrings across the datagen and pipeline packages.

Configuration

config

Configuration models for synthetic ERP/CRM data generation.

All generation behavior (volume, RNG seed, and "messiness" ratios) is centralized here as a validated pydantic model so that a run is fully described by a single, serializable object. This makes datasets reproducible and lets configs be checked into version control (as YAML) for repeatable demos and tests.

GenerationConfig

Bases: BaseModel

Top-level configuration for a single dataset generation run.

from_yaml classmethod 

from_yaml(path)

Load a config from a YAML file, falling back to defaults for omitted fields.

Source code in src/datagen/config.py 
92
93
94
95
96
@classmethod
def from_yaml(cls, path: Path) -> GenerationConfig:
    """Load a config from a YAML file, falling back to defaults for omitted fields."""
    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    return cls.model_validate(data)

to_yaml 

to_yaml(path)

Serialize this config to YAML (useful for recording exactly how a dataset was produced).

Source code in src/datagen/config.py 
 98
 99
100
101
102
def to_yaml(self, path: Path) -> None:
    """Serialize this config to YAML (useful for recording exactly how a dataset was produced)."""
    path.write_text(
        yaml.safe_dump(self.model_dump(mode="json"), sort_keys=False), encoding="utf-8"
    )

MessinessConfig

Bases: BaseModel

Ratios (0.0-1.0) controlling how much "bad data" is injected.

Each ratio is applied independently and per export target (ERP vs. CRM), so the two system exports diverge from each other realistically instead of being corrupted identically.

Canonical entities

identities

Canonical ("ground truth") entity models.

These represent the single source of truth for each business entity before it is projected into system-specific representations (ERP CSV vs. CRM JSON) and before cosmetic messiness (bad date formats, missing values, encoding issues) is layered on at export time.

Structural discrepancies that a reconciliation pipeline must detect -- records existing in only one system, or a payment amount that legitimately differs between systems -- are decided here, since they require knowledge of both systems at once. Cosmetic messiness is applied later, per system, in datagen.messiness.

Shared utilities

rng

Seeded random-number-generator utilities.

To keep dataset generation fully reproducible while still letting each component (customers, invoices, payments, per-system messiness, ...) draw independent random streams, we derive child generators from a single master seed via numpy.random.SeedSequence. This avoids the common pitfall of reusing one global RNG (which makes output order-sensitive) while still guaranteeing that a given master seed always reproduces the same dataset.

seeded_uuid4 

seeded_uuid4(rng)

Draw a random-looking UUID (version 4 bit layout) from a seeded RNG.

Using uuid.uuid4() directly would draw from OS randomness and break reproducibility; this instead derives the 128 bits from the caller's seeded generator so that a given master seed always yields identical entity IDs.

Source code in src/datagen/rng.py 
25
26
27
28
29
30
31
32
33
34
35
36
def seeded_uuid4(rng: np.random.Generator) -> UUID:
    """Draw a random-looking UUID (version 4 bit layout) from a seeded RNG.

    Using ``uuid.uuid4()`` directly would draw from OS randomness and break
    reproducibility; this instead derives the 128 bits from the caller's
    seeded generator so that a given master seed always yields identical
    entity IDs.
    """
    raw = bytearray(rng.bytes(16))
    raw[6] = (raw[6] & 0x0F) | 0x40  # version 4
    raw[8] = (raw[8] & 0x3F) | 0x80  # variant 10xx
    return UUID(bytes=bytes(raw))

spawn_rngs 

spawn_rngs(master_seed, names)

Create one independent, reproducible RNG per name from a master seed.

Source code in src/datagen/rng.py 
18
19
20
21
22
def spawn_rngs(master_seed: int, names: list[str]) -> dict[str, np.random.Generator]:
    """Create one independent, reproducible RNG per name from a master seed."""
    seed_sequence = np.random.SeedSequence(master_seed)
    children = seed_sequence.spawn(len(names))
    return {name: np.random.default_rng(child) for name, child in zip(names, children, strict=True)}

keys

Business-facing key formatting.

ERP and CRM systems virtually never agree on ID formatting even when they refer to the same underlying entity (one uses zero-padded numeric codes, another prefixes differently, etc). These helpers derive both systems' business keys from the same underlying sequence_number so a reconciliation pipeline can be built to match them (e.g. by normalizing digits) instead of relying on the internal correlation UUIDs, which are never exported.

Generators

customers

Faker-based synthetic customer generation.

generate_customers 

generate_customers(count, seed)

Generate count unique, deterministic customers for the given seed.

Source code in src/datagen/generators/customers.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def generate_customers(count: int, seed: int) -> list[CanonicalCustomer]:
    """Generate ``count`` unique, deterministic customers for the given seed."""
    faker = Faker()
    faker.seed_instance(seed)
    rng = np.random.default_rng(seed)

    today = date.today()
    seen_emails: set[str] = set()
    customers: list[CanonicalCustomer] = []

    for index in range(1, count + 1):
        first_name = faker.first_name()
        last_name = faker.last_name()
        domain = faker.free_email_domain()
        email = f"{first_name}.{last_name}@{domain}".lower()
        suffix = 1
        while email in seen_emails:
            suffix += 1
            email = f"{first_name}.{last_name}{suffix}@{domain}".lower()
        seen_emails.add(email)

        created_days_ago = int(rng.integers(30, 365 * 3))
        customers.append(
            CanonicalCustomer(
                customer_id=seeded_uuid4(rng),
                sequence_number=index,
                first_name=first_name,
                last_name=last_name,
                email=email,
                phone=faker.phone_number(),
                street=faker.street_address(),
                city=faker.city(),
                region=faker.state(),
                postal_code=faker.postcode(),
                country=faker.country(),
                created_at=today - timedelta(days=created_days_ago),
            )
        )

    return customers

invoices

mimesis-based synthetic invoice generation.

Structural cross-system discrepancies (an invoice existing in only one of ERP/CRM) are decided here since a reconciliation pipeline needs to be able to detect exactly this kind of divergence.

generate_invoices 

generate_invoices(customers, config, seed)

Generate invoices for each customer, per config volume and messiness settings.

Source code in src/datagen/generators/invoices.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def generate_invoices(
    customers: list[CanonicalCustomer], config: GenerationConfig, seed: int
) -> list[CanonicalInvoice]:
    """Generate invoices for each customer, per ``config`` volume and messiness settings."""
    finance = Finance(locale=Locale.EN)
    finance.reseed(seed)
    rng = np.random.default_rng(seed)

    invoices: list[CanonicalInvoice] = []
    sequence_number = 1

    for customer in customers:
        num_invoices = int(
            rng.integers(config.min_invoices_per_customer, config.max_invoices_per_customer + 1)
        )
        for _ in range(num_invoices):
            days_ago = int(rng.integers(1, 365 * 2))
            issue_date = date.today() - timedelta(days=days_ago)
            due_date = issue_date + timedelta(days=int(rng.choice([15, 30, 45, 60])))
            amount = _round_money(float(rng.uniform(50.0, 5000.0)))
            currency = finance.currency_iso_code()
            status_idx = rng.choice(len(_STATUS_CHOICES), p=_STATUS_WEIGHTS)
            status = _STATUS_CHOICES[status_idx]

            exists_in_erp = True
            exists_in_crm = True
            if rng.random() < config.messiness.orphan_ratio:
                if rng.random() < 0.5:
                    exists_in_erp = False
                else:
                    exists_in_crm = False

            invoices.append(
                CanonicalInvoice(
                    invoice_id=seeded_uuid4(rng),
                    customer_id=customer.customer_id,
                    sequence_number=sequence_number,
                    issue_date=issue_date,
                    due_date=due_date,
                    currency=str(currency),
                    amount=amount,
                    status=status,
                    exists_in_erp=exists_in_erp,
                    exists_in_crm=exists_in_crm,
                )
            )
            sequence_number += 1

    return invoices

payments

Synthetic payment generation.

Amount drift between ERP and CRM (representing real-world discrepancies such as bank fees or FX rounding) is decided here since it requires knowledge of both systems' views of the same payment.

generate_payments 

generate_payments(invoices, config, seed)

Generate payments against a subset of invoices, per config settings.

Source code in src/datagen/generators/payments.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def generate_payments(
    invoices: list[CanonicalInvoice], config: GenerationConfig, seed: int
) -> list[CanonicalPayment]:
    """Generate payments against a subset of ``invoices``, per ``config`` settings."""
    rng = np.random.default_rng(seed)

    payments: list[CanonicalPayment] = []
    sequence_number = 1

    for invoice in invoices:
        eligible = invoice.status.value in _PAID_STATUSES
        if not eligible or rng.random() > config.payment_coverage_ratio:
            continue

        is_partial = invoice.status.value == "partial"
        portion = (
            Decimal(str(round(float(rng.uniform(0.3, 0.9)), 2))) if is_partial else Decimal("1")
        )
        amount = _round_money(invoice.amount * portion)

        payment_date = invoice.issue_date + timedelta(days=int(rng.integers(0, 45)))
        method_idx = rng.choice(len(_METHOD_CHOICES), p=_METHOD_WEIGHTS)
        method = _METHOD_CHOICES[method_idx]

        exists_in_erp = invoice.exists_in_erp
        exists_in_crm = invoice.exists_in_crm
        if (
            invoice.exists_in_erp
            and invoice.exists_in_crm
            and rng.random() < config.messiness.orphan_ratio
        ):
            if rng.random() < 0.5:
                exists_in_erp = False
            else:
                exists_in_crm = False

        erp_amount_drift = Decimal("0")
        if rng.random() < config.messiness.amount_drift_ratio:
            drift_magnitude = Decimal(str(round(float(rng.uniform(0.01, 5.0)), 2)))
            erp_amount_drift = drift_magnitude if rng.random() < 0.5 else -drift_magnitude

        payments.append(
            CanonicalPayment(
                payment_id=seeded_uuid4(rng),
                invoice_id=invoice.invoice_id,
                customer_id=invoice.customer_id,
                sequence_number=sequence_number,
                payment_date=payment_date,
                amount=amount,
                method=method,
                exists_in_erp=exists_in_erp,
                exists_in_crm=exists_in_crm,
                erp_amount_drift=erp_amount_drift,
            )
        )
        sequence_number += 1

    return payments

Messiness injectors

dates

Inject inconsistent or invalid date formatting into date columns.

corrupt_dates 

corrupt_dates(df, columns, ratio, rng)

Rewrite a random subset of cells in columns to an inconsistent or invalid date format.

Source code in src/datagen/messiness/dates.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def corrupt_dates(
    df: pd.DataFrame, columns: list[str], ratio: float, rng: np.random.Generator
) -> pd.DataFrame:
    """Rewrite a random subset of cells in ``columns`` to an inconsistent or invalid date format."""
    if ratio <= 0.0:
        return df
    df = df.copy()
    for column in columns:
        if column not in df.columns:
            continue
        mask = rng.random(len(df)) < ratio
        for i in df.index[mask]:
            original = df.at[i, column]
            if pd.isna(original):
                continue
            df.at[i, column] = _corrupt_date_value(original, rng)
    return df

encoding

Inject mixed-encoding / mojibake corruption into text columns.

Simulates the classic real-world bug where UTF-8 encoded text is misinterpreted as a single-byte codepage (e.g. CP-1252/Latin-1), which is a frequent source of "dirty" data when files are exchanged between systems that don't agree on encoding.

corrupt_encoding 

corrupt_encoding(df, columns, ratio, rng)

Corrupt a random subset of cells in columns with mojibake-style encoding artifacts.

Source code in src/datagen/messiness/encoding.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def corrupt_encoding(
    df: pd.DataFrame, columns: list[str], ratio: float, rng: np.random.Generator
) -> pd.DataFrame:
    """Corrupt a random subset of cells in ``columns`` with mojibake-style encoding artifacts."""
    if ratio <= 0.0:
        return df
    df = df.copy()
    for column in columns:
        if column not in df.columns:
            continue
        mask = rng.random(len(df)) < ratio
        for i in df.index[mask]:
            value = df.at[i, column]
            if not isinstance(value, str) or not value:
                continue
            df.at[i, column] = _corrupt_text(value, rng)
    return df

missing

Inject missing/null-like values into otherwise-populated columns.

inject_missing 

inject_missing(df, columns, ratio, rng)

Replace a random subset of cells in columns with a missing-value marker.

Source code in src/datagen/messiness/missing.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def inject_missing(
    df: pd.DataFrame, columns: list[str], ratio: float, rng: np.random.Generator
) -> pd.DataFrame:
    """Replace a random subset of cells in ``columns`` with a missing-value marker."""
    if ratio <= 0.0:
        return df
    df = df.copy()
    for column in columns:
        if column not in df.columns:
            continue
        mask = rng.random(len(df)) < ratio
        for i in df.index[mask]:
            marker = MISSING_MARKERS[int(rng.integers(len(MISSING_MARKERS)))]
            df.at[i, column] = marker
    return df

duplicates

Inject near-duplicate rows into an export.

Real-world duplicate rows are rarely byte-identical (extra whitespace, case differences, re-keyed entries) so a random subset of columns on each duplicate row is lightly perturbed rather than copied verbatim.

add_duplicates 

add_duplicates(df, ratio, rng, exclude_columns=())

Append near-duplicates of a random subset of rows, then shuffle row order.

exclude_columns should list ID/foreign-key columns that must not be perturbed: corrupting a join key would silently orphan the duplicated record (e.g. detaching an invoice from its customer) instead of producing a realistic, detectable duplicate.

Source code in src/datagen/messiness/duplicates.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def add_duplicates(
    df: pd.DataFrame,
    ratio: float,
    rng: np.random.Generator,
    exclude_columns: tuple[str, ...] = (),
) -> pd.DataFrame:
    """Append near-duplicates of a random subset of rows, then shuffle row order.

    ``exclude_columns`` should list ID/foreign-key columns that must not be
    perturbed: corrupting a join key would silently orphan the duplicated
    record (e.g. detaching an invoice from its customer) instead of
    producing a realistic, detectable duplicate.
    """
    if ratio <= 0.0 or df.empty:
        return df.reset_index(drop=True)

    num_duplicates = round(len(df) * ratio)
    if num_duplicates == 0:
        return df.reset_index(drop=True)

    excluded = frozenset(exclude_columns)
    duplicate_positions = rng.choice(len(df), size=num_duplicates, replace=False)
    duplicates = df.iloc[duplicate_positions].apply(
        lambda row: _perturb_duplicate_row(row, rng, excluded), axis=1
    )

    combined = pd.concat([df, duplicates], ignore_index=True)
    shuffled_order = rng.permutation(len(combined))
    return combined.iloc[shuffled_order].reset_index(drop=True)

mismatches

Apply structural cross-system mismatches at export time.

Whether a payment amount should legitimately drift between ERP and CRM is decided during generation (see datagen.generators.payments); this module is where that decision is actually materialized into the ERP export's amount column, since the CRM export always shows the canonical/true amount.

apply_amount_drift 

apply_amount_drift(df, amount_column, drift_column)

Add drift_column into amount_column and drop the now-internal drift column.

Source code in src/datagen/messiness/mismatches.py 
14
15
16
17
18
19
20
def apply_amount_drift(df: pd.DataFrame, amount_column: str, drift_column: str) -> pd.DataFrame:
    """Add ``drift_column`` into ``amount_column`` and drop the now-internal drift column."""
    if drift_column not in df.columns:
        return df
    df = df.copy()
    df[amount_column] = df[amount_column] + df[drift_column]
    return df.drop(columns=[drift_column])

Exporters

erp_csv

ERP-style CSV export (legacy flat, upper-snake-case field names).

export_erp_csv 

export_erp_csv(
    customers, invoices, payments, config, seed, output_dir
)

Write ERP-style customers/invoices/payments CSVs under output_dir/erp.

Source code in src/datagen/exporters/erp_csv.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def export_erp_csv(
    customers: list[CanonicalCustomer],
    invoices: list[CanonicalInvoice],
    payments: list[CanonicalPayment],
    config: GenerationConfig,
    seed: int,
    output_dir: Path,
) -> None:
    """Write ERP-style customers/invoices/payments CSVs under ``output_dir/erp``."""
    rngs = spawn_rngs(seed, ["customers", "invoices", "payments"])
    messiness = config.messiness

    customers_by_id = {str(c.customer_id): c for c in customers}
    invoices_by_id = {str(i.invoice_id): i for i in invoices}

    customers_df = _build_customers_df(customers)
    customers_df = corrupt_encoding(
        customers_df,
        ["CUST_NAME", "ADDR_LINE1", "CITY", "REGION"],
        messiness.encoding_issue_ratio,
        rngs["customers"],
    )
    customers_df = inject_missing(
        customers_df,
        ["PHONE_NUM", "ADDR_LINE1", "REGION", "POSTAL_CD"],
        messiness.missing_value_ratio,
        rngs["customers"],
    )
    customers_df = corrupt_dates(
        customers_df, ["CREATED_DT"], messiness.bad_date_ratio, rngs["customers"]
    )
    customers_df = add_duplicates(
        customers_df, messiness.duplicate_ratio, rngs["customers"], exclude_columns=("CUST_ID",)
    )

    invoices_df = _build_invoices_df(invoices, customers_by_id)
    invoices_df = inject_missing(
        invoices_df, ["CURR_CD", "STATUS_CD"], messiness.missing_value_ratio, rngs["invoices"]
    )
    invoices_df = corrupt_dates(
        invoices_df, ["ISSUE_DT", "DUE_DT"], messiness.bad_date_ratio, rngs["invoices"]
    )
    invoices_df = add_duplicates(
        invoices_df,
        messiness.duplicate_ratio,
        rngs["invoices"],
        exclude_columns=("INV_NO", "CUST_ID"),
    )

    payments_df = _build_payments_df(payments, customers_by_id, invoices_by_id)
    payments_df = apply_amount_drift(payments_df, "AMT", "_AMT_DRIFT")
    payments_df = inject_missing(
        payments_df, ["PMT_METHOD_CD"], messiness.missing_value_ratio, rngs["payments"]
    )
    payments_df = corrupt_dates(payments_df, ["PMT_DT"], messiness.bad_date_ratio, rngs["payments"])
    payments_df = add_duplicates(
        payments_df,
        messiness.duplicate_ratio,
        rngs["payments"],
        exclude_columns=("PMT_NO", "INV_NO", "CUST_ID"),
    )

    erp_dir = output_dir / "erp"
    erp_dir.mkdir(parents=True, exist_ok=True)
    customers_df.to_csv(erp_dir / "customers.csv", index=False)
    invoices_df.to_csv(erp_dir / "invoices.csv", index=False)
    payments_df.to_csv(erp_dir / "payments.csv", index=False)

crm_json

CRM-style nested JSON export (camelCase field names, API-response shape).

export_crm_json 

export_crm_json(
    customers, invoices, payments, config, seed, output_dir
)

Write a CRM-style nested customers.json under output_dir/crm.

Source code in src/datagen/exporters/crm_json.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def export_crm_json(
    customers: list[CanonicalCustomer],
    invoices: list[CanonicalInvoice],
    payments: list[CanonicalPayment],
    config: GenerationConfig,
    seed: int,
    output_dir: Path,
) -> None:
    """Write a CRM-style nested customers.json under ``output_dir/crm``."""
    rngs = spawn_rngs(seed, ["customers", "invoices", "payments"])
    messiness = config.messiness

    customers_by_id = {str(c.customer_id): c for c in customers}
    invoices_by_id = {str(i.invoice_id): i for i in invoices}

    customers_df = _build_customers_df(customers)
    customers_df = corrupt_encoding(
        customers_df,
        ["customerName", "street", "city", "region"],
        messiness.encoding_issue_ratio,
        rngs["customers"],
    )
    customers_df = inject_missing(
        customers_df,
        ["phone", "street", "region", "postalCode"],
        messiness.missing_value_ratio,
        rngs["customers"],
    )
    customers_df = corrupt_dates(
        customers_df, ["createdAt"], messiness.bad_date_ratio, rngs["customers"]
    )
    customers_df = add_duplicates(
        customers_df, messiness.duplicate_ratio, rngs["customers"], exclude_columns=("customerId",)
    )
    # Duplicated customerIds must stay unique keys in the nested JSON tree, so
    # re-key any exact-duplicate rows produced by add_duplicates with a suffix.
    customers_df = _dedupe_join_key(customers_df, "customerId")

    invoices_df = _build_invoices_df(invoices, customers_by_id)
    invoices_df = inject_missing(
        invoices_df, ["currency", "status"], messiness.missing_value_ratio, rngs["invoices"]
    )
    invoices_df = corrupt_dates(
        invoices_df, ["issueDate", "dueDate"], messiness.bad_date_ratio, rngs["invoices"]
    )
    invoices_df = add_duplicates(
        invoices_df,
        messiness.duplicate_ratio,
        rngs["invoices"],
        exclude_columns=("invoiceNumber", "customerId"),
    )
    invoices_df = _dedupe_join_key(invoices_df, "invoiceNumber")

    payments_df = _build_payments_df(payments, invoices_by_id)
    payments_df = inject_missing(
        payments_df, ["method"], messiness.missing_value_ratio, rngs["payments"]
    )
    payments_df = corrupt_dates(
        payments_df, ["paymentDate"], messiness.bad_date_ratio, rngs["payments"]
    )
    payments_df = add_duplicates(
        payments_df,
        messiness.duplicate_ratio,
        rngs["payments"],
        exclude_columns=("invoiceNumber", "paymentNumber"),
    )

    invoice_records = _nest_payments_into_invoices(invoices_df, payments_df)
    customer_records = _nest_invoices_into_customers(customers_df, invoice_records)
    customer_records = _replace_nan_with_none(customer_records)

    crm_dir = output_dir / "crm"
    crm_dir.mkdir(parents=True, exist_ok=True)
    (crm_dir / "customers.json").write_text(
        json.dumps(customer_records, indent=2, default=str, allow_nan=False) + "\n",
        encoding="utf-8",
    )

Ground truth

ground_truth

Ground-truth reconciliation mapping.

Optionally emitted alongside a generated dataset, this file records the actual cross-system relationships the generator created -- which invoices/payments exist in both systems, which are orphaned in only one, and which payments have a legitimate ERP/CRM amount mismatch. A reconciliation pipeline's output can be scored against this file to measure precision and recall, since it captures answers that are otherwise undiscoverable purely by comparing the messy exports.

build_ground_truth 

build_ground_truth(invoices, payments)

Build a JSON-serializable ground-truth reconciliation mapping.

Source code in src/datagen/ground_truth.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def build_ground_truth(
    invoices: list[CanonicalInvoice], payments: list[CanonicalPayment]
) -> dict[str, Any]:
    """Build a JSON-serializable ground-truth reconciliation mapping."""
    invoice_entries = [
        {
            "invoice_id": str(inv.invoice_id),
            "erp_key": erp_invoice_key(inv.sequence_number) if inv.exists_in_erp else None,
            "crm_key": crm_invoice_key(inv.sequence_number) if inv.exists_in_crm else None,
            "exists_in_erp": inv.exists_in_erp,
            "exists_in_crm": inv.exists_in_crm,
            "expected_match": inv.exists_in_erp and inv.exists_in_crm,
        }
        for inv in invoices
    ]

    payment_entries = [
        {
            "payment_id": str(pmt.payment_id),
            "erp_key": erp_payment_key(pmt.sequence_number) if pmt.exists_in_erp else None,
            "crm_key": crm_payment_key(pmt.sequence_number) if pmt.exists_in_crm else None,
            "exists_in_erp": pmt.exists_in_erp,
            "exists_in_crm": pmt.exists_in_crm,
            "expected_match": pmt.exists_in_erp and pmt.exists_in_crm,
            "amount_drift": float(pmt.erp_amount_drift),
            "has_amount_mismatch": pmt.erp_amount_drift != 0,
        }
        for pmt in payments
    ]

    return {
        "invoices": invoice_entries,
        "payments": payment_entries,
        "summary": {
            "total_invoices": len(invoice_entries),
            "invoice_orphans": sum(1 for e in invoice_entries if not e["expected_match"]),
            "total_payments": len(payment_entries),
            "payment_orphans": sum(1 for e in payment_entries if not e["expected_match"]),
            "payment_amount_mismatches": sum(
                1 for e in payment_entries if e["has_amount_mismatch"]
            ),
        },
    }

datagen CLI

cli

Command-line interface for generating synthetic ERP/CRM datasets.

generate 

generate(
    seed=None,
    customers=None,
    min_invoices=None,
    max_invoices=None,
    payment_coverage=None,
    output_dir=None,
    config_file=None,
    with_ground_truth=None,
    missing_value_ratio=None,
    bad_date_ratio=None,
    encoding_issue_ratio=None,
    duplicate_ratio=None,
    orphan_ratio=None,
    amount_drift_ratio=None,
)

Generate a full synthetic ERP + CRM dataset (customers, invoices, payments).

Without --config, generation uses GenerationConfig defaults. Any other flag that is explicitly passed always overrides the base config (whether that base came from defaults or --config), so --config base.yaml --output-dir out2/ behaves as expected.

Source code in src/datagen/cli.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@app.command()
def generate(
    seed: Annotated[
        int | None,
        typer.Option(
            help="Override: master RNG seed; the same seed always reproduces the same dataset."
        ),
    ] = None,
    customers: Annotated[
        int | None, typer.Option(help="Override: number of customers to generate.")
    ] = None,
    min_invoices: Annotated[
        int | None, typer.Option(help="Override: minimum invoices per customer.")
    ] = None,
    max_invoices: Annotated[
        int | None, typer.Option(help="Override: maximum invoices per customer.")
    ] = None,
    payment_coverage: Annotated[
        float | None, typer.Option(help="Override: fraction of invoices that receive a payment.")
    ] = None,
    output_dir: Annotated[
        Path | None,
        typer.Option(
            "--output-dir",
            help="Override: directory to write erp/, crm/, and optional ground truth into.",
        ),
    ] = None,
    config_file: Annotated[
        Path | None,
        typer.Option(
            "--config",
            exists=True,
            help="Base YAML GenerationConfig; any other flag passed overrides it.",
        ),
    ] = None,
    with_ground_truth: Annotated[
        bool | None,
        typer.Option(
            "--with-ground-truth/--no-ground-truth",
            help="Override: also emit ground_truth.json for validating reconciliation results.",
        ),
    ] = None,
    missing_value_ratio: Annotated[
        float | None,
        typer.Option(help="Override: fraction of cells replaced with missing markers."),
    ] = None,
    bad_date_ratio: Annotated[
        float | None,
        typer.Option(help="Override: fraction of dates rewritten inconsistently/invalidly."),
    ] = None,
    encoding_issue_ratio: Annotated[
        float | None, typer.Option(help="Override: fraction of text cells corrupted with mojibake.")
    ] = None,
    duplicate_ratio: Annotated[
        float | None, typer.Option(help="Override: fraction of rows duplicated per export.")
    ] = None,
    orphan_ratio: Annotated[
        float | None,
        typer.Option(help="Override: fraction of invoices/payments existing in only one system."),
    ] = None,
    amount_drift_ratio: Annotated[
        float | None,
        typer.Option(help="Override: fraction of payments with an ERP/CRM amount mismatch."),
    ] = None,
) -> None:
    """Generate a full synthetic ERP + CRM dataset (customers, invoices, payments).

    Without ``--config``, generation uses `GenerationConfig`
    defaults. Any other flag that is explicitly passed always overrides the
    base config (whether that base came from defaults or ``--config``), so
    ``--config base.yaml --output-dir out2/`` behaves as expected.
    """
    config = (
        GenerationConfig.from_yaml(config_file) if config_file is not None else GenerationConfig()
    )

    top_level_overrides = {
        "seed": seed,
        "num_customers": customers,
        "min_invoices_per_customer": min_invoices,
        "max_invoices_per_customer": max_invoices,
        "payment_coverage_ratio": payment_coverage,
        "output_dir": output_dir,
        "with_ground_truth": with_ground_truth,
    }
    active_top_level_overrides = {k: v for k, v in top_level_overrides.items() if v is not None}
    if active_top_level_overrides:
        config = GenerationConfig.model_validate(
            {**config.model_dump(), **active_top_level_overrides}
        )

    messiness_overrides = {
        "missing_value_ratio": missing_value_ratio,
        "bad_date_ratio": bad_date_ratio,
        "encoding_issue_ratio": encoding_issue_ratio,
        "duplicate_ratio": duplicate_ratio,
        "orphan_ratio": orphan_ratio,
        "amount_drift_ratio": amount_drift_ratio,
    }
    active_messiness_overrides = {k: v for k, v in messiness_overrides.items() if v is not None}
    if active_messiness_overrides:
        config.messiness = config.messiness.model_copy(update=active_messiness_overrides)
        config = GenerationConfig.model_validate(config.model_dump())

    with console.status("[bold green]Generating synthetic data..."):
        customer_records = generate_customers(config.num_customers, config.seed)
        invoice_records = generate_invoices(customer_records, config, config.seed)
        payment_records = generate_payments(invoice_records, config, config.seed)

        config.output_dir.mkdir(parents=True, exist_ok=True)
        export_erp_csv(
            customer_records,
            invoice_records,
            payment_records,
            config,
            config.seed,
            config.output_dir,
        )
        export_crm_json(
            customer_records,
            invoice_records,
            payment_records,
            config,
            config.seed,
            config.output_dir,
        )

        if config.with_ground_truth:
            ground_truth = build_ground_truth(invoice_records, payment_records)
            write_ground_truth(ground_truth, config.output_dir / "ground_truth.json")

        config.to_yaml(config.output_dir / "generation_config.yaml")

    table = Table(title="Dataset generated")
    table.add_column("Entity")
    table.add_column("Count", justify="right")
    table.add_row("Customers", str(len(customer_records)))
    table.add_row("Invoices", str(len(invoice_records)))
    table.add_row("Payments", str(len(payment_records)))
    console.print(table)
    console.print(f"[bold]Output directory:[/bold] {config.output_dir.resolve()}")
    console.print("  erp/customers.csv, erp/invoices.csv, erp/payments.csv")
    console.print("  crm/customers.json")
    if config.with_ground_truth:
        console.print("  ground_truth.json")
    console.print("  generation_config.yaml  (records the exact config used, for reproducibility)")

show_config 

show_config(config_file=None)

Print the fully-resolved GenerationConfig as YAML, without generating anything.

Source code in src/datagen/cli.py 
176
177
178
179
180
181
182
183
184
185
186
187
@app.command("show-config")
def show_config(
    config_file: Annotated[
        Path | None,
        typer.Option("--config", exists=True, help="YAML config to resolve instead of defaults."),
    ] = None,
) -> None:
    """Print the fully-resolved GenerationConfig as YAML, without generating anything."""
    config = (
        GenerationConfig.from_yaml(config_file) if config_file is not None else GenerationConfig()
    )
    console.print(yaml.safe_dump(config.model_dump(mode="json"), sort_keys=False))

Pipeline models

models

Target clean-entity schemas, quarantine entries, and the crosswalk.

CleanCustomer / CleanInvoice / CleanPayment describe the post-cleaning record shape: one instance per source-system record (an ERP-side and a CRM-side customer are each validated into this same schema, tagged by source_system), validated before the record is allowed into entity resolution. A field is required here only if reconciliation cannot function without it (per CLAUDE.md's "Cleaning vs. quarantine" missing-values rule); everything else stays optional so a record with a gap still participates in resolution using whatever signal it does have.

CrosswalkEntry

Bases: BaseModel

One row per matched, orphan, or ambiguous entity -- never just the matched ones.

DuplicateMergeLog

Bases: BaseModel

Record of an intra-system near-duplicate cluster collapsed into one canonical row.

ReasonCode

Bases: StrEnum

Quarantine reason codes, one per failure mode in CLAUDE.md's "Resolution outcomes".

Ingestion

ingest

Raw ingestion of ERP CSV and CRM nested-JSON exports.

Reads files into polars DataFrames with their original column names and all-Utf8 dtypes -- no cleaning or type coercion happens here (see pipeline.cleaners for that). Keeping ingestion a pure I/O + flattening step means messy values (bad date formats, missing-value markers, mojibake) survive intact for the cleaning stage to handle explicitly, per CLAUDE.md's "Cleaning vs. quarantine" rule.

ingest_all 

ingest_all(data_dir)

Read both the ERP and CRM exports under data_dir into raw, all-string tables.

Source code in src/pipeline/ingest.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def ingest_all(data_dir: Path) -> RawTables:
    """Read both the ERP and CRM exports under `data_dir` into raw, all-string tables."""
    erp_customers, erp_invoices, erp_payments = read_erp_tables(data_dir / "erp")
    crm_customers, crm_invoices, crm_payments = read_crm_tables(data_dir / "crm")
    return RawTables(
        erp_customers=erp_customers,
        erp_invoices=erp_invoices,
        erp_payments=erp_payments,
        crm_customers=crm_customers,
        crm_invoices=crm_invoices,
        crm_payments=crm_payments,
    )

read_crm_tables 

read_crm_tables(crm_dir)

Read crm/customers.json and flatten it into three flat, all-string DataFrames.

Source code in src/pipeline/ingest.py 
79
80
81
82
83
84
85
86
87
def read_crm_tables(crm_dir: Path) -> tuple[pl.DataFrame, pl.DataFrame, pl.DataFrame]:
    """Read `crm/customers.json` and flatten it into three flat, all-string DataFrames."""
    raw_customers = json.loads((crm_dir / "customers.json").read_text(encoding="utf-8"))
    customer_records, invoice_records, payment_records = _flatten_crm_customers(raw_customers)
    return (
        _to_utf8_df(customer_records),
        _to_utf8_df(invoice_records),
        _to_utf8_df(payment_records),
    )

read_erp_tables 

read_erp_tables(erp_dir)

Read erp/customers.csv, erp/invoices.csv, and erp/payments.csv.

Source code in src/pipeline/ingest.py 
34
35
36
37
38
39
40
def read_erp_tables(erp_dir: Path) -> tuple[pl.DataFrame, pl.DataFrame, pl.DataFrame]:
    """Read `erp/customers.csv`, `erp/invoices.csv`, and `erp/payments.csv`."""
    return (
        _read_erp_csv(erp_dir / "customers.csv"),
        _read_erp_csv(erp_dir / "invoices.csv"),
        _read_erp_csv(erp_dir / "payments.csv"),
    )

Cleaners

duplicates

Collapse intra-system near-duplicate rows, before validation or resolution.

datagen's duplicate injection (see src/datagen/messiness/duplicates.py) never perturbs a duplicated row's ID/foreign-key columns -- only a random other string column, by whitespace-padding it or upper-casing it. The CRM JSON exporter is the one exception: since it nests rows into a tree keyed by that same business key, it re-keys each duplicate with a -DUPn suffix first so it doesn't collide (see datagen.exporters.crm_json._dedupe_join_key). Stripping that suffix before grouping means both the flat ERP CSVs and the nested CRM JSON collapse the same way. This is a separate, earlier problem than cross-system entity resolution in pipeline.resolve: collapsing here first means resolution never has to match against duplicated noise.

collapse_duplicates 

collapse_duplicates(df, key_column)

Collapse rows sharing the same key_column value into one canonical row.

Rows with a null/missing key are left untouched -- they can't be grouped by business key and are handled later as missing-required-field failures.

Source code in src/pipeline/cleaners/duplicates.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def collapse_duplicates(
    df: pl.DataFrame, key_column: str
) -> tuple[pl.DataFrame, list[DuplicateMergeLog]]:
    """Collapse rows sharing the same ``key_column`` value into one canonical row.

    Rows with a null/missing key are left untouched -- they can't be grouped
    by business key and are handled later as missing-required-field failures.
    """
    if df.is_empty() or key_column not in df.columns:
        return df, []

    groups: dict[str, list[dict[str, Any]]] = {}
    order: list[str] = []
    ungrouped: list[dict[str, Any]] = []

    for row in df.to_dicts():
        raw_key = row.get(key_column)
        if raw_key is None:
            ungrouped.append(row)
            continue
        key = _base_key(raw_key)
        if key not in groups:
            groups[key] = []
            order.append(key)
        groups[key].append(row)

    merge_logs: list[DuplicateMergeLog] = []
    merged_rows: list[dict[str, Any]] = []
    for key in order:
        group = groups[key]
        if len(group) == 1:
            merged_rows.append(group[0])
            continue
        canonical, differing_columns = _merge_group(group, key_column, key)
        merged_rows.append(canonical)
        merge_logs.append(
            DuplicateMergeLog(
                key_column=key_column,
                key_value=key,
                row_count=len(group),
                differing_columns=differing_columns,
            )
        )

    collapsed = pl.DataFrame(merged_rows + ungrouped, schema=df.schema)
    return collapsed, merge_logs

missing

Normalize missing-value markers to null.

datagen's missing-value injection (see src/datagen/messiness/missing.py) replaces cells with one of a fixed set of markers: a true null, an empty string, or a sentinel like "N/A". Recognizing all of them here means downstream per-field required/optional handling only ever has to deal with "present" vs. "null" -- not a grab bag of sentinel spellings.

clean_missing_markers 

clean_missing_markers(df, columns)

Normalize missing-value markers to null across columns.

Source code in src/pipeline/cleaners/missing.py 
26
27
28
29
30
31
32
33
34
35
def clean_missing_markers(df: pl.DataFrame, columns: list[str]) -> pl.DataFrame:
    """Normalize missing-value markers to null across ``columns``."""
    exprs = [
        pl.col(column).map_elements(normalize_missing_value, return_dtype=pl.Utf8).alias(column)
        for column in columns
        if column in df.columns
    ]
    if not exprs:
        return df
    return df.with_columns(exprs)

normalize_missing_value 

normalize_missing_value(value)

Map a known missing-value marker to None; pass any other value through.

Source code in src/pipeline/cleaners/missing.py 
19
20
21
22
23
def normalize_missing_value(value: str | None) -> str | None:
    """Map a known missing-value marker to ``None``; pass any other value through."""
    if value is None or value in MISSING_MARKERS:
        return None
    return value

dates

Parse messy date strings to ISO-8601, or flag them as unparseable.

datagen's date corruption (see src/datagen/messiness/dates.py) rewrites a clean ISO date into one of a fixed set of alternate formats, a Unix timestamp, or a known invalid placeholder ("0000-00-00", "TBD", etc.). Trying every one of those formats -- not just a summarized subset -- before giving up is what makes this "cleaning" rather than a validation gate; see CLAUDE.md's "Cleaning vs. quarantine" rule.

Assumes generic missing-value markers (see pipeline.cleaners.missing) have already been normalized to null: a null input here means "legitimately absent", not "invalid" -- only a non-null value that fails every format is flagged as invalid.

clean_date_column 

clean_date_column(df, column)

Parse column to ISO-8601 strings in place; add a {column}_invalid flag column.

A row is flagged invalid when its raw value was non-null but failed to parse under any known format -- that's the signal a later validation stage uses to decide whether to quarantine (see CLAUDE.md's "unparseable_date" reason code).

Source code in src/pipeline/cleaners/dates.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def clean_date_column(df: pl.DataFrame, column: str) -> pl.DataFrame:
    """Parse ``column`` to ISO-8601 strings in place; add a `{column}_invalid` flag column.

    A row is flagged invalid when its raw value was non-null but failed to
    parse under any known format -- that's the signal a later validation
    stage uses to decide whether to quarantine (see CLAUDE.md's
    "unparseable_date" reason code).
    """
    if column not in df.columns:
        return df

    parsed_values: list[str | None] = []
    invalid_flags: list[bool] = []
    for raw in df[column].to_list():
        if raw is None:
            parsed_values.append(None)
            invalid_flags.append(False)
            continue
        parsed = parse_date(raw)
        parsed_values.append(parsed.isoformat() if parsed is not None else None)
        invalid_flags.append(parsed is None)

    return df.with_columns(
        pl.Series(column, parsed_values),
        pl.Series(f"{column}_invalid", invalid_flags),
    )

parse_date 

parse_date(value)

Parse a single date string, trying every format the generator can produce.

Returns None for an unparseable string, including known invalid placeholders like "0000-00-00" or "TBD" (neither matches any known format, so no special-casing is needed for them).

Source code in src/pipeline/cleaners/dates.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def parse_date(value: str) -> date | None:
    """Parse a single date string, trying every format the generator can produce.

    Returns ``None`` for an unparseable string, including known invalid
    placeholders like ``"0000-00-00"`` or ``"TBD"`` (neither matches any
    known format, so no special-casing is needed for them).
    """
    value = value.strip()
    if not value:
        return None

    try:
        return date.fromisoformat(value)
    except ValueError:
        pass

    for fmt in _ALTERNATE_FORMATS:
        try:
            return datetime.strptime(value, fmt).date()
        except ValueError:
            continue

    if value.isdigit():
        try:
            # datagen builds this via datetime(y, m, d).timestamp(), which
            # interprets the naive datetime as local time -- decode the same
            # way (naive/local) so the round trip lands on the same calendar
            # day regardless of which local timezone either side runs in.
            return datetime.fromtimestamp(int(value)).date()
        except (OSError, OverflowError, ValueError):
            return None

    return None

encoding

Repair UTF-8-misread-as-CP-1252 mojibake, or flag it as unrecoverable.

datagen's encoding corruption (see src/datagen/messiness/encoding.py) inserts accented characters into otherwise-ASCII text, then encodes it as UTF-8 and decodes those bytes as CP-1252. Reversing that round trip (encode as CP-1252, decode as UTF-8) repairs it exactly; running the same round trip on text that was never corrupted either no-ops (pure ASCII) or raises (genuinely non-CP-1252-encodable text), so it's a safe repair attempt to always make.

clean_encoding_column 

clean_encoding_column(df, column)

Repair mojibake in column in place; add a {column}_invalid flag column.

A row is flagged invalid when its raw value was non-null but the mojibake repair was unrecoverable -- the signal a later validation stage uses to decide whether to quarantine (see CLAUDE.md's "unrecoverable_encoding" reason code).

Source code in src/pipeline/cleaners/encoding.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def clean_encoding_column(df: pl.DataFrame, column: str) -> pl.DataFrame:
    """Repair mojibake in ``column`` in place; add a `{column}_invalid` flag column.

    A row is flagged invalid when its raw value was non-null but the
    mojibake repair was unrecoverable -- the signal a later validation
    stage uses to decide whether to quarantine (see CLAUDE.md's
    "unrecoverable_encoding" reason code).
    """
    if column not in df.columns:
        return df

    repaired_values: list[str | None] = []
    invalid_flags: list[bool] = []
    for raw in df[column].to_list():
        if raw is None:
            repaired_values.append(None)
            invalid_flags.append(False)
            continue
        repaired = repair_mojibake(raw)
        repaired_values.append(raw if repaired is None else repaired)
        invalid_flags.append(repaired is None)

    return df.with_columns(
        pl.Series(column, repaired_values),
        pl.Series(f"{column}_invalid", invalid_flags),
    )

repair_mojibake 

repair_mojibake(text)

Attempt to reverse CP-1252/UTF-8 mojibake in text.

Returns the repaired text, the original text unchanged if it wasn't mojibake to begin with, or None if it's unrecoverable (the text already contains the Unicode replacement character, or repairing it would produce one).

Source code in src/pipeline/cleaners/encoding.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def repair_mojibake(text: str) -> str | None:
    """Attempt to reverse CP-1252/UTF-8 mojibake in ``text``.

    Returns the repaired text, the original text unchanged if it wasn't
    mojibake to begin with, or ``None`` if it's unrecoverable (the text
    already contains the Unicode replacement character, or repairing it
    would produce one).
    """
    if _REPLACEMENT_CHAR in text:
        return None
    try:
        repaired = text.encode("cp1252").decode("utf-8")
    except (UnicodeDecodeError, UnicodeEncodeError):
        return text
    return None if _REPLACEMENT_CHAR in repaired else repaired

Schema validation

validate

Wire the field-level cleaners into schema-validated records or quarantine entries.

For each source-specific raw table, cleans it (see pipeline.cleaners) and validates every row into the matching Clean* model from pipeline.models, or a QuarantineEntry when it can't be. A row is quarantined for the most specific reason available: a flagged unparseable date or unrecoverable encoding failure from the cleaning stage takes priority over a generic schema-validation failure, since it points at the actual root cause.

Entity resolution

resolve

Entity resolution: match ERP and CRM records into a canonical crosswalk.

Per CLAUDE.md's hard constraint, matching never relies on business-key format shortcuts (CUST_ID vs customerId are cosmetically different encodings of the same sequence number only because this is a synthetic dataset -- a real migration would not have that convenient correlation). Instead:

  • Customers are matched on full_name similarity (rapidfuzz) combined with normalized exact email/phone comparison into one confidence score per candidate pair. datagen.identities.CanonicalCustomer has no exists_in_erp/exists_in_crm split (unlike invoices/payments), so a genuine customer orphan is not a designed feature of the generator -- but it is still handled correctly here rather than assumed away, since a real pipeline can't rely on that.
  • Invoices are matched within an already-resolved customer's own invoices only, on exact (issue_date, currency, amount) equality. There is no legitimate cross-system date lag or amount drift for invoices (only payments drift), so this is an exact join, not a fuzzy score, scoped to the matched customer pair.
  • Payments are matched within an already-resolved invoice's own payments, on exact payment_date equality only. Amount is deliberately drifted for some ERP payments (erp_amount_drift) to simulate bank fees/FX rounding, so amount is never part of the match gate here -- flagging that drift is report.py's job, using the crosswalk this module produces to look the matched pair's amounts back up.

Every entity resolves to exactly one of four ResolutionOutcomes (see CLAUDE.md's "Resolution outcomes"), decided per-record via mutual top-match agreement to avoid many-to-one collisions:

  • Matched -- a record's unique top-scoring candidate (score >= threshold) also has it as its own unique top-scoring candidate: one CrosswalkEntry with both keys populated.
  • Orphan -- zero candidates score >= threshold: a CrosswalkEntry with the missing side null. Never quarantined.
  • Ambiguous -- the record has a candidate >= threshold but the mutual top-match agreement above doesn't hold (multiple tied top candidates on either side, or a non-reciprocal top pick): every record on both sides of that tie is quarantined with ReasonCode.AMBIGUOUS_MATCH, rather than guessing which one is correct.
  • Invalid -- handled upstream by validate.py; this module never sees those records.

The customer fuzzy-match confidence threshold below was calibrated per CLAUDE.md by sweeping eval/score.py data/raw --threshold <value> (500 customers, --with-ground-truth) from 0.1 to 1.0 in increments of 0.05: combined invoice+payment F1 is flat at its maximum (86.6%) across the entire 0.1-0.95 range and only drops (to 85.4%) at 0.99-1.0, where an exact-match-only gate starts rejecting genuine matches with a merely near-identical name. 0.75 sits well inside that plateau, away from that edge.

PairResolution dataclass 

PairResolution(
    matched, orphans_a, orphans_b, ambiguous_a, ambiguous_b
)

Result of resolving one side's records (A) against the other (B).

resolve_customers 

resolve_customers(
    erp_customers,
    crm_customers,
    threshold=CUSTOMER_MATCH_THRESHOLD,
)

Match ERP/CRM customers on name + email/phone confidence.

Returns (crosswalk_entries, quarantine_entries) covering every input customer exactly once (matched, orphan, or ambiguous -- never dropped).

Source code in src/pipeline/resolve.py 
246
247
248
249
250
251
252
253
254
255
256
257
def resolve_customers(
    erp_customers: Sequence[CleanCustomer],
    crm_customers: Sequence[CleanCustomer],
    threshold: float = CUSTOMER_MATCH_THRESHOLD,
) -> tuple[list[CrosswalkEntry], list[QuarantineEntry]]:
    """Match ERP/CRM customers on name + email/phone confidence.

    Returns `(crosswalk_entries, quarantine_entries)` covering every input
    customer exactly once (matched, orphan, or ambiguous -- never dropped).
    """
    result = _resolve_pairs(erp_customers, crm_customers, _customer_confidence, threshold)
    return _pairs_to_crosswalk_and_quarantine(result, EntityType.CUSTOMER, lambda c: c.business_key)

resolve_invoices 

resolve_invoices(
    erp_invoices, crm_invoices, customer_crosswalk
)

Match ERP/CRM invoices, scoped to each already-resolved customer pair.

An invoice whose customer wasn't matched (orphan or ambiguous on the customer side) has no counterpart to compare against, so it resolves straight to orphan -- there's nothing to be ambiguous about without a resolved parent.

Source code in src/pipeline/resolve.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def resolve_invoices(
    erp_invoices: Sequence[CleanInvoice],
    crm_invoices: Sequence[CleanInvoice],
    customer_crosswalk: Sequence[CrosswalkEntry],
) -> tuple[list[CrosswalkEntry], list[QuarantineEntry]]:
    """Match ERP/CRM invoices, scoped to each already-resolved customer pair.

    An invoice whose customer wasn't matched (orphan or ambiguous on the
    customer side) has no counterpart to compare against, so it resolves
    straight to orphan -- there's nothing to be ambiguous *about* without a
    resolved parent.
    """
    matched_customers = {
        entry.erp_key: entry.crm_key
        for entry in customer_crosswalk
        if entry.entity_type == EntityType.CUSTOMER
        and entry.erp_key is not None
        and entry.crm_key is not None
    }

    erp_by_customer: dict[str, list[CleanInvoice]] = defaultdict(list)
    for invoice in erp_invoices:
        erp_by_customer[invoice.customer_business_key].append(invoice)

    crm_by_customer: dict[str, list[CleanInvoice]] = defaultdict(list)
    for invoice in crm_invoices:
        crm_by_customer[invoice.customer_business_key].append(invoice)

    crosswalk: list[CrosswalkEntry] = []
    quarantine: list[QuarantineEntry] = []

    for erp_customer_key, group_erp in erp_by_customer.items():
        crm_customer_key = matched_customers.get(erp_customer_key)
        if crm_customer_key is None:
            for invoice in group_erp:
                crosswalk.append(_crosswalk_entry(EntityType.INVOICE, invoice.business_key, None))
            continue

        group_crm = crm_by_customer.pop(crm_customer_key, [])
        result = _resolve_pairs(group_erp, group_crm, _invoice_exact_match, threshold=1.0)
        group_crosswalk, group_quarantine = _pairs_to_crosswalk_and_quarantine(
            result, EntityType.INVOICE, lambda inv: inv.business_key
        )
        crosswalk.extend(group_crosswalk)
        quarantine.extend(group_quarantine)

    # CRM invoices left over: either their customer was never matched at all,
    # or it was matched but happens to have zero ERP invoices.
    for group_crm in crm_by_customer.values():
        for invoice in group_crm:
            crosswalk.append(_crosswalk_entry(EntityType.INVOICE, None, invoice.business_key))

    return crosswalk, quarantine

resolve_payments 

resolve_payments(
    erp_payments, crm_payments, invoice_crosswalk
)

Match ERP/CRM payments, scoped to each already-resolved invoice pair.

Matches on payment_date equality only -- amount is deliberately drifted for some ERP payments and is never a match gate here (see module docstring); report.py flags drift by looking the matched pair's amounts back up via the crosswalk this returns.

Source code in src/pipeline/resolve.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
def resolve_payments(
    erp_payments: Sequence[CleanPayment],
    crm_payments: Sequence[CleanPayment],
    invoice_crosswalk: Sequence[CrosswalkEntry],
) -> tuple[list[CrosswalkEntry], list[QuarantineEntry]]:
    """Match ERP/CRM payments, scoped to each already-resolved invoice pair.

    Matches on `payment_date` equality only -- amount is deliberately
    drifted for some ERP payments and is never a match gate here (see
    module docstring); report.py flags drift by looking the matched pair's
    amounts back up via the crosswalk this returns.
    """
    matched_invoices = {
        entry.erp_key: entry.crm_key
        for entry in invoice_crosswalk
        if entry.entity_type == EntityType.INVOICE
        and entry.erp_key is not None
        and entry.crm_key is not None
    }

    erp_by_invoice: dict[str, list[CleanPayment]] = defaultdict(list)
    for payment in erp_payments:
        erp_by_invoice[payment.invoice_business_key].append(payment)

    crm_by_invoice: dict[str, list[CleanPayment]] = defaultdict(list)
    for payment in crm_payments:
        crm_by_invoice[payment.invoice_business_key].append(payment)

    crosswalk: list[CrosswalkEntry] = []
    quarantine: list[QuarantineEntry] = []

    for erp_invoice_key, group_erp in erp_by_invoice.items():
        crm_invoice_key = matched_invoices.get(erp_invoice_key)
        if crm_invoice_key is None:
            for payment in group_erp:
                crosswalk.append(_crosswalk_entry(EntityType.PAYMENT, payment.business_key, None))
            continue

        group_crm = crm_by_invoice.pop(crm_invoice_key, [])
        result = _resolve_pairs(group_erp, group_crm, _payment_exact_match, threshold=1.0)
        group_crosswalk, group_quarantine = _pairs_to_crosswalk_and_quarantine(
            result, EntityType.PAYMENT, lambda pmt: pmt.business_key
        )
        crosswalk.extend(group_crosswalk)
        quarantine.extend(group_quarantine)

    for group_crm in crm_by_invoice.values():
        for payment in group_crm:
            crosswalk.append(_crosswalk_entry(EntityType.PAYMENT, None, payment.business_key))

    return crosswalk, quarantine

Crosswalk persistence

crosswalk

Persist the crosswalk table as JSON, with an idempotent upsert.

The crosswalk is keyed on (entity_type, erp_key, crm_key) (see CrosswalkEntry) rather than on canonical_id, since resolve.py mints a fresh UUID on every run. Re-running the pipeline on unchanged input must not duplicate records or reassign existing canonical IDs, so upsert_crosswalk uses a reject-on-conflict strategy: when an incoming entry's key already exists on disk, the existing entry (and its canonical_id) is kept as-is and the incoming one is discarded; only genuinely new keys are appended.

Known limitation (deliberate, not silently worked around): because the key includes the exact erp_key/crm_key shape, a previously-orphaned record whose missing side later gets resolved (e.g. a corrected export fixes a one-sided invoice) has a different key on the next run -- (entity_type, "INV-1", None) vs. (entity_type, "INV-1", "IN-1") are different keys -- and is written as a brand-new entry with a new canonical_id, rather than resolving forward onto the stale orphan entry, which is left in place.

upsert_crosswalk 

upsert_crosswalk(new_entries, path)

Merge new_entries into the crosswalk at path, keyed as described above.

Source code in src/pipeline/crosswalk.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def upsert_crosswalk(new_entries: list[CrosswalkEntry], path: Path) -> list[CrosswalkEntry]:
    """Merge `new_entries` into the crosswalk at `path`, keyed as described above."""
    existing = load_crosswalk(path)
    known_keys = {_key(entry) for entry in existing}

    merged = list(existing)
    for entry in new_entries:
        key = _key(entry)
        if key not in known_keys:
            merged.append(entry)
            known_keys.add(key)

    save_crosswalk(merged, path)
    return merged

Quarantine log

quarantine

Persist quarantined records to quarantine_log.json.

Every row that fails cleaning/validation, or resolves to the ambiguous outcome, must land here -- never be silently dropped (see CLAUDE.md's "Zero silent failures" rule). original_data may retain full PII since this file is the designated place for it; PII masking only applies to console/summary output (see pipeline.cli), not this file.

Postgres target schema

postgres

Load pipeline results into the Postgres target schema (see schema.sql).

This is an additional persistence layer alongside the JSON crosswalk and quarantine log, not a replacement -- pipeline.crosswalk and pipeline.quarantine remain the source of truth that reconcile run reads back on the next invocation. Mirrors their exact idempotency semantics:

  • customers / invoices / payments are upserted on (source_system, business_key) -- each run's clean projection replaces the previous one for that key.
  • crosswalk uses the same reject-on-conflict strategy as pipeline.crosswalk.upsert_crosswalk: a canonical_id is never reassigned once (entity_type, erp_key, crm_key) already exists.
  • quarantine_log is replaced in full on every run, matching pipeline.quarantine.write_quarantine_log's overwrite (not upsert) behavior -- it reflects only the most recent run's failures.

Callers should open the connection as a context manager so a failure partway through a load rolls back instead of leaving a half-written run:

with psycopg.connect(dsn) as conn:
    load_pipeline_result(conn, result)

create_schema 

create_schema(conn)

Create all pipeline tables if they don't already exist.

Source code in src/pipeline/postgres.py 
47
48
49
50
def create_schema(conn: Connection) -> None:
    """Create all pipeline tables if they don't already exist."""
    with conn.cursor() as cur:
        cur.execute(SCHEMA_SQL)

load_crosswalk 

load_crosswalk(conn, entries)

Upsert entries, preserving the existing canonical_id on a key collision.

Source code in src/pipeline/postgres.py 
118
119
120
121
122
123
124
125
126
127
128
129
def load_crosswalk(conn: Connection, entries: list[CrosswalkEntry]) -> None:
    """Upsert `entries`, preserving the existing `canonical_id` on a key collision."""
    with conn.cursor() as cur:
        for entry in entries:
            cur.execute(
                """
                INSERT INTO crosswalk (entity_type, canonical_id, erp_key, crm_key)
                VALUES (%(entity_type)s, %(canonical_id)s, %(erp_key)s, %(crm_key)s)
                ON CONFLICT (entity_type, erp_key, crm_key) DO NOTHING
                """,
                entry.model_dump(mode="json"),
            )

load_pipeline_result 

load_pipeline_result(conn, result)

Create the schema (if needed) and load a full pipeline run into Postgres.

Does not commit -- open conn as a context manager (with psycopg.connect(dsn) as conn:) so the whole run commits or rolls back atomically.

Source code in src/pipeline/postgres.py 
155
156
157
158
159
160
161
162
163
164
165
166
def load_pipeline_result(conn: Connection, result: PipelineResult) -> None:
    """Create the schema (if needed) and load a full pipeline run into Postgres.

    Does not commit -- open `conn` as a context manager (`with psycopg.connect(dsn) as
    conn:`) so the whole run commits or rolls back atomically.
    """
    create_schema(conn)
    load_customers(conn, result.erp_customers + result.crm_customers)
    load_invoices(conn, result.erp_invoices + result.crm_invoices)
    load_payments(conn, result.erp_payments + result.crm_payments)
    load_crosswalk(conn, result.crosswalk)
    replace_quarantine(conn, result.quarantine)

replace_quarantine 

replace_quarantine(conn, entries)

Replace the entire quarantine_log table with entries.

Source code in src/pipeline/postgres.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def replace_quarantine(conn: Connection, entries: list[QuarantineEntry]) -> None:
    """Replace the entire `quarantine_log` table with `entries`."""
    with conn.cursor() as cur:
        cur.execute("DELETE FROM quarantine_log")
        for entry in entries:
            payload: dict[str, Any] = entry.model_dump(mode="json")
            cur.execute(
                """
                INSERT INTO quarantine_log (entity_type, source_system, reason_code, stage,
                    original_data)
                VALUES (%(entity_type)s, %(source_system)s, %(reason_code)s, %(stage)s,
                    %(original_data)s)
                """,
                payload | {"original_data": _as_jsonb(payload["original_data"])},
            )

Reconciliation report

report

Build the Markdown reconciliation report from pipeline outputs.

Precision/recall against ground_truth.json is never computed here -- per CLAUDE.md's ground_truth.json eval-only rule, only eval/score.py (or tests) may read that file. When available, the caller passes in an already-computed PrecisionRecall to embed in the report; this module never imports or references ground_truth.json itself.

IngestCounts dataclass 

IngestCounts(
    erp_customers,
    erp_invoices,
    erp_payments,
    crm_customers,
    crm_invoices,
    crm_payments,
)

Raw row counts per source system, before cleaning/validation.

PrecisionRecall dataclass 

PrecisionRecall(precision, recall, f1)

Computed by eval/score.py against ground_truth.json -- never by this module.

compute_payment_drift 

compute_payment_drift(
    crosswalk, erp_payments, crm_payments
)

Flag matched payment pairs whose amounts disagree beyond the rounding epsilon.

Source code in src/pipeline/report.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def compute_payment_drift(
    crosswalk: list[CrosswalkEntry],
    erp_payments: list[CleanPayment],
    crm_payments: list[CleanPayment],
) -> list[PaymentDrift]:
    """Flag matched payment pairs whose amounts disagree beyond the rounding epsilon."""
    erp_by_key = {payment.business_key: payment for payment in erp_payments}
    crm_by_key = {payment.business_key: payment for payment in crm_payments}

    drifts: list[PaymentDrift] = []
    for entry in crosswalk:
        if (
            entry.entity_type != EntityType.PAYMENT
            or entry.erp_key is None
            or entry.crm_key is None
        ):
            continue
        erp = erp_by_key.get(entry.erp_key)
        crm = crm_by_key.get(entry.crm_key)
        if erp is None or crm is None:
            continue
        if abs(erp.amount - crm.amount) > AMOUNT_DRIFT_THRESHOLD:
            drifts.append(PaymentDrift(entry.erp_key, entry.crm_key, erp.amount, crm.amount))
    return drifts

Orchestration

orchestrate

End-to-end pipeline orchestration: ingest -> dedupe -> validate -> resolve.

Both eval/score.py and pipeline.cli need to run this exact same sequence of stages over a data directory, so it's factored out here once rather than duplicated between them.

run_pipeline 

run_pipeline(
    data_dir,
    customer_match_threshold=CUSTOMER_MATCH_THRESHOLD,
)

Run ingestion, intra-system dedup, cleaning/validation, and resolution over data_dir.

customer_match_threshold is exposed here (rather than hardcoded) so eval/score.py can sweep values for offline calibration.

Source code in src/pipeline/orchestrate.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def run_pipeline(
    data_dir: Path, customer_match_threshold: float = CUSTOMER_MATCH_THRESHOLD
) -> PipelineResult:
    """Run ingestion, intra-system dedup, cleaning/validation, and resolution over `data_dir`.

    `customer_match_threshold` is exposed here (rather than hardcoded) so
    `eval/score.py` can sweep values for offline calibration.
    """
    raw = ingest_all(data_dir)

    ingest_counts = IngestCounts(
        erp_customers=raw.erp_customers.height,
        erp_invoices=raw.erp_invoices.height,
        erp_payments=raw.erp_payments.height,
        crm_customers=raw.crm_customers.height,
        crm_invoices=raw.crm_invoices.height,
        crm_payments=raw.crm_payments.height,
    )

    duplicate_logs: list[DuplicateMergeLog] = []

    def dedupe(df: pl.DataFrame, key_column: str) -> pl.DataFrame:
        deduped, logs = collapse_duplicates(df, key_column)
        duplicate_logs.extend(logs)
        return deduped

    erp_customers_df = dedupe(raw.erp_customers, CUSTOMER_COLUMNS[SourceSystem.ERP]["business_key"])
    crm_customers_df = dedupe(raw.crm_customers, CUSTOMER_COLUMNS[SourceSystem.CRM]["business_key"])
    erp_invoices_df = dedupe(raw.erp_invoices, INVOICE_COLUMNS[SourceSystem.ERP]["business_key"])
    crm_invoices_df = dedupe(raw.crm_invoices, INVOICE_COLUMNS[SourceSystem.CRM]["business_key"])
    erp_payments_df = dedupe(raw.erp_payments, PAYMENT_COLUMNS[SourceSystem.ERP]["business_key"])
    crm_payments_df = dedupe(raw.crm_payments, PAYMENT_COLUMNS[SourceSystem.CRM]["business_key"])

    quarantine: list[QuarantineEntry] = []

    erp_customers, quarantined = validate_customers(erp_customers_df, SourceSystem.ERP)
    quarantine.extend(quarantined)
    crm_customers, quarantined = validate_customers(crm_customers_df, SourceSystem.CRM)
    quarantine.extend(quarantined)
    erp_invoices, quarantined = validate_invoices(erp_invoices_df, SourceSystem.ERP)
    quarantine.extend(quarantined)
    crm_invoices, quarantined = validate_invoices(crm_invoices_df, SourceSystem.CRM)
    quarantine.extend(quarantined)
    erp_payments, quarantined = validate_payments(erp_payments_df, SourceSystem.ERP)
    quarantine.extend(quarantined)
    crm_payments, quarantined = validate_payments(crm_payments_df, SourceSystem.CRM)
    quarantine.extend(quarantined)

    customer_crosswalk, quarantined = resolve_customers(
        erp_customers, crm_customers, threshold=customer_match_threshold
    )
    quarantine.extend(quarantined)
    invoice_crosswalk, quarantined = resolve_invoices(
        erp_invoices, crm_invoices, customer_crosswalk
    )
    quarantine.extend(quarantined)
    payment_crosswalk, quarantined = resolve_payments(erp_payments, crm_payments, invoice_crosswalk)
    quarantine.extend(quarantined)

    return PipelineResult(
        ingest_counts=ingest_counts,
        duplicate_logs=duplicate_logs,
        quarantine=quarantine,
        crosswalk=customer_crosswalk + invoice_crosswalk + payment_crosswalk,
        erp_customers=erp_customers,
        crm_customers=crm_customers,
        erp_invoices=erp_invoices,
        crm_invoices=crm_invoices,
        erp_payments=erp_payments,
        crm_payments=crm_payments,
    )

reconcile CLI

cli

Command-line interface for running the ERP/CRM reconciliation pipeline.

This CLI never reads ground_truth.json -- per CLAUDE.md's eval-only rule, that file is only ever consulted by eval/score.py or tests. Precision/ recall is therefore never part of this CLI's report; run eval/score.py separately against a dataset generated with --with-ground-truth for that.

run 

run(
    data_dir,
    crosswalk_path=Path("data/processed/crosswalk.json"),
    quarantine_path=Path(
        "data/processed/quarantine_log.json"
    ),
    duplicate_log_path=Path(
        "data/processed/duplicate_log.json"
    ),
    report_path=Path("RECONCILIATION_REPORT.md"),
    customer_threshold=CUSTOMER_MATCH_THRESHOLD,
    postgres_dsn=None,
)

Run ingest -> dedupe -> clean -> resolve over DATA_DIR and write the crosswalk/report.

Source code in src/pipeline/cli.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@app.command()
def run(
    data_dir: Annotated[
        Path,
        typer.Argument(
            exists=True,
            file_okay=False,
            help="Directory containing erp/ (CSV) and crm/ (JSON) subdirectories.",
        ),
    ],
    crosswalk_path: Annotated[
        Path,
        typer.Option(help="Where to upsert the crosswalk JSON (existing entries are preserved)."),
    ] = Path("data/processed/crosswalk.json"),
    quarantine_path: Annotated[
        Path, typer.Option(help="Where to write the quarantine log JSON.")
    ] = Path("data/processed/quarantine_log.json"),
    duplicate_log_path: Annotated[
        Path, typer.Option(help="Where to write the intra-system duplicate-merge log JSON.")
    ] = Path("data/processed/duplicate_log.json"),
    report_path: Annotated[
        Path, typer.Option(help="Where to write the Markdown reconciliation report.")
    ] = Path("RECONCILIATION_REPORT.md"),
    customer_threshold: Annotated[
        float, typer.Option(help="Confidence threshold for customer entity resolution.")
    ] = CUSTOMER_MATCH_THRESHOLD,
    postgres_dsn: Annotated[
        str | None,
        typer.Option(
            help="If set, also load the run into this Postgres DSN (see pipeline.postgres)."
        ),
    ] = None,
) -> None:
    """Run ingest -> dedupe -> clean -> resolve over DATA_DIR and write the crosswalk/report."""
    start = time.perf_counter()
    result = run_pipeline(data_dir, customer_match_threshold=customer_threshold)
    execution_time_seconds = time.perf_counter() - start

    merged_crosswalk = upsert_crosswalk(result.crosswalk, crosswalk_path)
    write_quarantine_log(result.quarantine, quarantine_path)
    write_duplicate_log(result.duplicate_logs, duplicate_log_path)

    report = render_report(
        ingest_counts=result.ingest_counts,
        crosswalk=result.crosswalk,
        quarantine=result.quarantine,
        erp_payments=result.erp_payments,
        crm_payments=result.crm_payments,
        duplicate_logs=result.duplicate_logs,
        execution_time_seconds=execution_time_seconds,
    )
    write_report(report, report_path)

    _print_summary(
        result,
        merged_crosswalk,
        report_path,
        crosswalk_path,
        quarantine_path,
        duplicate_log_path,
        execution_time_seconds,
    )

    if postgres_dsn is not None:
        _load_postgres(result, postgres_dsn)

Evaluation against ground truth

score

Score a pipeline run's crosswalk against ground_truth.json.

This is the ONLY module allowed to read ground_truth.json -- per CLAUDE.md's hard constraint, no module under src/pipeline/ may import, read, or reference it. This script runs the pipeline itself (via pipeline.orchestrate.run_pipeline) and scores the resulting crosswalk after the fact; the pipeline never consults the answer key at runtime.

ground_truth.json only covers invoices and payments -- customers have no exists_in_erp/exists_in_crm split in the generator, so there's nothing to score customer resolution against.

Usage::

uv run python eval/score.py path/to/data_dir
uv run python eval/score.py path/to/data_dir --threshold 0.8

The --threshold flag sweeps resolve_customers' confidence threshold for offline calibration (see pipeline.resolve.CUSTOMER_MATCH_THRESHOLD's docstring) -- it does not read ground_truth.json any differently, it just re-runs the pipeline with a different threshold before scoring.

score_crosswalk 

score_crosswalk(crosswalk, ground_truth)

Combined precision/recall across invoices and payments (all ground_truth.json covers).

Source code in eval/score.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def score_crosswalk(
    crosswalk: list[CrosswalkEntry], ground_truth: dict[str, Any]
) -> PrecisionRecall:
    """Combined precision/recall across invoices and payments (all `ground_truth.json` covers)."""
    expected = _ground_truth_pairs(
        EntityType.INVOICE, ground_truth["invoices"]
    ) | _ground_truth_pairs(EntityType.PAYMENT, ground_truth["payments"])
    predicted = _predicted_pairs(crosswalk, EntityType.INVOICE) | _predicted_pairs(
        crosswalk, EntityType.PAYMENT
    )
    return _precision_recall(predicted, expected)