From 21c954206ab861aa8d55f4c2d33640bdb8654c2b Mon Sep 17 00:00:00 2001 From: Samarth Uday Date: Fri, 28 Aug 2026 23:28:21 +0530 Subject: [PATCH] feat: Implement artifact-backed API for transaction risk prediction - Added a new Flask API in `src/api/app.py` to serve risk predictions based on transaction features. - Introduced health check endpoint to verify model availability. - Updated prediction endpoint to return risk probability and review requirements. - Refactored transaction processing in `src/api/simple_api_server.py` to use risk probability instead of risk score. - Modified real-time dashboard to display risk probability. - Created data loading functionality in `src/data/loader.py` for SAML-D dataset. - Added evaluation metrics and explainability functions in `src/evaluation/metrics.py` and `src/evaluation/explainability.py`. - Developed feature engineering functions for behavioral and transaction features in `src/features/behavioral_features.py` and `src/features/transaction_features.py`. - Implemented model training and calibration logic in `src/models/train.py` and `src/models/calibration.py`. - Added unit tests for feature engineering and inference in `tests/test_features.py` and `tests/test_inference.py`. - Updated utility scripts for system startup and transaction generation to reflect API changes. --- .firebaserc | 5 - .gitattributes | 2 - .gitignore | 119 +++------ SAML-D.csv | 3 - artifacts/.gitkeep | 0 data/README.md | 25 ++ data/raw/.gitkeep | 0 .../system.ipynb => notebooks/01_eda.ipynb | 0 .../02_model_research.ipynb | 0 notebooks/03_backtesting.ipynb | 19 ++ reports/.gitkeep | 0 reports/figures/.gitkeep | 0 requirements.txt | 34 +-- scripts/build_features.py | 40 +++ scripts/feature_ablation.py | 93 +++++++ scripts/train_model.py | 224 +++++++++++++++++ src/api/__init__.py | 1 + src/api/app.py | 94 +++++++ src/api/simple_api_server.py | 10 +- src/dashboard/real_time_dashboard.html | 2 +- src/data/__init__.py | 0 src/data/loader.py | 76 ++++++ src/evaluation/__init__.py | 0 src/evaluation/explainability.py | 18 ++ src/evaluation/metrics.py | 196 +++++++++++++++ src/features/__init__.py | 0 src/features/behavioral_features.py | 198 +++++++++++++++ src/features/transaction_features.py | 53 ++++ src/models/__init__.py | 0 src/models/baseline.py | 14 ++ src/models/calibration.py | 47 ++++ src/models/inference.py | 56 +++++ src/models/train.py | 232 ++++++++++++++++++ src/utils/simple_ingestion.py | 15 +- src/utils/start_system.py | 77 +----- src/utils/test_ingestion.py | 2 +- tests/test_features.py | 25 ++ tests/test_inference.py | 27 ++ tests/test_leakage.py | 13 + 39 files changed, 1527 insertions(+), 193 deletions(-) delete mode 100644 .firebaserc delete mode 100644 .gitattributes delete mode 100644 SAML-D.csv create mode 100644 artifacts/.gitkeep create mode 100644 data/README.md create mode 100644 data/raw/.gitkeep rename src/notebooks/system.ipynb => notebooks/01_eda.ipynb (100%) rename src/notebooks/Model.ipynb => notebooks/02_model_research.ipynb (100%) create mode 100644 notebooks/03_backtesting.ipynb create mode 100644 reports/.gitkeep create mode 100644 reports/figures/.gitkeep create mode 100644 scripts/build_features.py create mode 100644 scripts/feature_ablation.py create mode 100644 scripts/train_model.py create mode 100644 src/api/__init__.py create mode 100644 src/api/app.py create mode 100644 src/data/__init__.py create mode 100644 src/data/loader.py create mode 100644 src/evaluation/__init__.py create mode 100644 src/evaluation/explainability.py create mode 100644 src/evaluation/metrics.py create mode 100644 src/features/__init__.py create mode 100644 src/features/behavioral_features.py create mode 100644 src/features/transaction_features.py create mode 100644 src/models/__init__.py create mode 100644 src/models/baseline.py create mode 100644 src/models/calibration.py create mode 100644 src/models/inference.py create mode 100644 src/models/train.py create mode 100644 tests/test_features.py create mode 100644 tests/test_inference.py create mode 100644 tests/test_leakage.py diff --git a/.firebaserc b/.firebaserc deleted file mode 100644 index 228161c..0000000 --- a/.firebaserc +++ /dev/null @@ -1,5 +0,0 @@ -{ - "projects": { - "default": "risk-compliance-system" - } -} diff --git a/.gitattributes b/.gitattributes deleted file mode 100644 index 4a63abc..0000000 --- a/.gitattributes +++ /dev/null @@ -1,2 +0,0 @@ -*.csv filter=lfs diff=lfs merge=lfs -text -SAML-D.csv filter=lfs diff=lfs merge=lfs -text diff --git a/.gitignore b/.gitignore index 8f5bc22..721f485 100644 --- a/.gitignore +++ b/.gitignore @@ -1,102 +1,47 @@ -# Logs -logs -*.log -npm-debug.log* -yarn-debug.log* -yarn-error.log* -firebase-debug.log* -firebase-debug.*.log* - -# Firebase cache -.firebase/ - -# Firebase config - -# Uncomment this if you'd like others to create their own Firebase project. -# For a team working on the same Firebase project(s), it is recommended to leave -# it commented so all members can deploy to the same project(s) in .firebaserc. -# .firebaserc - -# Runtime data -pids -*.pid -*.seed -*.pid.lock - -# Directory for instrumented libs generated by jscoverage/JSCover -lib-cov - -# Coverage directory used by tools like istanbul -coverage - -# nyc test coverage -.nyc_output - -# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files) -.grunt - -# Bower dependency directory (https://bower.io/) -bower_components - -# node-waf configuration -.lock-wscript - -# Compiled binary addons (http://nodejs.org/api/addons.html) -build/Release - -# Dependency directories -node_modules/ - -# Optional npm cache directory -.npm - -# Optional eslint cache -.eslintcache - -# Optional REPL history -.node_repl_history - -# Output of 'npm pack' -*.tgz - -# Yarn Integrity file -.yarn-integrity - -# dotenv environment variables file -.env - -# dataconnect generated files -.dataconnect - - # Python __pycache__/ -*.pyc -*.pyo -*.pyd +*.py[cod] +*.so .Python venv/ +.venv/ env/ -# Firebase -firebase-credentials.json -.firebase/ - # Environment .env -# Models -*.pkl -src/models/*.pkl +# IDE +.vscode/ +.idea/ +.DS_Store + +# Jupyter +.ipynb_checkpoints/ + +# Raw datasets +data/raw/* +!data/raw/.gitkeep + +# Generated datasets +data/processed/ +*.parquet + +# Model artifacts +artifacts/* +!artifacts/.gitkeep + +# Generated figures +reports/figures/* +!reports/figures/.gitkeep # Logs logs/ *.log -# VS Code -.vscode/settings.json +# Testing +.pytest_cache/ +.coverage +htmlcov/ -# Data -data/ -*.csv -*.json \ No newline at end of file +# Large data +*.csv \ No newline at end of file diff --git a/SAML-D.csv b/SAML-D.csv deleted file mode 100644 index 8ecbb59..0000000 --- a/SAML-D.csv +++ /dev/null @@ -1,3 +0,0 @@ -version https://git-lfs.github.com/spec/v1 -oid sha256:5b71ce2ea7b47fe6f19da1aa151776b04ec74560a852c2c077df91d20b8b4ef9 -size 996168850 diff --git a/artifacts/.gitkeep b/artifacts/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/README.md b/data/README.md new file mode 100644 index 0000000..9d81f7c --- /dev/null +++ b/data/README.md @@ -0,0 +1,25 @@ +# Dataset + +This project uses the Synthetic Anti-Money Laundering Dataset (SAML-D). + +The raw dataset is intentionally not committed to this repository because of +its size and third-party licensing terms. + +## Dataset + +SAML-D contains approximately 9.5 million synthetic financial transactions, +including rare suspicious transaction patterns and multiple laundering +typologies. + +Expected raw file: + +data/raw/SAML-D.csv + +## Citation + +B. Oztas, D. Cetinkaya, F. Adedoyin, M. Budka, H. Dogan and G. Aksu, +"Enhancing Anti-Money Laundering: Development of a Synthetic Transaction +Monitoring Dataset," 2023 IEEE International Conference on e-Business +Engineering (ICEBE), 2023. + +Dataset license: CC BY-NC-SA 4.0. \ No newline at end of file diff --git a/data/raw/.gitkeep b/data/raw/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/src/notebooks/system.ipynb b/notebooks/01_eda.ipynb similarity index 100% rename from src/notebooks/system.ipynb rename to notebooks/01_eda.ipynb diff --git a/src/notebooks/Model.ipynb b/notebooks/02_model_research.ipynb similarity index 100% rename from src/notebooks/Model.ipynb rename to notebooks/02_model_research.ipynb diff --git a/notebooks/03_backtesting.ipynb b/notebooks/03_backtesting.ipynb new file mode 100644 index 0000000..63f130f --- /dev/null +++ b/notebooks/03_backtesting.ipynb @@ -0,0 +1,19 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Out-of-Time Backtesting\n", + "\n", + "Use this notebook for walk-forward evaluation, alert-budget metrics, threshold optimisation, drift monitoring, and laundering-typology analysis. Populate results only from executed experiments." + ] + } + ], + "metadata": { + "kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"}, + "language_info": {"name": "python", "version": "3.12"} + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/reports/.gitkeep b/reports/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/reports/figures/.gitkeep b/reports/figures/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/requirements.txt b/requirements.txt index e4a6d93..20256bd 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,16 +1,18 @@ -pandas>=1.3.0 -numpy>=1.21.0 -scikit-learn>=1.0.0 -xgboost>=1.5.0 -imbalanced-learn>=0.8.0 -flask>=2.0.0 -flask-cors>=3.0.10 -python-dotenv>=0.19.0 -firebase-admin>=6.0.0 -google-cloud-firestore>=2.7.0 -schedule>=1.1.0 -matplotlib>=3.4.0 -seaborn>=0.11.0 -plotly>=5.0.0 -pytest>=6.2.0 -python-dateutil>=2.8.0 \ No newline at end of file +pandas>=2.2,<3.0 +numpy>=1.26,<3.0 +scikit-learn>=1.5,<2.0 +xgboost>=2.1,<4.0 + +duckdb>=1.1,<2.0 +pyarrow>=17,<22 +joblib>=1.4,<2.0 + +matplotlib>=3.9,<4.0 +shap>=0.46,<1.0 + +flask>=3.0,<4.0 +flask-cors>=5.0,<7.0 +python-dotenv>=1.0,<2.0 +requests>=2.32,<3.0 + +pytest>=8.0,<9.0 \ No newline at end of file diff --git a/scripts/build_features.py b/scripts/build_features.py new file mode 100644 index 0000000..976fdc6 --- /dev/null +++ b/scripts/build_features.py @@ -0,0 +1,40 @@ +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.data.loader import load_saml_data +from src.features.behavioral_features import add_behavioral_features +from src.features.transaction_features import add_transaction_features + +RAW_PATH = PROJECT_ROOT / "data/raw/SAML-D.csv" +OUTPUT_PATH = PROJECT_ROOT / "data/processed/transactions_features.parquet" + + +def main(): + print("Loading SAML-D...") + df = load_saml_data(RAW_PATH) + + print(f"Loaded {len(df):,} transactions") + + print("Building transaction features...") + df = add_transaction_features(df) + + print("Building behavioral features...") + df = add_behavioral_features(df) + + OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True) + + df.to_parquet( + OUTPUT_PATH, + index=False, + compression="snappy", + ) + + print(f"Feature dataset written to {OUTPUT_PATH}") + print(f"Shape: {df.shape}") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/scripts/feature_ablation.py b/scripts/feature_ablation.py new file mode 100644 index 0000000..2c4f87d --- /dev/null +++ b/scripts/feature_ablation.py @@ -0,0 +1,93 @@ +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) + +import pandas as pd + +from src.evaluation.metrics import ( + evaluate_model, + precision_recall_at_alert_rate, +) +from src.models.calibration import ProbabilityCalibrator +from src.models.train import ( + ABLATION_FEATURE_SETS, + TARGET, + chronological_split, + fit_model, +) + +FEATURE_PATH = PROJECT_ROOT / "data/processed/transactions_features.parquet" +ALERT_RATE = 0.005 + + +def evaluate_feature_set(name, features, train, validation, test): + ( + preprocessor, + model, + X_validation_processed, + y_validation, + ) = fit_model(train, validation, features) + + validation_probabilities = model.predict_proba( + X_validation_processed + )[:, 1] + + calibrator = ProbabilityCalibrator().fit( + validation_probabilities, + y_validation, + ) + + test_probabilities = calibrator.predict( + model.predict_proba( + preprocessor.transform(test[features]) + )[:, 1] + ) + + metrics = evaluate_model( + test[TARGET], + test_probabilities, + ) + alert_metrics = precision_recall_at_alert_rate( + test[TARGET], + test_probabilities, + ALERT_RATE, + ) + + return { + "model": name, + "pr_auc": metrics["pr_auc"], + "recall_at_0.5%": alert_metrics["recall"], + } + + +def main(): + df = pd.read_parquet(FEATURE_PATH) + train, validation, test = chronological_split(df) + + results = [ + evaluate_feature_set( + name, + features, + train, + validation, + test, + ) + for name, features in ABLATION_FEATURE_SETS.items() + ] + + print("\nFEATURE ABLATION RESULTS") + print("=" * 55) + print(f"{'Model':<22} {'PR-AUC':>12} {'Recall@0.5%':>16}") + print("-" * 55) + for result in results: + print( + f"{result['model']:<22} " + f"{result['pr_auc']:>12.6f} " + f"{result['recall_at_0.5%']:>16.6f}" + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/train_model.py b/scripts/train_model.py new file mode 100644 index 0000000..5c0f0f7 --- /dev/null +++ b/scripts/train_model.py @@ -0,0 +1,224 @@ +import sys +from pathlib import Path + +import joblib +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.evaluation.metrics import ( + evaluate_model, + threshold_for_alert_rate, +) +from src.models.calibration import ( + ProbabilityCalibrator, +) +from src.models.train import ( + MODEL_FEATURES, + TARGET, + chronological_split, + fit_model, +) +from src.models.baseline import build_logistic_baseline + +FEATURE_PATH = PROJECT_ROOT / "data/processed/transactions_features.parquet" +ARTIFACT_PATH = PROJECT_ROOT / "artifacts/risk_model.joblib" + + +def print_split_stats(name, frame): + + positive = frame[TARGET].sum() + prevalence = frame[TARGET].mean() + + print( + f"{name}: " + f"{len(frame):,} rows | " + f"{positive:,} suspicious | " + f"{prevalence:.4%}" + ) + + +def main(): + + ARTIFACT_PATH.parent.mkdir(exist_ok=True) + + print("Loading feature dataset...") + + df = pd.read_parquet( + FEATURE_PATH + ) + + train, validation, test = ( + chronological_split(df) + ) + + print_split_stats( + "TRAIN", + train, + ) + + print_split_stats( + "VALIDATION", + validation, + ) + + print_split_stats( + "TEST", + test, + ) + + ( + preprocessor, + model, + X_validation_processed, + y_validation, + ) = fit_model( + train, + validation, + ) + + # --------------------------- + # CALIBRATION + # --------------------------- + + raw_validation_prob = ( + model.predict_proba( + X_validation_processed + )[:, 1] + ) + + calibrator = ProbabilityCalibrator() + + calibrator.fit( + raw_validation_prob, + y_validation, + ) + + calibrated_validation_prob = ( + calibrator.predict( + raw_validation_prob + ) + ) + + # Use a realistic operational alert capacity + alert_rate = 0.005 + + threshold = ( + threshold_for_alert_rate( + calibrated_validation_prob, + alert_rate, + ) + ) + + print( + f"\nThreshold for " + f"{alert_rate:.2%} alert rate: " + f"{threshold:.6f}" + ) + + # --------------------------- + # TEST + # --------------------------- + + X_test = test[MODEL_FEATURES] + y_test = test[TARGET] + + X_test_processed = ( + preprocessor.transform( + X_test + ) + ) + + raw_test_prob = ( + model.predict_proba( + X_test_processed + )[:, 1] + ) + + calibrated_test_prob = ( + calibrator.predict( + raw_test_prob + ) + ) + + metrics = evaluate_model( + y_test, + calibrated_test_prob, + ) + + print("\nOUT-OF-TIME TEST RESULTS") + print("=" * 50) + + for key, value in metrics.items(): + + if isinstance(value, float): + print( + f"{key:30s}: " + f"{value:.6f}" + ) + + else: + print( + f"{key:30s}: " + f"{value}" + ) + + # Benchmark the nonlinear model against a scalable logistic classifier on + # the exact same chronological partitions and preprocessing contract. + baseline = build_logistic_baseline() + baseline.fit(preprocessor.transform(train[MODEL_FEATURES]), train[TARGET]) + baseline_validation_prob = baseline.predict_proba( + X_validation_processed + )[:, 1] + baseline_calibrator = ProbabilityCalibrator().fit( + baseline_validation_prob, + y_validation, + ) + baseline_test_prob = baseline_calibrator.predict( + baseline.predict_proba(X_test_processed)[:, 1] + ) + baseline_metrics = evaluate_model(y_test, baseline_test_prob) + + print("\nLOGISTIC BASELINE TEST RESULTS") + print("=" * 50) + for key in ("pr_auc", "alert_0.500%_recall", "alert_0.500%_lift"): + print(f"{key:30s}: {baseline_metrics[key]:.6f}") + + # --------------------------- + # SAVE + # --------------------------- + + artifact = { + "preprocessor": preprocessor, + "model": model, + "calibrator": calibrator, + + "features": MODEL_FEATURES, + + "decision_threshold": threshold, + "alert_rate": alert_rate, + + "test_metrics": metrics, + "baseline_test_metrics": baseline_metrics, + "validation_probability_quantiles": [ + float(value) + for value in sorted(calibrated_validation_prob) + ], + + "model_version": "2.0.0", + } + + joblib.dump( + artifact, + ARTIFACT_PATH, + ) + + print( + "\nSaved model to " + f"{ARTIFACT_PATH.relative_to(PROJECT_ROOT)}" + ) + + +if __name__ == "__main__": + main() diff --git a/src/api/__init__.py b/src/api/__init__.py new file mode 100644 index 0000000..f295278 --- /dev/null +++ b/src/api/__init__.py @@ -0,0 +1 @@ +"""HTTP interfaces for the surveillance engine.""" diff --git a/src/api/app.py b/src/api/app.py new file mode 100644 index 0000000..f896c2e --- /dev/null +++ b/src/api/app.py @@ -0,0 +1,94 @@ +"""Artifact-backed API for feature-store supplied transaction features. + +This API intentionally does not calculate behavioural history from a single +raw transaction. That responsibility belongs to an online feature store. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import joblib +from flask import Flask, jsonify, request +from flask_cors import CORS + +from src.models.inference import ( + model_input_from_features, + predict_calibrated_probability, + probability_percentile, +) + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +MODEL_PATH = PROJECT_ROOT / "artifacts/risk_model.joblib" + + +def create_app(model_path: Path = MODEL_PATH) -> Flask: + app = Flask(__name__) + CORS(app) + + artifact: dict[str, Any] | None = None + load_error: str | None = None + try: + artifact = joblib.load(model_path) + except FileNotFoundError: + load_error = f"Model artifact not found at {model_path}. Train the offline model first." + except Exception as error: # pragma: no cover - defensive startup path + load_error = f"Unable to load model artifact: {error}" + + @app.get("/api/health") + def health(): + return jsonify( + { + "status": "healthy" if artifact else "model_unavailable", + "model_loaded": artifact is not None, + "timestamp": datetime.now(timezone.utc).isoformat(), + "detail": load_error, + } + ) + + @app.get("/api/model/info") + def model_info(): + if artifact is None: + return jsonify({"error": load_error}), 503 + return jsonify( + { + "model": "XGBoost", + "model_version": artifact["model_version"], + "alert_rate": artifact["alert_rate"], + "decision_threshold": artifact["decision_threshold"], + "test_metrics": artifact["test_metrics"], + } + ) + + @app.post("/api/predict") + def predict(): + if artifact is None: + return jsonify({"error": load_error}), 503 + payload = request.get_json(silent=True) + if not isinstance(payload, dict): + return jsonify({"error": "Request body must be a JSON object."}), 400 + try: + features = model_input_from_features(payload, artifact["features"]) + probability = predict_calibrated_probability(artifact, features) + except ValueError as error: + return jsonify({"error": str(error)}), 400 + + result: dict[str, Any] = { + "risk_probability": probability, + "requires_review": probability >= artifact["decision_threshold"], + } + percentile = probability_percentile(artifact, probability) + if percentile is not None: + result["risk_percentile"] = percentile + return jsonify(result) + + return app + + +app = create_app() + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=5000, debug=False) diff --git a/src/api/simple_api_server.py b/src/api/simple_api_server.py index b4fb6ae..2e95b91 100644 --- a/src/api/simple_api_server.py +++ b/src/api/simple_api_server.py @@ -227,7 +227,7 @@ def process_transaction(): if risk_level == 'HIGH': high_risk_transactions.append({ 'transaction_id': data['transaction_id'], - 'risk_score': risk_score, + 'risk_probability': risk_score, 'risk_level': risk_level, 'amount': data['amount'], 'sender_id': data['sender_id'], @@ -241,7 +241,7 @@ def process_transaction(): 'type': 'HIGH_RISK_TRANSACTION', 'severity': 'HIGH', 'message': f'High risk transaction detected: {data["transaction_id"]}', - 'risk_score': risk_score, + 'risk_probability': risk_score, 'amount': data['amount'], 'sender': data['sender_id'], 'receiver': data['receiver_id'], @@ -252,12 +252,11 @@ def process_transaction(): return jsonify({ 'transaction_id': data['transaction_id'], - 'risk_score': risk_score, + 'risk_probability': risk_score, 'risk_level': risk_level, 'compliance_status': 'PENDING' if requires_review else 'APPROVED', 'requires_review': requires_review, 'flagged_features': flagged_features, - 'confidence': abs(risk_score - 0.5) * 2, # Simple confidence calculation 'processed_at': datetime.now().isoformat() }) @@ -342,12 +341,11 @@ def bulk_process_transactions(): results.append({ 'transaction_id': tx_data['transaction_id'], - 'risk_score': risk_score, + 'risk_probability': risk_score, 'risk_level': risk_level, 'compliance_status': 'PENDING' if requires_review else 'APPROVED', 'requires_review': requires_review, 'flagged_features': flagged_features, - 'confidence': abs(risk_score - 0.5) * 2, 'processed_at': datetime.now().isoformat() }) diff --git a/src/dashboard/real_time_dashboard.html b/src/dashboard/real_time_dashboard.html index fc8e12b..3a51225 100644 --- a/src/dashboard/real_time_dashboard.html +++ b/src/dashboard/real_time_dashboard.html @@ -913,7 +913,7 @@

Recent Alerts

${alert.type || 'Risk Alert'}
Amount: ${amount}
- Risk Score: ${alert.risk_score ? alert.risk_score.toFixed(3) : 'N/A'}
+ Risk Probability: ${alert.risk_probability ? alert.risk_probability.toFixed(3) : 'N/A'}
${timestamp}
`; diff --git a/src/data/__init__.py b/src/data/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/data/loader.py b/src/data/loader.py new file mode 100644 index 0000000..f36b35a --- /dev/null +++ b/src/data/loader.py @@ -0,0 +1,76 @@ +from pathlib import Path + +import pandas as pd + +EXPECTED_COLUMNS = [ + "Time", + "Date", + "Sender_account", + "Receiver_account", + "Amount", + "Payment_currency", + "Received_currency", + "Sender_bank_location", + "Receiver_bank_location", + "Payment_type", + "Is_laundering", + "Laundering_type", +] + + +def load_saml_data( + path: str | Path = "data/raw/SAML-D.csv", +) -> pd.DataFrame: + """ + Load the SAML-D transaction dataset and construct a chronological timestamp. + """ + + path = Path(path) + + if not path.exists(): + raise FileNotFoundError( + f"SAML-D dataset not found at {path}. " + "Place the CSV inside data/raw/." + ) + + # A Git LFS pointer is text metadata, not the dataset itself. Detecting it + # here gives contributors an actionable error instead of a misleading + # missing-column failure from pandas. + if path.read_bytes()[:64].startswith(b"version https://git-lfs.github.com/spec/v1"): + raise FileNotFoundError( + f"{path} is a Git LFS pointer, not the SAML-D CSV. " + "Download the licensed dataset and place the actual CSV in data/raw/." + ) + + df = pd.read_csv( + path, + dtype={ + "Sender_account": "string", + "Receiver_account": "string", + "Amount": "float64", + "Payment_currency": "category", + "Received_currency": "category", + "Sender_bank_location": "category", + "Receiver_bank_location": "category", + "Payment_type": "category", + "Is_laundering": "int8", + "Laundering_type": "category", + }, + ) + + missing = set(EXPECTED_COLUMNS) - set(df.columns) + + if missing: + raise ValueError(f"Missing columns: {missing}") + + df["timestamp"] = pd.to_datetime( + df["Date"].astype(str) + " " + df["Time"].astype(str), + errors="coerce", + ) + + df = df.dropna(subset=["timestamp", "Amount"]) + + # Chronology is critical for financial backtesting. + df = df.sort_values("timestamp").reset_index(drop=True) + + return df diff --git a/src/evaluation/__init__.py b/src/evaluation/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/evaluation/explainability.py b/src/evaluation/explainability.py new file mode 100644 index 0000000..cb70136 --- /dev/null +++ b/src/evaluation/explainability.py @@ -0,0 +1,18 @@ +import shap + + +def create_shap_explainer(model): + return shap.TreeExplainer(model) + + +def calculate_shap_values( + model, + X, +): + explainer = create_shap_explainer( + model + ) + + return explainer.shap_values( + X + ) \ No newline at end of file diff --git a/src/evaluation/metrics.py b/src/evaluation/metrics.py new file mode 100644 index 0000000..9e9dca3 --- /dev/null +++ b/src/evaluation/metrics.py @@ -0,0 +1,196 @@ +import numpy as np +from sklearn.metrics import ( + average_precision_score, + brier_score_loss, + log_loss, + roc_auc_score, +) + + +def threshold_for_alert_rate( + probabilities, + alert_rate=0.005, +): + """ + Alert only the top X% highest-risk transactions. + """ + + probabilities = np.asarray(probabilities, dtype=float) + if not 0 < alert_rate <= 1: + raise ValueError("alert_rate must be in (0, 1].") + + return float( + np.quantile( + probabilities, + 1 - alert_rate, + ) + ) + + +def precision_recall_at_alert_rate( + y_true, + probabilities, + alert_rate=0.005, +): + y_true = np.asarray(y_true) + probabilities = np.asarray(probabilities) + + threshold = threshold_for_alert_rate( + probabilities, + alert_rate, + ) + + predicted = probabilities >= threshold + + tp = np.sum( + (predicted == 1) & + (y_true == 1) + ) + + fp = np.sum( + (predicted == 1) & + (y_true == 0) + ) + + fn = np.sum( + (predicted == 0) & + (y_true == 1) + ) + + precision = ( + tp / (tp + fp) + if tp + fp > 0 + else 0 + ) + + recall = ( + tp / (tp + fn) + if tp + fn > 0 + else 0 + ) + + base_rate = y_true.mean() + + lift = ( + precision / base_rate + if base_rate > 0 + else 0 + ) + + return { + "alert_rate": alert_rate, + "threshold": threshold, + "precision": precision, + "recall": recall, + "lift": lift, + "alerts": int(predicted.sum()), + } + + +def expected_decision_cost( + y_true, + probabilities, + threshold, + false_negative_cost=100, + false_positive_cost=1, +): + """Return the operational cost induced by a binary alert threshold.""" + + y_true = np.asarray(y_true) + predictions = (np.asarray(probabilities) >= threshold).astype(int) + + false_negatives = np.sum((predictions == 0) & (y_true == 1)) + false_positives = np.sum((predictions == 1) & (y_true == 0)) + + return float( + false_negative_cost * false_negatives + + false_positive_cost * false_positives + ) + +def calculate_risk_weighted_exposure( + amounts, + probabilities, +): + """ + Probability-weighted transaction amount. + + This is NOT called expected financial loss because + the dataset does not contain realized loss severity. + """ + + amounts = np.asarray(amounts) + probabilities = np.asarray(probabilities) + + return amounts * probabilities + + +def evaluate_model( + y_true, + probabilities, +): + + y_true = np.asarray(y_true) + if np.unique(y_true).size < 2: + raise ValueError("Evaluation requires both target classes.") + + probabilities = np.clip( + probabilities, + 1e-8, + 1 - 1e-8, + ) + + results = { + "pr_auc": + average_precision_score( + y_true, + probabilities, + ), + + "roc_auc": + roc_auc_score( + y_true, + probabilities, + ), + + "brier_score": + brier_score_loss( + y_true, + probabilities, + ), + + "log_loss": + log_loss( + y_true, + probabilities, + ), + } + + for rate in [ + 0.001, + 0.005, + 0.01, + ]: + + performance = ( + precision_recall_at_alert_rate( + y_true, + probabilities, + alert_rate=rate, + ) + ) + + prefix = f"alert_{rate:.3%}" + + results[ + f"{prefix}_precision" + ] = performance["precision"] + + results[ + f"{prefix}_recall" + ] = performance["recall"] + + results[ + f"{prefix}_lift" + ] = performance["lift"] + + return results diff --git a/src/features/__init__.py b/src/features/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/features/behavioral_features.py b/src/features/behavioral_features.py new file mode 100644 index 0000000..4d291a2 --- /dev/null +++ b/src/features/behavioral_features.py @@ -0,0 +1,198 @@ +import duckdb +import pandas as pd + + +def add_behavioral_features(df: pd.DataFrame) -> pd.DataFrame: + """ + Generate historical behavioral variables. + + IMPORTANT: + All rolling windows end immediately BEFORE the current transaction. + This prevents future information from leaking into the prediction. + """ + + con = duckdb.connect() + + con.register("transactions", df) + + query = """ + WITH history AS ( + SELECT + *, + + COUNT(*) OVER ( + PARTITION BY Sender_account + ORDER BY timestamp + RANGE BETWEEN INTERVAL '24 hours' PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ) AS sender_txn_count_24h, + + COALESCE( + SUM(Amount) OVER ( + PARTITION BY Sender_account + ORDER BY timestamp + RANGE BETWEEN INTERVAL '24 hours' PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ), + 0 + ) AS sender_amount_sum_24h, + + AVG(Amount) OVER ( + PARTITION BY Sender_account + ORDER BY timestamp + RANGE BETWEEN INTERVAL '30 days' PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ) AS sender_amount_mean_30d, + + STDDEV_SAMP(Amount) OVER ( + PARTITION BY Sender_account + ORDER BY timestamp + RANGE BETWEEN INTERVAL '30 days' PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ) AS sender_amount_std_30d, + + COUNT(*) OVER ( + PARTITION BY Receiver_account + ORDER BY timestamp + RANGE BETWEEN INTERVAL '24 hours' PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ) AS receiver_txn_count_24h, + + COALESCE( + SUM(Amount) OVER ( + PARTITION BY Receiver_account + ORDER BY timestamp + RANGE BETWEEN INTERVAL '24 hours' PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ), + 0 + ) AS receiver_amount_sum_24h, + + ROW_NUMBER() OVER ( + PARTITION BY Sender_account, Receiver_account + ORDER BY timestamp + ) - 1 AS sender_receiver_prior_count, + + LAG(timestamp) OVER ( + PARTITION BY Sender_account + ORDER BY timestamp + ) AS previous_sender_timestamp, + + COUNT(*) OVER ( + PARTITION BY Sender_account + ORDER BY timestamp + RANGE BETWEEN UNBOUNDED PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ) AS sender_out_degree, + + COUNT(*) OVER ( + PARTITION BY Receiver_account + ORDER BY timestamp + RANGE BETWEEN UNBOUNDED PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ) AS receiver_in_degree, + + COUNT(DISTINCT Receiver_account) OVER ( + PARTITION BY Sender_account + ORDER BY timestamp + RANGE BETWEEN UNBOUNDED PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ) AS sender_unique_counterparties, + + COUNT(DISTINCT Sender_account) OVER ( + PARTITION BY Receiver_account + ORDER BY timestamp + RANGE BETWEEN UNBOUNDED PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ) AS receiver_unique_counterparties, + + COUNT(*) OVER ( + PARTITION BY Sender_account, Receiver_account + ORDER BY timestamp + RANGE BETWEEN UNBOUNDED PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ) AS pair_transaction_count + + FROM transactions + ) + + SELECT + *, + + CASE + WHEN sender_amount_std_30d IS NULL + OR sender_amount_std_30d = 0 + THEN 0 + ELSE + (Amount - sender_amount_mean_30d) + / sender_amount_std_30d + END AS sender_amount_zscore, + + COALESCE( + DATE_DIFF( + 'second', + previous_sender_timestamp, + timestamp + ), + -1 + ) AS seconds_since_sender_txn + + FROM history + ORDER BY timestamp + """ + + result = con.execute(query).fetchdf() + + con.close() + + numerical_history_columns = [ + "sender_txn_count_24h", + "sender_amount_sum_24h", + "sender_amount_mean_30d", + "sender_amount_std_30d", + "receiver_txn_count_24h", + "receiver_amount_sum_24h", + "sender_receiver_prior_count", + "sender_amount_zscore", + "seconds_since_sender_txn", + ] + + result[numerical_history_columns] = ( + result[numerical_history_columns] + .replace([float("inf"), float("-inf")], 0) + .fillna(0) + ) + + sender_totals = {} + sender_squared_totals = {} + sender_receiver_totals = {} + concentration = [] + + for _, timestamp_group in result.groupby("timestamp", sort=False): + for row in timestamp_group.itertuples(index=False): + sender = row.Sender_account + receiver = row.Receiver_account + total = sender_totals.get(sender, 0.0) + concentration.append( + sender_squared_totals.get(sender, 0.0) / total**2 + if total > 0 + else 0.0 + ) + + for row in timestamp_group.itertuples(index=False): + sender = row.Sender_account + receiver = row.Receiver_account + amount = float(row.Amount) + pair_key = (sender, receiver) + pair_total = sender_receiver_totals.get(pair_key, 0.0) + sender_receiver_totals[pair_key] = pair_total + amount + sender_totals[sender] = sender_totals.get(sender, 0.0) + amount + sender_squared_totals[sender] = ( + sender_squared_totals.get(sender, 0.0) + + 2 * pair_total * amount + + amount**2 + ) + + result["sender_counterparty_hhi"] = concentration + + return result \ No newline at end of file diff --git a/src/features/transaction_features.py b/src/features/transaction_features.py new file mode 100644 index 0000000..66b6160 --- /dev/null +++ b/src/features/transaction_features.py @@ -0,0 +1,53 @@ +import numpy as np +import pandas as pd + + +def add_transaction_features(df: pd.DataFrame) -> pd.DataFrame: + """ + Features derived only from the current transaction. + No historical information is used here. + """ + + df = df.copy() + + ts = df["timestamp"] + + hour = ts.dt.hour + day_of_week = ts.dt.dayofweek + month = ts.dt.month + + # Heavy-tailed transaction amounts + df["log_amount"] = np.log1p(df["Amount"]) + + # Cyclical temporal encoding + df["hour_sin"] = np.sin(2 * np.pi * hour / 24) + df["hour_cos"] = np.cos(2 * np.pi * hour / 24) + + df["dow_sin"] = np.sin(2 * np.pi * day_of_week / 7) + df["dow_cos"] = np.cos(2 * np.pi * day_of_week / 7) + + df["month_sin"] = np.sin(2 * np.pi * month / 12) + df["month_cos"] = np.cos(2 * np.pi * month / 12) + + # Time risk indicators + df["is_weekend"] = (day_of_week >= 5).astype("int8") + df["is_night"] = ((hour >= 22) | (hour <= 6)).astype("int8") + + # Transaction structure + df["currency_mismatch"] = ( + df["Payment_currency"].astype(str) + != df["Received_currency"].astype(str) + ).astype("int8") + + df["cross_border"] = ( + df["Sender_bank_location"].astype(str) + != df["Receiver_bank_location"].astype(str) + ).astype("int8") + + # Round-value behavior + df["is_round_amount"] = np.isclose( + df["Amount"] % 1000, + 0, + ).astype("int8") + + return df \ No newline at end of file diff --git a/src/models/__init__.py b/src/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/models/baseline.py b/src/models/baseline.py new file mode 100644 index 0000000..ce59901 --- /dev/null +++ b/src/models/baseline.py @@ -0,0 +1,14 @@ +from sklearn.linear_model import SGDClassifier + + +def build_logistic_baseline(): + + return SGDClassifier( + loss="log_loss", + penalty="l2", + alpha=1e-4, + class_weight="balanced", + max_iter=1000, + tol=1e-3, + random_state=42, + ) \ No newline at end of file diff --git a/src/models/calibration.py b/src/models/calibration.py new file mode 100644 index 0000000..7c702d3 --- /dev/null +++ b/src/models/calibration.py @@ -0,0 +1,47 @@ +import numpy as np +from sklearn.linear_model import LogisticRegression + +EPSILON = 1e-6 + + +def probability_to_logit(probability): + probability = np.clip( + probability, + EPSILON, + 1 - EPSILON, + ) + + return np.log( + probability / (1 - probability) + ) + + +class ProbabilityCalibrator: + + def __init__(self): + self.model = LogisticRegression() + + def fit(self, probabilities, targets): + targets = np.asarray(targets) + if np.unique(targets).size < 2: + raise ValueError("Probability calibration requires both target classes.") + + logits = probability_to_logit( + np.asarray(probabilities) + ).reshape(-1, 1) + + self.model.fit( + logits, + targets, + ) + + return self + + def predict(self, probabilities): + logits = probability_to_logit( + np.asarray(probabilities) + ).reshape(-1, 1) + + return self.model.predict_proba( + logits + )[:, 1] diff --git a/src/models/inference.py b/src/models/inference.py new file mode 100644 index 0000000..2742e41 --- /dev/null +++ b/src/models/inference.py @@ -0,0 +1,56 @@ +"""Artifact-backed scoring helpers shared by offline and API inference.""" + +from __future__ import annotations + +from typing import Any, Mapping + +import numpy as np +import pandas as pd + + +def model_input_from_features( + transaction_features: Mapping[str, Any], + feature_names: list[str], +) -> pd.DataFrame: + """Validate and order an already-computed feature payload. + + Account identifiers are deliberately absent: a production feature store + must turn them into historical behavioural variables before this boundary. + """ + + forbidden = {"Sender_account", "Receiver_account", "Is_laundering", "Laundering_type"} + supplied = set(transaction_features) + forbidden_supplied = supplied & forbidden + if forbidden_supplied: + raise ValueError( + "Identifier or target fields are not model inputs: " + f"{sorted(forbidden_supplied)}" + ) + + missing = set(feature_names) - supplied + if missing: + raise ValueError( + "Missing model features. Historical behavioural features must be " + f"provided by the feature store: {sorted(missing)}" + ) + + return pd.DataFrame([{name: transaction_features[name] for name in feature_names}]) + + +def predict_calibrated_probability(artifact: Mapping[str, Any], features: pd.DataFrame) -> float: + """Score one feature row with the saved preprocessing, model, and calibrator.""" + + processed = artifact["preprocessor"].transform(features) + raw_probability = artifact["model"].predict_proba(processed)[:, 1] + return float(artifact["calibrator"].predict(raw_probability)[0]) + + +def probability_percentile(artifact: Mapping[str, Any], probability: float) -> float | None: + """Rank a probability against validation probabilities, when available.""" + + reference = artifact.get("validation_probability_quantiles") + if not reference: + return None + + values = np.asarray(reference, dtype=float) + return float(100 * np.searchsorted(values, probability, side="right") / len(values)) diff --git a/src/models/train.py b/src/models/train.py new file mode 100644 index 0000000..8647465 --- /dev/null +++ b/src/models/train.py @@ -0,0 +1,232 @@ +import numpy as np +import pandas as pd +import xgboost as xgb +from sklearn.compose import ColumnTransformer +from sklearn.impute import SimpleImputer +from sklearn.pipeline import Pipeline +from sklearn.preprocessing import OneHotEncoder + + +def chronological_split( + df: pd.DataFrame, + train_fraction: float = 0.70, + validation_fraction: float = 0.15, +): + df = df.sort_values("timestamp").reset_index(drop=True) + + n = len(df) + + train_end = int(n * train_fraction) + validation_end = int( + n * (train_fraction + validation_fraction) + ) + + train = df.iloc[:train_end].copy() + validation = df.iloc[train_end:validation_end].copy() + test = df.iloc[validation_end:].copy() + + return train, validation, test + +TARGET = "Is_laundering" + + +NUMERIC_FEATURES = [ + "Amount", + "log_amount", + + "hour_sin", + "hour_cos", + "dow_sin", + "dow_cos", + "month_sin", + "month_cos", + + "is_weekend", + "is_night", + + "currency_mismatch", + "cross_border", + "is_round_amount", + + "sender_txn_count_24h", + "sender_amount_sum_24h", + "sender_amount_mean_30d", + "sender_amount_std_30d", + "sender_amount_zscore", + + "receiver_txn_count_24h", + "receiver_amount_sum_24h", + + "sender_receiver_prior_count", + "seconds_since_sender_txn", + + "sender_out_degree", + "receiver_in_degree", + "sender_unique_counterparties", + "receiver_unique_counterparties", + "pair_transaction_count", + "sender_counterparty_hhi", +] + + +CATEGORICAL_FEATURES = [ + "Payment_type", + "Payment_currency", + "Received_currency", + "Sender_bank_location", + "Receiver_bank_location", +] + + +MODEL_FEATURES = NUMERIC_FEATURES + CATEGORICAL_FEATURES + +BASE_FEATURES = [ + "Amount", + "log_amount", + "hour_sin", + "hour_cos", + "dow_sin", + "dow_cos", + "month_sin", + "month_cos", + "is_weekend", + "is_night", + "currency_mismatch", + "cross_border", + "is_round_amount", +] + +BEHAVIORAL_FEATURES = [ + "sender_txn_count_24h", + "sender_amount_sum_24h", + "sender_amount_mean_30d", + "sender_amount_std_30d", + "sender_amount_zscore", + "receiver_txn_count_24h", + "receiver_amount_sum_24h", + "sender_receiver_prior_count", + "seconds_since_sender_txn", +] + +NETWORK_FEATURES = [ + "sender_out_degree", + "receiver_in_degree", + "sender_unique_counterparties", + "receiver_unique_counterparties", + "pair_transaction_count", + "sender_counterparty_hhi", +] + +ABLATION_FEATURE_SETS = { + "Base": BASE_FEATURES + CATEGORICAL_FEATURES, + "+ Behavioral": BASE_FEATURES + BEHAVIORAL_FEATURES + CATEGORICAL_FEATURES, + "+ Network": BASE_FEATURES + NETWORK_FEATURES + CATEGORICAL_FEATURES, + "All": MODEL_FEATURES, +} + + +def build_preprocessor(feature_names=MODEL_FEATURES): + numeric_features = [ + feature for feature in feature_names + if feature in NUMERIC_FEATURES + ] + categorical_features = [ + feature for feature in feature_names + if feature in CATEGORICAL_FEATURES + ] + + numeric_pipeline = Pipeline( + steps=[ + ( + "imputer", + SimpleImputer(strategy="median"), + ), + ] + ) + + categorical_pipeline = Pipeline( + steps=[ + ( + "imputer", + SimpleImputer(strategy="most_frequent"), + ), + ( + "onehot", + OneHotEncoder( + handle_unknown="ignore", + min_frequency=20, + ), + ), + ] + ) + + return ColumnTransformer( + transformers=[ + ( + "numeric", + numeric_pipeline, + numeric_features, + ), + ( + "categorical", + categorical_pipeline, + categorical_features, + ), + ], + remainder="drop", + ) + + +def build_xgboost_model(y_train): + positives = int(y_train.sum()) + negatives = len(y_train) - positives + + scale_pos_weight = negatives / max(positives, 1) + + print( + "scale_pos_weight:", + round(scale_pos_weight, 2), + ) + + return xgb.XGBClassifier( + objective="binary:logistic", + n_estimators=500, + max_depth=5, + learning_rate=0.05, + subsample=0.8, + colsample_bytree=0.8, + min_child_weight=5, + reg_alpha=0.1, + reg_lambda=2.0, + scale_pos_weight=scale_pos_weight, + eval_metric="aucpr", + tree_method="hist", + random_state=42, + n_jobs=-1, + ) + + +def fit_model(train, validation, feature_names=MODEL_FEATURES): + X_train = train[feature_names] + y_train = train[TARGET] + + X_validation = validation[feature_names] + y_validation = validation[TARGET] + + preprocessor = build_preprocessor(feature_names) + + X_train_processed = preprocessor.fit_transform(X_train) + X_validation_processed = preprocessor.transform(X_validation) + + model = build_xgboost_model(y_train) + model.fit( + X_train_processed, + y_train, + ) + + return ( + preprocessor, + model, + X_validation_processed, + y_validation, + ) \ No newline at end of file diff --git a/src/utils/simple_ingestion.py b/src/utils/simple_ingestion.py index 25bc473..5af7dbd 100644 --- a/src/utils/simple_ingestion.py +++ b/src/utils/simple_ingestion.py @@ -1,7 +1,8 @@ #!/usr/bin/env python3 -""" -Simple Real-Time Transaction Generator -Generates transactions continuously to demonstrate the system +"""Demo Transaction Stream Simulator. + +This random generator is only for demonstrating legacy UI/infrastructure; it +is not used for SAML-D model training, backtesting, or performance claims. """ import requests @@ -22,7 +23,7 @@ def generate_transaction(self): currencies = ['USD', 'EUR', 'GBP', 'JPY', 'CAD'] locations = ['US', 'UK', 'EU', 'JP', 'CA', 'AU', 'SG'] - # Generate realistic transaction data + # Deliberately simple demo data, not realistic financial behaviour. amount = random.uniform(100, 1000000) transaction_type = random.choice(transaction_types) @@ -60,7 +61,7 @@ def send_transaction(self, transaction): if response.status_code == 200: result = response.json() self.transaction_count += 1 - print(f"Transaction {self.transaction_count}: {transaction['transaction_id']} - Risk Score: {result['risk_score']:.3f} - Amount: ${transaction['amount']:,.2f}") + print(f"Transaction {self.transaction_count}: {transaction['transaction_id']} - Risk Probability: {result['risk_probability']:.3f} - Amount: ${transaction['amount']:,.2f}") return True else: print(f"Transaction failed: {response.status_code}") @@ -72,7 +73,7 @@ def send_transaction(self, transaction): def start_generation(self): """Start generating transactions continuously""" - print("Starting Simple Real-Time Transaction Generator...") + print("Starting Demo Transaction Stream Simulator...") print(f"API URL: {self.api_url}") print("Generating transactions every 2-5 seconds...") print("=" * 60) @@ -110,7 +111,7 @@ def stop_generation(self): def main(): """Main function""" print("=" * 60) - print("SIMPLE REAL-TIME TRANSACTION GENERATOR") + print("DEMO TRANSACTION STREAM SIMULATOR") print("=" * 60) # Check if API is running diff --git a/src/utils/start_system.py b/src/utils/start_system.py index 809665c..042da60 100644 --- a/src/utils/start_system.py +++ b/src/utils/start_system.py @@ -1,7 +1,8 @@ #!/usr/bin/env python3 -""" -Simple startup script for the Real-Time Compliance Monitoring System -Launches the API server and transaction generator +"""Start the artifact-backed surveillance API and static dashboard. + +Live transaction simulation remains intentionally disabled until an online +feature store supplies the model's behavioural history. """ import subprocess @@ -17,9 +18,9 @@ def start_api_server(): print("šŸš€ Starting API Server...") try: - print(f"šŸš€ Starting API server with command: {sys.executable} src/api/simple_api_server.py") + print(f"šŸš€ Starting API server with command: {sys.executable} -m src.api.app") process = subprocess.Popen( - [sys.executable, "src/api/simple_api_server.py"], + [sys.executable, "-m", "src.api.app"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True @@ -61,44 +62,6 @@ def start_api_server(): print(f"āŒ Error starting API Server: {e}") return None -def start_transaction_generator(): - """Start the transaction generator""" - print("šŸ“” Starting Transaction Generator...") - - try: - print(f"šŸš€ Starting transaction generator with command: {sys.executable} src/utils/simple_ingestion.py") - process = subprocess.Popen( - [sys.executable, "src/utils/simple_ingestion.py"], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True - ) - - # Check if process started - if process.poll() is not None: - print("āŒ Transaction generator process failed to start") - stdout, stderr = process.communicate() - print(f"STDOUT: {stdout}") - print(f"STDERR: {stderr}") - return None - - # Wait for generator to start - time.sleep(2) - - if process.poll() is None: - print("āœ… Transaction Generator started successfully") - return process - else: - print("āŒ Failed to start Transaction Generator") - stdout, stderr = process.communicate() - print(f"STDOUT: {stdout}") - print(f"STDERR: {stderr}") - return None - - except Exception as e: - print(f"āŒ Error starting Transaction Generator: {e}") - return None - def start_dashboard_server(): """Start the dashboard server""" print("šŸŒ Starting Dashboard Server...") @@ -138,18 +101,10 @@ def start_dashboard_server(): print(f"āŒ Error starting Dashboard Server: {e}") return None -def cleanup(api_process, generator_process, dashboard_process): +def cleanup(api_process, dashboard_process): """Clean up processes on exit""" print("\nšŸ›‘ Shutting down Real-Time Compliance System...") - if generator_process: - print("Stopping Transaction Generator...") - generator_process.terminate() - try: - generator_process.wait(timeout=5) - except subprocess.TimeoutExpired: - generator_process.kill() - if dashboard_process: print("Stopping Dashboard Server...") dashboard_process.terminate() @@ -171,13 +126,12 @@ def cleanup(api_process, generator_process, dashboard_process): def main(): """Main startup function""" print("=" * 60) - print("šŸš€ REAL-TIME COMPLIANCE MONITORING SYSTEM") + print("šŸš€ QUANTITATIVE TRANSACTION RISK SURVEILLANCE ENGINE") print("=" * 60) # Check if required files exist required_files = [ - "src/api/simple_api_server.py", - "src/utils/simple_ingestion.py", + "src/api/app.py", "src/dashboard/serve_dashboard.py", "src/dashboard/real_time_dashboard.html" ] @@ -190,7 +144,6 @@ def main(): print("āœ… All required files found") api_process = None - generator_process = None dashboard_process = None try: @@ -200,12 +153,6 @@ def main(): print("āŒ Failed to start API Server. Exiting...") sys.exit(1) - # Start transaction generator - generator_process = start_transaction_generator() - if not generator_process: - print("āŒ Failed to start Transaction Generator. Exiting...") - sys.exit(1) - # Start dashboard server dashboard_process = start_dashboard_server() if not dashboard_process: @@ -213,11 +160,11 @@ def main(): sys.exit(1) print("\n" + "=" * 60) - print("šŸŽ‰ REAL-TIME COMPLIANCE MONITORING SYSTEM IS RUNNING!") + print("šŸŽ‰ SURVEILLANCE ENGINE IS RUNNING!") print("=" * 60) print("šŸ“Š Dashboard: http://localhost:8082/real_time_dashboard.html") print("šŸ”Œ API Server: http://localhost:5000") - print("šŸ“” Transaction Generator: Running") + print("šŸ“” Demo stream: disabled until an online feature store is available") print("\nšŸ’” Press Ctrl+C to stop the system") print("=" * 60) @@ -230,7 +177,7 @@ def main(): except Exception as e: print(f"\nāŒ Unexpected error: {e}") finally: - cleanup(api_process, generator_process, dashboard_process) + cleanup(api_process, dashboard_process) if __name__ == "__main__": main() diff --git a/src/utils/test_ingestion.py b/src/utils/test_ingestion.py index 0708d43..9652ca4 100644 --- a/src/utils/test_ingestion.py +++ b/src/utils/test_ingestion.py @@ -40,7 +40,7 @@ def test_ingestion(): response = requests.post('http://localhost:5000/api/process_transaction', json=test_tx) if response.status_code == 200: result = response.json() - print(f"āœ… Test Transaction Processed: Risk Score {result['risk_score']:.3f}") + print(f"āœ… Test Transaction Processed: Risk Probability {result['risk_probability']:.3f}") else: print(f"āŒ Test Transaction Failed: {response.status_code}") return diff --git a/tests/test_features.py b/tests/test_features.py new file mode 100644 index 0000000..8182bb2 --- /dev/null +++ b/tests/test_features.py @@ -0,0 +1,25 @@ +import pandas as pd + +from src.features.transaction_features import add_transaction_features + + +def test_transaction_features_do_not_encode_account_ids(): + frame = pd.DataFrame( + { + "timestamp": pd.to_datetime(["2026-01-01 12:00:00"]), + "Sender_account": ["sender-001"], + "Receiver_account": ["receiver-001"], + "Amount": [1000.0], + "Payment_currency": ["UK pounds"], + "Received_currency": ["UK pounds"], + "Sender_bank_location": ["UK"], + "Receiver_bank_location": ["UK"], + } + ) + + result = add_transaction_features(frame) + + assert "log_amount" in result + assert "currency_mismatch" in result + assert result["Sender_account"].dtype == object + assert result["Receiver_account"].dtype == object diff --git a/tests/test_inference.py b/tests/test_inference.py new file mode 100644 index 0000000..3634e8a --- /dev/null +++ b/tests/test_inference.py @@ -0,0 +1,27 @@ +from pathlib import Path + +import pytest + +from src.api.app import create_app +from src.models.inference import model_input_from_features + + +def test_inference_rejects_identifiers_and_targets(): + with pytest.raises(ValueError, match="Identifier or target fields"): + model_input_from_features( + {"Amount": 100.0, "Sender_account": "not-a-feature"}, + ["Amount"], + ) + + +def test_inference_requires_feature_store_output(): + with pytest.raises(ValueError, match="Missing model features"): + model_input_from_features({"Amount": 100.0}, ["Amount", "sender_txn_count_24h"]) + + +def test_api_reports_missing_model_artifact(tmp_path: Path): + app = create_app(tmp_path / "absent.joblib") + response = app.test_client().get("/api/health") + + assert response.status_code == 200 + assert response.get_json()["model_loaded"] is False diff --git a/tests/test_leakage.py b/tests/test_leakage.py new file mode 100644 index 0000000..fc9019e --- /dev/null +++ b/tests/test_leakage.py @@ -0,0 +1,13 @@ +from src.models.train import MODEL_FEATURES + + +FORBIDDEN_FEATURES = { + "Is_laundering", + "Laundering_type", + "Sender_account", + "Receiver_account", +} + + +def test_no_identifier_or_target_leakage_features(): + assert not (set(MODEL_FEATURES) & FORBIDDEN_FEATURES)