Build a Simple Fraud Scoring Model#

You now have a point-in-time dataset. Before reaching for a large ML library, build a transparent baseline: a small risk score made only from information that was available when each payment was evaluated. It is simple enough for a fraud analyst to calculate by hand.

1. Load a point-in-time dataset#

We use a fraud-enabled configuration with fraud in the train, validation, and test periods. FraudTwin provides both the usable feature rows and a hidden answer for offline evaluation; the model sees only the former.

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 = load_config(project_root / "configs" / "benchmarks" / "camouflage-v1.yaml")
data = fraudtwin.generate(config)
dataset = data.require_dataset().frame

print("Rows:", dataset.height)
print("Features:", dataset.width)

Rows: 913
Features: 53

2. Choose the model inputs#

The inputs are deliberately ordinary signals: an unusual amount, a new device, a new merchant, and recent payment velocity. The important rule is not whether a feature sounds clever, but whether it was available before the decision was made.

feature_names = [
    "amount_vs_customer_avg",
    "new_device_flag",
    "new_merchant_flag",
    "transaction_count_1h",
]

dataset.select(feature_names).head()

shape: (5, 4)

amount_vs_customer_avg	new_device_flag	new_merchant_flag	transaction_count_1h
f64	bool	bool	i64
1.253695	false	false	0
0.484239	false	true	0
0.383891	false	true	0
0.668003	false	true	0
0.87436	false	true	0

3. Define a transparent risk score#

This is not a black box and it is not pretending to be production ML. Each weight says how much one signal contributes to suspicion, so the investigator can explain exactly why a payment received its score.

scored = dataset.with_columns(
    (
        0.5 * pl.col("amount_vs_customer_avg").clip(0, 5)
        + 1.0 * pl.col("new_device_flag").cast(pl.Float64)
        + 0.5 * pl.col("new_merchant_flag").cast(pl.Float64)
        + 0.1 * pl.col("transaction_count_1h").clip(0, 10)
    ).alias("risk_score")
)

scored.select(["payment_id", *feature_names, "risk_score"]).head()

shape: (5, 6)

payment_id	amount_vs_customer_avg	new_device_flag	new_merchant_flag	transaction_count_1h	risk_score
str	f64	bool	bool	i64	f64
"PAY-00000001"	1.253695	false	false	0	0.626847
"PAY-00000003"	0.484239	false	true	0	0.74212
"PAY-00000004"	0.383891	false	true	0	0.691946
"PAY-00000005"	0.668003	false	true	0	0.834001
"PAY-00000006"	0.87436	false	true	0	0.93718

4. Add an evaluation answer#

To measure the score, we attach the simulator’s fraud truth after scoring. This is an oracle evaluation field: it tells us whether the detector was right, but it must never be used to calculate the score itself.

fraud_payment_ids = {
    record.payment_id for record in data.behavior.fraud_records if record.fraud_truth
}

scored = scored.with_columns(
    pl.col("payment_id").is_in(fraud_payment_ids).alias("is_fraud"),
)

scored.select(["payment_id", "risk_score", "is_fraud", "split"]).head()

shape: (5, 4)

payment_id	risk_score	is_fraud	split
str	f64	bool	str
"PAY-00000001"	0.626847	false	"train"
"PAY-00000003"	0.74212	false	"validation"
"PAY-00000004"	0.691946	false	"train"
"PAY-00000005"	0.834001	false	"validation"
"PAY-00000006"	0.93718	false	"test"

5. Turn scores into decisions#

A threshold turns a continuous score into an operational decision. Here, scores of 0.5 or higher are sent for review; lowering the threshold catches more cases but also creates more work for investigators.

threshold = 0.5
scored = scored.with_columns((pl.col("risk_score") >= threshold).alias("predicted_fraud"))

scored.select(["risk_score", "is_fraud", "predicted_fraud"]).head()

shape: (5, 3)

risk_score	is_fraud	predicted_fraud
f64	bool	bool
0.626847	false	true
0.74212	false	true
0.691946	false	true
0.834001	false	true
0.93718	false	true

6. Measure precision and recall#

Precision tells us how many reviewed payments are truly fraud. Recall tells us how much known fraud we catch. Looking at both matters: a detector that flags everything has high recall but overwhelms the review team.

def metrics(rows: pl.DataFrame) -> dict[str, float]:
    true_positive = (
        ((pl.col("is_fraud")) & (pl.col("predicted_fraud"))).sum().alias("true_positive")
    )
    false_positive = (
        ((~pl.col("is_fraud")) & (pl.col("predicted_fraud"))).sum().alias("false_positive")
    )
    actual_positive = pl.col("is_fraud").sum().alias("actual_positive")
    values = rows.select([true_positive, false_positive, actual_positive]).row(0)
    tp, fp, positives = (int(value) for value in values)
    predicted = tp + fp
    return {
        "precision": tp / predicted if predicted else 0.0,
        "recall": tp / positives if positives else 0.0,
    }


results = []
for split in ("train", "validation", "test"):
    values = metrics(scored.filter(pl.col("split") == split))
    results.append({"split": split, **values})

pl.DataFrame(results)

shape: (3, 3)

split	precision	recall
str	f64	f64
"train"	0.114679	0.617284
"validation"	0.0	0.0
"test"	0.039216	0.5

7. Stress-test the baseline#

The baseline is now a reference point. Run 06 can provide harder, camouflaged, replayed, or graph-coordinated cases, and we can compare the same formula before investing in a more sophisticated model.

This baseline is not production-ready, and that is intentional. It teaches the complete loop—features, scoring, decisions, evaluation, and stress testing—without hiding the reasoning inside another dependency.

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': 949, 'events': 3165}

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': 949}

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': 949, 'events': 3165}

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': 949}