Skip to content

Week 6 — Fraud data at scale: PySpark and graphs

Why this week: anti-fraud and anti-scam teams work on millions of transactions and on links between accounts, numbers and websites. Their job postings ask for PySpark and graph data. This week you show both, on a fraud problem, where your ten years of rules and features count most.

By Friday you have: a notebook-free PySpark pipeline in pipelines/ that builds per-account features on 6 million PaySim transactions, a comparison of your rules against a model, a graph of synthetic scam reports with its clusters scored, and a new agent tool that says "this number appears in 26 linked reports".

Main online course/reading: the PySpark Getting Started pages (the DataFrame quickstart and the Spark SQL pages), and the NetworkX tutorial for graph basics. Both are free and short. No Claude calls are needed until Thursday.

Concepts

Spark, and why it needs Java. Spark is a program for working with data too big for one pandas DataFrame. It splits the data into chunks (partitions) and works on them in parallel, on one laptop or on hundreds of machines. Spark itself is written for the Java Virtual Machine (JVM), a program that runs Java-family code. PySpark is the Python front end: your Python sends instructions to a Java process that does the work. That's why you install Java once, and why Spark's error messages sometimes look like Java.

Local mode. master("local[*]") runs Spark on your laptop, using all its processor cores. The code is the same code a team runs on a cluster; only that one setting changes. Employers want to see the code, not the cluster.

Lazy evaluation. Most Spark steps (filter, add a column) are only a plan. Nothing runs until you ask for a result, such as count(), show() or writing a file. Spark then runs the whole plan at once, in the best order it can find. If something is slow or fails, it happens at the action, not at the line you just wrote.

Window functions. A calculation over a group of related rows that still returns one row per transaction. "How many transfers has this receiving account had in the 24 hours before this one?" is a window: the group is the account, the order is time, and the frame is the previous 24 hours. You know these from SQL as OVER (PARTITION BY ... ORDER BY ...).

Point-in-time features. A feature may only use what was known when the transaction happened. If "number of transfers to this account" counts later transfers too, your model sees the future and looks brilliant in testing, then fails in production. This is leakage. Every window this week ends at -1: strictly before the current hour.

Parquet. A file format that stores data by column, compressed, with the column types saved. Spark reads it much faster than CSV. Pipelines usually read the raw CSV once and save Parquet for every later step.

Time-based split and alert budget. Fraud models are tested on a later period than they are trained on, because that's how they are used. And fraud teams don't care much about accuracy; they care about how much fraud is in the alerts they can afford to review. "Of the top 1,000 alerts, how many were fraud?" is a number an operations manager understands.

A graph of reports. A graph has nodes (things) and edges (links between things). Here the nodes are scam reports and the identifiers they mention: phone numbers, links and bank accounts. An edge means "this report mentions this identifier". Two reports that share a bank account are now connected, even if they never share a phone number.

Connected components. A group of nodes you can travel between by following edges. In scam reports, a large component is a likely syndicate: reports linked by shared mule accounts and numbers. A component is only as good as its links: one identifier that appears everywhere, like an official hotline victims quote, can join unrelated groups into one meaningless blob.

Dependency groups. PySpark is a 450 MB download that the checker app never uses. Putting it in a separate uv group keeps it out of the app's install, so your CI next week stays fast and your Docker image in week 9 stays small.

Monday — Java, PySpark and the data

Learn (75 min):

  • PySpark installation, the "Dependencies" table only (5 min): which Java versions the current PySpark supports. When this page was written, PySpark 4.2 needed Java 17 or later, and Spark listed Java 17, 21 and 25.
  • Quickstart: DataFrame (45 min): read it, and type the examples into a scratch script if that helps.
  • The PaySim dataset page on Kaggle (15 min): what each column means. PaySim is synthetic: a simulator, built from real mobile-money logs, made it. Read the licence on that page too.

Build:

  1. Install Java 21, a long-term-support version, with Homebrew. No sudo needed:

    brew install openjdk@21

    Homebrew installs it "keg-only": it doesn't put it on your path, so the Mac doesn't see it yet. Tell your terminal where it is by adding two lines to ~/.zshrc:

    echo 'export JAVA_HOME=/opt/homebrew/opt/openjdk@21' >> ~/.zshrc
echo 'export PATH="$JAVA_HOME/bin:$PATH"' >> ~/.zshrc
source ~/.zshrc
java -version

    You should see openjdk version "21.... JAVA_HOME is the setting PySpark reads to find Java.

  2. Add PySpark and NetworkX to a separate dependency group called pipelines:

    uv add --group pipelines pyspark networkx

    From now on, pipeline scripts run with uv run --group pipelines .... Your app, tests and CI don't install them.

  3. Download PaySim into data/raw/, which is already in .gitignore. Either click "Download" on the Kaggle page (you need a free Kaggle account) and unzip the file into data/raw/, or use the Kaggle command line:

    uvx kaggle datasets download ealaxi/paysim1 -p data/raw
unzip data/raw/paysim1.zip -d data/raw
ls -lh data/raw

    You should see PS_20174392719_1491204439457_log.csv, about 470 MB. When this lesson was tested in October 2026, the download worked without logging in. If yours asks for credentials, create a token at kaggle.com/settings/api and run export KAGGLE_API_TOKEN=... in your terminal. Never paste the token into a file inside your repo; the repo is public.

  4. Make a 200,000-row sample for fast experiments. The full file takes Spark about 10 seconds to read on a recent Mac; the sample takes 2:

    head -n 200001 data/raw/PS_20174392719_1491204439457_log.csv > data/raw/paysim_sample.csv

  5. A first look, pipelines/explore_paysim.py:

    """A first look at PaySim with Spark.

Run with: uv run --group pipelines python pipelines/explore_paysim.py
"""

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[*]").appName("explore").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

df = spark.read.csv("data/raw/paysim_sample.csv", header=True, inferSchema=True)
df.printSchema()
df.show(5)

print("rows:", df.count())
df.groupBy("type").agg(
    F.count("*").alias("rows"),
    F.sum("isFraud").alias("fraud"),
).orderBy("rows", ascending=False).show()

print("distinct senders:", df.select("nameOrig").distinct().count())
print("distinct receivers:", df.select("nameDest").distinct().count())
spark.stop()

    uv run --group pipelines python pipelines/explore_paysim.py

    Spark prints a few warning lines as it starts; ignore them. Then change the file name to the full CSV and run it again.

  6. Your turn: answer three questions in README.md, with the numbers from the full file:

    • Which transaction types contain any fraud at all?
    • How many distinct senders are there, compared with the number of rows? What does that mean for "per-sender history" features?
    • What is step? (Hint: the dataset page says. Find the largest value.)

Done when: java -version works in a new terminal, the full file loads, and your README has the three answers.

Stuck?
  • Unable to locate a Java Runtime or JAVA_HOME is not set: open a new terminal so ~/.zshrc is read again, then echo $JAVA_HOME. It should print /opt/homebrew/opt/openjdk@21.
  • JAVA_GATEWAY_EXITED or Java gateway process exited before sending its port number: Spark couldn't start Java. Same fix as above; also check java -version says 17 or later.
  • FutureWarning: PySpark does not yet fully support pandas >= 3.0.0: harmless here. You never convert Spark data to pandas this week.
  • ModuleNotFoundError: No module named 'pyspark': you forgot --group pipelines after uv run.
  • uv sync removes PySpark, because it installs exactly the default groups. The next uv run --group pipelines ... puts it back from uv's cache in about a second. That's normal.
  • The three answers, from the full file: fraud appears only in TRANSFER (4,097) and CASH_OUT (4,116). There are 6,353,307 distinct senders in 6,362,620 rows, so nearly every sender appears once: sender history is empty and useless. Receivers repeat (2,722,362 distinct), so features go on the receiving account. step is one hour of simulated time; the largest is 743, about 31 days.

Tuesday — Clean, then build point-in-time features

Learn (60 min):

  • Window functions in Spark SQL (25 min): the PARTITION BY, ORDER BY and frame (RANGE BETWEEN) parts.
  • Window in the PySpark reference (15 min): the same ideas as Python methods. Read rangeBetween closely.
  • Re-read the "Point-in-time features" concept above, then write down three features you'd want for a mule account, from your own work.

Build:

  1. The feature pipeline, pipelines/paysim_features.py. Read every comment; the window definitions are the heart of it:

    """Load PaySim, clean it, and build point-in-time features with window functions.

Run with: uv run --group pipelines python pipelines/paysim_features.py
Writes: data/features/paysim/ (Parquet)
"""

import sys
import time

from pyspark.sql import DataFrame, SparkSession, Window
from pyspark.sql import functions as F
from pyspark.sql.types import (
    DoubleType,
    IntegerType,
    StringType,
    StructField,
    StructType,
)

RAW = "data/raw/PS_20174392719_1491204439457_log.csv"
OUT = "data/features/paysim"

SCHEMA = StructType(
    [
        StructField("step", IntegerType()),
        StructField("type", StringType()),
        StructField("amount", DoubleType()),
        StructField("nameOrig", StringType()),
        StructField("oldbalanceOrg", DoubleType()),
        StructField("newbalanceOrig", DoubleType()),
        StructField("nameDest", StringType()),
        StructField("oldbalanceDest", DoubleType()),
        StructField("newbalanceDest", DoubleType()),
        StructField("isFraud", IntegerType()),
        StructField("isFlaggedFraud", IntegerType()),
    ]
)


def load(spark: SparkSession, path: str) -> DataFrame:
    return spark.read.csv(path, header=True, schema=SCHEMA)


def clean(df: DataFrame) -> DataFrame:
    """Keep the two transaction types where fraud happens, and drop impossible rows."""
    return (
        df.filter(F.col("type").isin("TRANSFER", "CASH_OUT"))
        .filter(F.col("amount") > 0)
        .dropna(subset=["step", "amount", "nameOrig", "nameDest"])
    )


def add_features(df: DataFrame) -> DataFrame:
    """Per-account features that only look at the past (earlier steps)."""
    # Every earlier transaction into the same destination account.
    dest_history = (
        Window.partitionBy("nameDest")
        .orderBy("step")
        .rangeBetween(Window.unboundedPreceding, -1)
    )
    # Transactions into the same destination in the 24 hours (steps) before this one.
    dest_last_24h = Window.partitionBy("nameDest").orderBy("step").rangeBetween(-24, -1)
    dest_zero = (F.col("oldbalanceDest") == 0) & (F.col("newbalanceDest") == 0)

    return (
        df.withColumn("is_transfer", (F.col("type") == "TRANSFER").cast("int"))
        .withColumn("hour_of_day", F.col("step") % 24)
        .withColumn("log_amount", F.log1p("amount"))
        # Balance features: what an analyst would compute from the ledger.
        .withColumn(
            "drains_account", (F.col("amount") == F.col("oldbalanceOrg")).cast("int")
        )
        .withColumn(
            "orig_balance_error",
            F.col("oldbalanceOrg") - F.col("amount") - F.col("newbalanceOrig"),
        )
        .withColumn("dest_balance_zero", dest_zero.cast("int"))
        # Destination-account history: the "mule account" view.
        .withColumn("dest_prev_count", F.count("*").over(dest_history))
        .withColumn(
            "dest_prev_amount",
            F.coalesce(F.sum("amount").over(dest_history), F.lit(0.0)),
        )
        .withColumn("dest_count_24h", F.count("*").over(dest_last_24h))
        .withColumn(
            "dest_amount_24h",
            F.coalesce(F.sum("amount").over(dest_last_24h), F.lit(0.0)),
        )
        .withColumn(
            "hours_since_dest_seen",
            F.coalesce(F.col("step") - F.max("step").over(dest_history), F.lit(-1)),
        )
    )


def main() -> None:
    started = time.time()
    spark = (
        SparkSession.builder.master("local[*]")
        .appName("paysim-features")
        .config("spark.driver.memory", "4g")
        .config("spark.ui.showConsoleProgress", "false")
        .getOrCreate()
    )
    spark.sparkContext.setLogLevel("ERROR")

    raw_path = sys.argv[1] if len(sys.argv) > 1 else RAW
    raw = load(spark, raw_path)
    features = add_features(clean(raw))
    features.write.mode("overwrite").parquet(OUT)

    saved = spark.read.parquet(OUT)
    print(f"raw rows:      {raw.count():,}")
    fraud = saved.filter("isFraud = 1").count()
    print(f"feature rows:  {saved.count():,}  fraud: {fraud:,}")
    saved.select(
        "step",
        "type",
        "amount",
        "nameDest",
        "dest_prev_count",
        "dest_count_24h",
        "isFraud",
    ).show(5)
    print(f"done in {time.time() - started:.0f}s")
    spark.stop()


if __name__ == "__main__":
    main()

    A few things to notice:

    • The schema is written out instead of guessed (inferSchema). Guessing reads the file twice and can guess wrong; a written schema is also documentation.
    • rangeBetween(-24, -1) works on the value of step, not on row counts. It means "steps from 24 before to 1 before this one", so two transactions in the same hour never see each other. That's the point-in-time rule.
    • The functions are small and separate (load, clean, add_features), so you can test them on a tiny DataFrame later.
    • data/features/ should go in .gitignore, like data/raw/.

  2. Run it on the sample first, then on the full file:

    uv run --group pipelines python pipelines/paysim_features.py data/raw/paysim_sample.csv
uv run --group pipelines python pipelines/paysim_features.py

    On the full file you should see about this, in 10 to 30 seconds on a recent Mac:

    raw rows:      6,362,620
feature rows:  2,770,393  fraud: 8,197

    Look at the five sample rows. For account C100028684, dest_prev_count goes 0, 1, 2 as later transfers arrive. That's the window working.

  3. Your turn: add one feature from your own list. A good first one is amount_vs_dest_mean: this amount divided by the average of the account's earlier amounts, and -1 when there is no history. Then check whether it separates fraud from genuine transactions: group by isFraud and look at the median.

  4. Commit. Write in the README which features you built and why each one is point-in-time.

Done when: data/features/paysim/ exists from the full file, and your README lists each feature with one line on what it means to a fraud analyst.

Stuck? One version of the extra feature

Add this inside add_features, after dest_prev_amount:

.withColumn(
    "amount_vs_dest_mean",
    F.when(
        F.col("dest_prev_count") > 0,
        F.col("amount") / (F.col("dest_prev_amount") / F.col("dest_prev_count")),
    ).otherwise(F.lit(-1.0)),
)

Check it with df.groupBy("isFraud").agg(F.expr("percentile_approx(amount_vs_dest_mean, 0.5)")).show(). On the 200,000-row sample, the median for fraud was -1: most fraud goes to accounts with no earlier history. "Brand-new receiving account" is a real mule signal you'd recognise.

Other errors:

  • AnalysisException: ... cannot resolve 'nameDest': a column name is misspelt. PaySim spells it oldbalanceOrg (no "i") but newbalanceOrig.
  • java.lang.OutOfMemoryError: raise spark.driver.memory to "6g", or close other apps.
  • Path does not exist: run from the project folder, not from pipelines/.

Wednesday — Your rules against a model

Learn (75 min):

  • Classification and regression in Spark MLlib (30 min): read the "Gradient-boosted tree classifier" section. A gradient-boosted tree is a set of small decision trees, each correcting the last one's mistakes. It thinks in thresholds, like your rules do.
  • Precision-recall curves in the scikit-learn docs (20 min), for what PR-AUC means. You know precision and recall; PR-AUC summarises them across every threshold in one number.
  • Before you write code, write in your notebook (paper) the two rules you'd have shipped in your old job for this data.

Why Spark MLlib and not scikit-learn here: both would work. scikit-learn would need you to move 2.7 million rows into pandas first, and on a real 600-million-row table that step breaks. MLlib keeps everything in Spark, which is what the job postings mean by "PySpark". Mention in interviews that for a smaller table you'd often train with scikit-learn or LightGBM on aggregated features; it's faster to iterate.

Build:

  1. The comparison, pipelines/paysim_model.py. It trains on the first 400 hours and tests on the rest:

    """Compare rules with a gradient-boosted model on the PaySim features.

Run with: uv run --group pipelines python pipelines/paysim_model.py
Reads: data/features/paysim/ (from paysim_features.py)
"""

import time

from pyspark.ml import Pipeline
from pyspark.ml.classification import GBTClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.functions import vector_to_array
from pyspark.sql import Column, DataFrame, SparkSession
from pyspark.sql import functions as F

FEATURES = "data/features/paysim"
SPLIT_STEP = 400  # train on hours 1-400, test on hours 401-743
ALERT_BUDGET = 1000  # how many alerts an investigations team could review

BEHAVIOUR = [
    "log_amount",
    "is_transfer",
    "hour_of_day",
    "dest_prev_count",
    "dest_prev_amount",
    "dest_count_24h",
    "dest_amount_24h",
    "hours_since_dest_seen",
]
BALANCES = ["drains_account", "orig_balance_error", "dest_balance_zero"]


def rules() -> dict[str, Column]:
    """Each rule is a true/false column. Inside a function: F.col needs Spark running."""
    is_transfer = F.col("type") == "TRANSFER"
    return {
        "simulator flag (isFlaggedFraud)": F.col("isFlaggedFraud") == 1,
        "big transfer > 200k": is_transfer & (F.col("amount") > 200_000),
    }


def precision_recall(df: DataFrame, flagged: Column) -> tuple[float, float, int]:
    row = df.select(
        F.sum((flagged & (F.col("isFraud") == 1)).cast("int")).alias("tp"),
        F.sum(flagged.cast("int")).alias("alerts"),
        F.sum("isFraud").alias("fraud"),
    ).first()
    assert row is not None
    tp, alerts, fraud = row["tp"] or 0, row["alerts"] or 0, row["fraud"]
    return (tp / alerts if alerts else 0.0), tp / fraud, alerts


def train_and_score(train: DataFrame, test: DataFrame, columns: list[str]) -> DataFrame:
    assembler = VectorAssembler(inputCols=columns, outputCol="features")
    gbt = GBTClassifier(
        labelCol="isFraud", featuresCol="features", maxIter=30, maxDepth=5, seed=42
    )
    model = Pipeline(stages=[assembler, gbt]).fit(train)
    scored = model.transform(test)
    return scored.withColumn("score", vector_to_array("probability")[1])


def report(name: str, scored: DataFrame) -> None:
    pr_auc = BinaryClassificationEvaluator(
        labelCol="isFraud", rawPredictionCol="rawPrediction", metricName="areaUnderPR"
    ).evaluate(scored)
    p, r, alerts = precision_recall(scored, F.col("prediction") == 1)
    top = scored.orderBy(F.desc("score")).limit(ALERT_BUDGET)
    caught = top.filter("isFraud = 1").count()
    print(f"{name:34} precision {p:.3f}  recall {r:.3f}  alerts {alerts:>7,}")
    print(f"{'':34} PR-AUC {pr_auc:.3f}; fraud in top {ALERT_BUDGET:,}: {caught}")


def main() -> None:
    started = time.time()
    spark = (
        SparkSession.builder.master("local[*]")
        .appName("paysim-model")
        .config("spark.driver.memory", "4g")
        .config("spark.ui.showConsoleProgress", "false")
        .getOrCreate()
    )
    spark.sparkContext.setLogLevel("ERROR")

    df = spark.read.parquet(FEATURES)
    train = df.filter(F.col("step") <= SPLIT_STEP).cache()
    test = df.filter(F.col("step") > SPLIT_STEP).cache()
    for name, part in [("train", train), ("test", test)]:
        n, fraud = part.count(), part.filter("isFraud = 1").count()
        print(f"{name}: {n:,} rows, {fraud:,} fraud ({fraud / n:.2%})")

    print("\nRules, on the test period:")
    for name, rule in rules().items():
        p, r, alerts = precision_recall(test, rule)
        print(f"{name:34} precision {p:.3f}  recall {r:.3f}  alerts {alerts:>7,}")

    print("\nModels, on the test period:")
    report("GBT, behaviour only", train_and_score(train, test, BEHAVIOUR))
    with_balances = BEHAVIOUR + BALANCES
    report("GBT, behaviour + balances", train_and_score(train, test, with_balances))

    print(f"\ndone in {time.time() - started:.0f}s")
    spark.stop()


if __name__ == "__main__":
    main()

    The model is trained twice: once on behaviour features only (amount, time, receiving-account history), and once with the balance features added. Keep that in mind when you read the results.

  2. Your turn: add your two rules from this morning to rules(), then run it. It takes about a minute:

    uv run --group pipelines python pipelines/paysim_model.py

  3. Read the output like an analyst, not a data scientist. Write a short table in the README: each rule and model, its precision, recall and alert count. Then answer:

    • The test period has a much higher fraud rate than the training period (the first lines of output show it). Why might that be, and which numbers does it affect?
    • Which is better for an investigations team with 1,000 reviews a day: the best rule or the best model?
    • Is any result too good to be true? What would you check before trusting it on real data?

Done when: your README has the table and three short answers, and the comparison includes at least one rule you wrote.

Stuck? What the numbers looked like, and a rule that nearly solves it

With a rule called "account drained" (F.col("drains_account") == 1: the amount equals the sender's whole balance) added, the test period gave:

train: 2,529,532 rows, 4,473 fraud (0.18%)
test: 240,861 rows, 3,724 fraud (1.55%)

Rules, on the test period:
simulator flag (isFlaggedFraud)    precision 1.000  recall 0.003  alerts      12
big transfer > 200k                precision 0.034  recall 0.342  alerts  37,638
account drained                    precision 1.000  recall 0.976  alerts   3,633

Models, on the test period:
GBT, behaviour only                precision 0.334  recall 0.297  alerts   3,310
                                   PR-AUC 0.265; fraud in top 1,000: 114
GBT, behaviour + balances          precision 1.000  recall 0.998  alerts   3,718
                                   PR-AUC 0.998; fraud in top 1,000: 1000

One "account drained" rule catches 97.6% of fraud with no false alarms. That's a fact about the simulator: in PaySim, fraudsters always empty the account. Real fraud is never this clean, so a result like this on real data should make you look for leakage first. The honest story for an interview: "on PaySim the balance columns give it away; without them, behaviour features alone reach a PR-AUC of about 0.27, and the alert budget is what decides whether that's useful."

The fraud rate jumps in the test period because the simulator's normal traffic thins out after about hour 400 while fraud carries on. Precision depends on the base rate, so compare precision only within one period.

Other errors:

  • AssertionError from F.col at import time: you defined rules outside a function. F.col needs a running Spark session, which is why rules() is a function.
  • IllegalArgumentException: ... features ... does not exist: a name in BEHAVIOUR doesn't match a column you built yesterday.

Thursday — Scam reports as a graph, and a tool from it

Learn (60 min):

  • The NetworkX tutorial (40 min), up to "Analyzing graphs".
  • Components in the NetworkX reference (5 min): connected_components is the one you need.
  • Think back to your own case work: what made you sure two reports were the same group? What links fooled you?

There is no public dataset of Singapore scam reports with their phone numbers and bank accounts, for good reason. So you generate a synthetic one where you know the true groups, and use it to test your linking method before it ever meets real data.

Build:

  1. A generator, pipelines/make_scam_reports.py:

    """Generate synthetic scam reports that share phone numbers, links and bank accounts.

Run with: uv run --group pipelines python pipelines/make_scam_reports.py
Writes: data/scam_reports.csv

Each fake syndicate owns a small pool of identifiers and reuses them across
reports, the way real scam groups reuse mule accounts and phone numbers.
The true syndicate is kept in a column so you can check the clustering later.
Everything here is made up: numbers start with 8 or 9 but are random.
"""

import random

import pandas as pd

N_SYNDICATES = 40
N_LONE_REPORTS = 300
SCAM_TYPES = ["job", "phishing", "investment", "impersonation", "ecommerce", "love"]
OFFICIAL_HOTLINE = "1800 000 0000"  # stands in for an official hotline victims quote

rng = random.Random(42)


def phone() -> str:
    return f"{rng.choice('89')}{rng.randint(0, 9999999):07d}"


def url() -> str:
    word = rng.choice(["secure", "verify", "parcel", "jobs", "invest", "refund"])
    tld = rng.choice(["com", "net", "xyz", "top", "info"])
    return f"https://{word}-{rng.randint(100, 999)}.{tld}"


def bank_account() -> str:
    return f"{rng.randint(100, 999)}-{rng.randint(10000, 99999)}-{rng.randint(0, 9)}"


def make_report(n: int, syndicate: str, scam_type: str, pool: dict) -> dict:
    return {
        "report_id": f"R{n:05d}",
        "reported_day": rng.randint(1, 90),
        "scam_type": scam_type,
        # Each report mentions only some of the syndicate's identifiers.
        "phone": rng.choice(pool["phone"]) if rng.random() < 0.8 else "",
        "url": rng.choice(pool["url"]) if rng.random() < 0.5 else "",
        "bank_account": rng.choice(pool["bank_account"]) if rng.random() < 0.6 else "",
        "true_syndicate": syndicate,
    }


def main() -> None:
    rows = []
    for s in range(N_SYNDICATES):
        size = rng.choice([3, 5, 8, 12, 20, 30])  # a few big groups, many small ones
        pool = {
            "phone": [phone() for _ in range(rng.randint(1, 4))],
            "url": [url() for _ in range(rng.randint(1, 3))],
            "bank_account": [bank_account() for _ in range(rng.randint(1, 3))],
        }
        scam_type = rng.choice(SCAM_TYPES)
        for _ in range(size):
            rows.append(make_report(len(rows), f"S{s:02d}", scam_type, pool))

    for _ in range(N_LONE_REPORTS):  # one-off scammers with their own identifiers
        lone = {"phone": [phone()], "url": [url()], "bank_account": [bank_account()]}
        rows.append(make_report(len(rows), "none", rng.choice(SCAM_TYPES), lone))

    # Noise: some victims write down the official hotline they called afterwards.
    for row in rng.sample(rows, 25):
        row["phone"] = OFFICIAL_HOTLINE

    df = pd.DataFrame(rows).sample(frac=1, random_state=42)  # shuffle
    df.to_csv("data/scam_reports.csv", index=False)
    print(f"{len(df)} reports, {N_SYNDICATES} syndicates")
    print(df.head())


if __name__ == "__main__":
    main()

    uv run --group pipelines python pipelines/make_scam_reports.py

    It prints 805 reports, 40 syndicates. The hotline noise at the end is deliberate. You'll see why in a moment.

  2. The tool's lookup code comes first, because the graph needs it too: both must spell +65 9123 4567 and 91234567 the same way. Create src/scam_checker/tools/linked_reports.py:

    """Tool: how many linked scam reports mention this phone number, link or bank account."""

import re
from functools import lru_cache
from pathlib import Path

import pandas as pd

LINKS_PATH = Path("data/linked_reports.csv")


def normalise(value: str) -> str:
    """One spelling per identifier, so '+65 9123 4567' and '91234567' match."""
    value = value.strip().lower()
    if re.fullmatch(r"[\d\s+()-]+", value):  # looks like a phone or account number
        digits = re.sub(r"\D", "", value)
        if len(digits) == 10 and digits.startswith("65"):
            digits = digits[2:]  # drop Singapore's country code
        return digits
    value = re.sub(r"^https?://", "", value)
    value = re.sub(r"^www\.", "", value)
    return value.rstrip("/")


@lru_cache(maxsize=1)
def _links() -> pd.DataFrame:
    df = pd.read_csv(LINKS_PATH, dtype=str)
    return df.set_index("identifier")


def linked_reports(identifier: str) -> str:
    """Look up a phone number, link or bank account in community scam reports.

    Use this for any phone number, URL or bank account in the message.
    Returns how many reports mention it and how big its linked cluster is.
    """
    key = normalise(identifier)
    try:
        links = _links()
    except FileNotFoundError:
        return "Linked-report data is not available; do not rely on this check."
    if key not in links.index:
        return (
            f"'{identifier}' appears in no known scam reports. "
            "Absence is not proof it is safe."
        )
    row = links.loc[key]
    return (
        f"'{identifier}' appears in {row['identifier_reports']} scam reports. "
        f"Those reports link to a cluster of {row['cluster_reports']} reports sharing "
        f"{row['cluster_identifiers']} phone numbers, links and bank accounts."
    )

    @lru_cache loads the CSV once and keeps it in memory, so the agent can call the tool many times cheaply. The docstring is what the model reads to decide when to call the tool, so it's written as an instruction.

  3. Link the reports, pipelines/link_reports.py. Start with an empty IGNORE set:

    """Link scam reports that share identifiers, and find connected clusters.

Run with: uv run --group pipelines python pipelines/link_reports.py
Reads: data/scam_reports.csv     Writes: data/linked_reports.csv
"""

import networkx as nx
import pandas as pd

from scam_checker.tools.linked_reports import normalise

IDENTIFIER_COLUMNS = ["phone", "url", "bank_account"]
IGNORE: set[str] = set()


def build_graph(reports: pd.DataFrame) -> nx.Graph:
    """Two kinds of node: reports and identifiers. An edge means 'this report mentions it'."""
    graph = nx.Graph()
    for row in reports.itertuples():
        graph.add_node(row.report_id, kind="report")
        for column in IDENTIFIER_COLUMNS:
            value = getattr(row, column)
            if isinstance(value, str) and value and normalise(value) not in IGNORE:
                graph.add_node(normalise(value), kind=column)
                graph.add_edge(row.report_id, normalise(value))
    return graph


def cluster_table(graph: nx.Graph) -> pd.DataFrame:
    """One row per identifier: its own report count and its cluster's size."""
    rows = []
    for cluster_id, nodes in enumerate(nx.connected_components(graph)):
        reports = [n for n in nodes if graph.nodes[n]["kind"] == "report"]
        identifiers = [n for n in nodes if graph.nodes[n]["kind"] != "report"]
        for identifier in identifiers:
            rows.append(
                {
                    "identifier": identifier,
                    "kind": graph.nodes[identifier]["kind"],
                    "identifier_reports": graph.degree(identifier),
                    "cluster_id": cluster_id,
                    "cluster_reports": len(reports),
                    "cluster_identifiers": len(identifiers),
                }
            )
    return pd.DataFrame(rows)


def main() -> None:
    reports = pd.read_csv("data/scam_reports.csv", dtype=str)
    graph = build_graph(reports)
    table = cluster_table(graph)
    table.to_csv("data/linked_reports.csv", index=False)

    sizes = table.drop_duplicates("cluster_id")["cluster_reports"]
    print(f"{graph.number_of_nodes()} nodes, {graph.number_of_edges()} edges")
    print(f"{len(sizes)} clusters; {(sizes >= 3).sum()} with 3 or more reports")
    print("largest clusters (reports):", sorted(sizes, reverse=True)[:8])


if __name__ == "__main__":
    main()

    uv run --group pipelines python pipelines/link_reports.py

    You should see a largest cluster of 212 reports. No syndicate in the generator has more than 30.

  4. Score the clusters against the truth, pipelines/score_clusters.py. Purity is the share of a cluster's reports that come from its most common true syndicate; 1.00 means every cluster is one real group:

    """Check the clusters against the true syndicates (only possible on synthetic data).

Run with: uv run --group pipelines python pipelines/score_clusters.py
"""

import networkx as nx
import pandas as pd
from link_reports import build_graph

reports = pd.read_csv("data/scam_reports.csv", dtype=str).set_index("report_id")
graph = build_graph(reports.reset_index())

rows = []
for nodes in nx.connected_components(graph):
    ids = [n for n in nodes if graph.nodes[n]["kind"] == "report"]
    if len(ids) < 2:
        continue
    truth = reports.loc[ids, "true_syndicate"]
    purity = truth.value_counts(normalize=True).iloc[0]  # share from the top syndicate
    rows.append({"reports": len(ids), "purity": purity})

clusters = pd.DataFrame(rows)
weighted = (clusters.purity * clusters.reports).sum() / clusters.reports.sum()
print(f"{len(clusters)} clusters with 2+ reports")
print(f"weighted purity: {weighted:.2f}")
print(f"clusters below 90% pure: {(clusters.purity < 0.9).sum()}")

    With the empty IGNORE, weighted purity is about 0.63.

  5. Your turn: find out what joined unrelated syndicates into the 212-report cluster, then fix it in IGNORE. Look at the identifiers with the most reports, and ask what an investigator would say about each one. Re-run both scripts and record purity before and after in the README.

  6. Test the tool with the network and the real data out of the way, in tests/test_linked_reports.py. The fixture runs each test in an empty temporary folder with a tiny CSV:

    import pandas as pd
import pytest

from scam_checker.tools.linked_reports import _links, linked_reports, normalise


@pytest.fixture
def in_tmp_project(tmp_path, monkeypatch):
    """Run the test inside an empty folder with a tiny data/linked_reports.csv."""
    (tmp_path / "data").mkdir()
    row = {
        "identifier": "91234567",
        "kind": "phone",
        "identifier_reports": 14,
        "cluster_id": 3,
        "cluster_reports": 23,
        "cluster_identifiers": 9,
    }
    pd.DataFrame([row]).to_csv(tmp_path / "data" / "linked_reports.csv", index=False)
    monkeypatch.chdir(tmp_path)
    _links.cache_clear()  # forget data loaded by an earlier test
    yield tmp_path
    _links.cache_clear()


@pytest.mark.parametrize("raw", ["91234567", "+65 9123 4567", "9123-4567"])
def test_phone_formats_match(in_tmp_project, raw):
    assert "appears in 14 scam reports" in linked_reports(raw)


def test_unknown_number_is_not_called_safe(in_tmp_project):
    assert "not proof" in linked_reports("80000000")


def test_missing_data_file(in_tmp_project):
    (in_tmp_project / "data" / "linked_reports.csv").unlink()
    assert "not available" in linked_reports("91234567")


def test_urls_normalised():
    assert normalise("HTTPS://www.Parcel-123.xyz/") == "parcel-123.xyz"

    uv run pytest tests/test_linked_reports.py -q

  7. Add the tool to your agent. Register linked_reports in TOOLS the same way as your week 2 tools; if TOOLS maps names to functions, it's one line: "linked_reports": linked_reports. Then add one line to the agent's system prompt: "Run linked_reports on every phone number, link and bank account in the message, and if it finds reports, say how many in a red flag."

  8. Try it with a number from your data. Pick a busy one:

    uv run python -c "from scam_checker.tools.linked_reports import linked_reports; print(linked_reports('+65 9221 9862'))"

    With the seed above it prints '+65 9221 9862' appears in 26 scam reports. Those reports link to a cluster of 29 reports sharing 7 phone numbers, links and bank accounts. Your numbers may differ if you changed the generator. Then run the agent on a job-scam message containing that number, and check its saved run in runs/ to confirm the tool was called.

Done when: purity is above 0.95 with your fix, the tool tests pass, and one saved agent run shows linked_reports being called and its count quoted in a red flag.

Stuck? The giant cluster

Sort the output by identifier_reports: the top identifier is 18000000000, the stand-in hotline. Victims from many syndicates wrote it down because they called it afterwards, so it links groups that have nothing to do with each other. An investigator would never treat it as a link. Fix:

# Identifiers that link unrelated reports: official numbers victims quote.
IGNORE = {normalise("1800 000 0000")}

After the fix: largest clusters (reports): [30, 30, 29, 28, 28, 27, 20, 20], 43 clusters with 2 or more reports, weighted purity 1.00. On real data you'd keep a reviewed list of official numbers and domains (banks, government hotlines, gov.sg), and also be wary of any identifier linked to far more reports than a typical syndicate.

Other errors:

  • ModuleNotFoundError: No module named 'link_reports' in score_clusters.py: run it from the project folder as python pipelines/score_clusters.py; Python then looks in pipelines/ for imports.
  • The tool always says "no known scam reports": the CSV was read with numbers as integers, which drops leading zeros and breaks matching. Keep dtype=str.
  • The agent answers without calling the tool: make the instruction in the system prompt explicit, or call linked_reports yourself on every number before the agent starts and put the result in its prompt. That second pattern is less flexible but you can test it without a model.

Friday — Demo day, then apply

Learn (30 min): skim the graph section of a fraud job posting you want, and list which of this week's words appear in it (window features, PySpark, graph, entity resolution, connected components, mule accounts).

Build (before 9:00 or after demo day):

  1. Make sure the pipeline runs end to end from a clean state, in order:

    uv run --group pipelines python pipelines/paysim_features.py
uv run --group pipelines python pipelines/paysim_model.py
uv run --group pipelines python pipelines/make_scam_reports.py
uv run --group pipelines python pipelines/link_reports.py
uv run --group pipelines python pipelines/score_clusters.py

    Put these five lines in the README under "How to reproduce", with one sentence on where to get PaySim.

  2. Commit data/scam_reports.csv and data/linked_reports.csv: they're synthetic and small, and the tool needs the second one. Keep PaySim and the Parquet features out of git.

  3. Run uv run ruff check . and uv run pytest one last time, and commit.

Demo day (9:00–12:00):

  1. Record a 3-minute screen video: the pipeline running, the rules-versus-model table, the cluster purity before and after your fix, and the agent quoting a linked-report count.
  2. Write five lines on what you learned this week.
  3. Send both to Claude with the Friday check-in.

Apply now

This is the week to apply to anti-fraud and anti-scam engineering roles, such as GovTech's Singpass anti-fraud and anti-scam teams. In the application, lead with one sentence that joins your old work and this week: for example, "ten years writing fraud rules; I built a PySpark pipeline on 6 million transactions that compares my rules against a gradient-boosted model, and a graph that links scam reports into syndicates." Link the repo and the README table.

Deliverable checklist

  • Java 21 installed and PySpark running locally, in a pipelines dependency group
  • pipelines/paysim_features.py building point-in-time window features on the full PaySim file
  • pipelines/paysim_model.py comparing at least one rule you wrote with a model, and a README table of the results
  • Synthetic scam reports linked as a graph, with cluster purity before and after your fix
  • src/scam_checker/tools/linked_reports.py with tests, registered in the agent's TOOLS
  • A saved agent run that quotes a linked-report count
  • Applications sent to at least two anti-fraud or anti-scam roles
  • Demo video, five lines, and the Friday check-in sent

If you have time left: redo the clustering in Spark, so it would scale to millions of reports. GraphFrames adds graphs to Spark DataFrames and has a connected-components function; follow its install page for your Spark version (not tested for this course). Or drop the balance features and spend an hour improving the behaviour-only model: that's the version closest to real fraud work.