Measure drift by operational segment#

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.ml.drift import DriftConfig, compare_windows

rows = payments.to_dicts()
config_drift = DriftConfig(
    reference_name="baseline", comparison_name="current", minimum_samples=1, fields=("amount",)
)
segments = ["merchant_id", "payer_account_id"]
reports = {}
report_rows = []
for segment in segments:
    groups = sorted({str(r.get(segment)) for r in rows})[:3]
    reports[segment] = {}
    for group in groups:
        reference = [r for r in rows if str(r.get(segment)) == group]
        comparison = [dict(r, amount=float(r.get("amount") or 0) * 1.1) for r in reference]
        fingerprint = compare_windows(reference, comparison, config_drift).fingerprint
        reports[segment][group] = fingerprint
        report_rows.append(
            {
                "segment": segment,
                "group": group,
                "fingerprint_prefix": f"{fingerprint[:12]}...",
            }
        )

# Keep full fingerprints in reports, but show a readable summary in the tutorial.
display(pl.DataFrame(report_rows).sort(["segment", "group"]))
shape: (6, 3)
segmentgroupfingerprint_prefix
strstrstr
"merchant_id""MER-000001""46f6de5cdc11..."
"merchant_id""MER-000002""de769f110625..."
"merchant_id""MER-000003""b933a626fba8..."
"payer_account_id""ACC-000001""445849ba7649..."
"payer_account_id""ACC-000003""167d3b5bbad6..."
"payer_account_id""ACC-000004""89daa1c1f334..."

Run the core operation

display(
    pl.DataFrame(
        [
            {"segment": key, "groups": len(value), "status": "review"}
            for key, value in reports.items()
        ]
    )
)
shape: (2, 3)
segmentgroupsstatus
stri64str
"merchant_id"3"review"
"payer_account_id"3"review"

Measure and interpret the result

Feature drift is a P(X) change; segment mix is domain shift; degraded matured-label metrics indicate concept/performance drift.

Exercise a parameter or failure mode

alerts = [{"segment": key, "action": "investigate"} for key in reports]
display(pl.DataFrame(alerts))
shape: (2, 2)
segmentaction
strstr
"merchant_id""investigate"
"payer_account_id""investigate"

Write a compact artifact and fingerprint

from fraudtwin.reproducibility import sha256_json

assert reports
report_fingerprint = sha256_json(reports)
report_preview = pl.DataFrame(
    [
        {"segment": segment, "group": group, "fingerprint": f"{fingerprint[:12]}..."}
        for segment, groups in sorted(reports.items())
        for group, fingerprint in sorted(groups.items())
    ]
)
display(report_preview)
artifact_summary = {
    "segments": len(reports),
    "groups": sum(len(groups) for groups in reports.values()),
    "report_fingerprint": f"{report_fingerprint[:12]}...",
    "label_policy": "exclude_unresolved",
}
print(json.dumps(artifact_summary, indent=2))
shape: (6, 3)
segmentgroupfingerprint
strstrstr
"merchant_id""MER-000001""46f6de5cdc11..."
"merchant_id""MER-000002""de769f110625..."
"merchant_id""MER-000003""b933a626fba8..."
"payer_account_id""ACC-000001""445849ba7649..."
"payer_account_id""ACC-000003""167d3b5bbad6..."
"payer_account_id""ACC-000004""89daa1c1f334..."
{
  "segments": 2,
  "groups": 6,
  "report_fingerprint": "185d5136f64e...",
  "label_policy": "exclude_unresolved"
}

Verify invariants and clean up

Use a scheduled report and calibrate thresholds against seasonal variation.

Optional service integration

# 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

Review the expected outcome

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
}

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}