From Events to a Trustworthy ML Dataset#

You are asked to build a fraud detector after a suspicious payment incident. The simulator knows what happened, but events arrive imperfectly and labels become available later. Your task is to build a dataset using only information that was available at prediction time.

1. Start with a fraud-enabled run#

We use a versioned benchmark configuration so the investigation can be repeated exactly.

from pathlib import Path

import polars as pl

import fraudtwin
from fraudtwin.config import load_config

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"
config = load_config(config_path)

# Keep the fraud scenarios, but introduce realistic data-quality issues.
config = config.model_copy(
    update={"quality": config.quality.model_copy(update={"profile": "realistic"})}
)
data = fraudtwin.generate(config)

print("Run:", data.run_id)

Run: RUN-0b12923ed7256443

2. The incident begins with events#

Payment events are operational records. They include event time and source availability, which are not always identical.

events = pl.DataFrame(
    [
        {
            "Event": event.event_id,
            "Type": event.event_type,
            "Payment": event.payment_id,
            "Event time": event.event_time,
            "Available at": event.source_available_at,
        }
        for event in data.behavior.payment_events[:8]
    ]
)

events
shape: (8, 5)
EventTypePaymentEvent timeAvailable at
strstrstrdatetime[μs, UTC]datetime[μs, UTC]
"EVT-00000001""TRANSFER_COMPLETED""PAY-00000001"2026-01-02 12:43:00 UTC2026-01-02 12:43:05 UTC
"EVT-00000002""CARD_PAYMENT_INITIATED""PAY-00000002"2026-01-01 19:53:00 UTC2026-01-01 19:53:05 UTC
"EVT-00000002-02""CARD_AUTHORIZATION_REQUESTED""PAY-00000002"2026-01-01 19:53:01 UTC2026-01-01 19:53:06 UTC
"EVT-00000002-03""CARD_AUTHORIZED""PAY-00000002"2026-01-01 19:53:02 UTC2026-01-01 19:53:07 UTC
"EVT-00000002-04""CARD_CAPTURED""PAY-00000002"2026-01-01 19:53:07 UTC2026-01-01 19:53:12 UTC
"EVT-00000002-05""CARD_CLEARED""PAY-00000002"2026-01-01 19:53:37 UTC2026-01-01 19:53:42 UTC
"EVT-00000002-06""CARD_SETTLED""PAY-00000002"2026-01-01 19:54:37 UTC2026-01-01 19:54:42 UTC
"EVT-00000003""PIX_INITIATED""PAY-00000003"2026-01-01 06:50:00 UTC2026-01-01 06:50:02 UTC

3. The stream is imperfect#

Real pipelines contain duplicates, missing values, late events, and out-of-order records. FraudTwin records these faults instead of hiding them.

quality = pl.DataFrame(
    {"Issue": name, "Rows": count}
    for name, count in data.manifest.quality_fault_counts.items()
    if count > 0
)

quality.sort("Rows", descending=True)
shape: (6, 2)
IssueRows
stri64
"late_events"42
"out_of_order_events"16
"missing_optional_fields"14
"duplicate_events"3
"negative_amounts"2
"duplicate_records"1

4. Separate fraud truth from lookalikes#

The simulator’s truth table contains both real fraud and hard negatives. A detector must learn to distinguish them rather than rely on one obvious rule.

truth = (
    pl.DataFrame({"Record type": [record.record_type for record in data.behavior.fraud_records]})
    .group_by("Record type")
    .len()
    .rename({"len": "Records"})
)

truth
shape: (2, 2)
Record typeRecords
stru32
"HARD_NEGATIVE"9
"FRAUD"25

5. Observe the investigation workflow#

Alerts, cases, confirmations, and labels are produced progressively. The counts show how the incident becomes observable to downstream systems.

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

workflow
shape: (5, 2)
ArtifactRows
stri64
"Fraud records"34
"Alerts"34
"Cases"34
"Confirmations"34
"Labels"34

6. Labels arrive after the payment#

A label’s availability time is later than the time of the fraud itself. This delay is central to realistic fraud-model validation.

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]
    ]
)

labels
shape: (5, 4)
PaymentLabelFraud occurredLabel available
strstrdatetime[μs, UTC]datetime[μs, UTC]
"PAY-F01-000001-000001""FRAUD"2026-01-01 00:00:00 UTC2026-01-03 02:06:11 UTC
"PAY-F01-000001-000002""FRAUD"2026-01-01 00:00:02 UTC2026-01-03 01:06:13 UTC
"PAY-F01-000001-000003""FRAUD"2026-01-01 00:00:03 UTC2026-01-03 01:06:14 UTC
"PAY-HN-F01-000001-000001""LEGITIMATE"2026-01-01 00:00:00 UTC2026-01-02 01:06:07 UTC
"PAY-HN-F01-000004-000001""LEGITIMATE"2026-01-01 00:00:00 UTC2026-01-02 01:06:07 UTC

7. Build the point-in-time dataset#

FraudTwin builds each row using features and labels that were available at that row’s prediction time. This example keeps unresolved labels, so a null label means the outcome was not available yet at prediction_time; it is not a missing payment field.

ml_dataset = data.require_dataset().frame

print("Rows:", ml_dataset.height)
print("Columns:", ml_dataset.width)
print(
    {
        "labelled": ml_dataset.filter(pl.col("label").is_not_null()).height,
        "unresolved": ml_dataset.filter(pl.col("label").is_null()).height,
    }
)
preview = pl.concat(
    [
        ml_dataset.filter(pl.col("label").is_not_null()).head(3),
        ml_dataset.filter(pl.col("label").is_null()).head(2),
    ],
    how="vertical",
)
preview.select(["payment_id", "prediction_time", "label_available_at", "label", "split"])

Rows: 288
Columns: 53
{'labelled': 0, 'unresolved': 288}
shape: (2, 5)
payment_idprediction_timelabel_available_atlabelsplit
strdatetime[μs, UTC]datetime[μs, UTC]strstr
"PAY-00000001"2026-01-02 12:43:05 UTCnullnull"validation"
"PAY-00000002"2026-01-01 19:53:05 UTCnullnull"train"

8. Inspect the temporal splits#

The dataset uses time-aware splits so validation and test records represent future activity relative to training.

ml_dataset.group_by("split").len().rename({"len": "Rows"}).sort("split")
shape: (3, 2)
splitRows
stru32
"test"37
"train"225
"validation"26

9. Finish the investigation#

The generated run, quality diagnostics, fraud truth, delayed workflow records, and ML dataset are all tied together by stable IDs and a manifest. That makes the experiment reproducible and auditable.

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': 301, 'events': 1443}

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': 301, 'events': 1443}