Publish contracts and inspect Kafka delivery semantics

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.

!pip install confluent-kafka

Requirement already satisfied: confluent-kafka in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (2.15.1)

[notice] A new release of pip is available: 26.1 -> 26.2.1
[notice] To update, run: pip install --upgrade pip

Optional Kafka setup

The first cell installs the Python client. Start Kafka and Schema Registry only for broker-backed publication; the offline contract path does not require them.

docker compose --profile streaming up -d kafka schema-registry streaming-topics

Set FRAUDTWIN_KAFKA_BOOTSTRAP_SERVERS and FRAUDTWIN_SCHEMA_REGISTRY_URL when needed. Stop with docker compose --profile streaming down; unavailable services use the offline fallback.

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.contracts import contract_registry
from fraudtwin.kafka import publication_records
from fraudtwin.kafka_chaos import KafkaChaosConfig, simulate_delivery

registry = contract_registry()
records = publication_records(data.behavior, run_id, registry=registry)
print({"contracts": len(registry.subjects), "records": len(records)})

{'contracts': 6, 'records': 4949}

Run the core operation

display(
    pl.DataFrame(
        [
            {
                "subject": r.subject,
                "topic": r.topic,
                "record_id": r.record_id,
                "bytes": len(r.value),
            }
            for r in records[:12]
        ]
    )
)

shape: (12, 4)

subject	topic	record_id	bytes
str	str	str	i64
"payment-event"	"fraudsim.payment.events.v1"	"EVT-F01-000001-000001"	465
"payment-event"	"fraudsim.payment.events.v1"	"EVT-F02-000002-000001"	493
"payment-event"	"fraudsim.payment.events.v1"	"EVT-F03-000003-SIGNAL-01"	477
"payment-event"	"fraudsim.payment.events.v1"	"EVT-F04-000004-000001"	466
"payment-event"	"fraudsim.payment.events.v1"	"EVT-F05-000005-000001"	481
…	…	…	…
"payment-event"	"fraudsim.payment.events.v1"	"EVT-HN-F03-000003-SIGNAL-01"	465
"payment-event"	"fraudsim.payment.events.v1"	"EVT-HN-F04-000004-000001"	418
"payment-event"	"fraudsim.payment.events.v1"	"EVT-HN-F05-000005-000001"	436
"payment-event"	"fraudsim.payment.events.v1"	"EVT-F01-000001-000001-02"	496
"payment-event"	"fraudsim.payment.events.v1"	"EVT-F01-000001-000002"	465

Measure and interpret the result

chaos = simulate_delivery(
    records[:50],
    KafkaChaosConfig(
        seed=7,
        drop_probability=0.05,
        duplicate_probability=0.1,
        retry_probability=0.1,
        max_delay_seconds=30,
        reorder_window=5,
    ),
)
print(chaos.manifest["counts"])

{'sent': 50, 'acknowledged': 59, 'dropped': 2, 'retried': 5, 'duplicated': 5, 'late': 58, 'reordered': 27, 'deduplicated': 11}

Exercise a parameter or failure mode

assert all(r.record_id for r in chaos.envelopes)
print(
    {
        "input": chaos.input_count,
        "output": chaos.emitted_count,
        "fingerprint": f"{chaos.output_fingerprint[:12]}...",
    }
)

{'input': 50, 'output': 59, 'fingerprint': '3900e63e1e0c...'}

Write a compact artifact and fingerprint

import os

try:
    from confluent_kafka.admin import AdminClient
    from confluent_kafka.schema_registry import SchemaRegistryClient

    from fraudtwin.kafka import publisher_from_environment

    admin = AdminClient(
        {"bootstrap.servers": os.getenv("FRAUDTWIN_KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")}
    )
    topics = sorted(admin.list_topics(timeout=3).topics)
    registry_client = SchemaRegistryClient(
        {"url": os.getenv("FRAUDTWIN_SCHEMA_REGISTRY_URL", "http://localhost:8081")}
    )
    subjects = sorted(registry_client.get_subjects())
    published = publisher_from_environment().publish(data.behavior, run_id)
    print(
        {
            "connected": True,
            "topics": topics,
            "subjects": subjects,
            "published": published.record_counts,
        }
    )
except Exception as exc:
    print({"connected": False, "offline_fallback": True, "reason": type(exc).__name__})

{'connected': False, 'offline_fallback': True, 'reason': 'ModuleNotFoundError'}

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(json.dumps(summary, indent=2, default=str))

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

Review the expected outcome

print("Optional service cell: start Kafka and Schema Registry with the streaming guide.")

Optional service cell: start Kafka and Schema Registry with the streaming guide.

Record the generated shape and tutorial contract.

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

{'payments': 1092, 'events': 4699}

assert chaos.output_fingerprint
print({"offline_fallback": True, "chaos_fingerprint": f"{chaos.output_fingerprint[:12]}..."})

{'offline_fallback': True, 'chaos_fingerprint': '3900e63e1e0c...'}