Resume a scale run from a checkpoint#

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

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("Generated source run")
print(f"  run id: {run_id}")
print(f"  payments: {len(payments):,}")
print(f"  events: {len(data.behavior.payment_events):,}")

Generated source run
  run id: RUN-2a3ad02ee370aeb8
  payments: 1,092
  events: 4,699

Inspect the resolved scale plan

from fraudtwin.config import load_config
from fraudtwin.scale import require_scale_plan, shard_descriptors

scale_config = load_config(root / "configs" / "scale-dev.yaml")
plan = require_scale_plan(scale_config)
print("Scale plan")
print(f"  profile: {plan.profile}")
print(f"  target payments: {plan.target_payments:,}")
print(f"  shards: {plan.shard_count}  |  workers: {plan.worker_count}")
print(f"  chunk size: {plan.chunk_size:,} payments")
print(f"  checkpoint: every {plan.checkpoint_frequency_chunks} chunk(s)")
print(f"  partitioning: {plan.partition_mapping}")
print(f"  seed: {plan.seed}")
print(f"  configuration: {plan.configuration_hash[:12]}...")

Scale plan
  profile: dev
  target payments: 1,000
  shards: 2  |  workers: 2
  chunk size: 100 payments
  checkpoint: every 1 chunk(s)
  partitioning: stable_hash_v1
  seed: 42
  configuration: 74d79c06b059...

Review the deterministic shard layout

shards = shard_descriptors(plan)
display(pl.DataFrame([shard.model_dump(mode="json") for shard in shards]))
shape: (2, 3)
shard_indexshard_idmapping
i64strstr
0"SHARD-000000""stable_hash_v1"
1"SHARD-000001""stable_hash_v1"

Measure and interpret the result

from fraudtwin.scale import fingerprint_rows

rows = [{"logical_id": f"payment-{i:04d}", "amount": float(i + 1)} for i in range(20)]
full = fingerprint_rows(rows)
partial = fingerprint_rows(rows[:10])
print("Checkpoint fingerprints")
print(f"  partial checkpoint (10 rows): {partial[:12]}...")
print(f"  complete run (20 rows):       {full[:12]}...")

Checkpoint fingerprints
  partial checkpoint (10 rows): a88a36d486d7...
  complete run (20 rows):       566268e96f11...

Exercise a parameter or failure mode

resumed = rows[:10] + rows[10:]
assert fingerprint_rows(resumed) == full
print("Resume check")
print("  status: deterministic")
print(f"  reconciled rows: {len({r["logical_id"] for r in resumed}):,}")

Resume check
  status: deterministic
  reconciled rows: 20

Write a compact artifact and fingerprint

Checkpoint files should live in a temporary run directory and be retained only for recovery.

Verify invariants and clean up

# 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

Optional service integration

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("Generated dataset")
print(f"  run id: {summary['run_id']}")
print(f"  payments: {summary['payments']:,}")
print(f"  payment events: {summary['payment_events']:,}")
print(f"  fraud records: {summary['fraud_records']:,}")

Generated dataset
  run id: RUN-2a3ad02ee370aeb8
  payments: 1,092
  payment events: 4,699
  fraud records: 51

Record the generated shape and tutorial contract.#

summary = {
    "payments": len(data.behavior.payments),
    "events": len(data.behavior.payment_events),
}
assert summary["payments"] >= 0
print("Tutorial contract")
print(f"  payments: {summary['payments']:,}")
print(f"  events: {summary['events']:,}")

Tutorial contract
  payments: 1,092
  events: 4,699

Record the generated shape and tutorial contract.#

summary = {
    "payments": len(data.behavior.payments),
    "events": len(data.behavior.payment_events),
}
assert summary["payments"] >= 0
print("Tutorial contract")
print(f"  payments: {summary['payments']:,}")
print(f"  events: {summary['events']:,}")

Tutorial contract
  payments: 1,092
  events: 4,699

assert summary["events"] >= summary["payments"]
print("Checkpoint output contract passed")