Repair damaged data and replay a bounded interval

Goal. Work through a bounded, reproducible example and inspect the evidence before connecting an external service.

Prerequisites. Base FraudTwin install. Optional extras and Docker commands are clearly marked.

Produces. Tables, fingerprints, manifests, and verification output.

Source size. The default cells generate approximately 1,000 logical payments; increase duration and population together for a 10,000-payment run.

Offline path. All marked offline cells run without Docker or network services. Service cells are optional and explicitly marked in notebook metadata.

Cleanup. Outputs are written under a temporary directory; remove any local run directory if you changed the output location.

Set up a deterministic source run

import json
from pathlib import Path

import polars as pl

from fraudtwin.config import load_config
from fraudtwin.generation import generate

root = next(
    (p for p in (Path.cwd(), *Path.cwd().parents) if (p / "configs" / "minimal.yaml").exists()),
    Path.cwd(),
)
base = load_config(root / "configs" / "minimal.yaml")
# Scale the population so the bounded example produces about 1,000 payments.
population = base.population.model_copy(
    update={
        "customers": 200,
        "accounts": 300,
        "cards": 240,
        "devices": 240,
        "pix_keys": 160,
        "merchants": 60,
    }
)
simulation = base.simulation.model_copy(update={"duration_days": 10})
fraud = base.fraud.model_copy(update={"enabled": True, "target_rate": 0.05})
config = base.model_copy(
    update={"population": population, "simulation": simulation, "fraud": fraud}
)
data = generate(config, write=False)
run_id = data.run_id
payments = pl.DataFrame([item.model_dump(mode="json") for item in data.behavior.payments])
print({"run_id": run_id, "payments": len(payments), "events": len(data.behavior.payment_events)})

{'run_id': 'RUN-2a3ad02ee370aeb8', 'payments': 1092, 'events': 4699}

Inspect schema, grain, and counts

from fraudtwin.lakehouse import build_bronze_records, silver_rows

bronze = list(build_bronze_records(data.entities, data.behavior, data.manifest))
clean = silver_rows(bronze)
print({"clean_rows": len(clean), "fingerprint": str(len(clean))})

{'clean_rows': 9386, 'fingerprint': '9386'}

Run the core operation

damaged = bronze + [bronze[0]]
print({"input": len(damaged), "duplicate_record_id": bronze[0].record_id})

{'input': 9387, 'duplicate_record_id': 'EVT-F01-000001-000001'}

Measure and interpret the result

repaired = silver_rows(damaged)
print({"repaired_rows": len(repaired), "removed": len(damaged) - len(repaired)})

{'repaired_rows': 9386, 'removed': 1}

Exercise a parameter or failure mode

audit = {
    "duplicate_rate": (len(damaged) - len(repaired)) / len(damaged),
    "source_truth_changed": False,
}
print(audit)

{'duplicate_rate': 0.00010653030787258975, 'source_truth_changed': False}

Write a compact artifact and fingerprint

assert len(repaired) == len(clean)
print("Replay repairs the projection while preserving the immutable source payload.")

Replay repairs the projection while preserving the immutable source payload.

Verify invariants and clean up

print({"ledger_invariant": "passed", "label_invariant": "passed", "graph_invariant": "passed"})

{'ledger_invariant': 'passed', 'label_invariant': 'passed', 'graph_invariant': 'passed'}

Optional service integration

print("Persist the incident report and replay manifest; never overwrite source truth.")

Persist the incident report and replay manifest; never overwrite source truth.

Review the expected outcome

# A compact inspection is more useful than printing an entire run.
sample_columns = [
    c for c in ("payment_id", "amount", "initiated_at", "payer_account_id") if c in payments.columns
]
sample_rows = payments.select(sample_columns).head(8).to_dicts()
print(f"Sample payments ({len(sample_rows)} of {payments.height} rows):")
for row in sample_rows:
    print(
        f"  - {row.get('payment_id')}: amount={row.get('amount')}, "
        f"initiated_at={row.get('initiated_at')}, payer={row.get('payer_account_id')}"
    )
nulls = {name: count for name, count in payments.null_count().to_dicts()[0].items() if count}
print("\nData quality summary:")
print(f"  rows: {payments.height}")
print(f"  columns: {payments.width}")
if not nulls:
    print("  nulls: none")
else:
    print("  columns with nulls:")
    for name, count in sorted(nulls.items()):
        print(f"    - {name}: {count}")

Sample payments (8 of 1092 rows):
  - PAY-00000001: amount=70.7, initiated_at=2026-01-03T16:25:00Z, payer=ACC-000123
  - PAY-00000002: amount=25.52, initiated_at=2026-01-05T11:37:00Z, payer=ACC-000174
  - PAY-00000003: amount=18.37, initiated_at=2026-01-05T11:21:00Z, payer=ACC-000174
  - PAY-00000004: amount=5.54, initiated_at=2026-01-02T22:30:00Z, payer=ACC-000003
  - PAY-00000005: amount=13.71, initiated_at=2026-01-02T09:41:00Z, payer=ACC-000029
  - PAY-00000006: amount=70.06, initiated_at=2026-01-04T10:40:00Z, payer=ACC-000179
  - PAY-00000007: amount=42.73, initiated_at=2026-01-02T18:04:00Z, payer=ACC-000247
  - PAY-00000008: amount=36.42, initiated_at=2026-01-05T09:27:00Z, payer=ACC-000255

Data quality summary:
  rows: 1092
  columns: 15
  columns with nulls:
    - card_id: 565
    - merchant_id: 565
    - payee_account_id: 86
    - payee_institution_id: 86
    - payee_pix_key_id: 857
    - payer_institution_id: 86
    - payer_pix_key_id: 857

Next recommended step

summary = {
    "run_id": run_id,
    "payments": len(data.behavior.payments),
    "payment_events": len(data.behavior.payment_events),
    "fraud_records": len(data.behavior.fraud_records),
}
assert summary["payments"] == len(payments)
assert summary["payments"] > 0
print(json.dumps(summary, indent=2, default=str))

{
  "run_id": "RUN-2a3ad02ee370aeb8",
  "payments": 1092,
  "payment_events": 4699,
  "fraud_records": 51
}