Investigate Transaction Anomalies

Build a pandas pipeline to score, evaluate, and rank transaction anomalies.

Introduction

30 Second Summary

Transaction alerts can pile up faster than an investigator can review them. A useful alert system surfaces unusual behavior without burying real problems under false alarms.

In this project, you will build a transaction investigation engine powered by anomaly detection in Google Colab with NumPy plus pandas. The finished notebook ranks known injected anomalies into an investigation-ready CSV with a plain-language reason for every alert.

What You'll Build

One notebook run takes you from a visible relational join failure to a severity-ranked investigation report with explanations for every alert.

By the end of this project, you'll have:

  • A reproducible transaction simulation with 800 transactions across 40 customers. You'll catch duplicated rows plus missing account matches before validating the analysis table.
  • A detector scorecard where you compare three methods through confusion-matrix counts plus precision and recall.
  • A ranked alert report where every flagged transaction has a human-readable explanation before export to investigation_alerts.csv.
  • Secret Mission: Add customer-channel rarity to the severity score for an enhanced investigation report.

Do I need experience before starting?

Basic NumPy plus pandas experience is enough. Google Colab supplies the hosted runtime without local Python or VS Code.

Before We Start

Before the analysis begins, take a moment to connect the detector to the investigation problem. Your goal is to identify unusually large transactions using each customer's historical behavior and spending category as context.

Set Up the Pinned Colab Runtime

Your anomaly engine needs a stable runtime before any transaction data enters it. Google Colab gives the notebook a browser-based place to run.

Colab runtimes change over time. Pinning runtime 2026.07 keeps Python 3.12.13, NumPy 2.0.2, and pandas 2.2.2 aligned with the versions this notebook expects.

The version-check cell stops immediately if NumPy or pandas differs. That early failure protects every later join and rolling calculation from subtle package differences.

In this step, get ready to:
  • Create a Colab notebook named Transaction Anomaly Investigation Engine.
  • Pin the notebook to runtime 2026.07.
  • Build a version-check cell for NumPy and pandas.
Create and pin the notebook

A pinned runtime fixes the software environment that executes your cells. This gives every later result the same package foundation.

  • Go to Google Colab in your browser.
  • Create a blank notebook from the Colab landing page.
  • Click the notebook title in the upper-left corner.
  • Replace the current title with Transaction Anomaly Investigation Engine.

You're underway. The notebook now has a clear name that identifies the investigation workflow.

  • Open the Runtime menu.
  • Select Change runtime type.
  • Open the Runtime Version dropdown.
  • Select 2026.07.
  • Apply the runtime selection using the control at the bottom of the dialog.

The runtime field should now display 2026.07. This selection provides Python 3.12.13 with the package versions required by the notebook.

✔️ The pinned runtime is selected

The Runtime Version field shows 2026.07. Continue to the first code cell.

ⓧ Another runtime is selected

  • Open the Runtime menu.
  • Select Change runtime type.
  • Choose 2026.07 from the Runtime Version dropdown.
  • Apply the updated runtime selection using the control at the bottom of the dialog.

ⓧ The pinned runtime is missing

The runtime list can require a fresh page load before a past runtime appears. Keep the notebook title unchanged while you refresh the page.

  • Reload the notebook page in your browser.
  • Open the Runtime menu again.
  • Select Change runtime type.
  • Check the Runtime Version dropdown for 2026.07.
  • Read the official Colab runtime version guide if the pinned runtime still does not appear.
Add the imports and version checks

Aliases keep repeated library calls concise throughout the notebook. Assertions turn the required versions into executable checks.

  • Paste this complete runtime check into the notebook's first code cell:
# Cell 1: Verify the pinned Colab runtime
import numpy as np
import pandas as pd

assert np.__version__ == "2.0.2", f"Expected NumPy 2.0.2, found {np.__version__}"
assert pd.__version__ == "2.2.2", f"Expected pandas 2.2.2, found {pd.__version__}"
print("Python runtime ready")
print("NumPy", np.__version__)
print("pandas", pd.__version__)

What does this cell check?

  • The import numpy as np line loads NumPy under the conventional np alias.
  • The import pandas as pd line loads pandas under the conventional pd alias.
  • The two assert statements compare the installed versions with 2.0.2 and 2.2.2.
  • The three print calls provide a visible readiness result after both assertions pass.
Run the runtime check

Before you run the cell, do you expect all three readiness lines to print without an assertion stopping execution?

  • Run the first code cell using the circular play control to its left.

You'll see Python runtime ready on the first output line. The next lines show NumPy 2.0.2 and pandas 2.2.2.

That's the environment locked in. Your notebook can now run the investigation with the exact library versions its calculations expect.

Assertion stopping the cell?

An assertion stops the cell when the active runtime contains a different package version. The installed version appears in the assertion details.

  • Return to the Change runtime type dialog.
  • Select 2026.07 from the Runtime Version dropdown.
  • Run the first code cell again.

Still blocked? Help me diagnose my Colab runtime version check.

✔️ Awesome, I've got everything!

Your first cell matches the pinned runtime check. Colab stores the notebook in Google Drive.

ⓧ I'd like to double check the full code

  • Compare your first cell with this complete version.
# Cell 1: Verify the pinned Colab runtime
import numpy as np
import pandas as pd

assert np.__version__ == "2.0.2", f"Expected NumPy 2.0.2, found {np.__version__}"
assert pd.__version__ == "2.2.2", f"Expected pandas 2.2.2, found {pd.__version__}"
print("Python runtime ready")
print("NumPy", np.__version__)
print("pandas", pd.__version__)

Your reproducible runtime is ready. Next, you'll build the transaction tables that put its data checks to work.

Build the Tables and Expose the Bad Join

Your pinned Google Colab runtime is ready. The anomaly engine now needs trustworthy tables before any detector can score a transaction.

A relational join connects records through shared keys. Duplicate keys can multiply transactions. Missing keys can leave incomplete records.

NumPy generates reproducible transaction values. pandas stores those values in tables that expose the join problem.

In this step, get ready to:
  • Create customer and intentionally dirty account tables.
  • Generate 800 ordered transactions with known injected anomalies.
  • Run an unchecked join that exposes duplicated and missing account records.
Create the customer and account tables

Master tables describe stable entities such as customers or accounts. This project starts with a clean customer table plus an account table containing two deliberate data-quality defects.

  • Add a new code cell directly below Cell 1 for the customer table.
  • Create the seeded generator plus the customer table by pasting this first part into the new cell:
# Cell 2: Create customer and intentionally dirty account tables
rng = np.random.default_rng(42)

n_customers = 40
transactions_per_customer = 20
n_transactions = n_customers * transactions_per_customer

customer_ids = np.arange(1000, 1000 + n_customers)
segments = np.tile(["Starter", "Growth", "Premium", "Enterprise"], 10)
base_amounts = np.tile([25.0, 100.0, 400.0, 1600.0], 10)
regions = np.tile(["North", "South", "East", "West"], 10)

customers = pd.DataFrame(
    {
        "customer_id": customer_ids,
        "segment": segments,
        "region": regions,
        "expected_amount": base_amounts,
    }
)

What does this code do?

  • The seeded generator in `rng` makes the synthetic values reproducible whenever you run the notebook from the top.
  • The repeated arrays assign 40 customers across four spending segments plus four regions.
  • The `customers` table gives each customer an expected spending level that matches their segment.
  • Continue in the same cell by pasting the account-table code directly below the closing parenthesis:
account_ids = np.arange(5000, 5000 + n_customers)
account_types = np.where(
    np.arange(n_customers) % 5 == 0,
    "business",
    "personal",
)
tenure_months = 6 + np.arange(n_customers)

accounts_dirty = pd.DataFrame(
    {
        "account_id": list(account_ids[:-1]) + [account_ids[0]],
        "customer_id": list(customer_ids[:-1]) + [customer_ids[0]],
        "account_type": list(account_types[:-1]) + [account_types[0]],
        "tenure_months": list(tenure_months[:-1]) + [tenure_months[0]],
    }
)

Where are the account defects?

  • The intended account IDs run from `5000` through `5039`.
  • The slices stop before the final account record.
  • The appended values repeat the first account record in its place.
  • The resulting `accounts_dirty` table therefore contains 40 rows with one duplicated key plus one missing key.
  • Run Cell 2 to create `customers` plus `accounts_dirty`.

You will see Cell 2 finish without an exception. The runtime now holds 40 customer rows plus 40 intentionally dirty account rows.

Cell 2 not finishing?

Check that Cell 1 has already run in the current runtime. Cell 2 depends on the `np` and `pd` aliases created there.

Check the brackets around each dictionary inside the two DataFrame constructors. A missing bracket prevents the entire cell from running.

Still stuck? Help me troubleshoot the customer and account table cell.

Generate ordered transactions

Known anomaly labels make detector quality measurable later. Each customer receives 20 ordered transactions plus one injected anomaly at transaction order 18.

  • Add a new code cell directly below Cell 2 for the transaction generator.
  • Create the transaction positions plus category factors by pasting this first part into the new cell:
# Cell 3: Generate ordered transactions and inject known anomalies
customer_positions = np.repeat(np.arange(n_customers), transactions_per_customer)
transaction_order = np.tile(np.arange(transactions_per_customer), n_customers)
categories = np.tile(
    ["groceries", "fuel", "travel", "electronics"],
    n_transactions // 4,
)
channels = rng.choice(
    ["card", "mobile", "web", "branch"],
    size=n_transactions,
)
category_factors = np.select(
    [
        categories == "groceries",
        categories == "fuel",
        categories == "travel",
    ],
    [0.6, 1.0, 1.5],
    default=2.5,
)

How is each transaction shaped?

  • The repeated customer positions assign 20 consecutive transactions to each customer.
  • The transaction order cycles from `0` through `19` within every customer.
  • The categories cycle through groceries, fuel, travel, and electronics.
  • The category factors give each spending category a different typical amount.
  • Continue in the same cell by pasting the amount-generation code directly below the closing parenthesis:
normal_noise = rng.lognormal(mean=0.0, sigma=0.12, size=n_transactions)
normal_amounts = (
    np.repeat(base_amounts, transactions_per_customer)
    * category_factors
    * normal_noise
)
injected_mask = transaction_order == 18
amounts = np.where(injected_mask, normal_amounts * 4.0, normal_amounts)

transactions = pd.DataFrame(
    {
        "transaction_id": np.arange(1, n_transactions + 1),
        "account_id": account_ids[customer_positions],
        "transaction_order": transaction_order,
        "category": categories,
        "channel": channels,
        "amount": amounts,
        "is_injected_anomaly": injected_mask,
    }
)

How are the anomalies injected?

  • The lognormal noise gives ordinary transactions small variations around their customer and category baselines.
  • The `injected_mask` selects transaction order `18` for every customer.
  • The selected amounts become four times their normal value.
  • The `is_injected_anomaly` column preserves the ground-truth label for later detector evaluation.
  • Run Cell 3 to create the `transactions` table.

You will see Cell 3 finish without an exception. The runtime now holds ordered transaction records with seven analysis columns.

Transaction cell not finishing?

Run Cell 2 again if the runtime does not recognize `n_customers`, `n_transactions`, `base_amounts`, or `account_ids`.

Check that the category list contains exactly four values. The tiling calculation depends on that four-category cycle.

Need a hand? Help me troubleshoot the transaction generator.

Run the unchecked join

An unchecked merge accepts whatever key relationships it finds. Comparing the source count with the result count reveals whether that trust was justified.

  • Add a new code cell directly below Cell 3.
  • Build the naive join plus its audit summary by pasting this code into the new cell:
# Cell 4: Experience the unchecked-join failure
naive_join = transactions.merge(
    accounts_dirty,
    how="left",
    on="account_id",
)

naive_summary = pd.DataFrame(
    {
        "check": [
            "source transactions",
            "naive joined rows",
            "rows with missing account data",
        ],
        "count": [
            len(transactions),
            len(naive_join),
            int(naive_join["customer_id"].isna().sum()),
        ],
    }
)
print(naive_summary)

What does this audit measure?

  • The left merge keeps every transaction while attaching any matching account data.
  • The first count records the number of source transactions.
  • The second count records the number of rows produced by the merge.
  • The final count measures joined rows where account data could not supply a customer ID.

Before you run this cell, predict whether the joined row count stays at 800.

  • Run Cell 4 to reveal the effect of the unchecked join.

You will see `800` source transactions, `820` naive joined rows, and `20` rows with missing account data. The duplicate account multiplied 20 transactions. The omitted account left another 20 transactions incomplete.

This failure is intentional

The join completed without stopping, yet its output is unsafe for analysis. Row multiplication can inflate alert counts. Missing account data can break customer-specific baselines.

Summary counts look different?

Run Cell 2 before Cell 3. The generated transactions depend on the seeded state plus the arrays created in Cell 2.

Check that `accounts_dirty` uses `account_ids[:-1]` before appending `account_ids[0]`. Changing either slice changes the join failure.

Still seeing different counts? Help me trace the naive join summary.

✔️ Awesome, I've got everything!

You caught the silent data-quality failure. Your notebook now reproduces the dirty tables plus the unsafe join every time it runs from the top.

ⓧ I'd like to double check the full code

# Cell 1: Verify the pinned Colab runtime
import numpy as np
import pandas as pd

assert np.__version__ == "2.0.2", f"Expected NumPy 2.0.2, found {np.__version__}"
assert pd.__version__ == "2.2.2", f"Expected pandas 2.2.2, found {pd.__version__}"
print("Python runtime ready")
print("NumPy", np.__version__)
print("pandas", pd.__version__)

# Cell 2: Create customer and intentionally dirty account tables
rng = np.random.default_rng(42)

n_customers = 40
transactions_per_customer = 20
n_transactions = n_customers * transactions_per_customer

customer_ids = np.arange(1000, 1000 + n_customers)
segments = np.tile(["Starter", "Growth", "Premium", "Enterprise"], 10)
base_amounts = np.tile([25.0, 100.0, 400.0, 1600.0], 10)
regions = np.tile(["North", "South", "East", "West"], 10)

customers = pd.DataFrame(
    {
        "customer_id": customer_ids,
        "segment": segments,
        "region": regions,
        "expected_amount": base_amounts,
    }
)

account_ids = np.arange(5000, 5000 + n_customers)
account_types = np.where(
    np.arange(n_customers) % 5 == 0,
    "business",
    "personal",
)
tenure_months = 6 + np.arange(n_customers)

accounts_dirty = pd.DataFrame(
    {
        "account_id": list(account_ids[:-1]) + [account_ids[0]],
        "customer_id": list(customer_ids[:-1]) + [customer_ids[0]],
        "account_type": list(account_types[:-1]) + [account_types[0]],
        "tenure_months": list(tenure_months[:-1]) + [tenure_months[0]],
    }
)

# Cell 3: Generate ordered transactions and inject known anomalies
customer_positions = np.repeat(np.arange(n_customers), transactions_per_customer)
transaction_order = np.tile(np.arange(transactions_per_customer), n_customers)
categories = np.tile(
    ["groceries", "fuel", "travel", "electronics"],
    n_transactions // 4,
)
channels = rng.choice(
    ["card", "mobile", "web", "branch"],
    size=n_transactions,
)
category_factors = np.select(
    [
        categories == "groceries",
        categories == "fuel",
        categories == "travel",
    ],
    [0.6, 1.0, 1.5],
    default=2.5,
)
normal_noise = rng.lognormal(mean=0.0, sigma=0.12, size=n_transactions)
normal_amounts = (
    np.repeat(base_amounts, transactions_per_customer)
    * category_factors
    * normal_noise
)
injected_mask = transaction_order == 18
amounts = np.where(injected_mask, normal_amounts * 4.0, normal_amounts)

transactions = pd.DataFrame(
    {
        "transaction_id": np.arange(1, n_transactions + 1),
        "account_id": account_ids[customer_positions],
        "transaction_order": transaction_order,
        "category": categories,
        "channel": channels,
        "amount": amounts,
        "is_injected_anomaly": injected_mask,
    }
)

# Cell 4: Experience the unchecked-join failure
naive_join = transactions.merge(
    accounts_dirty,
    how="left",
    on="account_id",
)

naive_summary = pd.DataFrame(
    {
        "check": [
            "source transactions",
            "naive joined rows",
            "rows with missing account data",
        ],
        "count": [
            len(transactions),
            len(naive_join),
            int(naive_join["customer_id"].isna().sum()),
        ],
    }
)
print(naive_summary)

Your bad join is exposed with evidence from the row counts. Next, you will repair the account records plus enforce the intended relationships before analysis begins.

Repair and Validate the Data Model

Your unchecked naive_join exposed duplicate account records that multiplied transactions. It also exposed missing account records that left transactions unmatched.

Those defects make every downstream anomaly score untrustworthy. This step repairs the account data before enforcing the intended relationships in your Google Colab notebook.

Join cardinality defines how many records each key may match. You will use pandas to turn that expectation into a rule that the analysis table must pass.

In this step, get ready to:
  • Identify the duplicate account records.
  • Restore a canonical account table with all 40 accounts.
  • Enforce complete many-to-one relationships across the analysis table.
Audit and repair the account table

Duplicate keys make a master table ambiguous. A canonical account table gives each account_id one authoritative record.

The repair keeps the first record for account 5000. It also reconstructs the omitted account 5039 from the account arrays already in the notebook.

  • Add a code cell below the cell that prints naive_summary.
  • Paste the duplicate audit and account repair logic into the new cell:
# Cell 5: Repair duplicate and missing account records
all_duplicate_account_rows = accounts_dirty[
    accounts_dirty.duplicated(subset=["account_id"], keep=False)
]
accounts = accounts_dirty.drop_duplicates(
    subset=["account_id"],
    keep="first",
).copy()

missing_account = pd.DataFrame(
    {
        "account_id": [account_ids[-1]],
        "customer_id": [customer_ids[-1]],
        "account_type": [account_types[-1]],
        "tenure_months": [tenure_months[-1]],
    }
)
accounts = pd.concat([accounts, missing_account], ignore_index=True)

print("Duplicate account rows found:", len(all_duplicate_account_rows))
print("Clean account rows:", len(accounts))

What Does This Repair Do?

  • The duplicate audit uses keep=False to capture every record associated with the repeated account key.
  • The accounts table keeps one canonical record for each existing account key.
  • The missing_account table reconstructs the omitted final account.
  • The pd.concat() call appends the restored record to the canonical table.
  • Run the account repair cell.

You should see Duplicate account rows found: 2 followed by Clean account rows: 40.

Good catch. Your account master table now contains 40 canonical records.

Still Seeing the Wrong Account Count?

  • Rerun the earlier cells if a runtime restart removed accounts_dirty from memory.
  • Check that drop_duplicates() uses account_id as its subset.
  • Check that missing_account uses the final positions from the account arrays.

If the counts still differ, ask for help tracing the account repair logic.

Enforce many-to-one relationships

A validated merge turns the expected relationship into an executable rule. The validate="many_to_one" setting rejects duplicate keys on the right side of each merge.

Named indicators preserve the match status for every transaction. Sorting the finished table prepares each customer's transactions for prior-history analysis.

  • Add a code cell below the account repair cell.
  • Paste the validated merges and completeness audit into the new cell:
# Cell 6: Enforce join cardinality and audit completeness
analysis = transactions.merge(
    accounts,
    how="left",
    on="account_id",
    validate="many_to_one",
    indicator="account_match",
)
analysis = analysis.merge(
    customers,
    how="left",
    on="customer_id",
    validate="many_to_one",
    indicator="customer_match",
)
analysis = analysis.sort_values(
    by=["customer_id", "transaction_order"],
    ignore_index=True,
)

join_audit = analysis[["account_match", "customer_match"]].value_counts(
    dropna=False
)
assert len(analysis) == len(transactions)
assert analysis["account_id"].nunique() == n_customers
assert (analysis["account_match"] == "both").all()
assert (analysis["customer_match"] == "both").all()
print(join_audit)

What Does This Validation Enforce?

  • The first merge requires each transaction to match at most one account record.
  • The second merge requires each account to match at most one customer record.
  • The account_match indicator records the result of the account join.
  • The customer_match indicator records the result of the customer join.
  • The assertions stop the cell if the table gains rows.
  • The assertions also stop the cell if any transaction lacks a complete match.
Verify the repaired model

The assertions provide pass-or-fail guardrails for the data model. The printed audit provides visible evidence of the relationship that passed.

Before you run the cell, do you expect any unmatched records to remain?

  • Run the validated merge cell.

The cell should finish without an assertion error. You should see one indicator combination with a count of 800.

  • Confirm account_match shows only both.
  • Confirm customer_match shows only both.
  • Confirm the matched indicator combination has a count of 800.
  • Confirm the account repair output shows Clean account rows: 40.

That is the hard part secured. Every transaction now follows one complete customer-account path.

Did a Join Check Fail?

  • Rerun the account repair cell if the validated merge reports a relationship failure.
  • Check that accounts contains the restored missing_account record.
  • Check that the account merge keeps how="left" on account_id.
  • Check that the customer merge keeps how="left" on customer_id.

If an assertion still fails, ask for help locating the broken join condition.

✔️ Awesome, I've got everything!

Your notebook now contains the repaired account table and the validated 800-row analysis table. Double-check that the notebook has saved your latest cells.

ⓧ I'd like to double check the full code

# Cell 1: Verify the pinned Colab runtime
import numpy as np
import pandas as pd

assert np.__version__ == "2.0.2", f"Expected NumPy 2.0.2, found {np.__version__}"
assert pd.__version__ == "2.2.2", f"Expected pandas 2.2.2, found {pd.__version__}"
print("Python runtime ready")
print("NumPy", np.__version__)
print("pandas", pd.__version__)

# Cell 2: Create customer and intentionally dirty account tables
rng = np.random.default_rng(42)

n_customers = 40
transactions_per_customer = 20
n_transactions = n_customers * transactions_per_customer

customer_ids = np.arange(1000, 1000 + n_customers)
segments = np.tile(["Starter", "Growth", "Premium", "Enterprise"], 10)
base_amounts = np.tile([25.0, 100.0, 400.0, 1600.0], 10)
regions = np.tile(["North", "South", "East", "West"], 10)

customers = pd.DataFrame(
    {
        "customer_id": customer_ids,
        "segment": segments,
        "region": regions,
        "expected_amount": base_amounts,
    }
)

account_ids = np.arange(5000, 5000 + n_customers)
account_types = np.where(
    np.arange(n_customers) % 5 == 0,
    "business",
    "personal",
)
tenure_months = 6 + np.arange(n_customers)

accounts_dirty = pd.DataFrame(
    {
        "account_id": list(account_ids[:-1]) + [account_ids[0]],
        "customer_id": list(customer_ids[:-1]) + [customer_ids[0]],
        "account_type": list(account_types[:-1]) + [account_types[0]],
        "tenure_months": list(tenure_months[:-1]) + [tenure_months[0]],
    }
)

# Cell 3: Generate ordered transactions and inject known anomalies
customer_positions = np.repeat(np.arange(n_customers), transactions_per_customer)
transaction_order = np.tile(np.arange(transactions_per_customer), n_customers)
categories = np.tile(
    ["groceries", "fuel", "travel", "electronics"],
    n_transactions // 4,
)
channels = rng.choice(
    ["card", "mobile", "web", "branch"],
    size=n_transactions,
)
category_factors = np.select(
    [
        categories == "groceries",
        categories == "fuel",
        categories == "travel",
    ],
    [0.6, 1.0, 1.5],
    default=2.5,
)
normal_noise = rng.lognormal(mean=0.0, sigma=0.12, size=n_transactions)
normal_amounts = (
    np.repeat(base_amounts, transactions_per_customer)
    * category_factors
    * normal_noise
)
injected_mask = transaction_order == 18
amounts = np.where(injected_mask, normal_amounts * 4.0, normal_amounts)

transactions = pd.DataFrame(
    {
        "transaction_id": np.arange(1, n_transactions + 1),
        "account_id": account_ids[customer_positions],
        "transaction_order": transaction_order,
        "category": categories,
        "channel": channels,
        "amount": amounts,
        "is_injected_anomaly": injected_mask,
    }
)

# Cell 4: Experience the unchecked-join failure
naive_join = transactions.merge(
    accounts_dirty,
    how="left",
    on="account_id",
)

naive_summary = pd.DataFrame(
    {
        "check": [
            "source transactions",
            "naive joined rows",
            "rows with missing account data",
        ],
        "count": [
            len(transactions),
            len(naive_join),
            int(naive_join["customer_id"].isna().sum()),
        ],
    }
)
print(naive_summary)

# Cell 5: Repair duplicate and missing account records
all_duplicate_account_rows = accounts_dirty[
    accounts_dirty.duplicated(subset=["account_id"], keep=False)
]
accounts = accounts_dirty.drop_duplicates(
    subset=["account_id"],
    keep="first",
).copy()

missing_account = pd.DataFrame(
    {
        "account_id": [account_ids[-1]],
        "customer_id": [customer_ids[-1]],
        "account_type": [account_types[-1]],
        "tenure_months": [tenure_months[-1]],
    }
)
accounts = pd.concat([accounts, missing_account], ignore_index=True)

print("Duplicate account rows found:", len(all_duplicate_account_rows))
print("Clean account rows:", len(accounts))

# Cell 6: Enforce join cardinality and audit completeness
analysis = transactions.merge(
    accounts,
    how="left",
    on="account_id",
    validate="many_to_one",
    indicator="account_match",
)
analysis = analysis.merge(
    customers,
    how="left",
    on="customer_id",
    validate="many_to_one",
    indicator="customer_match",
)
analysis = analysis.sort_values(
    by=["customer_id", "transaction_order"],
    ignore_index=True,
)

join_audit = analysis[["account_match", "customer_match"]].value_counts(
    dropna=False
)
assert len(analysis) == len(transactions)
assert analysis["account_id"].nunique() == n_customers
assert (analysis["account_match"] == "both").all()
assert (analysis["customer_match"] == "both").all()
print(join_audit)

Your analysis table is clean enough to trust. Next, you will compare one global rule with behavior tailored to each customer.

Build Global and Rolling Detectors

The validated joins from the last step give every transaction one complete customer record. Your 800-row analysis table is now safe to score.

A global threshold compares every customer with the same cutoff. You will test what that assumption does to customers with very different spending levels.

Then pandas will build a prior-only customer baseline. The comparison reveals what personalized history changes.

In this step, get ready to:
  • Create vectorized features that compare each transaction with its category history.
  • Test a global percentile threshold against the injected anomalies.
  • Build a prior-only rolling detector for each customer.
Create vectorized features and a global baseline

Vectorized feature engineering calculates complete columns without processing rows one at a time. These features expose transaction order plus each amount's relationship with its customer-category median.

  • Add a new code cell directly beneath Cell 6.
  • Create the vectorized features plus the global baseline by pasting this code into the new cell:
# Cell 7: Create vectorized features and the flawed global detector
analysis["transaction_number"] = (
    analysis.groupby("customer_id", sort=False).cumcount() + 1
)
analysis["category_typical_amount"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["amount"].transform("median")
analysis["amount_to_category_typical"] = (
    analysis["amount"] / analysis["category_typical_amount"]
)

global_threshold = analysis["amount"].quantile(0.95)
analysis["global_flag"] = analysis["amount"] > global_threshold

What Does This Code Do?

  • The transaction_number column records each transaction's position inside its customer group.
  • The category_typical_amount column gives every row its customer-category median.
  • The amount_to_category_typical column turns each amount into a ratio against that median.
  • The global_threshold value is the amount column's 95th percentile. The global_flag column marks every amount above it.
  • Run Cell 7.
  • Confirm the cell finishes without an error.

Having Trouble Creating the Features?

Confirm Cell 6 ran successfully before Cell 7. The analysis table must already contain the customer plus category columns used for grouping.

Check each column name against the code above. A missing underscore or mismatched quote prevents pandas from finding the column.

Still stuck? Help me debug the vectorized feature cell.

Test the global baseline

The global detector uses one cutoff for every spending segment. A quick pair of checks reveals how many alerts it creates plus how many known anomalies it catches.

Before you inspect the counts, do you expect one population-wide cutoff to catch the injected anomaly for every customer?

  • Add a temporary code cell beneath Cell 7.
  • Count the global alerts by running this expression:
int(analysis["global_flag"].sum())

What Does This Check Measure?

Boolean values can be summed because true flags count as one. Converting the result to an integer gives a plain alert count.

You should see 40 global alerts. The 95th-percentile rule selects the highest-spending slice of the complete table.

  • Replace the expression in the temporary cell with this check:
int(
    (
        analysis["global_flag"]
        & analysis["is_injected_anomaly"]
    ).sum()
)

What Does This Check Measure?

The ampersand keeps rows where the global detector fired plus the injected label is true. The resulting count shows how many of the 40 known anomalies the global rule catches.

  • Run the temporary cell again.

You'll see a count below 40. The global detector misses injected anomalies even though it raises 40 alerts.

Why Does the Global Rule Miss Anomalies?

Enterprise transactions naturally occupy much of the dataset's upper tail. Anomalous Starter or Growth transactions can still sit below the shared cutoff.

This shortfall is intentional. It shows why transaction size needs the context of each customer's own history.

  • Delete the temporary code cell.

Seeing an Unexpected Global Count?

Rerun Cells 2 through 7 in order if the notebook was reconnected. The seeded dataset must be regenerated before the detector count is reproducible.

Confirm the threshold uses 0.95 plus the comparison uses the greater-than symbol.

Need another set of eyes? Help me investigate the unexpected global detector count.

Build a prior-only rolling detector

A rolling baseline follows each customer's recent behavior. Shifting the amount first keeps the current transaction out of the history used to judge it.

  • Add a new code cell directly beneath Cell 7.
  • Build the customer-specific rolling detector by pasting this first part of Cell 8:
# Cell 8: Build a prior-only customer rolling z-score detector
analysis["prior_amount"] = analysis.groupby(
    "customer_id",
    sort=False,
)["amount"].shift(1)
analysis["rolling_mean"] = analysis.groupby(
    "customer_id",
    sort=False,
)["prior_amount"].transform(
    lambda values: values.rolling(window=8, min_periods=5).mean()
)
analysis["rolling_std"] = analysis.groupby(
    "customer_id",
    sort=False,
)["prior_amount"].transform(
    lambda values: values.rolling(window=8, min_periods=5).std(ddof=1)
)
analysis["z_score"] = np.where(
    analysis["rolling_std"] > 0,
    (analysis["amount"] - analysis["rolling_mean"])
    / analysis["rolling_std"],
    np.nan,
)
analysis["z_flag"] = analysis["z_score"] > 3.0

How Does the Rolling Detector Work?

  • The grouped shift places each customer's previous amount in prior_amount. This prevents the transaction under review from changing its own baseline.
  • The rolling mean plus standard deviation use up to 8 prior transactions. At least 5 observations are required before a score is calculated.
  • The standard deviation uses ddof=1 to calculate the sample statistic.
  • NumPy's np.where calculates a standard-deviation score only when the rolling spread is positive.
  • The z_flag column marks scores above 3.0.
  • Run the partial Cell 8.
  • Confirm the rolling detector code finishes without an error.

Rolling Columns Failing?

Confirm analysis remains sorted by customer_id plus transaction_order. Rolling history depends on that order.

Check that both grouped transforms read from prior_amount. Reading from the current amount would leak the transaction into its own baseline.

Need help? Help me debug the customer rolling detector.

The detector now exists. The final part of Cell 8 turns both detector results into one comparison table.

  • Place the cursor on a new line beneath the z_flag assignment in Cell 8.
  • Complete the detector comparison by pasting this code:
baseline_snapshot = pd.DataFrame(
    {
        "detector": ["global_threshold", "rolling_z_score"],
        "alerts": [
            int(analysis["global_flag"].sum()),
            int(analysis["z_flag"].sum()),
        ],
        "injected_anomalies_caught": [
            int(
                (
                    analysis["global_flag"]
                    & analysis["is_injected_anomaly"]
                ).sum()
            ),
            int(
                (
                    analysis["z_flag"]
                    & analysis["is_injected_anomaly"]
                ).sum()
            ),
        ],
    }
)
print(baseline_snapshot)

What Does the Snapshot Compare?

Each row represents one detector. The columns compare total alerts with the number of injected anomalies caught.

This separates alert volume from useful detection. A detector can raise many alerts while still missing known anomalies.

Before you run the completed cell, do you expect the customer-specific detector to produce the same alert count as the global threshold?

  • Run the completed Cell 8.
  • Confirm the output contains rows for global_threshold plus rolling_z_score.

You'll see that the global detector catches fewer than 40 injected anomalies. The rolling detector produces a different alert count because every transaction is compared with that customer's prior behavior.

You have now proved the value of a customer-specific baseline. Your notebook can distinguish unusual behavior from naturally high spending.

Baseline Snapshot Missing?

Confirm the snapshot code sits inside Cell 8 beneath the z_flag assignment. Running only the first part creates the scores without printing the comparison.

Rerun Cell 7 before Cell 8 if the global columns are missing. The snapshot needs both flag columns.

Still not seeing two detector rows? Help me fix the baseline snapshot.

✔️ Awesome, I've got everything!

  • Confirm Cell 8 has finished running.
  • Keep the notebook open for the next step.

ⓧ I'd like to double check the full code

# Cell 1: Verify the pinned Colab runtime
import numpy as np
import pandas as pd

assert np.__version__ == "2.0.2", f"Expected NumPy 2.0.2, found {np.__version__}"
assert pd.__version__ == "2.2.2", f"Expected pandas 2.2.2, found {pd.__version__}"
print("Python runtime ready")
print("NumPy", np.__version__)
print("pandas", pd.__version__)

# Cell 2: Create customer and intentionally dirty account tables
rng = np.random.default_rng(42)

n_customers = 40
transactions_per_customer = 20
n_transactions = n_customers * transactions_per_customer

customer_ids = np.arange(1000, 1000 + n_customers)
segments = np.tile(["Starter", "Growth", "Premium", "Enterprise"], 10)
base_amounts = np.tile([25.0, 100.0, 400.0, 1600.0], 10)
regions = np.tile(["North", "South", "East", "West"], 10)

customers = pd.DataFrame(
    {
        "customer_id": customer_ids,
        "segment": segments,
        "region": regions,
        "expected_amount": base_amounts,
    }
)

account_ids = np.arange(5000, 5000 + n_customers)
account_types = np.where(
    np.arange(n_customers) % 5 == 0,
    "business",
    "personal",
)
tenure_months = 6 + np.arange(n_customers)

accounts_dirty = pd.DataFrame(
    {
        "account_id": list(account_ids[:-1]) + [account_ids[0]],
        "customer_id": list(customer_ids[:-1]) + [customer_ids[0]],
        "account_type": list(account_types[:-1]) + [account_types[0]],
        "tenure_months": list(tenure_months[:-1]) + [tenure_months[0]],
    }
)

# Cell 3: Generate ordered transactions and inject known anomalies
customer_positions = np.repeat(np.arange(n_customers), transactions_per_customer)
transaction_order = np.tile(np.arange(transactions_per_customer), n_customers)
categories = np.tile(
    ["groceries", "fuel", "travel", "electronics"],
    n_transactions // 4,
)
channels = rng.choice(
    ["card", "mobile", "web", "branch"],
    size=n_transactions,
)
category_factors = np.select(
    [
        categories == "groceries",
        categories == "fuel",
        categories == "travel",
    ],
    [0.6, 1.0, 1.5],
    default=2.5,
)
normal_noise = rng.lognormal(mean=0.0, sigma=0.12, size=n_transactions)
normal_amounts = (
    np.repeat(base_amounts, transactions_per_customer)
    * category_factors
    * normal_noise
)
injected_mask = transaction_order == 18
amounts = np.where(injected_mask, normal_amounts * 4.0, normal_amounts)

transactions = pd.DataFrame(
    {
        "transaction_id": np.arange(1, n_transactions + 1),
        "account_id": account_ids[customer_positions],
        "transaction_order": transaction_order,
        "category": categories,
        "channel": channels,
        "amount": amounts,
        "is_injected_anomaly": injected_mask,
    }
)

# Cell 4: Experience the unchecked-join failure
naive_join = transactions.merge(
    accounts_dirty,
    how="left",
    on="account_id",
)

naive_summary = pd.DataFrame(
    {
        "check": [
            "source transactions",
            "naive joined rows",
            "rows with missing account data",
        ],
        "count": [
            len(transactions),
            len(naive_join),
            int(naive_join["customer_id"].isna().sum()),
        ],
    }
)
print(naive_summary)

# Cell 5: Repair duplicate and missing account records
all_duplicate_account_rows = accounts_dirty[
    accounts_dirty.duplicated(subset=["account_id"], keep=False)
]
accounts = accounts_dirty.drop_duplicates(
    subset=["account_id"],
    keep="first",
).copy()

missing_account = pd.DataFrame(
    {
        "account_id": [account_ids[-1]],
        "customer_id": [customer_ids[-1]],
        "account_type": [account_types[-1]],
        "tenure_months": [tenure_months[-1]],
    }
)
accounts = pd.concat([accounts, missing_account], ignore_index=True)

print("Duplicate account rows found:", len(all_duplicate_account_rows))
print("Clean account rows:", len(accounts))

# Cell 6: Enforce join cardinality and audit completeness
analysis = transactions.merge(
    accounts,
    how="left",
    on="account_id",
    validate="many_to_one",
    indicator="account_match",
)
analysis = analysis.merge(
    customers,
    how="left",
    on="customer_id",
    validate="many_to_one",
    indicator="customer_match",
)
analysis = analysis.sort_values(
    by=["customer_id", "transaction_order"],
    ignore_index=True,
)

join_audit = analysis[["account_match", "customer_match"]].value_counts(
    dropna=False
)
assert len(analysis) == len(transactions)
assert analysis["account_id"].nunique() == n_customers
assert (analysis["account_match"] == "both").all()
assert (analysis["customer_match"] == "both").all()
print(join_audit)

# Cell 7: Create vectorized features and the flawed global detector
analysis["transaction_number"] = (
    analysis.groupby("customer_id", sort=False).cumcount() + 1
)
analysis["category_typical_amount"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["amount"].transform("median")
analysis["amount_to_category_typical"] = (
    analysis["amount"] / analysis["category_typical_amount"]
)

global_threshold = analysis["amount"].quantile(0.95)
analysis["global_flag"] = analysis["amount"] > global_threshold

# Cell 8: Build a prior-only customer rolling z-score detector
analysis["prior_amount"] = analysis.groupby(
    "customer_id",
    sort=False,
)["amount"].shift(1)
analysis["rolling_mean"] = analysis.groupby(
    "customer_id",
    sort=False,
)["prior_amount"].transform(
    lambda values: values.rolling(window=8, min_periods=5).mean()
)
analysis["rolling_std"] = analysis.groupby(
    "customer_id",
    sort=False,
)["prior_amount"].transform(
    lambda values: values.rolling(window=8, min_periods=5).std(ddof=1)
)
analysis["z_score"] = np.where(
    analysis["rolling_std"] > 0,
    (analysis["amount"] - analysis["rolling_mean"])
    / analysis["rolling_std"],
    np.nan,
)
analysis["z_flag"] = analysis["z_score"] > 3.0

baseline_snapshot = pd.DataFrame(
    {
        "detector": ["global_threshold", "rolling_z_score"],
        "alerts": [
            int(analysis["global_flag"].sum()),
            int(analysis["z_flag"].sum()),
        ],
        "injected_anomalies_caught": [
            int(
                (
                    analysis["global_flag"]
                    & analysis["is_injected_anomaly"]
                ).sum()
            ),
            int(
                (
                    analysis["z_flag"]
                    & analysis["is_injected_anomaly"]
                ).sum()
            ),
        ],
    }
)
print(baseline_snapshot)

Your detector comparison now exposes the global rule's blind spot. Next, you'll add a robust customer-category score before turning the strongest alerts into an investigation report.

Add Robust Scoring and Export Alerts

Your Google Colab notebook now compares every transaction with its customer's recent behavior. The rolling detector has already shown why personal history beats one global threshold.

Extreme amounts can distort means and standard deviations. A grouped pandas baseline built from the median and interquartile range is more resistant to that skew. Precision and recall then reveal the trade-offs between all three detectors.

In this step, get ready to:
  • Build a robust prior-only baseline for each customer and category.
  • Compare all three detectors against the injected anomaly labels.
  • Produce an explained and severity-ranked CSV investigation report.
Build a robust customer-category baseline

The rolling z-score compares a transaction with all recent purchases from the same customer. Adding category groups creates a tighter comparison between similar transactions such as travel purchases or grocery purchases.

  • Add a new code cell directly below Cell 8.
  • Build the prior-only median and quartile baselines by pasting this first part into the new cell:
# Cell 9: Build a robust customer-and-category rolling detector
analysis["prior_category_amount"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["amount"].shift(1)
analysis["rolling_median"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["prior_category_amount"].transform(
    lambda values: values.rolling(window=4, min_periods=3).median()
)
analysis["rolling_q1"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["prior_category_amount"].transform(
    lambda values: values.rolling(window=4, min_periods=3).quantile(0.25)
)
analysis["rolling_q3"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["prior_category_amount"].transform(
    lambda values: values.rolling(window=4, min_periods=3).quantile(0.75)
)
analysis["rolling_iqr"] = analysis["rolling_q3"] - analysis["rolling_q1"]

How does the robust baseline work?

  • The shift(1) operation removes the current transaction from its own baseline.
  • The rolling_median uses the previous four matching category amounts when at least three prior values exist.
  • The rolling_q1 and rolling_q3 columns describe the middle half of that recent history.
  • The rolling_iqr measures the distance between those quartiles.
  • Append the robust score and detector flag to the same cell by pasting this code below the existing lines:
analysis["robust_score"] = np.where(
    analysis["rolling_iqr"] > 0,
    (analysis["amount"] - analysis["rolling_median"])
    / analysis["rolling_iqr"],
    np.nan,
)
analysis["robust_flag"] = analysis["robust_score"] > 3.0

What does the robust score measure?

  • The score measures how far the current amount sits above its recent customer-category median.
  • The NumPy expression np.where keeps the score undefined when the interquartile range is zero.
  • The robust_flag column selects scores above 3.0.
  • Run Cell 9.

The cell completes without an exception. Your analysis table now holds a prior-only robust baseline for every eligible customer-category transaction.

Robust scoring cell not completing?

Confirm Cell 8 has already run in the current runtime. Cell 9 depends on the existing analysis table.

Check that the grouping columns are spelled customer_id and category everywhere in the cell.

Still stuck? Help me debug my grouped rolling median and IQR code.

Compare detector quality

The injected anomaly labels provide known ground truth. Comparing each detector with those labels turns alert counts into measurable investigation quality.

  • Add a new code cell directly below Cell 9.
  • Define the reusable detector evaluation function by pasting this first part into the new cell:
# Cell 10: Evaluate every detector against the injected labels
def detector_metrics(frame, flag_column):
    actual = frame["is_injected_anomaly"]
    predicted = frame[flag_column].fillna(False)

    true_positives = int((actual & predicted).sum())
    false_positives = int((~actual & predicted).sum())
    false_negatives = int((actual & ~predicted).sum())
    precision = (
        true_positives / (true_positives + false_positives)
        if true_positives + false_positives
        else 0.0
    )
    recall = (
        true_positives / (true_positives + false_negatives)
        if true_positives + false_negatives
        else 0.0
    )

    return {
        "detector": flag_column,
        "true_positives": true_positives,
        "false_positives": false_positives,
        "false_negatives": false_negatives,
        "precision": precision,
        "recall": recall,
    }

What does the evaluation function calculate?

  • True positives are injected anomalies that the detector flags.
  • False positives are normal transactions that the detector flags.
  • False negatives are injected anomalies that the detector misses.
  • Precision measures how many generated alerts are correct. Recall measures how many injected anomalies are caught.
  • Apply the evaluation function to all three detectors by pasting this code at the bottom of the same cell:
metrics = pd.DataFrame(
    [
        detector_metrics(analysis, "global_flag"),
        detector_metrics(analysis, "z_flag"),
        detector_metrics(analysis, "robust_flag"),
    ]
).sort_values(by=["recall", "precision"], ascending=[False, False])
print(metrics)

How is the comparison organized?

  • The metrics table contains one row for each detector.
  • The rows are sorted by recall first. Precision breaks ties.
  • The comparison exposes both missed anomalies and unnecessary investigation work.

Before you run the comparison, which detector do you expect to recover more injected anomalies?

  • Run Cell 10.

You'll see a three-row metrics table containing global_flag, z_flag, and robust_flag. Each row reports true positives, false positives, false negatives, precision, and recall.

Metrics table not appearing?

Confirm Cell 9 completed before Cell 10. The evaluation requires the robust_flag column.

Check that detector_metrics is defined before the metrics table calls it.

Need another pair of eyes? Help me troubleshoot my detector metrics table.

Rank and export investigation alerts

An investigator needs a focused queue instead of hundreds of scored transactions. Combining the customer-specific detectors keeps the alert set relevant. A severity score places the strongest deviations first.

  • Add a new code cell directly below Cell 10.
  • Create the combined detector and vectorized explanations by pasting this first part into the new cell:
# Cell 11: Rank, explain, and export investigation alerts
analysis["combined_flag"] = analysis["z_flag"] | analysis["robust_flag"]
analysis["severity_score"] = np.maximum(
    np.abs(analysis["z_score"].fillna(0)),
    np.abs(analysis["robust_score"].fillna(0)),
)
analysis["explanation"] = np.select(
    [
        analysis["robust_flag"] & analysis["z_flag"],
        analysis["robust_flag"],
        analysis["z_flag"],
    ],
    [
        "Large deviation from both customer and category baselines",
        "Large deviation from the customer-category median and IQR",
        "Large deviation from the customer rolling mean and standard deviation",
    ],
    default="Not selected for investigation",
)

How are alerts combined and explained?

  • The combined_flag selects a transaction when either customer-specific detector flags it.
  • The severity_score keeps the larger absolute score from the two detectors.
  • The explanation column describes which baseline identified the transaction.
  • Run Cell 11.

The cell completes without an exception. The analysis table now contains combined decisions, severity values, and human-readable explanations.

  • Append the alert columns and ranking logic to Cell 11 by pasting this code below the existing lines:
alert_columns = [
    "transaction_id",
    "customer_id",
    "account_id",
    "segment",
    "category",
    "channel",
    "amount",
    "z_score",
    "robust_score",
    "severity_score",
    "global_flag",
    "z_flag",
    "robust_flag",
    "is_injected_anomaly",
    "explanation",
]
alerts = analysis[analysis["combined_flag"]][alert_columns].copy()
alerts = alerts.sort_values(
    by=["severity_score", "transaction_id"],
    ascending=[False, True],
    ignore_index=True,
)

What goes into the investigation queue?

  • The alerts table keeps only transactions selected by the combined detector.
  • Severity is sorted from highest to lowest.
  • Transaction ID provides a stable ascending order when two alerts share the same severity.
  • The selected columns preserve the transaction context, detector evidence, injected label, and explanation.
  • Run Cell 11 again.

The cell completes successfully. The ranked alerts table now contains only customer-specific detector alerts.

  • Finish Cell 11 with the CSV export and output checks by pasting this code at the bottom:
alerts.to_csv(
    "investigation_alerts.csv",
    index=False,
    float_format="%.2f",
)

assert alerts["explanation"].ne("").all()
print(alerts.head(10))
print("CSV rows:", len(alerts))
print("Created investigation_alerts.csv")

What does the export finish?

  • The export writes investigation_alerts.csv without a DataFrame index column.
  • The %.2f format keeps floating-point values at two decimal places.
  • The assertion stops the cell if any alert explanation is empty.
  • The final prints show the first ten alerts, the exported row count, and the file creation confirmation.

✔️ Awesome, I've got everything!

Great. Double-check that Cells 9 through 11 are complete in your notebook.

ⓧ I'd like to double check the full code

# Cell 1: Verify the pinned Colab runtime
import numpy as np
import pandas as pd

assert np.__version__ == "2.0.2", f"Expected NumPy 2.0.2, found {np.__version__}"
assert pd.__version__ == "2.2.2", f"Expected pandas 2.2.2, found {pd.__version__}"
print("Python runtime ready")
print("NumPy", np.__version__)
print("pandas", pd.__version__)

# Cell 2: Create customer and intentionally dirty account tables
rng = np.random.default_rng(42)

n_customers = 40
transactions_per_customer = 20
n_transactions = n_customers * transactions_per_customer

customer_ids = np.arange(1000, 1000 + n_customers)
segments = np.tile(["Starter", "Growth", "Premium", "Enterprise"], 10)
base_amounts = np.tile([25.0, 100.0, 400.0, 1600.0], 10)
regions = np.tile(["North", "South", "East", "West"], 10)

customers = pd.DataFrame(
    {
        "customer_id": customer_ids,
        "segment": segments,
        "region": regions,
        "expected_amount": base_amounts,
    }
)

account_ids = np.arange(5000, 5000 + n_customers)
account_types = np.where(
    np.arange(n_customers) % 5 == 0,
    "business",
    "personal",
)
tenure_months = 6 + np.arange(n_customers)

accounts_dirty = pd.DataFrame(
    {
        "account_id": list(account_ids[:-1]) + [account_ids[0]],
        "customer_id": list(customer_ids[:-1]) + [customer_ids[0]],
        "account_type": list(account_types[:-1]) + [account_types[0]],
        "tenure_months": list(tenure_months[:-1]) + [tenure_months[0]],
    }
)

# Cell 3: Generate ordered transactions and inject known anomalies
customer_positions = np.repeat(np.arange(n_customers), transactions_per_customer)
transaction_order = np.tile(np.arange(transactions_per_customer), n_customers)
categories = np.tile(
    ["groceries", "fuel", "travel", "electronics"],
    n_transactions // 4,
)
channels = rng.choice(
    ["card", "mobile", "web", "branch"],
    size=n_transactions,
)
category_factors = np.select(
    [
        categories == "groceries",
        categories == "fuel",
        categories == "travel",
    ],
    [0.6, 1.0, 1.5],
    default=2.5,
)
normal_noise = rng.lognormal(mean=0.0, sigma=0.12, size=n_transactions)
normal_amounts = (
    np.repeat(base_amounts, transactions_per_customer)
    * category_factors
    * normal_noise
)
injected_mask = transaction_order == 18
amounts = np.where(injected_mask, normal_amounts * 4.0, normal_amounts)

transactions = pd.DataFrame(
    {
        "transaction_id": np.arange(1, n_transactions + 1),
        "account_id": account_ids[customer_positions],
        "transaction_order": transaction_order,
        "category": categories,
        "channel": channels,
        "amount": amounts,
        "is_injected_anomaly": injected_mask,
    }
)

# Cell 4: Experience the unchecked-join failure
naive_join = transactions.merge(
    accounts_dirty,
    how="left",
    on="account_id",
)

naive_summary = pd.DataFrame(
    {
        "check": [
            "source transactions",
            "naive joined rows",
            "rows with missing account data",
        ],
        "count": [
            len(transactions),
            len(naive_join),
            int(naive_join["customer_id"].isna().sum()),
        ],
    }
)
print(naive_summary)

# Cell 5: Repair duplicate and missing account records
all_duplicate_account_rows = accounts_dirty[
    accounts_dirty.duplicated(subset=["account_id"], keep=False)
]
accounts = accounts_dirty.drop_duplicates(
    subset=["account_id"],
    keep="first",
).copy()

missing_account = pd.DataFrame(
    {
        "account_id": [account_ids[-1]],
        "customer_id": [customer_ids[-1]],
        "account_type": [account_types[-1]],
        "tenure_months": [tenure_months[-1]],
    }
)
accounts = pd.concat([accounts, missing_account], ignore_index=True)

print("Duplicate account rows found:", len(all_duplicate_account_rows))
print("Clean account rows:", len(accounts))

# Cell 6: Enforce join cardinality and audit completeness
analysis = transactions.merge(
    accounts,
    how="left",
    on="account_id",
    validate="many_to_one",
    indicator="account_match",
)
analysis = analysis.merge(
    customers,
    how="left",
    on="customer_id",
    validate="many_to_one",
    indicator="customer_match",
)
analysis = analysis.sort_values(
    by=["customer_id", "transaction_order"],
    ignore_index=True,
)

join_audit = analysis[["account_match", "customer_match"]].value_counts(
    dropna=False
)
assert len(analysis) == len(transactions)
assert analysis["account_id"].nunique() == n_customers
assert (analysis["account_match"] == "both").all()
assert (analysis["customer_match"] == "both").all()
print(join_audit)

# Cell 7: Create vectorized features and the flawed global detector
analysis["transaction_number"] = (
    analysis.groupby("customer_id", sort=False).cumcount() + 1
)
analysis["category_typical_amount"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["amount"].transform("median")
analysis["amount_to_category_typical"] = (
    analysis["amount"] / analysis["category_typical_amount"]
)

global_threshold = analysis["amount"].quantile(0.95)
analysis["global_flag"] = analysis["amount"] > global_threshold

# Cell 8: Build a prior-only customer rolling z-score detector
analysis["prior_amount"] = analysis.groupby(
    "customer_id",
    sort=False,
)["amount"].shift(1)
analysis["rolling_mean"] = analysis.groupby(
    "customer_id",
    sort=False,
)["prior_amount"].transform(
    lambda values: values.rolling(window=8, min_periods=5).mean()
)
analysis["rolling_std"] = analysis.groupby(
    "customer_id",
    sort=False,
)["prior_amount"].transform(
    lambda values: values.rolling(window=8, min_periods=5).std(ddof=1)
)
analysis["z_score"] = np.where(
    analysis["rolling_std"] > 0,
    (analysis["amount"] - analysis["rolling_mean"])
    / analysis["rolling_std"],
    np.nan,
)
analysis["z_flag"] = analysis["z_score"] > 3.0

baseline_snapshot = pd.DataFrame(
    {
        "detector": ["global_threshold", "rolling_z_score"],
        "alerts": [
            int(analysis["global_flag"].sum()),
            int(analysis["z_flag"].sum()),
        ],
        "injected_anomalies_caught": [
            int(
                (
                    analysis["global_flag"]
                    & analysis["is_injected_anomaly"]
                ).sum()
            ),
            int(
                (
                    analysis["z_flag"]
                    & analysis["is_injected_anomaly"]
                ).sum()
            ),
        ],
    }
)
print(baseline_snapshot)

# Cell 9: Build a robust customer-and-category rolling detector
analysis["prior_category_amount"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["amount"].shift(1)
analysis["rolling_median"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["prior_category_amount"].transform(
    lambda values: values.rolling(window=4, min_periods=3).median()
)
analysis["rolling_q1"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["prior_category_amount"].transform(
    lambda values: values.rolling(window=4, min_periods=3).quantile(0.25)
)
analysis["rolling_q3"] = analysis.groupby(
    ["customer_id", "category"],
    sort=False,
)["prior_category_amount"].transform(
    lambda values: values.rolling(window=4, min_periods=3).quantile(0.75)
)
analysis["rolling_iqr"] = analysis["rolling_q3"] - analysis["rolling_q1"]
analysis["robust_score"] = np.where(
    analysis["rolling_iqr"] > 0,
    (analysis["amount"] - analysis["rolling_median"])
    / analysis["rolling_iqr"],
    np.nan,
)
analysis["robust_flag"] = analysis["robust_score"] > 3.0

# Cell 10: Evaluate every detector against the injected labels
def detector_metrics(frame, flag_column):
    actual = frame["is_injected_anomaly"]
    predicted = frame[flag_column].fillna(False)

    true_positives = int((actual & predicted).sum())
    false_positives = int((~actual & predicted).sum())
    false_negatives = int((actual & ~predicted).sum())
    precision = (
        true_positives / (true_positives + false_positives)
        if true_positives + false_positives
        else 0.0
    )
    recall = (
        true_positives / (true_positives + false_negatives)
        if true_positives + false_negatives
        else 0.0
    )

    return {
        "detector": flag_column,
        "true_positives": true_positives,
        "false_positives": false_positives,
        "false_negatives": false_negatives,
        "precision": precision,
        "recall": recall,
    }

metrics = pd.DataFrame(
    [
        detector_metrics(analysis, "global_flag"),
        detector_metrics(analysis, "z_flag"),
        detector_metrics(analysis, "robust_flag"),
    ]
).sort_values(by=["recall", "precision"], ascending=[False, False])
print(metrics)

# Cell 11: Rank, explain, and export investigation alerts
analysis["combined_flag"] = analysis["z_flag"] | analysis["robust_flag"]
analysis["severity_score"] = np.maximum(
    np.abs(analysis["z_score"].fillna(0)),
    np.abs(analysis["robust_score"].fillna(0)),
)
analysis["explanation"] = np.select(
    [
        analysis["robust_flag"] & analysis["z_flag"],
        analysis["robust_flag"],
        analysis["z_flag"],
    ],
    [
        "Large deviation from both customer and category baselines",
        "Large deviation from the customer-category median and IQR",
        "Large deviation from the customer rolling mean and standard deviation",
    ],
    default="Not selected for investigation",
)

alert_columns = [
    "transaction_id",
    "customer_id",
    "account_id",
    "segment",
    "category",
    "channel",
    "amount",
    "z_score",
    "robust_score",
    "severity_score",
    "global_flag",
    "z_flag",
    "robust_flag",
    "is_injected_anomaly",
    "explanation",
]
alerts = analysis[analysis["combined_flag"]][alert_columns].copy()
alerts = alerts.sort_values(
    by=["severity_score", "transaction_id"],
    ascending=[False, True],
    ignore_index=True,
)
alerts.to_csv(
    "investigation_alerts.csv",
    index=False,
    float_format="%.2f",
)

assert alerts["explanation"].ne("").all()
print(alerts.head(10))
print("CSV rows:", len(alerts))
print("Created investigation_alerts.csv")

Before you run the completed export cell, which transactions do you expect to appear at the top of the investigation queue?

  • Run the completed Cell 11.
  • Confirm the output ends with Created investigation_alerts.csv.

You'll see the first ten severity-ranked alerts and a CSV rows: count. Colab also lists investigation_alerts.csv among the active runtime files.

That's the investigation workflow complete. Every exported row now has detector evidence, a severity score, and a readable reason for review.

CSV not created?

Confirm the completed Cell 11 ran after Cells 9 and 10. The export depends on the robust score and combined alert columns.

Check that the filename is exactly investigation_alerts.csv and that the export call remains inside the active Colab cell.

Still missing the report? Help me troubleshoot my investigation alert export.

Secret mission

Add Channel Rarity to Severity

Amount anomalies tell you how unusual a transaction is. Add customer-channel rarity to move unusual channels higher in the investigation queue and export an enhanced report.

Clean Up Your Resources

Clean Up Your Resources

This project uses the free Google Colab hosted runtime. Your next choice is to keep the notebook with its generated CSV files, pause your work, or remove every project resource.

Resources you used:

  • The Transaction_Anomaly_Investigation_Engine.ipynb notebook stored in Google Drive.
  • The investigation_alerts.csv file stored in the active Colab runtime.
  • The investigation_alerts_enhanced.csv file stored in the active Colab runtime.

Keep everything running

No action is needed for the notebook. Google Colab stores it in Google Drive.

Runtime files are temporary. Download each CSV while the active runtime remains available.

  • Leave Transaction_Anomaly_Investigation_Engine.ipynb in Google Drive.
  • Download investigation_alerts.csv from the active Colab runtime.
  • Download investigation_alerts_enhanced.csv from the active Colab runtime.

Pause - I'll come back to this later

This option leaves the notebook saved in Google Drive. The hosted virtual machine may be deleted after it sits idle.

  • Download investigation_alerts.csv from the active Colab runtime.
  • Download investigation_alerts_enhanced.csv from the active Colab runtime.
  • Close the browser tab containing your Colab notebook.

Delete - I don't want to use this again

This option removes the notebook from Google Drive. It also discards the active runtime with both generated CSV files.

Deleting the runtime feels final, which is why the download steps come first.

  • Download investigation_alerts.csv if you want to keep the original investigation report.
  • Download investigation_alerts_enhanced.csv if you want to keep the enhanced investigation report.
  • In Google Drive, locate Transaction_Anomaly_Investigation_Engine.ipynb.
  • Open the notebook's context menu.
  • Move Transaction_Anomaly_Investigation_Engine.ipynb to trash.
  • Switch back to the open Google Colab notebook.
  • Select Runtime from the top menu.
  • Select Disconnect and delete runtime.

The notebook is now in Google Drive's trash. The active runtime with both generated CSV files is discarded.

Nice Work!

Nice Work!

You did it! Your Google Colab notebook now turns synthetic transaction data into ranked investigation alerts with human-readable explanations.

You've learned how to:

  • Build a validated analytical table from transaction records. Enforce many-to-one cardinality for every master-data join.
  • Generate reproducible transaction data with NumPy. Build prior-only grouped rolling detectors with pandas.
  • Compare all three detectors against known injected anomalies using precision plus recall. Turn the evaluation into a severity-ranked investigation report in investigation_alerts.csv with an explanation for every alert.
  • Secret Mission: Added a customer-channel rarity signal to the severity ranking. Exported investigation_alerts_enhanced.csv with rare-channel explanations.

Ready to quiz yourself?