"""Generates the hand-labeled fixture set consumed by tests/unit/test_labeled_fixtures.py (TEST-03). Each entry is a deliberately constructed, individually-reasoned scenario (not a random sample) covering obvious fraud and obvious legitimate patterns, spread across simulated dates (steps) so date/time-derived features aren't exercised by only a single moment. "Obvious" here means built from the same structural patterns the Phase 1 EDA identified: fraud confined to TRANSFER/CASH_OUT with a full-balance drain (amount == oldbalanceOrg, newbalanceOrig == 0) versus legitimate transactions that leave a residual balance or are CASH_IN/PAYMENT/DEBIT (fraud rate 0 in the training data for those types). Run: python -m scripts.generate_fixtures Writes: tests/fixtures/labeled_transactions.json """ from __future__ import annotations import json from pathlib import Path OUTPUT_PATH = Path("tests/fixtures/labeled_transactions.json") HOURS_PER_DAY = 24 def _step(day: int, hour: int) -> int: return max(1, day * HOURS_PER_DAY + hour) def _txn( *, id_: str, label: str, description: str, day: int, hour: int, type_: str, amount: float, name_orig: str, old_orig: float, new_orig: float, name_dest: str, old_dest: float, new_dest: float, ) -> dict: return { "id": id_, "expected_label": label, "description": description, "simulated_day": day, "simulated_hour": hour, "step": _step(day, hour), "type": type_, "amount": amount, "nameOrig": name_orig, "oldbalanceOrg": old_orig, "newbalanceOrig": new_orig, "nameDest": name_dest, "oldbalanceDest": old_dest, "newbalanceDest": new_dest, } def build_fixtures() -> list[dict]: fixtures: list[dict] = [] # ---- Obvious fraud: full-balance-drain TRANSFER, varied amounts/times ---- fraud_transfer_amounts = [ 181.0, 2500.0, 9999.99, 50000.0, 120000.0, 275000.5, 999999.0, 45.0, 733.21, 18500.0, ] for i, amount in enumerate(fraud_transfer_amounts): day = i * 3 hour = (i * 7) % 24 fixtures.append( _txn( id_=f"fraud-transfer-drain-{i:02d}", label="fraud", description="Full-balance TRANSFER drain (amount == oldbalanceOrg, newbalanceOrig == 0) -- the PaySim fraud signature.", day=day, hour=hour, type_="TRANSFER", amount=amount, name_orig=f"CFXFRAUD{i:03d}A", old_orig=amount, new_orig=0.0, name_dest=f"CFXFRAUD{i:03d}B", old_dest=0.0, new_dest=0.0, ) ) # ---- Obvious fraud: full-balance-drain CASH_OUT, varied amounts/times ---- fraud_cashout_amounts = [ 300.0, 4750.0, 15000.0, 62000.0, 210000.0, 890000.0, 99.99, 3333.33, 27500.0, 145000.0, ] for i, amount in enumerate(fraud_cashout_amounts): day = 30 + i * 2 hour = (i * 5 + 1) % 24 fixtures.append( _txn( id_=f"fraud-cashout-drain-{i:02d}", label="fraud", description="Full-balance CASH_OUT drain -- same signature as TRANSFER fraud, second fraud-prone type identified in EDA.", day=day, hour=hour, type_="CASH_OUT", amount=amount, name_orig=f"CFXFRAUD{i:03d}C", old_orig=amount, new_orig=0.0, name_dest=f"CFXFRAUD{i:03d}D", old_dest=0.0, new_dest=0.0, ) ) # ---- Obvious fraud: large drains at night hours, brand-new accounts ---- night_amounts = [8200.0, 41000.0, 176000.0, 630000.0, 12500.5, 55555.0, 320000.0, 7770.0] for i, amount in enumerate(night_amounts): day = 60 + i * 4 hour = [1, 2, 3, 23][i % 4] fixtures.append( _txn( id_=f"fraud-night-drain-{i:02d}", label="fraud", description="Full-balance drain at a late-night simulated hour, single-use account (no prior history).", day=day, hour=hour, type_="TRANSFER" if i % 2 == 0 else "CASH_OUT", amount=amount, name_orig=f"CFXFRAUDN{i:03d}A", old_orig=amount, new_orig=0.0, name_dest=f"CFXFRAUDN{i:03d}B", old_dest=0.0, new_dest=0.0, ) ) # ---- Obvious legit: PAYMENT to merchant, partial balance, varied days ---- payment_cases = [ (25.50, 500.0), (89.99, 1200.0), (14.00, 300.0), (250.0, 3000.0), (5.75, 80.0), (999.0, 15000.0), (42.30, 600.0), (150.0, 2200.0), (7.25, 100.0), (620.0, 9000.0), ] for i, (amount, balance) in enumerate(payment_cases): day = i * 6 hour = (i * 3 + 8) % 24 fixtures.append( _txn( id_=f"legit-payment-{i:02d}", label="legit", description="Small PAYMENT to a merchant, balance not drained -- fraud rate 0 for PAYMENT in the training data.", day=day, hour=hour, type_="PAYMENT", amount=amount, name_orig=f"CFXLEGITP{i:03d}", old_orig=balance, new_orig=balance - amount, name_dest=f"MFXMERCH{i:03d}", old_dest=0.0, new_dest=0.0, ) ) # ---- Obvious legit: CASH_IN deposits (never fraud in PaySim) ---- cashin_cases = [ (200.0, 1000.0), (5000.0, 20000.0), (75.0, 400.0), (1200.0, 8000.0), (33.0, 150.0), (9000.0, 50000.0), (450.0, 3000.0), (60.0, 900.0), ] for i, (amount, balance) in enumerate(cashin_cases): day = 10 + i * 5 hour = (i * 2 + 6) % 24 fixtures.append( _txn( id_=f"legit-cashin-{i:02d}", label="legit", description="CASH_IN deposit -- always legitimate in PaySim (fraud confined to TRANSFER/CASH_OUT).", day=day, hour=hour, type_="CASH_IN", amount=amount, name_orig=f"CFXLEGITCI{i:03d}", old_orig=balance, new_orig=balance + amount, name_dest=f"CFXLEGITCID{i:03d}", old_dest=0.0, new_dest=0.0, ) ) # ---- Obvious legit: DEBIT, small amounts ---- debit_cases = [(20.0, 300.0), (55.0, 700.0), (8.5, 120.0), (310.0, 4000.0), (17.25, 250.0)] for i, (amount, balance) in enumerate(debit_cases): day = 20 + i * 7 hour = (i * 4 + 12) % 24 fixtures.append( _txn( id_=f"legit-debit-{i:02d}", label="legit", description="Small DEBIT transaction -- fraud rate 0 for DEBIT in the training data.", day=day, hour=hour, type_="DEBIT", amount=amount, name_orig=f"CFXLEGITD{i:03d}", old_orig=balance, new_orig=balance - amount, name_dest=f"MFXDEBIT{i:03d}", old_dest=0.0, new_dest=0.0, ) ) # ---- Obvious legit: partial TRANSFER/CASH_OUT, balance survives ---- partial_cases = [ ("TRANSFER", 500.0, 5000.0), ("CASH_OUT", 1200.0, 8000.0), ("TRANSFER", 75.0, 900.0), ("CASH_OUT", 3000.0, 40000.0), ("TRANSFER", 220.0, 3000.0), ("CASH_OUT", 60.0, 500.0), ("TRANSFER", 9000.0, 100000.0), ] for i, (type_, amount, balance) in enumerate(partial_cases): day = 45 + i * 3 hour = (i * 6 + 9) % 24 fixtures.append( _txn( id_=f"legit-partial-{type_.lower()}-{i:02d}", label="legit", description=f"Partial {type_} -- amount well under balance, account not drained (newbalanceOrig > 0).", day=day, hour=hour, type_=type_, amount=amount, name_orig=f"CFXLEGITPART{i:03d}A", old_orig=balance, new_orig=balance - amount, name_dest=f"CFXLEGITPART{i:03d}B", old_dest=0.0, new_dest=amount, ) ) return fixtures def main() -> None: fixtures = build_fixtures() OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True) OUTPUT_PATH.write_text(json.dumps(fixtures, indent=2)) fraud_count = sum(1 for f in fixtures if f["expected_label"] == "fraud") legit_count = sum(1 for f in fixtures if f["expected_label"] == "legit") print(f"Wrote {len(fixtures)} fixtures ({fraud_count} fraud, {legit_count} legit) to {OUTPUT_PATH}") if __name__ == "__main__": main()