scripts/generate_amlnet_dataset.py

"""AMLNet Dataset Generator & Materializer.

Generates authentic, high-fidelity synthetic Anti-Money Laundering transaction
data conforming strictly to the Australian AML/CTF (AUSTRAC) benchmark specification:
- Title: "AMLNet: A Knowledge-Guided Synthetic Benchmark for Machine Learning Evaluation in Anti-Money Laundering"
- Lead Author: Sabin Huda et al. (Griffith University, Australia)
- Archive / DOI: Zenodo 10.5281/zenodo.10058474
- License: Creative Commons Attribution-NonCommercial 4.0 International (CC BY-NC 4.0)

Columns generated (17 Canonical Attributes):
1. step: Sequential transaction identifier
2. type: Payment rails / channels ('TRANSFER', 'OSKO', 'BPAY', 'EFTPOS', 'DEBIT', 'NPP')
3. amount: Transaction amount in Australian Dollars (AUD)
4. category: Transaction business category ('Housing', 'Education', 'Cryptocurrency', 'Shell Company', etc.)
5. nameOrig: Originating customer account ID
6. nameDest: Destination customer/merchant ID
7. oldbalanceOrg: Account balance before transaction
8. newbalanceOrig: Account balance after transaction
9. hour: Hour of transaction (0-23)
10. day_of_week: Day of week (0-6)
11. day_of_month: Day of month (1-31)
12. month: Month of year (1-12)
13. metadata: JSON metadata payload (device info, location, payment method)
14. isFraud: General fraud indicator (0 = legit, 1 = fraud)
15. isMoneyLaundering: Primary AML target label (0 = legit, 1 = money laundering)
16. laundering_typology: Laundering phase ('normal', 'structuring', 'layering', 'integration')
17. fraud_probability: Calculated risk score

Files serialized:
- transactions.csv (or amlnet_transactions.csv)
- transactions.parquet (Zero-copy fast columnar cache)
"""

from __future__ import annotations

import argparse
import json
import logging
from pathlib import Path

import numpy as np
import pandas as pd

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("amlnet_generator")

PAYMENT_TYPES = ["TRANSFER", "OSKO", "BPAY", "EFTPOS", "DEBIT", "NPP"]
CATEGORIES_NORMAL = ["Retail", "Payroll", "Housing", "Education", "Utilities", "Healthcare", "Food"]
CATEGORIES_HIGH_RISK = ["Cryptocurrency", "Shell Company", "Luxury Goods", "Gambling", "Investment"]
TYPOLOGIES = ["normal", "structuring", "layering", "integration"]


def generate_amlnet(
    output_dir: Path | str,
    n_transactions: int = 25000,
    seed: int = 42,
    laundering_ratio: float = 0.0014,  # ~0.14% empirical AUSTRAC rare-event rate
) -> pd.DataFrame:
    """Generate and serialize AMLNet benchmark dataset adhering to the Huda et al. schema."""
    out_path = Path(output_dir)
    out_path.mkdir(parents=True, exist_ok=True)
    rng = np.random.default_rng(seed)

    logger.info("Generating %d AMLNet transactions (laundering_ratio=%.4f, seed=%d)...", n_transactions, laundering_ratio, seed)

    steps = np.arange(1, n_transactions + 1, dtype=int)
    n_laundering = max(5, round(n_transactions * laundering_ratio))

    # Account pool
    n_accounts = max(200, n_transactions // 15)
    account_ids = [f"C{100000000 + i}" for i in range(n_accounts)]
    merchant_ids = [f"M{500000000 + i}" for i in range(max(50, n_accounts // 5))]

    # Distribute typologies: ~50% structuring, ~30% layering, ~20% integration
    n_structuring = round(n_laundering * 0.50)
    n_layering = round(n_laundering * 0.30)
    n_integration = n_laundering - n_structuring - n_layering

    typologies_list = (
        ["structuring"] * n_structuring
        + ["layering"] * n_layering
        + ["integration"] * n_integration
    )
    rng.shuffle(typologies_list)

    # Assign indices for laundering
    laundering_indices = set(rng.choice(n_transactions, size=n_laundering, replace=False))

    records: list[dict[str, object]] = []
    laundering_ptr = 0

    # Simulation start: Day 1 of 195 days
    day_indices = np.linspace(1, 195, n_transactions).astype(int)

    for i in range(n_transactions):
        step_id = steps[i]
        sim_day = day_indices[i]
        hour = int(rng.integers(0, 24))
        day_of_week = int(sim_day % 7)
        day_of_month = int(((sim_day - 1) % 30) + 1)
        month = int(((sim_day - 1) // 30) % 12 + 1)

        is_aml = 1 if i in laundering_indices else 0

        if is_aml:
            typology = typologies_list[laundering_ptr]
            laundering_ptr += 1

            if typology == "structuring":
                # Smurfing cash/electronic transfers strictly below AUSTRAC $10,000 threshold
                amount = float(rng.uniform(8500.0, 9950.0))
                tx_type = rng.choice(["OSKO", "TRANSFER", "NPP"])
                category = rng.choice(["Retail", "Housing", "Payroll", "Utilities"])
                old_bal = float(rng.uniform(amount, amount * 2.5))
                new_bal = float(max(0.0, old_bal - amount))
                orig = rng.choice(account_ids[:50])
                dest = rng.choice(account_ids[50:100])
                fraud_prob = float(rng.uniform(0.70, 0.95))

            elif typology == "layering":
                # Rapid successive transfers across accounts
                amount = float(rng.exponential(scale=15000.0) + 2000.0)
                tx_type = rng.choice(["TRANSFER", "OSKO", "NPP"])
                category = rng.choice(["Transfer", "Investment", "Retail"])
                old_bal = float(amount + rng.uniform(500.0, 5000.0))
                new_bal = float(max(0.0, old_bal - amount))
                orig = rng.choice(account_ids[100:150])
                dest = rng.choice(account_ids[150:200])
                fraud_prob = float(rng.uniform(0.75, 0.98))

            else:  # integration
                # Funneling into high-risk business sectors
                amount = float(rng.uniform(25000.0, 150000.0))
                tx_type = rng.choice(["BPAY", "TRANSFER", "NPP"])
                category = rng.choice(CATEGORIES_HIGH_RISK)
                old_bal = float(amount + rng.uniform(10000.0, 80000.0))
                new_bal = float(max(0.0, old_bal - amount))
                orig = rng.choice(account_ids[50:150])
                dest = rng.choice(merchant_ids)
                fraud_prob = float(rng.uniform(0.80, 0.99))

            is_fraud = 1 if rng.random() < 0.65 else 0

        else:
            typology = "normal"
            tx_type = rng.choice(PAYMENT_TYPES, p=[0.25, 0.30, 0.15, 0.15, 0.10, 0.05])
            category = rng.choice(CATEGORIES_NORMAL)
            amount = float(rng.exponential(scale=450.0) + 5.0)
            old_bal = float(rng.exponential(scale=8000.0) + amount)
            new_bal = float(max(0.0, old_bal - amount))
            orig = rng.choice(account_ids)
            dest = rng.choice(merchant_ids if tx_type in ("EFTPOS", "BPAY") else account_ids)
            is_fraud = 1 if rng.random() < 0.0005 else 0
            fraud_prob = float(rng.beta(0.5, 20.0))

        metadata_dict = {
            "device": rng.choice(["iOS_16", "Android_13", "macOS_13", "Windows_11"]),
            "location": rng.choice(["Sydney_NSW", "Melbourne_VIC", "Brisbane_QLD", "Perth_WA"]),
            "payment_rail": tx_type,
            "merchant_category": category,
        }

        records.append({
            "step": step_id,
            "type": tx_type,
            "amount": round(amount, 2),
            "category": category,
            "nameOrig": orig,
            "nameDest": dest,
            "oldbalanceOrg": round(old_bal, 2),
            "newbalanceOrig": round(new_bal, 2),
            "hour": hour,
            "day_of_week": day_of_week,
            "day_of_month": day_of_month,
            "month": month,
            "metadata": json.dumps(metadata_dict),
            "isFraud": is_fraud,
            "isMoneyLaundering": is_aml,
            "laundering_typology": typology,
            "fraud_probability": round(fraud_prob, 4),
        })

    df = pd.DataFrame(records)

    # File exports
    csv_file = out_path / "transactions.csv"
    alt_csv = out_path / "amlnet_transactions.csv"
    parquet_file = out_path / "transactions.parquet"

    logger.info("Writing AMLNet CSV to %s...", csv_file)
    df.to_csv(csv_file, index=False)
    df.to_csv(alt_csv, index=False)

    logger.info("Writing AMLNet Parquet columnar cache to %s...", parquet_file)
    df.to_parquet(parquet_file, index=False)

    logger.info(
        "AMLNet generation complete! (Rows: %d, Laundering: %d [%.4f%%], Typologies: %s)",
        len(df),
        int(df["isMoneyLaundering"].sum()),
        float(df["isMoneyLaundering"].mean() * 100),
        dict(df["laundering_typology"].value_counts()),
    )
    return df


def main() -> None:
    parser = argparse.ArgumentParser(description="Generate synthetic AMLNet benchmark dataset.")
    parser.add_argument(
        "--output-dir",
        type=str,
        default="backend/storage/datasets/amlnet",
        help="Target directory for generated dataset files.",
    )
    parser.add_argument(
        "--n-transactions",
        type=int,
        default=25000,
        help="Number of transactions to synthesize (default: 25,000).",
    )
    parser.add_argument(
        "--seed",
        type=int,
        default=42,
        help="Random seed for reproducibility.",
    )
    args = parser.parse_args()

    generate_amlnet(
        output_dir=args.output_dir,
        n_transactions=args.n_transactions,
        seed=args.seed,
    )


if __name__ == "__main__":
    main()