Explore Fraud and Delayed Labels#

FraudTwin keeps the simulator’s ground truth separate from what an operational fraud system can observe. This tutorial explores fraud scenarios, hard negatives, alerts, cases, and delayed labels.

1. Generate fraud-enabled data#

This tutorial uses the fraud-enabled benchmark configuration. It generates fraud and legitimate lookalikes so the two populations can be compared.

from pathlib import Path

import polars as pl

import fraudtwin

project_root = next(
    path
    for path in (Path.cwd(), *Path.cwd().parents)
    if (path / "configs" / "minimal.yaml").is_file()
)
config_path = project_root / "configs" / "benchmarks" / "difficulty-v1.yaml"
data = fraudtwin.generate(config_path)

print("Run:", data.run_id)

Run: RUN-642709d737867d7f

2. Compare fraud records#

A fraud record describes simulator truth. Hard negatives are legitimate records deliberately generated to resemble fraud.

fraud_records = pl.DataFrame(
    [
        {
            "Record type": record.record_type,
            "Fraud truth": record.fraud_truth,
            "Scenario": record.scenario_type,
            "Payment": record.payment_id,
            "Amount": record.amount,
        }
        for record in data.behavior.fraud_records
    ]
)

fraud_records.group_by(["Record type", "Fraud truth"]).len().rename({"len": "Records"})

shape: (2, 3)

Record type	Fraud truth	Records
str	bool	u32
"HARD_NEGATIVE"	false	9
"FRAUD"	true	25

3. Inspect what becomes observable#

Alerts and cases are operational observations derived from fraud records. They are not the same as the simulator’s original fraud truth.

workflow_counts = pl.DataFrame(
    {
        "Artifact": ["Fraud records", "Alerts", "Cases", "Labels"],
        "Rows": [
            len(data.behavior.fraud_records),
            len(data.behavior.alerts),
            len(data.behavior.fraud_cases),
            len(data.behavior.fraud_labels),
        ],
    }
)

workflow_counts

shape: (4, 2)

Artifact	Rows
str	i64
"Fraud records"	34
"Alerts"	34
"Cases"	34
"Labels"	34

4. See label delay#

A label becomes available after the fraud is observed and investigated. The delay is why future information must not be used when constructing historical ML data.

labels = pl.DataFrame(
    [
        {
            "Payment": label.payment_id,
            "Label": label.label,
            "Fraud occurred": label.fraud_occurred_at,
            "Label available": label.label_available_at,
        }
        for label in data.behavior.fraud_labels[:5]
    ]
)

display(labels)
delay_hours = [
    (label.label_available_at - label.fraud_occurred_at).total_seconds() / 3600
    for label in data.behavior.fraud_labels
    if label.label_available_at is not None and label.fraud_occurred_at is not None
]
if delay_hours:
    try:
        import matplotlib.pyplot as plt

        fig, axis = plt.subplots(figsize=(8, 4))
        axis.hist(
            delay_hours, bins=min(20, max(5, len(delay_hours) // 4)), color="#7c3aed", alpha=0.8
        )
        axis.set(
            title="Fraud-label availability delay",
            xlabel="hours from fraud to label",
            ylabel="labels",
        )
        fig.tight_layout()
        plt.show()
        plt.close(fig)
    except ImportError:
        print("Install matplotlib to render the label-delay plot.")
else:
    print("No completed label delays are available to plot.")

shape: (5, 4)

Payment	Label	Fraud occurred	Label available
str	str	datetime[μs, UTC]	datetime[μs, UTC]
"PAY-F01-000001-000001"	"FRAUD"	2026-01-01 00:00:00 UTC	2026-01-03 01:06:11 UTC
"PAY-F01-000001-000002"	"FRAUD"	2026-01-01 00:00:02 UTC	2026-01-03 01:06:13 UTC
"PAY-F01-000001-000003"	"FRAUD"	2026-01-01 00:00:03 UTC	2026-01-03 01:06:14 UTC
"PAY-HN-F01-000001-000001"	"LEGITIMATE"	2026-01-01 00:00:00 UTC	2026-01-02 01:06:07 UTC
"PAY-HN-F01-000004-000001"	"LEGITIMATE"	2026-01-01 00:00:00 UTC	2026-01-02 01:06:07 UTC

../_images/1183e9388ca7f6facae92231c4ae7dc6e8b03766e264a9f6cbb7e1b38952c67b.png

FraudTwin now shows the complete observation path: fraud truth is generated first, workflow artifacts appear later, and labels become available only after their configured delay.

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': 300, 'events': 1440}

Inspect stable payment identities.#

ids = [item.payment_id for item in data.behavior.payments]
assert len(ids) == len(set(ids))
print({"unique_payment_ids": len(ids)})

{'unique_payment_ids': 300}

Compare event-time coverage.#

times = [event.event_time for event in data.behavior.payment_events]
print(
    {
        "first_event": min(times).isoformat() if times else None,
        "last_event": max(times).isoformat() if times else None,
    }
)

{'first_event': '2026-01-01T00:00:00+00:00', 'last_event': '2026-01-02T23:48:02+00:00'}

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': 300, 'events': 1440}

Inspect stable payment identities.#

ids = [item.payment_id for item in data.behavior.payments]
assert len(ids) == len(set(ids))
print({"unique_payment_ids": len(ids)})

{'unique_payment_ids': 300}

Compare event-time coverage.#

times = [event.event_time for event in data.behavior.payment_events]
print(
    {
        "first_event": min(times).isoformat() if times else None,
        "last_event": max(times).isoformat() if times else None,
    }
)

{'first_event': '2026-01-01T00:00:00+00:00', 'last_event': '2026-01-02T23:48:02+00:00'}