Build a Reproducible Fraud Benchmark#
A fraud detector is useful only when its results can be reproduced and challenged. In this tutorial, we create one deterministic mixed benchmark, keep point-in-time features separate from oracle labels, train a dependency-free logistic regression model, compare it with a transparent heuristic, and inspect the evidence written to disk. The goal is not to make the learned model win; it is to make the comparison honest enough to reveal when it does not.
1. Prepare a small benchmark#
The mixed suite combines temporal regimes, feature and relation camouflage, graph campaigns, counterfactual requests, label observation, and calibration. This notebook uses a small local run with a fixed seed, so it is a reproducibility smoke benchmark—not evidence that a detector will generalise in production.
import json
import math
from pathlib import Path
from tempfile import TemporaryDirectory
import polars as pl
import fraudtwin
from fraudtwin.benchmark import BenchmarkRequest, build_suite_config, run_benchmark
from fraudtwin.ml import (
BaselineEvaluationConfig,
PredictionRecord,
evaluate_predictions,
heuristic_predictions,
write_evaluation,
)
workspace = TemporaryDirectory(prefix="fraudtwin-tutorial-08-")
work_dir = Path(workspace.name)
# A small generated reference lets the integrated suite exercise calibration too.
reference_run = fraudtwin.generate()
reference = pl.DataFrame(
{
"amount": [item.amount for item in reference_run.behavior.payments],
"event_time": [item.initiated_at for item in reference_run.behavior.payments],
"customer_id": [item.payer_account_id for item in reference_run.behavior.payments],
}
)
reference_path = work_dir / "reference.parquet"
reference.write_parquet(reference_path)
profile = fraudtwin.fit_calibration_profile(reference_path)
profile_path = fraudtwin.write_calibration_profile(profile, work_dir / "profile.yaml")
config = build_suite_config(
"mixed",
difficulty=7,
seed=42,
calibration_profile=profile_path,
)
benchmark = run_benchmark(
BenchmarkRequest(
suite="mixed",
difficulty=7,
seed=42,
output_dir=work_dir / "benchmarks",
calibration_profile=profile_path,
)
)
data = fraudtwin.generate(config)
written_run = fraudtwin.generate(config, write=True, output_dir=work_dir / "runs")
print("Benchmark:", benchmark.benchmark_id)
print("Run:", data.run_id)
print("Payments:", len(data.behavior.payments))
dataset = data.require_dataset()
print("Dataset rows:", dataset.count)
Benchmark: BM-8d3fb3b8af13749f
Run: RUN-9c26c6aa36fc34a9
Payments: 735
Dataset rows: 730
The row counts describe what the suite exercised; they are not model-quality metrics. The benchmark also records its configuration, seed, descriptors, and output fingerprints. The next table gives the benchmark’s own held-out baseline result and the scenario descriptors that make the difficulty inspectable.
benchmark_results = (
pl.DataFrame(benchmark.results)
.select(
[
"model",
"suite",
"difficulty",
"prediction_count",
"pr_auc",
"recall_at_fixed_fpr",
"f1",
"drop_vs_baseline",
]
)
.with_columns(
pl.col("pr_auc").cast(pl.Float64, strict=False).round(3),
pl.col("recall_at_fixed_fpr").cast(pl.Float64, strict=False).round(3),
pl.col("f1").cast(pl.Float64, strict=False).round(3),
pl.col("drop_vs_baseline").cast(pl.Float64, strict=False).round(3),
)
)
benchmark_results
| model | suite | difficulty | prediction_count | pr_auc | recall_at_fixed_fpr | f1 | drop_vs_baseline |
|---|---|---|---|---|---|---|---|
| str | str | i64 | i64 | f64 | f64 | f64 | f64 |
| "deterministic_heuristic" | "mixed" | 7 | 730 | null | 0.0 | 0.0 | null |
2. Make labels and chronological splits explicit#
The simulator keeps latent fraud truth out of ordinary observable dataset rows. For this controlled offline experiment, we join that oracle label onto a separate analysis copy; it is never included among the model features. This makes the exercise measurable, but it is not the same as having labels available at prediction time.
fraud_payment_ids = {item.payment_id for item in data.behavior.fraud_records if item.fraud_truth}
rows = sorted(dataset.rows, key=lambda row: row["prediction_time"])
train_end = int(len(rows) * 0.40)
validation_end = int(len(rows) * 0.70)
labelled_rows = []
for index, row in enumerate(rows):
split = "train" if index < train_end else "validation" if index < validation_end else "test"
labelled_rows.append(
{
**row,
"label": "FRAUD" if row["payment_id"] in fraud_payment_ids else "LEGITIMATE",
"split": split,
}
)
split_summary = pl.DataFrame(
[
{
"split": split,
"rows": len(split_rows := [row for row in labelled_rows if row["split"] == split]),
"fraud": sum(row["label"] == "FRAUD" for row in split_rows),
"fraud_rate": sum(row["label"] == "FRAUD" for row in split_rows) / len(split_rows),
"start": str(min(row["prediction_time"] for row in split_rows))[:19],
"end": str(max(row["prediction_time"] for row in split_rows))[:19],
}
for split in ("train", "validation", "test")
]
)
split_summary = split_summary.with_columns(pl.col("fraud_rate").round(3))
display(split_summary)
try:
import matplotlib.pyplot as plt
fig, axis = plt.subplots(figsize=(7, 4))
axis.bar(
split_summary["split"].to_list(), split_summary["fraud_rate"].to_list(), color="#9333ea"
)
axis.set(title="Fraud rate by chronological split", xlabel="split", ylabel="fraud rate")
axis.set_ylim(0, max(split_summary["fraud_rate"].to_list()) * 1.25)
fig.tight_layout()
plt.show()
plt.close(fig)
except ImportError:
print("Install matplotlib to render the split fraud-rate plot.")
| split | rows | fraud | fraud_rate | start | end |
|---|---|---|---|---|---|
| str | i64 | i64 | f64 | str | str |
| "train" | 292 | 32 | 0.11 | "2026-01-01 00:28:05" | "2026-01-06 10:26:02" |
| "validation" | 218 | 24 | 0.11 | "2026-01-06 10:54:05" | "2026-01-08 11:57:39" |
| "test" | 220 | 8 | 0.036 | "2026-01-08 11:57:39" | "2026-01-12 23:38:05" |
The split is chronological: earlier observations train the model, the middle period selects operating thresholds, and the final period is held out until the end. The fraud counts and rates are printed because a test set with very few positives makes point estimates unstable. For a production claim, repeat this comparison across seeds and longer time windows.
for split in ("train", "validation", "test"):
split_rows = [row for row in labelled_rows if row["split"] == split]
assert {row["label"] for row in split_rows} == {
"FRAUD",
"LEGITIMATE",
}, f"{split} must contain both classes for this demonstration"
assert set(row["split"] for row in labelled_rows) == {"train", "validation", "test"}
3. Train a dependency-free logistic regression#
This small implementation uses only Python’s standard library. It learns four point-in-time features with gradient descent. The feature list is fixed before evaluation; the model sees training labels, but never test labels or future-derived features while fitting. Its purpose is transparency, not competitive performance.
FEATURES = (
"amount_vs_customer_avg",
"new_device_flag",
"new_merchant_flag",
"transaction_count_1h",
)
def vector(row):
return [
float(row.get("amount_vs_customer_avg") or 0.0),
float(bool(row.get("new_device_flag"))),
float(bool(row.get("new_merchant_flag"))),
float(row.get("transaction_count_1h") or 0.0),
]
def sigmoid(value):
value = max(-30.0, min(30.0, value))
return 1.0 / (1.0 + math.exp(-value))
def fit_logistic_regression(rows, *, epochs=600, learning_rate=0.08, l2=0.01):
training = [row for row in rows if row["split"] == "train"]
raw = [vector(row) for row in training]
means = [sum(values[j] for values in raw) / len(raw) for j in range(len(FEATURES))]
scales = [
max(1e-12, (sum((values[j] - means[j]) ** 2 for values in raw) / len(raw)) ** 0.5)
for j in range(len(FEATURES))
]
normalised = [
[(value - means[j]) / scales[j] for j, value in enumerate(values)] for values in raw
]
weights = [0.0] * len(FEATURES)
intercept = 0.0
for _ in range(epochs):
gradients = [0.0] * len(FEATURES)
intercept_gradient = 0.0
for values, row in zip(normalised, training, strict=True):
expected = 1.0 if row["label"] == "FRAUD" else 0.0
error = (
sigmoid(
intercept
+ sum(weight * value for weight, value in zip(weights, values, strict=False))
)
- expected
)
intercept_gradient += error
for j, value in enumerate(values):
gradients[j] += error * value
size = len(training)
intercept -= learning_rate * intercept_gradient / size
for j in range(len(weights)):
weights[j] -= learning_rate * (gradients[j] / size + l2 * weights[j])
return {
"features": list(FEATURES),
"means": means,
"scales": scales,
"weights": weights,
"intercept": intercept,
}
def logistic_predictions(rows, model):
predictions = []
for row in rows:
values = vector(row)
normalised = [
(value - model["means"][j]) / model["scales"][j] for j, value in enumerate(values)
]
score = sigmoid(
model["intercept"]
+ sum(
weight * value for weight, value in zip(model["weights"], normalised, strict=False)
)
)
predictions.append(
PredictionRecord(
event_id=str(row["event_id"]),
prediction_timestamp=row["prediction_time"],
fraud_score=score,
)
)
return tuple(predictions)
model = fit_logistic_regression(labelled_rows)
print("Learned logistic coefficients:")
for feature, weight in zip(model["features"], model["weights"], strict=True):
print(f" {feature}: {weight:+.3f}")
print(f" intercept: {model['intercept']:+.3f}")
Learned logistic coefficients:
amount_vs_customer_avg: +0.549
new_device_flag: +0.438
new_merchant_flag: -0.082
transaction_count_1h: -0.375
intercept: -2.213
4. Compare the model with the heuristic#
FraudTwin’s evaluator applies the same metrics to both scores. Ranking metrics (ROC-AUC, PR-AUC, and precision/recall at the top 1%) do not depend on a chosen cutoff. Thresholded precision, recall, and F1 use a cutoff selected on validation; the test period is then scored once without retuning.
evaluation_config = BaselineEvaluationConfig(
models=("deterministic_heuristic",),
label_policy="exclude_unresolved",
)
logistic_result = evaluate_predictions(
labelled_rows,
logistic_predictions(labelled_rows, model),
evaluation_config,
model_id="dependency_free_logistic_regression",
)
heuristic_result = evaluate_predictions(
labelled_rows,
heuristic_predictions(labelled_rows),
evaluation_config,
model_id="deterministic_heuristic",
)
def metric_table(result):
rows = []
for item in result.metrics:
if item["partition"] not in {"train", "validation", "test"} or "segment_dimension" in item:
continue
prevalence = item["positive_count"] / item["labelled_row_count"]
top_1pct_precision = item["precision_at_k"]
rows.append(
{
"model": item["model_id"]
.replace("dependency_free_", "")
.replace("deterministic_", ""),
"split": item["partition"],
"rows": item["row_count"],
"fraud": item["positive_count"],
"fraud_rate": prevalence,
"roc_auc": item["roc_auc"],
"pr_auc": item["pr_auc"],
"precision@1%": top_1pct_precision,
"recall@1%": item["recall_at_k"],
"lift@1%": top_1pct_precision / prevalence if prevalence else None,
"precision": item["precision"],
"recall": item["recall"],
"f1": item["f1"],
"brier": item["brier_score"],
}
)
return rows
metrics = pl.DataFrame(metric_table(logistic_result) + metric_table(heuristic_result))
test_comparison = (
metrics.filter(pl.col("split") == "test")
.select(
[
"model",
"rows",
"fraud",
"fraud_rate",
"roc_auc",
"pr_auc",
"precision@1%",
"recall@1%",
"lift@1%",
"precision",
"recall",
"f1",
"brier",
]
)
.sort("pr_auc", descending=True)
.with_columns(pl.all().exclude(["model", "rows", "fraud"]).round(3))
)
test_comparison
| model | rows | fraud | fraud_rate | roc_auc | pr_auc | precision@1% | recall@1% | lift@1% | precision | recall | f1 | brier |
|---|---|---|---|---|---|---|---|---|---|---|---|---|
| str | i64 | i64 | f64 | f64 | f64 | f64 | f64 | f64 | f64 | f64 | f64 | f64 |
| "heuristic" | 220 | 8 | 0.036 | 0.541 | 0.126 | 0.333 | 0.125 | 9.167 | 0.167 | 0.25 | 0.2 | 0.048 |
| "logistic_regression" | 220 | 8 | 0.036 | 0.443 | 0.046 | 0.0 | 0.0 | 0.0 | 0.032 | 0.375 | 0.059 | 0.044 |
for result in (logistic_result, heuristic_result):
for item in result.metrics:
if item["partition"] in {"train", "validation", "test"} and "segment_dimension" not in item:
for name in ("roc_auc", "pr_auc", "precision", "recall", "f1", "brier_score"):
assert item[name] is not None and math.isfinite(float(item[name]))
validation_thresholds = pl.DataFrame(
[
{
"model": result.manifest["model_id"]
.replace("dependency_free_", "")
.replace("deterministic_", ""),
"f1_cutoff": result.manifest["thresholds"]["f1"],
"fpr_1pct_cutoff": result.manifest["thresholds"]["fixed_fpr"],
"recall_80pct_cutoff": result.manifest["thresholds"]["fixed_recall"],
}
for result in (logistic_result, heuristic_result)
]
).with_columns(pl.all().exclude("model").cast(pl.Float64, strict=False).round(3))
print("Validation-selected thresholds (rounded):")
for row in validation_thresholds.iter_rows(named=True):
print(
f" {row['model']}: F1 cutoff={row['f1_cutoff']:.3f}; "
f"FPR@1% cutoff={row['fpr_1pct_cutoff']:.3f}; "
f"Recall@80% cutoff={row['recall_80pct_cutoff']:.3f}"
)
Validation-selected thresholds (rounded):
logistic_regression: F1 cutoff=0.081; FPR@1% cutoff=1.000; Recall@80% cutoff=0.002
heuristic: F1 cutoff=0.571; FPR@1% cutoff=1.000; Recall@80% cutoff=0.000
5. Save evidence and check reproducibility#
A benchmark is more than a score. We save the predictions, metrics, learned weights, manifests, descriptors, and source run so the result can be audited or reproduced later. A fingerprint is useful because it lets us detect a changed result; it does not certify that the detector is good.
heuristic_paths = write_evaluation(heuristic_result, work_dir / "evaluations" / "heuristic")
logistic_paths = write_evaluation(logistic_result, work_dir / "evaluations" / "logistic_regression")
model_path = work_dir / "evaluations" / "logistic_regression" / "model.json"
model_path.write_text(json.dumps(model, indent=2) + "\n", encoding="utf-8")
repeat = fraudtwin.generate(config)
assert repeat.run_id == data.run_id
assert repeat.manifest.model_dump(mode="json") == data.manifest.model_dump(mode="json")
artifact_paths = {
"benchmark_manifest": benchmark.manifest_path,
"benchmark_descriptors": benchmark.descriptors_path,
"benchmark_results": benchmark.results_path,
"source_manifest": written_run.manifest_path,
"dataset": written_run.dataset_path,
"logistic_predictions": logistic_paths[0],
"logistic_metrics": logistic_paths[1],
"logistic_manifest": logistic_paths[2],
"learned_model": model_path,
}
assert all(path is not None and path.is_file() for path in artifact_paths.values())
artifact_summary = pl.DataFrame(
{
"artifact": list(artifact_paths),
"exists": [path is not None and path.is_file() for path in artifact_paths.values()],
"bytes": [path.stat().st_size for path in artifact_paths.values()],
}
)
print(
"Source and repeat manifests match:",
repeat.manifest.model_dump(mode="json") == data.manifest.model_dump(mode="json"),
)
print("Source fingerprint:", data.manifest.model_dump(mode="json").get("output_fingerprint"))
artifact_summary
Source and repeat manifests match: True
Source fingerprint: None
| artifact | exists | bytes |
|---|---|---|
| str | bool | i64 |
| "benchmark_manifest" | true | 279591 |
| "benchmark_descriptors" | true | 844 |
| "benchmark_results" | true | 4063 |
| "source_manifest" | true | 182807 |
| "dataset" | true | 93209 |
| "logistic_predictions" | true | 14094 |
| "logistic_metrics" | true | 21916 |
| "logistic_manifest" | true | 3325 |
| "learned_model" | true | 510 |
What this result means#
Interpret the test table in two layers: PR-AUC and lift@1% measure ranking quality for a limited review queue, while precision, recall, and F1 use the threshold chosen on validation. Compare the learned model with the heuristic on the test row—not on training—and remember that PR-AUC depends on the printed fraud prevalence.
This run may show the heuristic ahead of the learned model. That is a useful benchmark finding: four simple features and one seed are not enough to justify deployment, and a training win would not prove generalisation. The test fraud count is deliberately visible because a small count makes the estimate noisy; repeat across seeds or increase the time window before making a performance claim.
The oracle labels support this offline experiment; a production workflow would replace them with labels available through the investigation process. The matching manifests and artifact fingerprints demonstrate reproducibility, not predictive validity.
For the simpler version of the scoring idea, see Build a Simple Fraud Scoring Model.
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': 735, 'events': 3066}