Scale benchmarking, payment reconciliation, and experiment packaging#
Goal. Plan a bounded scale run, reconcile multi-rail payment projections, and package a complete reproducible experiment. This is the operational hand-off from a synthetic run to an auditable benchmark.
Audience. Data/ML scientists, fraud analysts, platform engineers, and researchers.
Prerequisites. Python 3.12+, a clean checkout, and the base FraudTwin install. The workflow is deterministic and runs offline; service integrations are deliberately out of scope here.
Source size. 1,000–10,000 logical payments. Every section writes only compact summaries, manifests, or fingerprints to a temporary directory.
Interpretation. Synthetic evidence demonstrates mechanics and invariants, not production prevalence or model performance guarantees.
Scale and resource benchmarking#
from pathlib import Path
import pandas as pd
from fraudtwin.config import load_config
from fraudtwin.reproducibility import sha256_json
from fraudtwin.scale import require_scale_plan, shard_descriptors
root = next(
(p for p in (Path.cwd(), *Path.cwd().parents) if (p / "configs" / "minimal.yaml").is_file()),
Path.cwd(),
)
config = load_config(root / "configs" / "scale-dev.yaml")
plan = require_scale_plan(config, run_id="BENCH-SCALE-DEV")
shards = shard_descriptors(plan)
print(
{
"profile": plan.profile,
"target_payments": plan.target_payments,
"shards": plan.shard_count,
"chunk_size": plan.chunk_size,
}
)
assert plan.target_payments >= 1000
{'profile': 'dev', 'target_payments': 1000, 'shards': 2, 'chunk_size': 100}
rows = [{"shard_id": x.shard_id, "index": x.shard_index, "mapping": x.mapping} for x in shards]
print(rows)
assert [x["index"] for x in rows] == list(range(plan.shard_count))
[{'shard_id': 'SHARD-000000', 'index': 0, 'mapping': 'stable_hash_v1'}, {'shard_id': 'SHARD-000001', 'index': 1, 'mapping': 'stable_hash_v1'}]
chunk_count = (plan.target_payments + plan.chunk_size - 1) // plan.chunk_size
print({"chunks": chunk_count, "checkpoint_frequency": plan.checkpoint_frequency_chunks})
assert chunk_count > 0
{'chunks': 10, 'checkpoint_frequency': 1}
profiles = {
name: {"target": target, "chunks": (target + plan.chunk_size - 1) // plan.chunk_size}
for name, target in (("dev", 1000), ("small", 100000), ("medium", 1000000))
}
print(profiles)
{'dev': {'target': 1000, 'chunks': 10}, 'small': {'target': 100000, 'chunks': 1000}, 'medium': {'target': 1000000, 'chunks': 10000}}
checkpoint = {
"run_id": plan.run_id,
"configuration_hash": plan.configuration_hash,
"completed_chunks": tuple(range(min(3, chunk_count))),
}
print(checkpoint)
assert max(checkpoint["completed_chunks"], default=-1) < chunk_count
{'run_id': 'BENCH-SCALE-DEV', 'configuration_hash': '74d79c06b059231b3b45338b65cf6dcec810dab32cb937f67cf0503ea25d98c4', 'completed_chunks': (0, 1, 2)}
remaining = chunk_count - len(checkpoint["completed_chunks"])
print({"completed": len(checkpoint["completed_chunks"]), "remaining": remaining})
assert remaining >= 0
{'completed': 3, 'remaining': 7}
logical_ids = tuple(f"PAY-{i:08d}" for i in range(plan.target_payments))
print({"logical_ids": len(logical_ids), "duplicates": len(logical_ids) - len(set(logical_ids))})
assert len(logical_ids) == len(set(logical_ids))
{'logical_ids': 1000, 'duplicates': 0}
manifest = {
"profile": plan.profile,
"target": plan.target_payments,
"shards": plan.shard_count,
"chunk_size": plan.chunk_size,
"completed": checkpoint["completed_chunks"],
"remaining": remaining,
}
manifest["fingerprint"] = sha256_json(manifest)
print(
{
"profile": manifest["profile"],
"target": manifest["target"],
"shards": manifest["shards"],
"completed": len(manifest["completed"]),
"remaining": manifest["remaining"],
"fingerprint": f"{manifest['fingerprint'][:12]}...",
}
)
{'profile': 'dev', 'target': 1000, 'shards': 2, 'completed': 3, 'remaining': 7, 'fingerprint': 'f33144a59d83...'}
same = require_scale_plan(config, run_id="BENCH-SCALE-DEV")
print({"same_configuration": same.configuration_hash == plan.configuration_hash})
assert same.configuration_hash == plan.configuration_hash
{'same_configuration': True}
assert manifest["fingerprint"] == sha256_json(
{k: v for k, v in manifest.items() if k != "fingerprint"}
)
print("The scale plan makes resume and reconciliation measurable before a large run is scheduled.")
The scale plan makes resume and reconciliation measurable before a large run is scheduled.
Multi-rail payment reconciliation#
from collections import Counter, defaultdict
from pathlib import Path
from fraudtwin.config import load_config
from fraudtwin.generation import generate
from fraudtwin.reproducibility import sha256_json
base = load_config(root / "configs" / "minimal.yaml")
config = base.model_copy(
update={"payments": base.payments.model_copy(update={"daily_target": 1000})}
)
data = generate(config, write=False)
payments = data.behavior.payments
events = data.behavior.payment_events
ledger = data.behavior.ledger_entries
print(
{"run_id": data.run_id, "payments": len(payments), "events": len(events), "ledger": len(ledger)}
)
assert len(payments) >= 1000
{'run_id': 'RUN-f0981dbfd03c1bf5', 'payments': 1000, 'events': 4506, 'ledger': 1914}
rail_counts = Counter(p.payment_rail for p in payments)
print(dict(rail_counts))
assert set(rail_counts) == {"CARD", "PIX", "ACCOUNT_TRANSFER"}
{'ACCOUNT_TRANSFER': 262, 'PIX': 130, 'CARD': 608}
event_counts = Counter(e.payment_rail for e in events)
print(dict(event_counts))
assert sum(event_counts.values()) == len(events)
{'ACCOUNT_TRANSFER': 262, 'PIX': 776, 'CARD': 3468}
status_counts = Counter(p.current_status for p in payments)
print(dict(status_counts))
assert status_counts
{'COMPLETED': 262, 'RECEIVED': 117, 'SETTLED': 476, 'REJECTED': 6, 'DECLINED': 61, 'REVERSED': 27, 'REFUNDED': 44, 'RETURNED': 7}
ledger_by_payment = defaultdict(float)
for e in ledger:
ledger_by_payment[e.payment_id] += e.amount if e.entry_type == "CREDIT" else -e.amount
matched = sum(p.payment_id in ledger_by_payment for p in payments)
print({"payments_with_ledger": matched, "coverage": round(matched / len(payments), 3)})
assert matched > 0
{'payments_with_ledger': 906, 'coverage': 0.906}
debit = round(sum(e.amount for e in ledger if e.entry_type == "DEBIT"), 2)
credit = round(sum(e.amount for e in ledger if e.entry_type == "CREDIT"), 2)
print({"debit_total": debit, "credit_total": credit, "difference": round(debit - credit, 2)})
{'debit_total': 18896.97, 'credit_total': 18896.97, 'difference': 0.0}
event_ids = {e.payment_id for e in events}
payment_ids = {p.payment_id for p in payments}
print(
{
"events_without_payment": len(event_ids - payment_ids),
"payments_without_events": len(payment_ids - event_ids),
}
)
assert event_ids <= payment_ids
{'events_without_payment': 0, 'payments_without_events': 0}
special = Counter(
e.event_type
for e in events
if any(x in e.event_type for x in ("REFUND", "RETURN", "REVERSAL", "CHARGEBACK"))
)
print({"special_lifecycle_events": dict(special)})
{'special_lifecycle_events': {'CARD_REFUNDED': 44, 'PIX_RETURN_REQUESTED': 7, 'PIX_RETURNED': 7}}
reconciliation = {
"rail_counts": dict(rail_counts),
"event_counts": dict(event_counts),
"payments": len(payments),
"ledger": len(ledger),
"ledger_coverage": matched / len(payments),
}
reconciliation["fingerprint"] = sha256_json(reconciliation)
pd.DataFrame(reconciliation)
| rail_counts | event_counts | payments | ledger | ledger_coverage | fingerprint | |
|---|---|---|---|---|---|---|
| ACCOUNT_TRANSFER | 262 | 262 | 1000 | 1914 | 0.906 | 158e03d77760e28f36d13d0f6cb375935f0a40c37e6b32... |
| PIX | 130 | 776 | 1000 | 1914 | 0.906 | 158e03d77760e28f36d13d0f6cb375935f0a40c37e6b32... |
| CARD | 608 | 3468 | 1000 | 1914 | 0.906 | 158e03d77760e28f36d13d0f6cb375935f0a40c37e6b32... |
assert reconciliation["fingerprint"] == sha256_json(
{k: v for k, v in reconciliation.items() if k != "fingerprint"}
)
assert reconciliation["ledger_coverage"] > 0
print("Lifecycle reconciliation uses domain IDs and does not rewrite source records.")
Lifecycle reconciliation uses domain IDs and does not rewrite source records.
Reproducible experiment packaging#
import json
from pathlib import Path
from tempfile import TemporaryDirectory
from fraudtwin import __version__
from fraudtwin.config import load_config
from fraudtwin.generation import generate
from fraudtwin.reproducibility import sha256_json
base = load_config(root / "configs" / "minimal.yaml")
config = base.model_copy(
update={"payments": base.payments.model_copy(update={"daily_target": 1000})}
)
data = generate(config, write=False)
counts = {
"payments": len(data.behavior.payments),
"events": len(data.behavior.payment_events),
"ledger": len(data.behavior.ledger_entries),
"dataset": data.require_dataset().frame.height,
}
print({"version": __version__, "run_id": data.run_id, **counts})
assert counts["payments"] >= 1000
{'version': '0.34.0', 'run_id': 'RUN-f0981dbfd03c1bf5', 'payments': 1000, 'events': 4506, 'ledger': 1914, 'dataset': 928}
schema_versions = {
"events": sorted({e.schema_version for e in data.behavior.payment_events}),
"manifest": data.manifest.schema_versions,
}
assert schema_versions["events"]
print(schema_versions["events"])
pd.Series(schema_versions["manifest"])
['5']
customers 1
institutions 1
accounts 1
cards 1
merchants 1
devices 1
pix_keys 1
behavior_profiles 1
payments 2
payment_events 5
ledger_entries 1
fraud_records 1
fraud_alerts 1
fraud_cases 1
case_confirmations 1
customer_disputes 1
fraud_labels 1
dtype: object
identity = {
"run_id": data.run_id,
"seed": config.simulation.seed,
"configuration_hash": data.manifest.scenario_config_hash,
"counts": counts,
"schema_versions": schema_versions,
}
identity["data_fingerprint"] = sha256_json(
{
"payments": tuple(p.payment_id for p in data.behavior.payments),
"events": tuple(e.event_id for e in data.behavior.payment_events),
}
)
pd.DataFrame(identity)
| run_id | seed | configuration_hash | counts | schema_versions | data_fingerprint | |
|---|---|---|---|---|---|---|
| payments | RUN-f0981dbfd03c1bf5 | 42 | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 1000.0 | NaN | 6a0d08f1596918f9f6672a6857124907fb14305605d435... |
| events | RUN-f0981dbfd03c1bf5 | 42 | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 4506.0 | [5] | 6a0d08f1596918f9f6672a6857124907fb14305605d435... |
| ledger | RUN-f0981dbfd03c1bf5 | 42 | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 1914.0 | NaN | 6a0d08f1596918f9f6672a6857124907fb14305605d435... |
| dataset | RUN-f0981dbfd03c1bf5 | 42 | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 928.0 | NaN | 6a0d08f1596918f9f6672a6857124907fb14305605d435... |
| manifest | RUN-f0981dbfd03c1bf5 | 42 | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | NaN | {'customers': '1', 'institutions': '1', 'accou... | 6a0d08f1596918f9f6672a6857124907fb14305605d435... |
experiment = {
"package": "fraudtwin",
"version": __version__,
"identity": identity,
"purpose": "bounded reproducibility",
"limitations": "synthetic data; local timing only",
}
experiment["fingerprint"] = sha256_json(experiment)
print(f"experiment_fingerprint: {experiment['fingerprint']}")
experiment_fingerprint: 640bfa0721c667dccffba0ce9b81b7c661235e37af61e87dd56bc7beb08ef393
with TemporaryDirectory() as tmp:
path = Path(tmp) / "experiment.json"
path.write_text(json.dumps(experiment, sort_keys=True, indent=2, default=str), encoding="utf-8")
loaded = json.loads(path.read_text(encoding="utf-8"))
print({"artifact": path.name, "keys": sorted(loaded)})
assert loaded["fingerprint"] == experiment["fingerprint"]
{'artifact': 'experiment.json', 'keys': ['fingerprint', 'identity', 'limitations', 'package', 'purpose', 'version']}
repeat = generate(config, write=False)
print(
{
"same_run_id": repeat.run_id == data.run_id,
"same_events": tuple(e.event_id for e in repeat.behavior.payment_events)
== tuple(e.event_id for e in data.behavior.payment_events),
}
)
assert repeat.run_id == data.run_id
{'same_run_id': True, 'same_events': True}
release = {
"experiment_fingerprint": experiment["fingerprint"],
"configuration_hash": data.manifest.scenario_config_hash,
"seed": config.simulation.seed,
"counts": counts,
}
pd.DataFrame(release)
| experiment_fingerprint | configuration_hash | seed | counts | |
|---|---|---|---|---|
| payments | 640bfa0721c667dccffba0ce9b81b7c661235e37af61e8... | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 42 | 1000 |
| events | 640bfa0721c667dccffba0ce9b81b7c661235e37af61e8... | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 42 | 4506 |
| ledger | 640bfa0721c667dccffba0ce9b81b7c661235e37af61e8... | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 42 | 1914 |
| dataset | 640bfa0721c667dccffba0ce9b81b7c661235e37af61e8... | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 42 | 928 |
release["fingerprint"] = sha256_json(release)
assert release["fingerprint"]
pd.DataFrame(release)
| experiment_fingerprint | configuration_hash | seed | counts | fingerprint | |
|---|---|---|---|---|---|
| payments | 640bfa0721c667dccffba0ce9b81b7c661235e37af61e8... | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 42 | 1000 | 5cea7fcb11298124234b8ea9410775f13342fdd6a3262e... |
| events | 640bfa0721c667dccffba0ce9b81b7c661235e37af61e8... | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 42 | 4506 | 5cea7fcb11298124234b8ea9410775f13342fdd6a3262e... |
| ledger | 640bfa0721c667dccffba0ce9b81b7c661235e37af61e8... | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 42 | 1914 | 5cea7fcb11298124234b8ea9410775f13342fdd6a3262e... |
| dataset | 640bfa0721c667dccffba0ce9b81b7c661235e37af61e8... | f0981dbfd03c1bf57a1c8f6ac331dbaebbcbc4eef0b7ea... | 42 | 928 | 5cea7fcb11298124234b8ea9410775f13342fdd6a3262e... |
print({"reproducibility_fields": sorted(experiment["identity"])})
{'reproducibility_fields': ['configuration_hash', 'counts', 'data_fingerprint', 'run_id', 'schema_versions', 'seed']}
assert release["fingerprint"] == sha256_json(
{k: v for k, v in release.items() if k != "fingerprint"}
)
Configuration, seed, package version, schema, counts, and fingerprints are ready to archive.
Verification and next step#
Re-run the offline cells from a clean checkout and compare the printed fingerprints. For service-backed publication, continue with the relevant integration runbook after this notebook; do not treat synthetic metrics as a deployment SLO.