diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bba71de..d38b4ed 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,10 +14,10 @@ jobs: steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: ${{ matrix.python-version }} cache: pip @@ -25,7 +25,10 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -r requirements.txt + pip install -e ".[dev]" + + - name: Lint + run: ruff check . - name: Run test suite run: python -m pytest -v diff --git a/README.md b/README.md index 7da8f46..306b174 100644 --- a/README.md +++ b/README.md @@ -2,273 +2,248 @@ # Quantitative Transaction Risk Modeling -**A research-grade framework for rare-event financial transaction surveillance using temporal, behavioral, and network signals.** +### A research-oriented framework for rare-event financial transaction surveillance using temporal, behavioral, and network signals. -[![Python](https://img.shields.io/badge/python-3.10%2B-blue.svg)](https://www.python.org/) -[![License: MIT](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE) -[![Build](https://img.shields.io/badge/build-passing-brightgreen.svg)](#) -[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black) -[![PRs Welcome](https://img.shields.io/badge/PRs-welcome-brightgreen.svg)](#contributing) - -`Python` · `XGBoost` · `scikit-learn` · `DuckDB` · `SHAP` · `Flask` - -[Overview](#overview) • -[Research Design](#research-design) • -[Quick Start](#quick-start) • -[API](#inference-api) • -[Results](#results) • -[Contributing](#contributing) +**Python · XGBoost · scikit-learn · DuckDB · SHAP · Flask** ---- +## Research Question -## Overview +Can historical transaction behavior, temporal dynamics, and network structure improve the identification of rare suspicious transactions under a constrained investigation budget, and does that improvement remain stable out of time? -**Quantitative Transaction Risk Modeling** is an end-to-end framework for modeling rare suspicious activity in high-volume financial transaction data. It is built as a research artifact first and a production prototype second — every design choice is made to keep the evaluation honest under severe class imbalance and realistic operational constraints. +The framework uses the synthetic SAML-D dataset, containing approximately 9.5 million transactions with a suspicious-event rate of roughly 0.1%. It treats surveillance as a ranking and decision problem rather than a conventional balanced classification task. -**Research question:** +For transaction $i$, the model estimates: -> Can historical transaction behavior, temporal dynamics, and network structure improve the identification of rare suspicious transactions under a constrained investigation budget? +$$ +P(Y_i = 1 \mid X_i) +$$ -The framework is validated on the **SAML-D** synthetic transaction dataset — approximately **9.5 million transactions** with a suspicious-event rate of roughly **0.1%** — and emphasizes: +where $X_i$ contains only information available before the transaction occurs. -- Strict chronological development (no shuffling, no lookahead) -- Leakage-aware historical feature construction -- Probability calibration, not just raw model scores -- Alert-budget–constrained decision thresholds -- Out-of-time evaluation on a held-out future window -- Reproducible, versioned inference artifacts +## Workflow -## Why This Exists +```mermaid +flowchart LR + A[SAML-D CSV] --> B[DuckDB Feature Construction] + B --> C[Timestamp-Boundary Splits] + C --> D[Logistic Benchmark] + C --> E[XGBoost] + E --> F[Probability Calibration] + F --> G[Alert-Budget Evaluation] + G --> H[Walk-Forward Backtesting] + H --> I[Reports and Artifact] + I --> J[Optional Inference API] +``` -Most fraud/AML tutorials optimize ROC-AUC on a random train/test split and call it done. That approach silently leaks the future into the past and produces metrics that don't survive contact with a real compliance team, who can only investigate a fixed number of alerts per day. This project is an attempt to do it properly: chronological splits, calibrated probabilities, and evaluation metrics chosen because they hold up under a ~0.1% base rate and a fixed alert budget. +Historical windows terminate strictly before the current transaction. Preprocessing is fitted on training data only, while the final test period preserves the natural class prevalence. -## Research Design +## Features -Transaction surveillance is treated as a **ranking and decision problem**, not a plain classification problem. For transaction $i$, the model estimates: +- **Transaction:** amount, log amount, cyclical time, weekend/night indicators, currency mismatch, geographic mismatch, and round amounts. +- **Behavioral:** currency-aware rolling sender/receiver activity, account-relative z-scores, time since the previous sender transaction, and historical pair frequency. +- **Network:** lifetime transaction counts, unique counterparties, pair frequency, and sender counterparty concentration using an HHI-style measure. $$ -P(Y_i = 1 \mid X_i) +HHI_i = \sum_j p_{ij}^{2} $$ -where $Y_i = 1$ denotes suspicious activity and $X_i$ contains only information available *before* the transaction occurs. +Here $p_{ij}$ is the fraction of a sender's historical transfer value sent to counterparty $j$. -```mermaid -flowchart LR - A[SAML-D Transactions] --> B[Chronological Ordering] - B --> C[Transaction Features] - B --> D[Behavioral Features] - B --> E[Network Features] - C --> F[Leakage-Safe Preprocessing] - D --> F - E --> F - F --> G[Logistic Baseline] - F --> H[XGBoost] - H --> I[Probability Calibration] - I --> J[Alert-Budget Threshold] - J --> K[Out-of-Time Evaluation] - K --> L[Model Artifact] - L --> M[Inference API] -``` +## Evaluation -Historical variables use windows ending strictly before the current transaction. The dataset is split chronologically into **70% training**, **15% validation/calibration**, and **15% untouched out-of-time test** data — the test period is never seen during feature fitting, calibration, or threshold selection. +The project reports PR-AUC, ROC-AUC, Brier score, log loss, precision, recall, and lift at explicit alert budgets including 0.1%, 0.5%, and 1.0%. Alert metrics select exactly the top-$K$ ranked transactions, including when probabilities tie. -## Feature Engineering +Feature ablation compares transaction-only, transaction plus behavioral, transaction plus network, and full feature sets. Walk-forward evaluation measures whether performance is stable across future periods. Generated results are written to `reports/` after a dataset-backed run. -| Group | Examples | -| --- | --- | -| **Transaction** | amount, log amount, cyclical time-of-day, weekend/night indicators, currency mismatch, geographic mismatch, round-amount flags | -| **Behavioral** | rolling sender/receiver activity, currency-aware amount statistics, account-relative z-scores, time since previous sender transaction, historical sender–receiver interaction counts | -| **Network** | historical transaction counts, unique counterparties, pair frequency, sender counterparty concentration | +## Results -Network features treat accounts as nodes and transactions as directed edges. Counterparty concentration is measured with a Herfindahl–Hirschman-style index: +The model was trained and evaluated on the SAML-D dataset (9.5M transactions, synthetic). Below are the key metrics from the full pipeline run: -$$ -HHI_i = \sum_j p_{ij}^{2} -$$ +### Performance Metrics -where $p_{ij}$ is the fraction of a sender's historical transfer value sent to counterparty $j$. +| Metric | Score | +|--------|-------| +| **PR-AUC** | **0.9859** | +| **ROC-AUC** | **0.9998** | +| Brier Score | 0.0001 | +| Log Loss | 0.0004 | +| Precision @ 0.1% alert rate | 100% | +| Recall @ 0.1% alert rate | 83.9% | +| Lift @ 0.1% alert rate | **838.8x** | -## Modeling & Evaluation +### Feature Ablation Results -A regularized logistic classifier provides an interpretable linear benchmark. **XGBoost** is the primary nonlinear model, trained with class weighting to account for the rarity of the positive class. +Model performance improves dramatically as feature categories are added: -Because the event rate is ~0.1%, evaluation deliberately avoids metrics that look good by default under imbalance: +| Feature Set | PR-AUC | Recall @ 0.5% | +|---|---|---| +| Base (transaction features only) | 0.0898 | 12.9% | +| + Behavioral (activity patterns) | 0.2002 | 45.2% | +| + Network (graph concentration) | **0.9435** | **97.7%** | +| All (+ categorical) | **0.9859** | **99.3%** | -| Metric | Purpose | -| --- | --- | -| PR-AUC | Rare-event ranking quality | -| ROC-AUC | Overall discrimination | -| Brier score | Probability calibration | -| Log loss | Probabilistic accuracy | -| Precision@K | Suspicious share within the alert budget | -| Recall@K | Suspicious activity captured within the alert budget | -| Lift@K | Concentration relative to the base rate | +Network features provide the most significant lift (0.0898 → 0.9435), indicating that transaction graph structure is critical for AML detection. -Feature-ablation experiments compare **transaction-only**, **transaction + behavioral**, **transaction + network**, and **full feature** sets on identical chronological partitions, so the marginal value of behavioral and network signal can be measured directly rather than assumed. +### Behavioral Typology Detection -## Results +The model detects specific money-laundering patterns with high recall: + +- **100% detection**: Structuring, Smurfing, Cash Withdrawal, Fan-In, Layered Fan-In, Fan-Out, Bipartite +- **98%+ detection**: Behavioural Change, Cycle, Deposit-Send, Scatter-Gather, Gather-Scatter, Stacked Bipartite, Single Large +- **Limitations**: Over-Invoicing (86% recall) detected less reliably due to low transaction prevalence -> Fill this section in with the numbers from your latest `reports/` run before publishing — reviewers look here first. +### Analysis Visualizations -| Feature Set | PR-AUC | Precision@1% | Recall@1% | Lift@1% | -| --- | --- | --- | --- | --- | -| Transaction-only | — | — | — | — | -| + Behavioral | — | — | — | — | -| + Network | — | — | — | — | -| Full | — | — | — | — | +![ROC Curve](docs/assets/roc_curve.png) + +![Precision-Recall Curve](docs/assets/precision_recall_curve.png) + +![Feature Importance](docs/assets/feature_importance.png) + +![Ablation Comparison](docs/assets/ablation_comparison.png) + +![Typology Detection](docs/assets/typology_detection.png) + +![Calibration Curve](docs/assets/calibration_curve.png) ## Repository Structure ```text Quantitative-Transaction-Risk-Modeling/ -├── artifacts/ # Saved model artifacts (preprocessor, model, calibrator, metadata) -├── data/ # Raw and processed data (gitignored beyond samples) -├── docs/assets/ # Diagrams, plots, and images used in documentation -├── reports/ # Evaluation reports and ablation results +├── artifacts/ +├── data/ +│ ├── raw/ +│ └── processed/ +├── docs/ +│ └── assets/ # Committed model outputs, figures, metrics +├── public/ # Frontend SPA (HTML/CSS/JS) +├── reports/ +│ ├── figures/ # Generated visualization PNGs +│ ├── ablation_results.csv +│ ├── typology_results.csv +│ └── model_metrics.json ├── scripts/ -│ ├── build_features.py # Chronological, leakage-safe feature construction -│ ├── feature_ablation.py # Transaction / behavioral / network ablation study -│ └── train_model.py # Training, calibration, threshold selection, evaluation +│ ├── build_features.py +│ ├── feature_ablation.py +│ ├── generate_report_figures.py +│ ├── run_experiments.py +│ ├── train_model.py +│ └── walk_forward_backtest.py ├── src/ -│ ├── api/app.py # Flask inference API -│ ├── data/loader.py # Raw data loading utilities -│ ├── evaluation/ # Metrics and evaluation harness -│ ├── features/ # Feature engineering modules -│ └── models/ # Model wrappers and calibration logic -├── tests/ # Unit and integration tests +│ ├── api/ +│ │ ├── main.py # FastAPI application +│ │ ├── models.py # Pydantic request/response models +│ │ ├── inference.py # Async inference layer +│ │ └── results_loader.py +│ ├── data/ +│ ├── evaluation/ +│ ├── features/ +│ └── models/ +├── tests/ ├── README.md ├── pyproject.toml └── requirements.txt ``` -## Quick Start +## Quick Start & Reproduction ### Prerequisites - Python 3.11+ - ~4 GB free disk space for the SAML-D dataset and derived features + ### Installation ```bash -git clone https://github.com//Quantitative-Transaction-Risk-Modeling.git +git clone https://github.com/Samarthuday/Quantitative-Transaction-Risk-Modeling.git cd Quantitative-Transaction-Risk-Modeling - python3 -m venv venv -source venv/bin/activate # Windows: venv\Scripts\activate -pip install --upgrade pip -pip install -r requirements.txt +source venv/bin/activate +pip install -e ".[dev]" ``` -### Run the pipeline +Obtain SAML-D separately and place it at `data/raw/SAML-D.csv`, then run the complete workflow: + +```bash +venv/bin/python scripts/run_experiments.py +venv/bin/python -m pytest -v +``` -Download the [SAML-D dataset](https://www.kaggle.com/datasets/berkanoztas/synthetic-transaction-monitoring-dataset-aml) and place it at `data/raw/SAML-D.csv`, then: +For a development smoke run using a small feature file: ```bash -venv/bin/python scripts/build_features.py # chronological, leakage-safe features -venv/bin/python scripts/train_model.py # train, calibrate, select threshold, evaluate -venv/bin/python scripts/feature_ablation.py # transaction vs. behavioral vs. network study -venv/bin/python -m pytest -v # run the test suite +venv/bin/python scripts/train_model.py --features data/processed/features.parquet --fast ``` -`train_model.py` fits preprocessing on the training period only, calibrates validation probabilities, selects an alert-budget threshold, evaluates the untouched out-of-time test period, benchmarks against the logistic baseline, and writes `artifacts/risk_model.joblib`. +The workflow produces `data/processed/transactions_features.parquet`, `artifacts/risk_model.joblib`, `reports/model_metrics.json`, `reports/ablation_results.csv`, and `reports/walk_forward_results.csv`. -## Inference API +## Optional Inference Interface -The saved artifact bundles the preprocessor, XGBoost model, probability calibrator, feature contract, decision threshold, evaluation metrics, and reproducibility metadata into a single versioned object. +The FastAPI service scores precomputed feature vectors with the serialized preprocessor, model, and calibrator. It includes an interactive dashboard displaying model metrics, analysis charts, feature ablation, and behavioral typology detection rates. + +### Starting the Server ```bash -venv/bin/python -m src.api.app +uvicorn src.api.main:app --host 0.0.0.0 --port 8000 ``` +Open `http://localhost:8000` in your browser to see the dashboard with all analysis visualizations. + +### API Endpoints + | Method | Endpoint | Description | | --- | --- | --- | -| `GET` | `/api/health` | Service and model health check | -| `GET` | `/api/model/info` | Model metadata, feature contract, and evaluation metrics | -| `POST` | `/api/predict` | Calibrated transaction-risk inference | +| GET | `/` | Interactive dashboard (SPA) | +| GET | `/api/health` | Service and model health | +| GET | `/api/model/info` | Model metadata and evaluation metrics | +| GET | `/api/results` | Analysis results (ablation, typology, figures as base64) | +| POST | `/api/predict` | Calibrated probability for a feature vector | -
-Example request/response +### Example: Score a Transaction ```bash -curl -X POST http://localhost:5000/api/predict \ +curl -X POST http://localhost:8000/api/predict \ -H "Content-Type: application/json" \ -d '{ - "amount": 15230.50, - "sender_id": "ACC-10245", - "receiver_id": "ACC-88213", - "currency": "USD", - "timestamp": "2026-08-28T14:32:00Z" - }' + "Amount": 1000.0, + "log_amount": 6.908, + "hour_sin": 0.5, + "hour_cos": 0.866, + ... + }' ``` +**Response:** ```json { - "risk_score": 0.0421, - "flagged": false, - "threshold": 0.0387, - "model_version": "2026-08-01" + "risk_probability": 0.0002, + "requires_review": false, + "risk_percentile": 98.9 } ``` -
- -> **Note:** The API expects the engineered feature schema used during training. Historical behavioral variables must be supplied by an upstream feature store rather than reconstructed from a single raw transaction. - -## Reproducibility - -- Preprocessing is fitted using training observations only. -- Historical windows exclude the current transaction (no lookahead). -- Currency-specific amount statistics avoid mixing incomparable nominal values. -- Final evaluation preserves natural rare-event prevalence — no oversampling of the test set. -- Calibration and operational threshold selection are kept separate from final testing. -- The saved artifact records training dates, class prevalence, model hyperparameters, package versions, and compact validation probability quantiles for drift monitoring. - -## Roadmap - -- [ ] Streaming/online feature computation for near-real-time scoring -- [ ] Graph neural network baseline for the network-feature branch -- [ ] Model card and datasheet for the released artifact -- [ ] Dockerized inference service - -## Contributing +### Dashboard Features -Contributions are welcome. Please: +- **Overview tab**: Key metrics (PR-AUC, ROC-AUC, alert threshold, model status) +- **Analysis Charts tab**: ROC curve, Precision-Recall curve, feature importance, ablation comparison, typology detection, calibration curve +- **Feature Ablation tab**: Performance improvement as feature categories are added +- **Behavioral Typology tab**: Detection rates for 27 AML patterns with search/filter +- **Provenance tab**: Dataset size (9.5M transactions), model version, training metadata +- **Dark mode**: Toggle in header, saved to localStorage -1. Open an issue describing the change before large PRs. -2. Run `pytest` and `black .` before submitting. -3. Keep new features leakage-safe — anything derived from data must respect the chronological cutoff. +**Note:** Online historical feature computation is outside the current research scope. The API does not accept raw account identifiers or reconstruct behavioral history from a single transaction. -See [`CONTRIBUTING.md`](CONTRIBUTING.md) for the full guide. +## Limitations -## Citation - -If you use this framework or the accompanying analysis, please cite: - -```bibtex -@software{quant_transaction_risk_modeling, - title = {Quantitative Transaction Risk Modeling}, - author = {}, - year = {2026}, - url = {https://github.com//Quantitative-Transaction-Risk-Modeling} -} -``` +- SAML-D is synthetic; results do not establish performance at real financial institutions. +- Historical relationships may differ in real transaction networks. +- No causal interpretation is claimed. +- Calibration may drift under distribution shift. +- Risk probabilities are model estimates, not financial-loss probabilities. +- The online feature-store layer is outside the current research scope. ## Dataset Reference -This project uses the **Synthetic Anti-Money Laundering Dataset (SAML-D)**: - -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)*. - -## License - -Distributed under the [MIT License](LICENSE). - ---- - -
- -Built by [Your Name](https://github.com/) — feedback and issues welcome. - -
\ No newline at end of file +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). diff --git a/docs/assets/ablation_comparison.png b/docs/assets/ablation_comparison.png new file mode 100644 index 0000000..daa3cb7 Binary files /dev/null and b/docs/assets/ablation_comparison.png differ diff --git a/docs/assets/calibration_curve.png b/docs/assets/calibration_curve.png new file mode 100644 index 0000000..0aa2b25 Binary files /dev/null and b/docs/assets/calibration_curve.png differ diff --git a/docs/assets/feature_importance.png b/docs/assets/feature_importance.png new file mode 100644 index 0000000..77fc1f9 Binary files /dev/null and b/docs/assets/feature_importance.png differ diff --git a/docs/assets/model_metrics.json b/docs/assets/model_metrics.json new file mode 100644 index 0000000..9434212 --- /dev/null +++ b/docs/assets/model_metrics.json @@ -0,0 +1,41 @@ +{ + "model_version": "2.1.0", + "test_metrics": { + "pr_auc": 0.9859286355765932, + "roc_auc": 0.9998338278230887, + "brier_score": 9.91499618976377e-05, + "log_loss": 0.00044060166692361236, + "alert_0.100%_precision": 1.0, + "alert_0.100%_recall": 0.8388429752066116, + "alert_0.100%_lift": 838.7916174734357, + "alert_0.500%_precision": 0.23673469387755103, + "alert_0.500%_recall": 0.9929161747343566, + "alert_0.500%_lift": 198.5710767896297, + "alert_1.000%_precision": 0.11857846586910627, + "alert_1.000%_recall": 0.9946871310507674, + "alert_1.000%_lift": 99.46262318386624 + }, + "baseline_test_metrics": { + "pr_auc": 0.041376159513901396, + "roc_auc": 0.974922958712314, + "brier_score": 0.0011776858560728673, + "log_loss": 0.006451417472600472, + "alert_0.100%_precision": 0.025334271639690358, + "alert_0.100%_recall": 0.021251475796930343, + "alert_0.100%_lift": 21.250174686167266, + "alert_0.500%_precision": 0.07881773399014778, + "alert_0.500%_recall": 0.3305785123966942, + "alert_0.500%_lift": 66.11165457918705, + "alert_1.000%_precision": 0.05214637579169599, + "alert_1.000%_recall": 0.4374262101534829, + "alert_1.000%_lift": 43.73994289569429 + }, + "training_prevalence": 0.001003416857597292, + "test_prevalence": 0.0011921912179000402, + "calibration": { + "raw_brier_score": 0.0016988252755254507, + "calibrated_brier_score": 4.299887950764969e-05, + "raw_log_loss": 0.006217300426214933, + "calibrated_log_loss": 0.0002347317640669644 + } +} diff --git a/docs/assets/precision_recall_curve.png b/docs/assets/precision_recall_curve.png new file mode 100644 index 0000000..9d0128e Binary files /dev/null and b/docs/assets/precision_recall_curve.png differ diff --git a/docs/assets/roc_curve.png b/docs/assets/roc_curve.png new file mode 100644 index 0000000..1a38d49 Binary files /dev/null and b/docs/assets/roc_curve.png differ diff --git a/docs/assets/typology_detection.png b/docs/assets/typology_detection.png new file mode 100644 index 0000000..50be116 Binary files /dev/null and b/docs/assets/typology_detection.png differ diff --git a/public/index.html b/public/index.html new file mode 100644 index 0000000..18742a8 --- /dev/null +++ b/public/index.html @@ -0,0 +1,798 @@ + + + + + + Transaction Risk Model - Dashboard + + + + +
+
+
+

🚀 Transaction Risk Model

+

Real-time risk prediction & analysis dashboard

+
+
+ +
+
+ +
+ +
+ + + + + +
+ + +
+
+
+

Model Status

+
—
+
Current State
+
+
+ +
+

PR-AUC Score

+
—
+
Precision-Recall
+
+ +
+

ROC-AUC Score

+
—
+
Receiver Operating
+
+ +
+

Alert Threshold

+
—
+
Decision Threshold
+
+ +
+

Alert Rate

+
—
+
Top-K Percentage
+
+ +
+

Model Version

+
—
+
Training Version
+
+
+ +
+

📊 Performance Metrics

+
+
+
+ + +
+
+
+
+
+ + +
+
+

🔬 Feature Ablation Study

+

+ Shows how each feature category (behavioral, network) contributes to model performance. +

+ + + + + + + + + + + + +
Feature SetPR-AUCRecall @ 0.5% Alert RateImprovement
+
+
+ + +
+
+

🎯 Behavioral Typology Analysis

+

+ Detection rates across 27 money laundering behaviors. Green = high-risk behavior detected, Gray = normal transactions. +

+
+ +
+ + + + + + + + + + + + +
Behavior / PatternTransactionsSuspiciousDetection Rate
+
+
+ + +
+
+

📋 Model Provenance & Dataset Information

+

+ Complete information about the dataset, training configuration, and model generation. +

+
+
+
+
+
+
+
+ +
+

🔒 Quantitative Transaction Risk Modeling | FastAPI Dashboard

+

Last updated: —

+
+
+ + + + diff --git a/pyproject.toml b/pyproject.toml index 031f497..5104c29 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,9 +4,31 @@ build-backend = "setuptools.build_meta" [project] name = "quantitative-transaction-risk-modeling" -version = "2.0.0" +version = "2.1.0" description = "Rare-event financial transaction risk modeling with temporal, behavioral, and network features." requires-python = ">=3.11" +dependencies = [ + "numpy>=1.26,<3.0", + "pandas>=2.2,<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", + "fastapi>=0.109,<1.0", + "uvicorn[standard]>=0.29,<1.0", + "pydantic>=2.0,<3.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0,<9.0", + "pytest-asyncio>=0.23,<1.0", + "httpx>=0.26,<1.0", + "ruff>=0.6,<1.0", +] [tool.pytest.ini_options] testpaths = ["tests"] @@ -17,3 +39,7 @@ line-length = 88 [tool.ruff.lint] select = ["E", "F", "I"] +ignore = ["E501"] + +[tool.ruff.lint.per-file-ignores] +"scripts/*.py" = ["E402"] diff --git a/reports/model_metrics.json b/reports/model_metrics.json new file mode 100644 index 0000000..9434212 --- /dev/null +++ b/reports/model_metrics.json @@ -0,0 +1,41 @@ +{ + "model_version": "2.1.0", + "test_metrics": { + "pr_auc": 0.9859286355765932, + "roc_auc": 0.9998338278230887, + "brier_score": 9.91499618976377e-05, + "log_loss": 0.00044060166692361236, + "alert_0.100%_precision": 1.0, + "alert_0.100%_recall": 0.8388429752066116, + "alert_0.100%_lift": 838.7916174734357, + "alert_0.500%_precision": 0.23673469387755103, + "alert_0.500%_recall": 0.9929161747343566, + "alert_0.500%_lift": 198.5710767896297, + "alert_1.000%_precision": 0.11857846586910627, + "alert_1.000%_recall": 0.9946871310507674, + "alert_1.000%_lift": 99.46262318386624 + }, + "baseline_test_metrics": { + "pr_auc": 0.041376159513901396, + "roc_auc": 0.974922958712314, + "brier_score": 0.0011776858560728673, + "log_loss": 0.006451417472600472, + "alert_0.100%_precision": 0.025334271639690358, + "alert_0.100%_recall": 0.021251475796930343, + "alert_0.100%_lift": 21.250174686167266, + "alert_0.500%_precision": 0.07881773399014778, + "alert_0.500%_recall": 0.3305785123966942, + "alert_0.500%_lift": 66.11165457918705, + "alert_1.000%_precision": 0.05214637579169599, + "alert_1.000%_recall": 0.4374262101534829, + "alert_1.000%_lift": 43.73994289569429 + }, + "training_prevalence": 0.001003416857597292, + "test_prevalence": 0.0011921912179000402, + "calibration": { + "raw_brier_score": 0.0016988252755254507, + "calibrated_brier_score": 4.299887950764969e-05, + "raw_log_loss": 0.006217300426214933, + "calibrated_log_loss": 0.0002347317640669644 + } +} diff --git a/requirements.txt b/requirements.txt index 20256bd..2877d2b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,9 +10,11 @@ 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 +fastapi>=0.109,<1.0 +uvicorn[standard]>=0.29,<1.0 +pydantic>=2.0,<3.0 python-dotenv>=1.0,<2.0 requests>=2.32,<3.0 -pytest>=8.0,<9.0 \ No newline at end of file +pytest>=8.0,<9.0 +httpx>=0.26,<1.0 \ No newline at end of file diff --git a/scripts/build_features.py b/scripts/build_features.py index 976fdc6..93e56be 100644 --- a/scripts/build_features.py +++ b/scripts/build_features.py @@ -1,3 +1,4 @@ +import argparse import sys from pathlib import Path @@ -7,33 +8,43 @@ 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 +from src.logging_config import setup_logging + +logger = setup_logging(__name__) 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) + parser = argparse.ArgumentParser() + parser.add_argument("--input", type=Path, default=RAW_PATH) + parser.add_argument("--output", type=Path, default=OUTPUT_PATH) + args = parser.parse_args() - print(f"Loaded {len(df):,} transactions") + logger.info(f"Loading SAML-D from {args.input}...") + df = load_saml_data(args.input) + logger.info(f"Loaded {len(df):,} transactions") - print("Building transaction features...") + logger.info("Building transaction features...") df = add_transaction_features(df) + logger.debug(f"Transaction features: {[c for c in df.columns if c.startswith(('hour_', 'dow_', 'month_', 'is_', 'currency_', 'cross_'))]}") - print("Building behavioral features...") + logger.info("Building behavioral and network features...") df = add_behavioral_features(df) + logger.debug(f"Final dataset shape: {df.shape[0]:,} rows x {df.shape[1]} columns") - OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True) + args.output.parent.mkdir(parents=True, exist_ok=True) + logger.info(f"Writing feature dataset to {args.output}...") df.to_parquet( - OUTPUT_PATH, + args.output, index=False, compression="snappy", ) - print(f"Feature dataset written to {OUTPUT_PATH}") - print(f"Shape: {df.shape}") + logger.info(f"Feature engineering completed. Output: {args.output}") + logger.info(f"Dataset shape: {df.shape[0]:,} rows x {df.shape[1]} columns") if __name__ == "__main__": diff --git a/scripts/feature_ablation.py b/scripts/feature_ablation.py index 2c4f87d..d47ed9c 100644 --- a/scripts/feature_ablation.py +++ b/scripts/feature_ablation.py @@ -1,3 +1,5 @@ +import argparse +import csv import sys from pathlib import Path @@ -6,29 +8,35 @@ import pandas as pd +from src.config import get_config from src.evaluation.metrics import ( evaluate_model, precision_recall_at_alert_rate, ) +from src.logging_config import setup_logging from src.models.calibration import ProbabilityCalibrator from src.models.train import ( ABLATION_FEATURE_SETS, TARGET, - chronological_split, fit_model, + temporal_split, ) +logger = setup_logging(__name__) + FEATURE_PATH = PROJECT_ROOT / "data/processed/transactions_features.parquet" ALERT_RATE = 0.005 -def evaluate_feature_set(name, features, train, validation, test): +def evaluate_feature_set(name, features, train, calibration, validation, test, config): + logger.info(f"Evaluating feature set: '{name}' ({len(features)} features)") ( preprocessor, model, X_validation_processed, y_validation, - ) = fit_model(train, validation, features) + ) = fit_model(train, calibration, features, config=config) + logger.debug(f"Model trained for '{name}'") validation_probabilities = model.predict_proba( X_validation_processed @@ -63,31 +71,52 @@ def evaluate_feature_set(name, features, train, validation, test): def main(): - df = pd.read_parquet(FEATURE_PATH) - train, validation, test = chronological_split(df) + parser = argparse.ArgumentParser() + parser.add_argument("--features", type=Path, default=FEATURE_PATH) + parser.add_argument("--output", type=Path, default=PROJECT_ROOT / "reports/ablation_results.csv") + parser.add_argument("--fast", action="store_true") + args = parser.parse_args() + + logger.info(f"Loading feature dataset from {args.features}...") + df = pd.read_parquet(args.features) + logger.info(f"Loaded {len(df):,} transactions") + + logger.info("Performing temporal split...") + train, calibration, validation, test = temporal_split(df) + config = get_config(fast=args.fast) + if args.fast: + logger.info("Using FAST mode configuration") + + logger.info(f"Evaluating {len(ABLATION_FEATURE_SETS)} feature sets...") results = [ evaluate_feature_set( name, features, train, + calibration, validation, test, + config, ) 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) + logger.info("FEATURE ABLATION RESULTS:") for result in results: - print( - f"{result['model']:<22} " - f"{result['pr_auc']:>12.6f} " - f"{result['recall_at_0.5%']:>16.6f}" + logger.info( + f" {result['model']:<22} PR-AUC: {result['pr_auc']:.6f} | " + f"Recall@0.5%: {result['recall_at_0.5%']:.6f}" ) + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=results[0].keys()) + writer.writeheader() + writer.writerows(results) + + logger.info(f"Ablation results written to {args.output.relative_to(PROJECT_ROOT)}") + if __name__ == "__main__": main() diff --git a/scripts/generate_report_figures.py b/scripts/generate_report_figures.py new file mode 100644 index 0000000..e1108e3 --- /dev/null +++ b/scripts/generate_report_figures.py @@ -0,0 +1,187 @@ +"""Generate report figures (ROC, PR, feature importance, ablation, typology) from existing artifacts.""" + +import csv +import sys +from pathlib import Path + +import joblib +import matplotlib.pyplot as plt +import numpy as np +import pandas as pd +from sklearn.metrics import roc_curve, precision_recall_curve, auc + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.logging_config import setup_logging +from src.models.calibration import ProbabilityCalibrator +from src.models.train import MODEL_FEATURES, TARGET, temporal_split + +logger = setup_logging(__name__) + +ARTIFACT_PATH = PROJECT_ROOT / "artifacts/risk_model.joblib" +FEATURE_PATH = PROJECT_ROOT / "data/processed/transactions_features.parquet" +FIGURES_DIR = PROJECT_ROOT / "reports/figures" +ABLATION_PATH = PROJECT_ROOT / "reports/ablation_results.csv" +TYPOLOGY_PATH = PROJECT_ROOT / "reports/typology_results.csv" + + +def main(): + FIGURES_DIR.mkdir(parents=True, exist_ok=True) + + logger.info(f"Loading artifact from {ARTIFACT_PATH}...") + artifact = joblib.load(ARTIFACT_PATH) + preprocessor = artifact["preprocessor"] + model = artifact["model"] + calibrator = artifact["calibrator"] + + logger.info(f"Loading features from {FEATURE_PATH}...") + df = pd.read_parquet(FEATURE_PATH) + + logger.info("Performing temporal split to recover test set...") + _, _, _, test = temporal_split(df) + X_test = test[MODEL_FEATURES] + y_test = test[TARGET] + + logger.info("Running inference on test set...") + 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) + + # ===== ROC Curve ===== + logger.info("Generating ROC curve...") + fpr, tpr, _ = roc_curve(y_test, calibrated_test_prob) + roc_auc = auc(fpr, tpr) + + fig, ax = plt.subplots(figsize=(8, 8)) + ax.plot(fpr, tpr, color="darkorange", lw=2.5, label=f"ROC curve (AUC = {roc_auc:.4f})") + ax.plot([0, 1], [0, 1], color="navy", lw=2, linestyle="--", label="Random Classifier") + ax.set_xlim([0.0, 1.0]) + ax.set_ylim([0.0, 1.05]) + ax.set_xlabel("False Positive Rate", fontsize=12, fontweight="bold") + ax.set_ylabel("True Positive Rate", fontsize=12, fontweight="bold") + ax.set_title("ROC Curve (Test Set)", fontsize=14, fontweight="bold", pad=20) + ax.legend(loc="lower right", fontsize=11) + ax.grid(alpha=0.3) + fig.tight_layout() + fig.savefig(FIGURES_DIR / "roc_curve.png", dpi=150, bbox_inches="tight") + plt.close(fig) + logger.info(f"Saved ROC curve to {FIGURES_DIR / 'roc_curve.png'}") + + # ===== Precision-Recall Curve ===== + logger.info("Generating Precision-Recall curve...") + precision, recall, _ = precision_recall_curve(y_test, calibrated_test_prob) + pr_auc = auc(recall, precision) + + fig, ax = plt.subplots(figsize=(8, 8)) + ax.plot(recall, precision, color="#06b6d4", lw=2.5, label=f"PR curve (AUC = {pr_auc:.4f})") + ax.fill_between(recall, precision, alpha=0.2, color="#06b6d4") + ax.set_xlim([0.0, 1.0]) + ax.set_ylim([0.0, 1.05]) + ax.set_xlabel("Recall (True Positive Rate)", fontsize=12, fontweight="bold") + ax.set_ylabel("Precision", fontsize=12, fontweight="bold") + ax.set_title("Precision-Recall Curve (Test Set)", fontsize=14, fontweight="bold", pad=20) + ax.legend(loc="upper right", fontsize=11) + ax.grid(alpha=0.3) + fig.tight_layout() + fig.savefig(FIGURES_DIR / "precision_recall_curve.png", dpi=150, bbox_inches="tight") + plt.close(fig) + logger.info(f"Saved PR curve to {FIGURES_DIR / 'precision_recall_curve.png'}") + + # ===== Feature Importance (Top 15) ===== + logger.info("Generating feature importance chart...") + feature_names = preprocessor.get_feature_names_out() + importances = model.feature_importances_ + feature_importance_df = pd.DataFrame({ + "feature": feature_names, + "importance": importances + }).sort_values("importance", ascending=False).head(15) + + fig, ax = plt.subplots(figsize=(10, 6)) + bars = ax.barh(range(len(feature_importance_df)), feature_importance_df["importance"], color="#10b981") + ax.set_yticks(range(len(feature_importance_df))) + ax.set_yticklabels(feature_importance_df["feature"], fontsize=10) + ax.set_xlabel("Importance Score", fontsize=12, fontweight="bold") + ax.set_title("Top 15 Most Important Features", fontsize=14, fontweight="bold", pad=20) + ax.invert_yaxis() + ax.grid(axis="x", alpha=0.3) + for i, bar in enumerate(bars): + width = bar.get_width() + ax.text(width, bar.get_y() + bar.get_height() / 2, f" {width:.3f}", + ha="left", va="center", fontsize=9, fontweight="bold") + fig.tight_layout() + fig.savefig(FIGURES_DIR / "feature_importance.png", dpi=150, bbox_inches="tight") + plt.close(fig) + logger.info(f"Saved feature importance to {FIGURES_DIR / 'feature_importance.png'}") + + # ===== Ablation Comparison ===== + logger.info("Generating ablation comparison chart...") + ablation_data = [] + with ABLATION_PATH.open() as f: + reader = csv.DictReader(f) + for row in reader: + ablation_data.append({ + "model": row["model"], + "pr_auc": float(row["pr_auc"]) + }) + + if ablation_data: + fig, ax = plt.subplots(figsize=(10, 6)) + models = [d["model"] for d in ablation_data] + pr_aucs = [d["pr_auc"] for d in ablation_data] + colors = ["#94a3b8", "#a78bfa", "#06b6d4", "#10b981"] + bars = ax.bar(models, pr_aucs, color=colors[:len(models)], edgecolor="black", linewidth=1.5) + ax.set_ylabel("PR-AUC Score", fontsize=12, fontweight="bold") + ax.set_title("Feature Ablation Study - PR-AUC Improvement", fontsize=14, fontweight="bold", pad=20) + ax.set_ylim([0, 1.0]) + ax.grid(axis="y", alpha=0.3) + for bar in bars: + height = bar.get_height() + ax.text(bar.get_x() + bar.get_width() / 2, height, f"{height:.4f}", + ha="center", va="bottom", fontsize=10, fontweight="bold") + fig.tight_layout() + fig.savefig(FIGURES_DIR / "ablation_comparison.png", dpi=150, bbox_inches="tight") + plt.close(fig) + logger.info(f"Saved ablation comparison to {FIGURES_DIR / 'ablation_comparison.png'}") + + # ===== Typology Detection ===== + logger.info("Generating typology detection chart...") + typology_data = [] + with TYPOLOGY_PATH.open() as f: + reader = csv.DictReader(f) + for row in reader: + if int(row["positives"]) > 0: + typology_data.append({ + "typology": row["typology"], + "recall": float(row["recall_at_0.5%"]) + }) + + typology_data = sorted(typology_data, key=lambda x: x["recall"], reverse=True) + + if typology_data: + fig, ax = plt.subplots(figsize=(10, 10)) + typologies = [d["typology"] for d in typology_data] + recalls = [d["recall"] for d in typology_data] + colors = ["#10b981" if r >= 0.9 else "#f59e0b" if r >= 0.5 else "#ef4444" for r in recalls] + bars = ax.barh(range(len(typologies)), recalls, color=colors, edgecolor="black", linewidth=1) + ax.set_yticks(range(len(typologies))) + ax.set_yticklabels(typologies, fontsize=9) + ax.set_xlabel("Detection Rate (Recall @ 0.5% Alert Budget)", fontsize=11, fontweight="bold") + ax.set_title("AML Behavioral Pattern Detection Rates", fontsize=14, fontweight="bold", pad=20) + ax.set_xlim([0, 1.05]) + ax.invert_yaxis() + ax.grid(axis="x", alpha=0.3) + for i, bar in enumerate(bars): + width = bar.get_width() + ax.text(width, bar.get_y() + bar.get_height() / 2, f" {width:.1%}", + ha="left", va="center", fontsize=8, fontweight="bold") + fig.tight_layout() + fig.savefig(FIGURES_DIR / "typology_detection.png", dpi=150, bbox_inches="tight") + plt.close(fig) + logger.info(f"Saved typology detection to {FIGURES_DIR / 'typology_detection.png'}") + + logger.info("✓ All report figures generated successfully") + + +if __name__ == "__main__": + main() diff --git a/scripts/run_experiments.py b/scripts/run_experiments.py new file mode 100644 index 0000000..2d7aff1 --- /dev/null +++ b/scripts/run_experiments.py @@ -0,0 +1,37 @@ +import argparse +import subprocess +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[1] + + +def run(command): + print(f"\n$ {' '.join(command)}") + subprocess.run(command, cwd=PROJECT_ROOT, check=True) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--input", type=Path, default=PROJECT_ROOT / "data/raw/SAML-D.csv") + parser.add_argument("--features", type=Path, default=PROJECT_ROOT / "data/processed/transactions_features.parquet") + parser.add_argument("--artifact", type=Path, default=PROJECT_ROOT / "artifacts/risk_model.joblib") + parser.add_argument("--fast", action="store_true") + args = parser.parse_args() + python = sys.executable + + run([python, "scripts/build_features.py", "--input", str(args.input), "--output", str(args.features)]) + train = [python, "scripts/train_model.py", "--features", str(args.features), "--artifact", str(args.artifact)] + ablation = [python, "scripts/feature_ablation.py", "--features", str(args.features)] + walk_forward = [python, "scripts/walk_forward_backtest.py", "--features", str(args.features)] + if args.fast: + train.append("--fast") + ablation.append("--fast") + walk_forward.append("--fast") + run(train) + run(ablation) + run(walk_forward) + + +if __name__ == "__main__": + main() diff --git a/scripts/train_model.py b/scripts/train_model.py index c25b14a..aad4d1a 100644 --- a/scripts/train_model.py +++ b/scripts/train_model.py @@ -1,3 +1,6 @@ +import argparse +import csv +import json import sys from importlib.metadata import version from pathlib import Path @@ -5,24 +8,30 @@ import joblib import numpy as np import pandas as pd +from sklearn.calibration import calibration_curve PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT)) +from src.config import get_config +from src.logging_config import setup_logging + +logger = setup_logging(__name__) + from src.evaluation.metrics import ( evaluate_model, - threshold_for_alert_rate, + top_k_alert_mask, ) +from src.models.baseline import build_logistic_baseline from src.models.calibration import ( ProbabilityCalibrator, ) from src.models.train import ( MODEL_FEATURES, TARGET, - chronological_split, fit_model, + temporal_split, ) -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" @@ -41,7 +50,7 @@ def print_split_stats(name, frame): positive = frame[TARGET].sum() prevalence = frame[TARGET].mean() - print( + logger.info( f"{name}: " f"{len(frame):,} rows | " f"{positive:,} suspicious | " @@ -50,24 +59,39 @@ def print_split_stats(name, frame): def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--features", type=Path, default=FEATURE_PATH) + parser.add_argument("--artifact", type=Path, default=ARTIFACT_PATH) + parser.add_argument("--fast", action="store_true") + args = parser.parse_args() - ARTIFACT_PATH.parent.mkdir(exist_ok=True) + args.artifact.parent.mkdir(parents=True, exist_ok=True) - print("Loading feature dataset...") + logger.info(f"Loading feature dataset from {args.features}...") df = pd.read_parquet( - FEATURE_PATH + args.features ) + logger.info(f"Loaded {len(df):,} transactions with {df.shape[1]} features") - train, validation, test = ( - chronological_split(df) - ) + logger.info("Performing temporal split (65% train / 10% calibration / 10% validation / 15% test)...") + train, calibration, validation, test = temporal_split(df) + + config = get_config(fast=args.fast) + if args.fast: + logger.info("Using FAST mode configuration") + logger.info(f"XGBoost config: n_estimators={config.xgboost.n_estimators}, max_depth={config.xgboost.max_depth}") print_split_stats( "TRAIN", train, ) + print_split_stats( + "CALIBRATION", + calibration, + ) + print_split_stats( "VALIDATION", validation, @@ -85,14 +109,15 @@ def main(): y_validation, ) = fit_model( train, - validation, + calibration, + config=config, ) # --------------------------- # CALIBRATION # --------------------------- - raw_validation_prob = ( + raw_calibration_prob = ( model.predict_proba( X_validation_processed )[:, 1] @@ -101,30 +126,64 @@ def main(): calibrator = ProbabilityCalibrator() calibrator.fit( - raw_validation_prob, + raw_calibration_prob, y_validation, ) calibrated_validation_prob = ( - calibrator.predict( - raw_validation_prob - ) + calibrator.predict(raw_calibration_prob) ) + calibration_report = { + "raw_brier_score": float( + evaluate_model(y_validation, raw_calibration_prob)["brier_score"] + ), + "calibrated_brier_score": float( + evaluate_model(y_validation, calibrated_validation_prob)["brier_score"] + ), + "raw_log_loss": float( + evaluate_model(y_validation, raw_calibration_prob)["log_loss"] + ), + "calibrated_log_loss": float( + evaluate_model(y_validation, calibrated_validation_prob)["log_loss"] + ), + } + figure_dir = PROJECT_ROOT / "reports/figures" + figure_dir.mkdir(parents=True, exist_ok=True) + import matplotlib.pyplot as plt + + figure, axis = plt.subplots(figsize=(6, 6)) + for probabilities, label in ( + (raw_calibration_prob, "Raw XGBoost"), + (calibrated_validation_prob, "Calibrated XGBoost"), + ): + observed, predicted = calibration_curve( + y_validation, + probabilities, + n_bins=10, + strategy="quantile", + ) + axis.plot(predicted, observed, marker="o", label=label) + axis.plot([0, 1], [0, 1], linestyle="--", color="black", label="Perfect") + axis.set_xlabel("Mean predicted probability") + axis.set_ylabel("Observed frequency") + axis.set_title("Probability calibration") + axis.legend() + figure.tight_layout() + figure.savefig(figure_dir / "calibration_curve.png", dpi=150) + plt.close(figure) + # Use a realistic operational alert capacity alert_rate = 0.005 - threshold = ( - threshold_for_alert_rate( - calibrated_validation_prob, - alert_rate, - ) + calibration_alerts = top_k_alert_mask( + calibrated_validation_prob, + alert_rate, ) + threshold = float(calibrated_validation_prob[calibration_alerts].min()) - print( - f"\nThreshold for " - f"{alert_rate:.2%} alert rate: " - f"{threshold:.6f}" + logger.info( + f"Threshold for {alert_rate:.2%} alert rate: {threshold:.6f}" ) # --------------------------- @@ -157,25 +216,39 @@ def main(): calibrated_test_prob, ) - print("\nOUT-OF-TIME TEST RESULTS") - print("=" * 50) + typology_results = [] + if "Laundering_type" in test: + alert_mask = top_k_alert_mask(calibrated_test_prob, alert_rate) + for typology, group in test.groupby("Laundering_type", dropna=False): + group_mask = group.index.isin(test.index[alert_mask]) + positives = group[TARGET].sum() + typology_results.append( + { + "typology": str(typology), + "transactions": int(len(group)), + "positives": int(positives), + "recall_at_0.5%": float( + group_mask[group[TARGET].to_numpy() == 1].mean() + if positives + else 0 + ), + } + ) + typology_path = PROJECT_ROOT / "reports/typology_results.csv" + with typology_path.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=typology_results[0].keys()) + writer.writeheader() + writer.writerows(typology_results) - for key, value in metrics.items(): + logger.info("OUT-OF-TIME TEST RESULTS:") + for key, value in metrics.items(): if isinstance(value, float): - print( - f"{key:30s}: " - f"{value:.6f}" - ) - + logger.info(f" {key:30s}: {value:.6f}") else: - print( - f"{key:30s}: " - f"{value}" - ) + logger.info(f" {key:30s}: {value}") - # Benchmark the nonlinear model against a scalable logistic classifier on - # the exact same chronological partitions and preprocessing contract. + logger.info("Training logistic baseline for comparison...") baseline = build_logistic_baseline() baseline.fit(preprocessor.transform(train[MODEL_FEATURES]), train[TARGET]) baseline_validation_prob = baseline.predict_proba( @@ -190,10 +263,9 @@ def main(): ) baseline_metrics = evaluate_model(y_test, baseline_test_prob) - print("\nLOGISTIC BASELINE TEST RESULTS") - print("=" * 50) + logger.info("LOGISTIC BASELINE TEST RESULTS:") for key in ("pr_auc", "alert_0.500%_recall", "alert_0.500%_lift"): - print(f"{key:30s}: {baseline_metrics[key]:.6f}") + logger.info(f" {key:30s}: {baseline_metrics[key]:.6f}") # --------------------------- # SAVE @@ -203,6 +275,7 @@ def main(): "preprocessor": preprocessor, "model": model, "calibrator": calibrator, + "calibration_metrics": calibration_report, "features": MODEL_FEATURES, @@ -229,18 +302,26 @@ def main(): "model_parameters": model.get_params(), "package_versions": package_versions(), - "model_version": "2.0.0", + "model_version": "2.1.0", } - joblib.dump( - artifact, - ARTIFACT_PATH, - ) + logger.info(f"Saving model artifact to {args.artifact}...") + joblib.dump(artifact, args.artifact) + logger.info(f"Model saved successfully") - print( - "\nSaved model to " - f"{ARTIFACT_PATH.relative_to(PROJECT_ROOT)}" - ) + logger.info(f"Writing metrics report...") + report_path = PROJECT_ROOT / "reports/model_metrics.json" + report_path.parent.mkdir(parents=True, exist_ok=True) + report = { + "model_version": artifact["model_version"], + "test_metrics": metrics, + "baseline_test_metrics": baseline_metrics, + "training_prevalence": artifact["training_prevalence"], + "test_prevalence": artifact["test_prevalence"], + "calibration": calibration_report, + } + report_path.write_text(json.dumps(report, indent=2) + "\n") + logger.info(f"Metrics report written to {report_path.relative_to(PROJECT_ROOT)}") if __name__ == "__main__": diff --git a/scripts/walk_forward_backtest.py b/scripts/walk_forward_backtest.py new file mode 100644 index 0000000..84e9044 --- /dev/null +++ b/scripts/walk_forward_backtest.py @@ -0,0 +1,118 @@ +import argparse +import csv +import sys +from pathlib import Path + +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.config import get_config +from src.evaluation.metrics import evaluate_model, precision_recall_at_alert_rate +from src.logging_config import setup_logging +from src.models.calibration import ProbabilityCalibrator +from src.models.train import MODEL_FEATURES, TARGET, fit_model + +logger = setup_logging(__name__) + + +def run_backtest(frame, config): + logger.info(f"Starting walk-forward backtest on {len(frame):,} transactions") + ordered = frame.sort_values("timestamp").reset_index(drop=True) + timestamps = ordered["timestamp"].drop_duplicates().sort_values().tolist() + block_count = 14 + if len(timestamps) < block_count: + raise ValueError("walk-forward backtesting requires at least 14 unique timestamps") + + logger.info(f"Found {len(timestamps)} unique timestamps, creating rolling windows...") + results = [] + for window in range(4): + train_end = 4 + window * 2 + calibration_end = train_end + 2 + test_end = calibration_end + 2 + if test_end > len(timestamps): + continue + train = ordered[ordered["timestamp"] < timestamps[train_end]] + calibration = ordered[ + (ordered["timestamp"] >= timestamps[train_end]) + & (ordered["timestamp"] < timestamps[calibration_end]) + ] + test = ordered[ + (ordered["timestamp"] >= timestamps[calibration_end]) + & (ordered["timestamp"] < timestamps[test_end]) + ] + if calibration[TARGET].nunique() < 2: + logger.debug(f"Window {window + 1}: Skipping (insufficient target variance in calibration)") + continue + if train[TARGET].nunique() < 2 or test[TARGET].nunique() < 2: + logger.debug(f"Window {window + 1}: Skipping (insufficient target variance in train/test)") + continue + + logger.info(f"Processing window {window + 1}: train={len(train):,}, calib={len(calibration):,}, test={len(test):,}") + preprocessor, model, calibration_x, calibration_y = fit_model( + train, + calibration, + MODEL_FEATURES, + config=config, + ) + calibration_raw = model.predict_proba(calibration_x)[:, 1] + calibrator = ProbabilityCalibrator().fit(calibration_raw, calibration_y) + test_x = preprocessor.transform(test[MODEL_FEATURES]) + probabilities = calibrator.predict(model.predict_proba(test_x)[:, 1]) + metrics = evaluate_model(test[TARGET], probabilities) + budget = precision_recall_at_alert_rate( + test[TARGET], probabilities, alert_rate=0.005 + ) + results.append( + { + "window": f"W{window + 1}", + "pr_auc": metrics["pr_auc"], + "recall_at_0.5%": budget["recall"], + "precision_at_0.5%": budget["precision"], + "lift_at_0.5%": budget["lift"], + } + ) + return results + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--features", type=Path, default=PROJECT_ROOT / "data/processed/transactions_features.parquet") + parser.add_argument("--output", type=Path, default=PROJECT_ROOT / "reports/walk_forward_results.csv") + parser.add_argument("--fast", action="store_true") + args = parser.parse_args() + + logger.info(f"Loading feature dataset from {args.features}...") + df = pd.read_parquet(args.features) + logger.info(f"Loaded {len(df):,} transactions") + + config = get_config(fast=args.fast) + if args.fast: + logger.info("Using FAST mode configuration") + + results = run_backtest(df, config=config) + + if not results: + raise RuntimeError("no valid walk-forward windows were available") + + logger.info(f"Completed {len(results)} walk-forward windows") + logger.info("WALK-FORWARD BACKTEST RESULTS:") + for result in results: + logger.info( + f" {result['window']}: PR-AUC={result['pr_auc']:.6f} | " + f"Recall@0.5%={result['recall_at_0.5%']:.6f} | " + f"Lift@0.5%={result['lift_at_0.5%']:.6f}" + ) + + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=results[0].keys()) + writer.writeheader() + writer.writerows(results) + + logger.info(f"Results written to {args.output.relative_to(PROJECT_ROOT)}") + + +if __name__ == "__main__": + main() diff --git a/src/api/app.py b/src/api/app.py index f896c2e..ef3e0fd 100644 --- a/src/api/app.py +++ b/src/api/app.py @@ -11,9 +11,12 @@ from typing import Any import joblib -from flask import Flask, jsonify, request +from flask import Flask, jsonify, request, render_template from flask_cors import CORS +from marshmallow import ValidationError + +from src.api.schemas import validate_feature_vector from src.models.inference import ( model_input_from_features, predict_calibrated_probability, @@ -22,10 +25,11 @@ PROJECT_ROOT = Path(__file__).resolve().parents[2] MODEL_PATH = PROJECT_ROOT / "artifacts/risk_model.joblib" +TEMPLATE_DIR = Path(__file__).resolve().parent / "templates" def create_app(model_path: Path = MODEL_PATH) -> Flask: - app = Flask(__name__) + app = Flask(__name__, template_folder=str(TEMPLATE_DIR)) CORS(app) artifact: dict[str, Any] | None = None @@ -37,6 +41,10 @@ def create_app(model_path: Path = MODEL_PATH) -> Flask: except Exception as error: # pragma: no cover - defensive startup path load_error = f"Unable to load model artifact: {error}" + @app.get("/") + def dashboard(): + return render_template("dashboard.html") + @app.get("/api/health") def health(): return jsonify( @@ -66,11 +74,22 @@ def model_info(): def predict(): if artifact is None: return jsonify({"error": load_error}), 503 - payload = request.get_json(silent=True) + + try: + payload = request.get_json(force=True) + except Exception as error: + return jsonify({"error": f"Invalid JSON: {str(error)}"}), 400 + if not isinstance(payload, dict): return jsonify({"error": "Request body must be a JSON object."}), 400 + + try: + validated_features = validate_feature_vector(payload) + except ValidationError as error: + return jsonify({"error": error.messages}), 400 + try: - features = model_input_from_features(payload, artifact["features"]) + features = model_input_from_features(validated_features, artifact["features"]) probability = predict_calibrated_probability(artifact, features) except ValueError as error: return jsonify({"error": str(error)}), 400 diff --git a/src/api/inference.py b/src/api/inference.py new file mode 100644 index 0000000..bdc04f8 --- /dev/null +++ b/src/api/inference.py @@ -0,0 +1,85 @@ +"""Async-aware inference layer for FastAPI.""" + +import asyncio +from typing import Any, Mapping, Optional + +import numpy as np +import pandas as pd + + +async def model_input_from_features( + transaction_features: Mapping, + feature_names: list[str], +) -> pd.DataFrame: + """ + Validate and extract model features from transaction features. + + Rejects forbidden keys (account identifiers, target). + Raises ValueError if required features are missing. + """ + forbidden_keys = {"Sender_account", "Receiver_account", "Is_laundering", "Laundering_type"} + supplied_keys = set(transaction_features.keys()) + overlap = supplied_keys & forbidden_keys + + if overlap: + raise ValueError( + f"Identifier or target fields are not model inputs: {overlap}" + ) + + feature_set = set(feature_names) + missing = feature_set - supplied_keys + + if missing: + raise ValueError( + f"Missing model features. Required: {missing}" + ) + + record = {feature: transaction_features[feature] for feature in feature_names} + return pd.DataFrame([record]) + + +async def predict_calibrated_probability( + artifact: dict[str, Any], + features: pd.DataFrame, +) -> float: + """ + Run inference and return calibrated probability. + + CPU-bound operations (preprocessing, model.predict_proba, calibration) + are dispatched to the event loop's default thread pool. + """ + + def _preprocess(): + return artifact["preprocessor"].transform(features) + + def _predict(X_processed): + return artifact["model"].predict_proba(X_processed)[:, 1] + + def _calibrate(raw_probs): + return artifact["calibrator"].predict(raw_probs)[0] + + loop = asyncio.get_event_loop() + + X_processed = await loop.run_in_executor(None, _preprocess) + raw_prob = await loop.run_in_executor(None, _predict, X_processed) + calibrated = await loop.run_in_executor(None, _calibrate, raw_prob) + + return float(calibrated) + + +def probability_percentile( + artifact: dict[str, Any], + probability: float, +) -> Optional[float]: + """ + Calculate percentile rank of a probability against validation quantiles. + + Lightweight operation (no async needed); pure Python array indexing. + """ + quantiles = artifact.get("validation_probability_quantiles") + + if not quantiles: + return None + + percentile = np.searchsorted(quantiles, probability) * 100 / len(quantiles) + return float(percentile) diff --git a/src/api/main.py b/src/api/main.py new file mode 100644 index 0000000..0b8b851 --- /dev/null +++ b/src/api/main.py @@ -0,0 +1,138 @@ +"""FastAPI application for transaction risk prediction inference.""" + +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +import joblib +from fastapi import FastAPI, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from fastapi.staticfiles import StaticFiles + +from src.api.inference import ( + model_input_from_features, + predict_calibrated_probability, + probability_percentile, +) +from src.api.models import ( + FeatureVector, + HealthResponse, + ModelInfoResponse, + PredictionResponse, +) +from src.api.results_loader import ( + get_ablation_results, + get_typology_results, + get_all_figures, +) + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +MODEL_PATH = PROJECT_ROOT / "artifacts/risk_model.joblib" +PUBLIC_DIR = PROJECT_ROOT / "public" + + +def create_app(model_path: Path = MODEL_PATH) -> FastAPI: + """Create and configure FastAPI application with optional model artifact.""" + + app = FastAPI( + title="Transaction Risk Prediction API", + description="XGBoost-based inference for transaction risk scoring", + version="2.1.0", + ) + + # Enable CORS + app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_methods=["*"], + allow_headers=["*"], + ) + + # Load model artifact at startup + artifact: Optional[dict[str, Any]] = None + load_error: Optional[str] = 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: + load_error = f"Unable to load model artifact: {error}" + + # Routes + + @app.get("/api/health", response_model=HealthResponse) + async def health() -> HealthResponse: + """Service health check.""" + return HealthResponse( + 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", response_model=ModelInfoResponse) + async def model_info() -> ModelInfoResponse: + """Model metadata and evaluation metrics.""" + if artifact is None: + raise HTTPException(status_code=503, detail=load_error) + + return ModelInfoResponse( + model="XGBoost", + model_version=artifact["model_version"], + alert_rate=artifact["alert_rate"], + decision_threshold=artifact["decision_threshold"], + test_metrics=artifact["test_metrics"], + ) + + @app.get("/api/results") + async def get_results() -> dict[str, Any]: + """Analysis results: ablation study, behavioral typology, and all report figures.""" + return { + "ablation": get_ablation_results(), + "typology": get_typology_results(), + "figures": get_all_figures(), + } + + @app.post("/api/predict", response_model=PredictionResponse) + async def predict(features: FeatureVector) -> PredictionResponse: + """Predict transaction risk probability.""" + if artifact is None: + raise HTTPException(status_code=503, detail=load_error) + + try: + # Validate features and construct input DataFrame + feature_dict = features.model_dump(mode="json") + model_features = artifact["features"] + X = await model_input_from_features(feature_dict, model_features) + + # Run inference asynchronously + probability = await predict_calibrated_probability(artifact, X) + + # Calculate percentile rank + percentile = probability_percentile(artifact, probability) + + return PredictionResponse( + risk_probability=probability, + requires_review=probability >= artifact["decision_threshold"], + risk_percentile=percentile, + ) + + except ValueError as error: + raise HTTPException(status_code=400, detail=str(error)) + + # Serve static SPA from public/ directory + if PUBLIC_DIR.exists(): + app.mount("", StaticFiles(directory=str(PUBLIC_DIR), html=True), name="static") + + return app + + +# Create module-level app instance for `uvicorn src.api.main:app` +app = create_app() + + +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=5000) diff --git a/src/api/models.py b/src/api/models.py new file mode 100644 index 0000000..266bba8 --- /dev/null +++ b/src/api/models.py @@ -0,0 +1,77 @@ +"""Pydantic models for API request/response validation.""" + +from typing import Optional, Dict, Any +from pydantic import BaseModel, Field + + +class FeatureVector(BaseModel): + """Transaction feature vector for risk prediction.""" + + Amount: float = Field(..., ge=0, description="Transaction amount in base currency") + log_amount: float = Field(..., description="Log(1 + Amount)") + + hour_sin: float = Field(..., ge=-1, le=1, description="Cyclical hour encoding") + hour_cos: float = Field(..., ge=-1, le=1) + dow_sin: float = Field(..., ge=-1, le=1, description="Day of week cyclical encoding") + dow_cos: float = Field(..., ge=-1, le=1) + month_sin: float = Field(..., ge=-1, le=1, description="Month cyclical encoding") + month_cos: float = Field(..., ge=-1, le=1) + + is_weekend: int = Field(..., ge=0, le=1) + is_night: int = Field(..., ge=0, le=1) + currency_mismatch: int = Field(..., ge=0, le=1) + cross_border: int = Field(..., ge=0, le=1) + is_round_amount: int = Field(..., ge=0, le=1) + + sender_txn_count_24h: int = Field(..., ge=0) + sender_amount_sum_24h: float = Field(..., ge=0) + sender_amount_mean_30d: Optional[float] = Field(None) + sender_amount_std_30d: Optional[float] = Field(None) + sender_amount_zscore: float + + receiver_txn_count_24h: int = Field(..., ge=0) + receiver_amount_sum_24h: float = Field(..., ge=0) + + seconds_since_sender_txn: int + + sender_txn_count_lifetime: int = Field(..., ge=0) + receiver_txn_count_lifetime: int = Field(..., ge=0) + sender_out_degree: int = Field(..., ge=0) + receiver_in_degree: int = Field(..., ge=0) + pair_transaction_count: int = Field(..., ge=0) + sender_counterparty_hhi: float = Field(..., ge=0, le=1) + + Payment_type: str + Payment_currency: str + Received_currency: str + Sender_bank_location: str + Receiver_bank_location: str + + model_config = {"extra": "forbid"} + + +class HealthResponse(BaseModel): + """Service health status.""" + + status: str = Field(..., description="healthy or model_unavailable") + model_loaded: bool + timestamp: str + detail: Optional[str] = None + + +class ModelInfoResponse(BaseModel): + """Model metadata and evaluation metrics.""" + + model: str + model_version: str + alert_rate: float + decision_threshold: float + test_metrics: Dict[str, Any] + + +class PredictionResponse(BaseModel): + """Risk prediction result.""" + + risk_probability: float = Field(..., ge=0, le=1) + requires_review: bool + risk_percentile: Optional[float] = Field(None, ge=0, le=100) diff --git a/src/api/openapi.yaml b/src/api/openapi.yaml new file mode 100644 index 0000000..4d5395a --- /dev/null +++ b/src/api/openapi.yaml @@ -0,0 +1,431 @@ +openapi: 3.0.0 +info: + title: Transaction Risk Prediction API + description: | + Inference API for predicting transaction risk using XGBoost model. + + **Note:** This API accepts pre-computed feature vectors only. Raw account identifiers + and behavioral history computation are outside scope—that responsibility belongs to an + online feature store. + version: "2.1.0" + contact: + name: Project Repository + url: https://github.com/Samarthuday/Quantitative-Transaction-Risk-Modeling + +servers: + - url: http://localhost:5000 + description: Local development + - url: https://api.example.com + description: Production + +paths: + /api/health: + get: + summary: Service health check + description: | + Returns the current health status of the service and model availability. + operationId: getHealth + tags: + - Health + responses: + '200': + description: Service is healthy + content: + application/json: + schema: + $ref: '#/components/schemas/HealthResponse' + '503': + description: Service unavailable or model failed to load + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + + /api/model/info: + get: + summary: Model metadata and evaluation metrics + description: | + Returns information about the trained model including version, + decision threshold, and test set performance metrics. + operationId: getModelInfo + tags: + - Model + responses: + '200': + description: Model information retrieved successfully + content: + application/json: + schema: + $ref: '#/components/schemas/ModelInfoResponse' + '503': + description: Model not available + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + + /api/predict: + post: + summary: Predict transaction risk probability + description: | + Predicts the risk probability for a transaction given its pre-computed feature vector. + + **Important:** Input must be a complete feature vector with all required fields. + Missing or invalid fields will result in a 400 error. + operationId: predict + tags: + - Prediction + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/FeatureVector' + examples: + transaction: + summary: Example feature vector + value: + Amount: 1000.0 + log_amount: 6.908 + hour_sin: 0.5 + hour_cos: 0.866 + dow_sin: 0.0 + dow_cos: 1.0 + month_sin: 0.259 + month_cos: 0.966 + is_weekend: 0 + is_night: 0 + currency_mismatch: 0 + cross_border: 0 + is_round_amount: 1 + sender_txn_count_24h: 5 + sender_amount_sum_24h: 5000.0 + sender_amount_mean_30d: 1000.0 + sender_amount_std_30d: 500.0 + sender_amount_zscore: 0.0 + receiver_txn_count_24h: 3 + receiver_amount_sum_24h: 3000.0 + seconds_since_sender_txn: 3600 + sender_txn_count_lifetime: 100 + receiver_txn_count_lifetime: 50 + sender_out_degree: 25 + receiver_in_degree: 20 + pair_transaction_count: 5 + sender_counterparty_hhi: 0.2 + Payment_type: "Transfer" + Payment_currency: "USD" + Received_currency: "USD" + Sender_bank_location: "US" + Receiver_bank_location: "US" + responses: + '200': + description: Risk prediction successful + content: + application/json: + schema: + $ref: '#/components/schemas/PredictionResponse' + '400': + description: Invalid request (missing/invalid features) + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + '503': + description: Model not available + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + +components: + schemas: + HealthResponse: + type: object + required: + - status + - model_loaded + - timestamp + properties: + status: + type: string + enum: + - healthy + - model_unavailable + description: Overall service health status + model_loaded: + type: boolean + description: Whether the model artifact is available + timestamp: + type: string + format: date-time + description: Current UTC timestamp + detail: + type: string + nullable: true + description: Error details if model failed to load + + ModelInfoResponse: + type: object + required: + - model + - model_version + - alert_rate + - decision_threshold + - test_metrics + properties: + model: + type: string + example: "XGBoost" + description: Model type identifier + model_version: + type: string + example: "2.1.0" + description: Semantic version of the model + alert_rate: + type: number + format: float + example: 0.005 + description: Alert rate used for threshold selection (0.5%) + decision_threshold: + type: number + format: float + example: 0.156 + description: Calibrated probability threshold for alerts + test_metrics: + type: object + description: Model performance metrics on held-out test set + properties: + pr_auc: + type: number + format: float + description: Precision-Recall AUC + roc_auc: + type: number + format: float + description: ROC AUC + brier_score: + type: number + format: float + description: Brier score (lower is better) + log_loss: + type: number + format: float + description: Logarithmic loss + alert_0.1%_precision: + type: number + format: float + alert_0.1%_recall: + type: number + format: float + alert_0.5%_precision: + type: number + format: float + alert_0.5%_recall: + type: number + format: float + alert_1.0%_precision: + type: number + format: float + alert_1.0%_recall: + type: number + format: float + + FeatureVector: + type: object + required: + - 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 + - seconds_since_sender_txn + - sender_txn_count_lifetime + - receiver_txn_count_lifetime + - sender_out_degree + - receiver_in_degree + - pair_transaction_count + - sender_counterparty_hhi + - Payment_type + - Payment_currency + - Received_currency + - Sender_bank_location + - Receiver_bank_location + properties: + Amount: + type: number + format: float + minimum: 0 + description: Transaction amount in base currency + log_amount: + type: number + format: float + description: Log(1 + Amount) + hour_sin: + type: number + format: float + minimum: -1 + maximum: 1 + description: Cyclical encoding of hour + hour_cos: + type: number + format: float + minimum: -1 + maximum: 1 + dow_sin: + type: number + format: float + minimum: -1 + maximum: 1 + description: Day of week cyclical encoding + dow_cos: + type: number + format: float + minimum: -1 + maximum: 1 + month_sin: + type: number + format: float + minimum: -1 + maximum: 1 + description: Month cyclical encoding + month_cos: + type: number + format: float + minimum: -1 + maximum: 1 + is_weekend: + type: integer + enum: [0, 1] + is_night: + type: integer + enum: [0, 1] + currency_mismatch: + type: integer + enum: [0, 1] + description: 1 if payment_currency != received_currency + cross_border: + type: integer + enum: [0, 1] + description: 1 if sender_location != receiver_location + is_round_amount: + type: integer + enum: [0, 1] + description: 1 if amount is divisible by 1000 + sender_txn_count_24h: + type: integer + minimum: 0 + sender_amount_sum_24h: + type: number + format: float + minimum: 0 + sender_amount_mean_30d: + type: number + format: float + nullable: true + sender_amount_std_30d: + type: number + format: float + nullable: true + sender_amount_zscore: + type: number + format: float + receiver_txn_count_24h: + type: integer + minimum: 0 + receiver_amount_sum_24h: + type: number + format: float + minimum: 0 + seconds_since_sender_txn: + type: integer + description: Time since sender's last transaction (-1 if none) + sender_txn_count_lifetime: + type: integer + minimum: 0 + receiver_txn_count_lifetime: + type: integer + minimum: 0 + sender_out_degree: + type: integer + minimum: 0 + description: Unique counterparties sender has transacted with + receiver_in_degree: + type: integer + minimum: 0 + description: Unique senders who have transacted with receiver + pair_transaction_count: + type: integer + minimum: 0 + description: Number of prior transactions between sender-receiver pair + sender_counterparty_hhi: + type: number + format: float + minimum: 0 + maximum: 1 + description: HHI concentration measure for sender's counterparties + Payment_type: + type: string + example: "Transfer" + Payment_currency: + type: string + example: "USD" + Received_currency: + type: string + example: "USD" + Sender_bank_location: + type: string + example: "US" + Receiver_bank_location: + type: string + example: "US" + + PredictionResponse: + type: object + required: + - risk_probability + - requires_review + properties: + risk_probability: + type: number + format: float + minimum: 0 + maximum: 1 + description: Calibrated risk probability (0 = low risk, 1 = high risk) + requires_review: + type: boolean + description: Whether transaction exceeds alert threshold + risk_percentile: + type: number + format: float + minimum: 0 + maximum: 100 + nullable: true + description: Percentile rank against validation set (0-100) + + ErrorResponse: + type: object + required: + - error + properties: + error: + type: string + description: Error message describing what went wrong + detail: + type: object + nullable: true + description: Additional error details (e.g., validation errors per field) diff --git a/src/api/results_loader.py b/src/api/results_loader.py new file mode 100644 index 0000000..36f77db --- /dev/null +++ b/src/api/results_loader.py @@ -0,0 +1,105 @@ +"""Load and cache analysis results (ablation, typology, figures).""" + +import base64 +import csv +from pathlib import Path +from typing import Any, Optional + +PROJECT_ROOT = Path(__file__).resolve().parents[2] + + +def load_ablation_results() -> list[dict[str, Any]]: + """Load feature ablation study results from CSV.""" + ablation_path = PROJECT_ROOT / "reports/ablation_results.csv" + if not ablation_path.exists(): + return [] + + results = [] + with ablation_path.open() as f: + reader = csv.DictReader(f) + for row in reader: + results.append({ + "model": row["model"], + "pr_auc": float(row["pr_auc"]), + "recall_at_0.5%": float(row["recall_at_0.5%"]), + }) + return results + + +def load_typology_results() -> list[dict[str, Any]]: + """Load behavioral typology analysis results from CSV.""" + typology_path = PROJECT_ROOT / "reports/typology_results.csv" + if not typology_path.exists(): + return [] + + results = [] + with typology_path.open() as f: + reader = csv.DictReader(f) + for row in reader: + results.append({ + "typology": row["typology"], + "transactions": int(row["transactions"]), + "positives": int(row["positives"]), + "recall_at_0.5%": float(row["recall_at_0.5%"]), + }) + return results + + +def encode_all_figures() -> dict[str, str]: + """Load all report figures from reports/figures/*.png and encode to base64.""" + figures_dir = PROJECT_ROOT / "reports/figures" + if not figures_dir.exists(): + return {} + + figures = {} + figure_mapping = { + "calibration_curve": "Calibration Curve", + "roc_curve": "ROC Curve", + "precision_recall_curve": "Precision-Recall Curve", + "feature_importance": "Feature Importance", + "ablation_comparison": "Ablation Comparison", + "typology_detection": "Typology Detection", + } + + for png_path in sorted(figures_dir.glob("*.png")): + stem = png_path.stem + with png_path.open("rb") as f: + image_bytes = f.read() + base64_str = base64.b64encode(image_bytes).decode("utf-8") + label = figure_mapping.get(stem, stem.replace("_", " ").title()) + figures[stem] = { + "base64": base64_str, + "label": label, + } + + return figures + + +# Cache results at module level +_ablation_cache: Optional[list] = None +_typology_cache: Optional[list] = None +_figures_cache: Optional[dict] = None + + +def get_ablation_results() -> list[dict[str, Any]]: + """Get cached ablation results.""" + global _ablation_cache + if _ablation_cache is None: + _ablation_cache = load_ablation_results() + return _ablation_cache + + +def get_typology_results() -> list[dict[str, Any]]: + """Get cached typology results.""" + global _typology_cache + if _typology_cache is None: + _typology_cache = load_typology_results() + return _typology_cache + + +def get_all_figures() -> dict[str, dict[str, str]]: + """Get cached all report figures as base64 with labels.""" + global _figures_cache + if _figures_cache is None: + _figures_cache = encode_all_figures() + return _figures_cache diff --git a/src/api/schemas.py b/src/api/schemas.py new file mode 100644 index 0000000..a14c9ba --- /dev/null +++ b/src/api/schemas.py @@ -0,0 +1,70 @@ +"""Request validation schemas for the inference API.""" + +from marshmallow import Schema, fields, ValidationError, validate + + +class FeatureVectorSchema(Schema): + """Schema for validating feature vectors in prediction requests.""" + + Amount = fields.Float(required=True, validate=validate.Range(min=0)) + log_amount = fields.Float(required=True) + + hour_sin = fields.Float(required=True, validate=validate.Range(min=-1, max=1)) + hour_cos = fields.Float(required=True, validate=validate.Range(min=-1, max=1)) + + dow_sin = fields.Float(required=True, validate=validate.Range(min=-1, max=1)) + dow_cos = fields.Float(required=True, validate=validate.Range(min=-1, max=1)) + + month_sin = fields.Float(required=True, validate=validate.Range(min=-1, max=1)) + month_cos = fields.Float(required=True, validate=validate.Range(min=-1, max=1)) + + is_weekend = fields.Int(required=True, validate=validate.OneOf([0, 1])) + is_night = fields.Int(required=True, validate=validate.OneOf([0, 1])) + + currency_mismatch = fields.Int(required=True, validate=validate.OneOf([0, 1])) + cross_border = fields.Int(required=True, validate=validate.OneOf([0, 1])) + is_round_amount = fields.Int(required=True, validate=validate.OneOf([0, 1])) + + sender_txn_count_24h = fields.Int(required=True, validate=validate.Range(min=0)) + sender_amount_sum_24h = fields.Float(required=True, validate=validate.Range(min=0)) + sender_amount_mean_30d = fields.Float(required=True, allow_none=True) + sender_amount_std_30d = fields.Float(required=True, allow_none=True) + sender_amount_zscore = fields.Float(required=True) + + receiver_txn_count_24h = fields.Int(required=True, validate=validate.Range(min=0)) + receiver_amount_sum_24h = fields.Float(required=True, validate=validate.Range(min=0)) + + seconds_since_sender_txn = fields.Int(required=True) + + sender_txn_count_lifetime = fields.Int(required=True, validate=validate.Range(min=0)) + receiver_txn_count_lifetime = fields.Int(required=True, validate=validate.Range(min=0)) + sender_out_degree = fields.Int(required=True, validate=validate.Range(min=0)) + receiver_in_degree = fields.Int(required=True, validate=validate.Range(min=0)) + pair_transaction_count = fields.Int(required=True, validate=validate.Range(min=0)) + sender_counterparty_hhi = fields.Float(required=True, validate=validate.Range(min=0, max=1)) + + Payment_type = fields.Str(required=True) + Payment_currency = fields.Str(required=True) + Received_currency = fields.Str(required=True) + Sender_bank_location = fields.Str(required=True) + Receiver_bank_location = fields.Str(required=True) + + class Meta: + unknown = "raise" + + +def validate_feature_vector(payload): + """ + Validate a feature vector against the schema. + + Args: + payload: Dictionary with feature values + + Returns: + Tuple of (cleaned_data, errors) + + Raises: + ValidationError: If validation fails + """ + schema = FeatureVectorSchema() + return schema.load(payload) diff --git a/src/api/templates/dashboard.html b/src/api/templates/dashboard.html new file mode 100644 index 0000000..acf7185 --- /dev/null +++ b/src/api/templates/dashboard.html @@ -0,0 +1,425 @@ + + + + + + Transaction Risk Model - Dashboard + + + +
+
+

🚀 Transaction Risk Model

+

Real-time monitoring dashboard for XGBoost risk prediction model

+
+ +
+ +
+
+

Model Status

+
—
+
Current State
+
+
+ +
+

Test Metrics

+
—
+
PR-AUC Score
+
+ +
+

ROC-AUC

+
—
+
ROC AUC Score
+
+ +
+

Alert Threshold

+
—
+
Decision Threshold
+
+ +
+

Alert Rate

+
—
+
Top-K Percentage
+
+ +
+

Model Version

+
—
+
Training Version
+
+
+ +
+

📊 Performance Metrics

+
+
+ +
+

🎯 Feature Sets

+
+
+ Transaction + Amount, time, cyclical +
+
+ Behavioral + Activity, patterns +
+
+ Network + Graph, concentration +
+
+ Categorical + Type, currency, location +
+
+
+ +
+

🔒 Quantitative Transaction Risk Modeling | Real-time Dashboard

+

Last updated: —

+
+
+ + + + diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..ecdec7a --- /dev/null +++ b/src/config.py @@ -0,0 +1,92 @@ +"""Configuration management for model training and experiments.""" + +from dataclasses import dataclass, asdict + + +@dataclass +class XGBoostConfig: + """XGBoost hyperparameter configuration.""" + + n_estimators: int = 500 + max_depth: int = 5 + learning_rate: float = 0.05 + subsample: float = 0.8 + colsample_bytree: float = 0.8 + min_child_weight: int = 5 + reg_alpha: float = 0.1 + reg_lambda: float = 2.0 + tree_method: str = "hist" + eval_metric: str = "aucpr" + random_state: int = 42 + + def to_dict(self): + """Convert to dictionary for XGBoost initialization.""" + return asdict(self) + + +@dataclass +class PreprocessingConfig: + """Data preprocessing configuration.""" + + numeric_impute_strategy: str = "median" + categorical_impute_strategy: str = "most_frequent" + categorical_min_frequency: int = 20 + categorical_handle_unknown: str = "ignore" + + +@dataclass +class TemporalSplitConfig: + """Temporal split configuration.""" + + train_fraction: float = 0.65 + calibration_fraction: float = 0.10 + validation_fraction: float = 0.10 + + +@dataclass +class AlertConfig: + """Alert threshold and budget configuration.""" + + alert_rate: float = 0.005 # 0.5% + alert_rates_for_metrics: list = None + + def __post_init__(self): + if self.alert_rates_for_metrics is None: + self.alert_rates_for_metrics = [0.001, 0.005, 0.01] + + +@dataclass +class TrainingConfig: + """Complete training configuration.""" + + xgboost: XGBoostConfig = None + preprocessing: PreprocessingConfig = None + temporal_split: TemporalSplitConfig = None + alert: AlertConfig = None + + def __post_init__(self): + if self.xgboost is None: + self.xgboost = XGBoostConfig() + if self.preprocessing is None: + self.preprocessing = PreprocessingConfig() + if self.temporal_split is None: + self.temporal_split = TemporalSplitConfig() + if self.alert is None: + self.alert = AlertConfig() + + +def get_fast_config() -> TrainingConfig: + """Return configuration for fast/smoke testing.""" + config = TrainingConfig() + config.xgboost.n_estimators = 10 # Minimal trees for fast testing + return config + + +def get_production_config() -> TrainingConfig: + """Return configuration for production training.""" + return TrainingConfig() + + +def get_config(fast: bool = False) -> TrainingConfig: + """Get configuration based on mode.""" + return get_fast_config() if fast else get_production_config() diff --git a/src/data/validation.py b/src/data/validation.py new file mode 100644 index 0000000..132ceca --- /dev/null +++ b/src/data/validation.py @@ -0,0 +1,156 @@ +"""Data quality validation and checks.""" + +import logging +import pandas as pd +import numpy as np + +from src.models.train import MODEL_FEATURES, TARGET + +logger = logging.getLogger(__name__) + + +class DataQualityValidator: + """Validate feature dataset quality and schema.""" + + def __init__(self, nan_threshold: float = 0.05): + """ + Initialize validator. + + Args: + nan_threshold: Acceptable NaN percentage (0.05 = 5%) + """ + self.nan_threshold = nan_threshold + self.issues = [] + + def validate(self, df: pd.DataFrame) -> bool: + """ + Perform all validation checks. + + Args: + df: Feature dataset to validate + + Returns: + True if all checks pass, False otherwise + """ + self.issues = [] + + self._check_schema(df) + self._check_nans(df) + self._check_chronological_order(df) + self._check_target_distribution(df) + self._check_numeric_ranges(df) + + if self.issues: + logger.warning(f"Data quality issues found: {len(self.issues)}") + for issue in self.issues: + logger.warning(f" - {issue}") + return False + + logger.info("Data quality validation passed") + return True + + def _check_schema(self, df: pd.DataFrame): + """Check that all required features are present.""" + expected_features = set(MODEL_FEATURES) + actual_features = set(df.columns) + + missing = expected_features - actual_features + if missing: + self.issues.append(f"Missing features: {missing}") + + if TARGET not in df.columns: + self.issues.append(f"Target column '{TARGET}' not found") + + if "timestamp" not in df.columns: + self.issues.append("timestamp column not found") + + def _check_nans(self, df: pd.DataFrame): + """Check NaN values in each column.""" + for col in df.columns: + nan_count = df[col].isna().sum() + nan_pct = nan_count / len(df) + + if nan_pct > self.nan_threshold: + self.issues.append( + f"{col}: {nan_pct:.2%} NaN values (threshold: {self.nan_threshold:.2%})" + ) + + def _check_chronological_order(self, df: pd.DataFrame): + """Verify data is chronologically ordered.""" + if "timestamp" not in df.columns: + return + + if not df["timestamp"].is_monotonic_increasing: + self.issues.append("Data is not chronologically sorted") + + def _check_target_distribution(self, df: pd.DataFrame): + """Check target class distribution.""" + if TARGET not in df.columns: + return + + if df[TARGET].nunique() < 2: + self.issues.append(f"Target '{TARGET}' has fewer than 2 classes") + + prevalence = df[TARGET].mean() + if prevalence < 0.0001 or prevalence > 0.9999: + logger.warning( + f"Target prevalence is extreme: {prevalence:.6f} " + "(may indicate data leakage or sampling bias)" + ) + + def _check_numeric_ranges(self, df: pd.DataFrame): + """Check numeric features for unreasonable values.""" + numeric_cols = df.select_dtypes(include=[np.number]).columns + + for col in numeric_cols: + # Skip target and special columns + if col in [TARGET] or col.startswith("sender_") or col.startswith("receiver_"): + continue + + # Check for infinity values + inf_count = np.isinf(df[col]).sum() + if inf_count > 0: + self.issues.append(f"{col}: {inf_count} infinite values") + + # Check for extreme outliers + q1 = df[col].quantile(0.25) + q3 = df[col].quantile(0.75) + iqr = q3 - q1 + + if iqr > 0: + lower_bound = q1 - 10 * iqr + upper_bound = q3 + 10 * iqr + + outliers = ((df[col] < lower_bound) | (df[col] > upper_bound)).sum() + if outliers > len(df) * 0.05: # More than 5% outliers + logger.warning( + f"{col}: {outliers} extreme outliers ({outliers / len(df):.2%})" + ) + + +def validate_features( + df: pd.DataFrame, + nan_threshold: float = 0.05, +) -> bool: + """ + Validate a feature dataset. + + Args: + df: Feature dataset + nan_threshold: Acceptable NaN percentage + + Returns: + True if valid, False otherwise + + Raises: + ValueError: If critical validation checks fail + """ + validator = DataQualityValidator(nan_threshold=nan_threshold) + + if not validator.validate(df): + if validator.issues: + first_issue = validator.issues[0] + if "Missing" in first_issue or "not found" in first_issue: + raise ValueError(f"Critical validation failure: {first_issue}") + + return True diff --git a/src/evaluation/explainability.py b/src/evaluation/explainability.py index cb70136..8ce6404 100644 --- a/src/evaluation/explainability.py +++ b/src/evaluation/explainability.py @@ -1,18 +1,103 @@ +"""Model explainability using SHAP values.""" + +from typing import Union + +import numpy as np +import pandas as pd import shap +import xgboost as xgb + +def create_shap_explainer(model: xgb.XGBClassifier) -> shap.TreeExplainer: + """ + Create a SHAP TreeExplainer for an XGBoost model. -def create_shap_explainer(model): + Args: + model: Fitted XGBoost classifier + + Returns: + SHAP TreeExplainer instance + """ 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 + model: xgb.XGBClassifier, + X: Union[np.ndarray, pd.DataFrame], +) -> np.ndarray: + """ + Calculate SHAP values for model predictions. + + Args: + model: Fitted XGBoost classifier + X: Input features + + Returns: + SHAP values array + """ + explainer = create_shap_explainer(model) + return explainer.shap_values(X) + + +def feature_importance_summary( + model: xgb.XGBClassifier, + X: Union[np.ndarray, pd.DataFrame], + feature_names: list = None, +) -> pd.DataFrame: + """ + Calculate mean absolute SHAP values (global feature importance). + + Args: + model: Fitted XGBoost classifier + X: Input features for explanation + feature_names: List of feature names + + Returns: + DataFrame with features and importance scores, sorted descending + """ + shap_values = calculate_shap_values(model, X) + + mean_abs_shap = np.abs(shap_values).mean(axis=0) + + if feature_names is None: + feature_names = [f"Feature_{i}" for i in range(len(mean_abs_shap))] + + return pd.DataFrame({ + "feature": feature_names, + "importance": mean_abs_shap, + }).sort_values("importance", ascending=False) + + +def explain_prediction_instance( + model: xgb.XGBClassifier, + X: Union[np.ndarray, pd.DataFrame], + instance_idx: int, + feature_names: list = None, +) -> pd.DataFrame: + """ + Explain a single prediction using SHAP values. + + Args: + model: Fitted XGBoost classifier + X: Input features + instance_idx: Index of instance to explain + feature_names: List of feature names + + Returns: + DataFrame with feature contributions to prediction + """ + explainer = create_shap_explainer(model) + shap_values = explainer.shap_values(X) + + instance_shap = shap_values[instance_idx] + + if feature_names is None: + feature_names = [f"Feature_{i}" for i in range(len(instance_shap))] + + contributions = pd.DataFrame({ + "feature": feature_names, + "shap_value": instance_shap, + "abs_shap_value": np.abs(instance_shap), + }).sort_values("abs_shap_value", ascending=False) + + return contributions \ No newline at end of file diff --git a/src/evaluation/metrics.py b/src/evaluation/metrics.py index 9e9dca3..9b5c786 100644 --- a/src/evaluation/metrics.py +++ b/src/evaluation/metrics.py @@ -1,3 +1,7 @@ +"""Evaluation metrics for risk models.""" + +from typing import Dict, Union + import numpy as np from sklearn.metrics import ( average_precision_score, @@ -8,9 +12,9 @@ def threshold_for_alert_rate( - probabilities, - alert_rate=0.005, -): + probabilities: Union[np.ndarray, list], + alert_rate: float = 0.005, +) -> float: """ Alert only the top X% highest-risk transactions. """ @@ -27,20 +31,35 @@ def threshold_for_alert_rate( ) +def top_k_alert_mask( + probabilities: Union[np.ndarray, list], + alert_rate: float = 0.005, +) -> np.ndarray: + """Select top-K transactions based on alert rate.""" + probabilities = np.asarray(probabilities, dtype=float) + if probabilities.ndim != 1 or len(probabilities) == 0: + raise ValueError("probabilities must be a non-empty one-dimensional array.") + if not 0 < alert_rate <= 1: + raise ValueError("alert_rate must be in (0, 1].") + + alert_count = max(1, int(np.ceil(len(probabilities) * alert_rate))) + order = np.argsort(-probabilities, kind="stable") + selected = order[:alert_count] + mask = np.zeros(len(probabilities), dtype=bool) + mask[selected] = True + return mask + + def precision_recall_at_alert_rate( - y_true, - probabilities, - alert_rate=0.005, -): + y_true: Union[np.ndarray, list], + probabilities: Union[np.ndarray, list], + alert_rate: float = 0.005, +) -> Dict[str, float]: y_true = np.asarray(y_true) probabilities = np.asarray(probabilities) - threshold = threshold_for_alert_rate( - probabilities, - alert_rate, - ) - - predicted = probabilities >= threshold + predicted = top_k_alert_mask(probabilities, alert_rate) + threshold = float(probabilities[predicted].min()) tp = np.sum( (predicted == 1) & @@ -88,12 +107,12 @@ def precision_recall_at_alert_rate( def expected_decision_cost( - y_true, - probabilities, - threshold, - false_negative_cost=100, - false_positive_cost=1, -): + y_true: Union[np.ndarray, list], + probabilities: Union[np.ndarray, list], + threshold: float, + false_negative_cost: float = 100, + false_positive_cost: float = 1, +) -> float: """Return the operational cost induced by a binary alert threshold.""" y_true = np.asarray(y_true) @@ -108,9 +127,9 @@ def expected_decision_cost( ) def calculate_risk_weighted_exposure( - amounts, - probabilities, -): + amounts: Union[np.ndarray, list], + probabilities: Union[np.ndarray, list], +) -> np.ndarray: """ Probability-weighted transaction amount. @@ -125,9 +144,9 @@ def calculate_risk_weighted_exposure( def evaluate_model( - y_true, - probabilities, -): + y_true: Union[np.ndarray, list], + probabilities: Union[np.ndarray, list], +) -> Dict[str, float]: y_true = np.asarray(y_true) if np.unique(y_true).size < 2: diff --git a/src/features/behavioral_features.py b/src/features/behavioral_features.py index df742d9..6ba0ece 100644 --- a/src/features/behavioral_features.py +++ b/src/features/behavioral_features.py @@ -3,144 +3,146 @@ 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. - """ - + """Generate point-in-time behavioral and network features.""" con = duckdb.connect() - con.register("transactions", df) query = """ WITH history AS ( SELECT *, - COUNT(*) OVER ( PARTITION BY Sender_account, Payment_currency 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, Payment_currency - ORDER BY timestamp - RANGE BETWEEN INTERVAL '24 hours' PRECEDING - AND INTERVAL '1 microsecond' PRECEDING - ), - 0 - ) AS sender_amount_sum_24h, - + COALESCE(SUM(Amount) OVER ( + PARTITION BY Sender_account, Payment_currency + 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, Payment_currency 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, Payment_currency 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, Received_currency 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, Received_currency - ORDER BY timestamp - RANGE BETWEEN INTERVAL '24 hours' PRECEDING - AND INTERVAL '1 microsecond' PRECEDING - ), - 0 - ) AS receiver_amount_sum_24h, - + COALESCE(SUM(Amount) OVER ( + PARTITION BY Receiver_account, Received_currency + ORDER BY timestamp + RANGE BETWEEN INTERVAL '24 hours' PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ), 0) AS receiver_amount_sum_24h, 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_txn_count_lifetime, - COUNT(*) OVER ( PARTITION BY Receiver_account ORDER BY timestamp RANGE BETWEEN UNBOUNDED PRECEDING AND INTERVAL '1 microsecond' PRECEDING ) AS receiver_txn_count_lifetime, - COUNT(DISTINCT Receiver_account) OVER ( PARTITION BY Sender_account ORDER BY timestamp RANGE BETWEEN UNBOUNDED PRECEDING AND INTERVAL '1 microsecond' PRECEDING ) AS sender_out_degree, - COUNT(DISTINCT Sender_account) OVER ( PARTITION BY Receiver_account ORDER BY timestamp RANGE BETWEEN UNBOUNDED PRECEDING AND INTERVAL '1 microsecond' PRECEDING ) AS receiver_in_degree, - 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 + ), + pair_history AS ( + SELECT + Sender_account, + Payment_currency, + Receiver_account, + timestamp, + SUM(Amount) AS pair_amount + FROM transactions + GROUP BY Sender_account, Payment_currency, Receiver_account, timestamp + ), + pair_cumulative AS ( + SELECT + *, + COALESCE(SUM(pair_amount) OVER ( + PARTITION BY Sender_account, Payment_currency, Receiver_account + ORDER BY timestamp + RANGE BETWEEN UNBOUNDED PRECEDING + AND INTERVAL '1 microsecond' PRECEDING + ), 0) AS historical_pair_amount + FROM pair_history + ), + sender_hhi AS ( + SELECT + Sender_account, + Payment_currency, + timestamp, + CASE + WHEN SUM(historical_pair_amount) = 0 THEN 0 + ELSE SUM(POWER(historical_pair_amount, 2)) + / POWER(SUM(historical_pair_amount), 2) + END AS sender_counterparty_hhi + FROM pair_cumulative + GROUP BY Sender_account, Payment_currency, timestamp ) - SELECT - *, - + history.*, 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 + ELSE (Amount - sender_amount_mean_30d) / sender_amount_std_30d END AS sender_amount_zscore, - COALESCE( - DATE_DIFF( - 'second', - previous_sender_timestamp, - timestamp - ), + DATE_DIFF('second', previous_sender_timestamp, history.timestamp), -1 - ) AS seconds_since_sender_txn - + ) AS seconds_since_sender_txn, + COALESCE(sender_hhi.sender_counterparty_hhi, 0) + AS sender_counterparty_hhi FROM history - ORDER BY timestamp + LEFT JOIN sender_hhi + ON history.Sender_account = sender_hhi.Sender_account + AND history.Payment_currency = sender_hhi.Payment_currency + AND history.timestamp = sender_hhi.timestamp + ORDER BY history.timestamp """ result = con.execute(query).fetchdf() - con.close() - numerical_history_columns = [ + numeric_columns = [ "sender_txn_count_24h", "sender_amount_sum_24h", "sender_amount_mean_30d", @@ -154,48 +156,11 @@ def add_behavioral_features(df: pd.DataFrame) -> pd.DataFrame: "pair_transaction_count", "sender_amount_zscore", "seconds_since_sender_txn", + "sender_counterparty_hhi", ] - - result[numerical_history_columns] = ( - result[numerical_history_columns] + result[numeric_columns] = ( + result[numeric_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 - currency = row.Payment_currency - sender_key = (sender, currency) - total = sender_totals.get(sender_key, 0.0) - concentration.append( - sender_squared_totals.get(sender_key, 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 - currency = row.Payment_currency - amount = float(row.Amount) - sender_key = (sender, currency) - pair_key = (sender, receiver, currency) - pair_total = sender_receiver_totals.get(pair_key, 0.0) - sender_receiver_totals[pair_key] = pair_total + amount - sender_totals[sender_key] = sender_totals.get(sender_key, 0.0) + amount - sender_squared_totals[sender_key] = ( - sender_squared_totals.get(sender_key, 0.0) - + 2 * pair_total * amount - + amount**2 - ) - - result["sender_counterparty_hhi"] = concentration - - return result \ No newline at end of file + return result diff --git a/src/logging_config.py b/src/logging_config.py new file mode 100644 index 0000000..ca2c1fc --- /dev/null +++ b/src/logging_config.py @@ -0,0 +1,74 @@ +"""Structured logging configuration for the project.""" + +import logging +import sys +from pathlib import Path +from typing import Optional + + +def setup_logging( + name: str, + level: int = logging.INFO, + log_file: Optional[Path] = None, +) -> logging.Logger: + """ + Configure structured logging for a module. + + Args: + name: Logger name (typically __name__) + level: Logging level (default INFO) + log_file: Optional path to write logs to file + + Returns: + Configured logger instance + """ + logger = logging.getLogger(name) + logger.setLevel(level) + + if logger.hasHandlers(): + return logger + + formatter = logging.Formatter( + "%(asctime)s - %(name)s - %(levelname)s - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setLevel(level) + console_handler.setFormatter(formatter) + logger.addHandler(console_handler) + + if log_file: + log_file.parent.mkdir(parents=True, exist_ok=True) + file_handler = logging.FileHandler(log_file) + file_handler.setLevel(level) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + return logger + + +class ProgressLogger: + """Context manager for tracking progress through large operations.""" + + def __init__(self, logger: logging.Logger, total: int, message: str): + self.logger = logger + self.total = total + self.message = message + self.processed = 0 + + def update(self, count: int = 1, interval: int = 10000): + """Update progress and log at specified interval.""" + self.processed += count + if self.processed % interval == 0: + pct = (self.processed / self.total) * 100 + self.logger.info( + f"{self.message}: {self.processed:,} / {self.total:,} ({pct:.1f}%)" + ) + + def __enter__(self): + self.logger.info(f"Starting: {self.message} ({self.total:,} items)") + return self + + def __exit__(self, *args): + self.logger.info(f"Completed: {self.message}") diff --git a/src/models/train.py b/src/models/train.py index 96fc59c..b316dbb 100644 --- a/src/models/train.py +++ b/src/models/train.py @@ -1,4 +1,3 @@ -import numpy as np import pandas as pd import xgboost as xgb from sklearn.compose import ColumnTransformer @@ -6,6 +5,8 @@ from sklearn.pipeline import Pipeline from sklearn.preprocessing import OneHotEncoder +from src.config import TrainingConfig, get_config + def chronological_split( df: pd.DataFrame, @@ -27,6 +28,41 @@ def chronological_split( return train, validation, test + +def temporal_split( + df: pd.DataFrame, + train_fraction: float = 0.65, + calibration_fraction: float = 0.10, + validation_fraction: float = 0.10, +): + ordered = df.sort_values("timestamp").reset_index(drop=True) + if train_fraction <= 0 or calibration_fraction <= 0 or validation_fraction <= 0: + raise ValueError("split fractions must be positive.") + if train_fraction + calibration_fraction + validation_fraction >= 1: + raise ValueError("split fractions must leave observations for test data.") + + timestamps = ordered["timestamp"].drop_duplicates().sort_values().to_numpy() + timestamp_count = len(timestamps) + train_end = max(1, int(timestamp_count * train_fraction)) + calibration_end = max(train_end + 1, int(timestamp_count * (train_fraction + calibration_fraction))) + validation_end = max(calibration_end + 1, int(timestamp_count * (train_fraction + calibration_fraction + validation_fraction))) + if validation_end >= timestamp_count: + raise ValueError("not enough distinct timestamps for four temporal periods.") + + boundaries = timestamps[[train_end, calibration_end, validation_end]] + train = ordered[ordered["timestamp"] < boundaries[0]].copy() + calibration = ordered[ + (ordered["timestamp"] >= boundaries[0]) + & (ordered["timestamp"] < boundaries[1]) + ].copy() + validation = ordered[ + (ordered["timestamp"] >= boundaries[1]) + & (ordered["timestamp"] < boundaries[2]) + ].copy() + test = ordered[ordered["timestamp"] >= boundaries[2]].copy() + + return train, calibration, validation, test + TARGET = "Is_laundering" @@ -123,7 +159,10 @@ def chronological_split( } -def build_preprocessor(feature_names=MODEL_FEATURES): +def build_preprocessor(feature_names=MODEL_FEATURES, config: TrainingConfig = None): + if config is None: + config = get_config() + numeric_features = [ feature for feature in feature_names if feature in NUMERIC_FEATURES @@ -137,7 +176,7 @@ def build_preprocessor(feature_names=MODEL_FEATURES): steps=[ ( "imputer", - SimpleImputer(strategy="median"), + SimpleImputer(strategy=config.preprocessing.numeric_impute_strategy), ), ] ) @@ -146,13 +185,13 @@ def build_preprocessor(feature_names=MODEL_FEATURES): steps=[ ( "imputer", - SimpleImputer(strategy="most_frequent"), + SimpleImputer(strategy=config.preprocessing.categorical_impute_strategy), ), ( "onehot", OneHotEncoder( - handle_unknown="ignore", - min_frequency=20, + handle_unknown=config.preprocessing.categorical_handle_unknown, + min_frequency=config.preprocessing.categorical_min_frequency, ), ), ] @@ -175,48 +214,41 @@ def build_preprocessor(feature_names=MODEL_FEATURES): ) -def build_xgboost_model(y_train): +def build_xgboost_model(y_train, config: TrainingConfig = None): + if config is None: + config = get_config() + 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), - ) + model_params = config.xgboost.to_dict() + model_params.update({ + "objective": "binary:logistic", + "scale_pos_weight": scale_pos_weight, + "n_jobs": -1, + }) + + return xgb.XGBClassifier(**model_params) - 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, config: TrainingConfig = None): + if config is None: + config = get_config() -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) + preprocessor = build_preprocessor(feature_names, config=config) X_train_processed = preprocessor.fit_transform(X_train) X_validation_processed = preprocessor.transform(X_validation) - model = build_xgboost_model(y_train) + model = build_xgboost_model(y_train, config=config) model.fit( X_train_processed, y_train, diff --git a/src/monitoring/__init__.py b/src/monitoring/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/monitoring/drift.py b/src/monitoring/drift.py new file mode 100644 index 0000000..ee1b165 --- /dev/null +++ b/src/monitoring/drift.py @@ -0,0 +1,132 @@ +"""Model drift and data distribution monitoring.""" + +from typing import Dict, Optional, List + +import numpy as np +import pandas as pd + + +class DriftDetector: + """Detect prediction distribution shifts and feature drift.""" + + def __init__(self, baseline_quantiles: List[float]): + """ + Initialize drift detector. + + Args: + baseline_quantiles: Baseline prediction probability quantiles + """ + self.baseline_quantiles = np.array(baseline_quantiles) + self.baseline_median = np.percentile(baseline_quantiles, 50) + self.baseline_std = np.std(baseline_quantiles) + + def detect_prediction_drift( + self, + predictions: np.ndarray, + threshold_std: float = 2.0, + ) -> Dict[str, float]: + """ + Detect shift in prediction distribution. + + Args: + predictions: Array of predicted probabilities + threshold_std: Standard deviation threshold for drift detection + + Returns: + Dictionary with drift metrics + """ + predictions = np.asarray(predictions) + + current_median = np.median(predictions) + current_std = np.std(predictions) + + median_shift = abs(current_median - self.baseline_median) + std_ratio = current_std / max(self.baseline_std, 1e-8) + + median_drift = median_shift > (threshold_std * self.baseline_std) + variance_drift = std_ratio > 2.0 or std_ratio < 0.5 + + return { + "baseline_median": float(self.baseline_median), + "current_median": float(current_median), + "median_shift": float(median_shift), + "baseline_std": float(self.baseline_std), + "current_std": float(current_std), + "std_ratio": float(std_ratio), + "median_drift_detected": bool(median_drift), + "variance_drift_detected": bool(variance_drift), + "drift_detected": bool(median_drift or variance_drift), + } + + def detect_feature_drift( + self, + feature_df: pd.DataFrame, + baseline_stats: Dict[str, Dict[str, float]], + threshold: float = 0.1, + ) -> Dict[str, Dict[str, float]]: + """ + Detect shifts in feature distributions. + + Args: + feature_df: DataFrame with features + baseline_stats: Dictionary of baseline statistics per feature + threshold: Percentage change threshold for drift + + Returns: + Dictionary with feature drift metrics + """ + drift_results = {} + + for col in feature_df.columns: + if col not in baseline_stats: + continue + + current_mean = feature_df[col].mean() + current_std = feature_df[col].std() + baseline_mean = baseline_stats[col].get("mean", 0) + baseline_std = baseline_stats[col].get("std", 1) + + mean_change = abs(current_mean - baseline_mean) / max(abs(baseline_mean), 1e-8) + std_change = abs(current_std - baseline_std) / max(baseline_std, 1e-8) + + drift_results[col] = { + "baseline_mean": float(baseline_mean), + "current_mean": float(current_mean), + "mean_change_pct": float(mean_change * 100), + "baseline_std": float(baseline_std), + "current_std": float(current_std), + "std_change_pct": float(std_change * 100), + "drift_detected": bool(mean_change > threshold or std_change > threshold), + } + + return drift_results + + +def compute_baseline_stats( + df: pd.DataFrame, + features: List[str], +) -> Dict[str, Dict[str, float]]: + """ + Compute baseline statistics for feature drift detection. + + Args: + df: Training/baseline dataset + features: List of features to compute stats for + + Returns: + Dictionary with mean and std for each feature + """ + stats = {} + + for col in features: + if col not in df.columns: + continue + + stats[col] = { + "mean": float(df[col].mean()), + "std": float(df[col].std()), + "min": float(df[col].min()), + "max": float(df[col].max()), + } + + return stats diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..3dd6d44 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,191 @@ +"""Tests for FastAPI inference API.""" + +import json +import pytest +from pathlib import Path + +from fastapi.testclient import TestClient + +from src.api.main import create_app + + +@pytest.fixture +def client(tmp_path): + """Create a FastAPI test client without a trained model.""" + # Use a non-existent path to ensure no model is loaded + app = create_app(tmp_path / "nonexistent.joblib") + return TestClient(app) + + +def get_valid_payload(): + """Return a valid feature vector for testing.""" + return { + "Amount": 1000.0, + "log_amount": 6.908, + "hour_sin": 0.5, + "hour_cos": 0.866, + "dow_sin": 0.0, + "dow_cos": 1.0, + "month_sin": 0.259, + "month_cos": 0.966, + "is_weekend": 0, + "is_night": 0, + "currency_mismatch": 0, + "cross_border": 0, + "is_round_amount": 1, + "sender_txn_count_24h": 5, + "sender_amount_sum_24h": 5000.0, + "sender_amount_mean_30d": 1000.0, + "sender_amount_std_30d": 500.0, + "sender_amount_zscore": 0.0, + "receiver_txn_count_24h": 3, + "receiver_amount_sum_24h": 3000.0, + "seconds_since_sender_txn": 3600, + "sender_txn_count_lifetime": 100, + "receiver_txn_count_lifetime": 50, + "sender_out_degree": 25, + "receiver_in_degree": 20, + "pair_transaction_count": 5, + "sender_counterparty_hhi": 0.2, + "Payment_type": "Transfer", + "Payment_currency": "USD", + "Received_currency": "USD", + "Sender_bank_location": "US", + "Receiver_bank_location": "US", + } + + +def test_health_check(client): + """Test health endpoint.""" + response = client.get("/api/health") + + assert response.status_code == 200 + data = response.json() + assert "status" in data + assert "model_loaded" in data + assert "timestamp" in data + + +def test_health_check_format(client): + """Test that health response has correct format.""" + response = client.get("/api/health") + data = response.json() + + assert isinstance(data, dict) + assert data["status"] in ["healthy", "model_unavailable"] + assert isinstance(data["model_loaded"], bool) + assert "timestamp" in data + + +def test_model_info_without_model(client): + """Test model info endpoint when model is unavailable.""" + response = client.get("/api/model/info") + # Without a trained model artifact, should get 503 + assert response.status_code in [503, 500] # FastAPI error handling + + +def test_predict_missing_required_field(client): + """Test prediction with missing required field.""" + payload = get_valid_payload() + del payload["Amount"] + + response = client.post("/api/predict", json=payload) + + assert response.status_code == 422 + data = response.json() + assert "detail" in data + + +def test_predict_invalid_feature_type(client): + """Test prediction with invalid feature type.""" + payload = get_valid_payload() + payload["Amount"] = "not_a_number" + + response = client.post("/api/predict", json=payload) + + assert response.status_code == 422 + data = response.json() + assert "detail" in data + + +def test_predict_extra_fields_rejected(client): + """Test that extra unknown fields are rejected.""" + payload = get_valid_payload() + payload["extra_field"] = "should_fail" + + response = client.post("/api/predict", json=payload) + + assert response.status_code == 422 + + +def test_predict_negative_amount(client): + """Test that negative amounts are rejected.""" + payload = get_valid_payload() + payload["Amount"] = -100.0 + + response = client.post("/api/predict", json=payload) + + assert response.status_code == 422 + + +def test_predict_invalid_binary_field(client): + """Test that invalid binary field values are rejected.""" + payload = get_valid_payload() + payload["is_weekend"] = 2 + + response = client.post("/api/predict", json=payload) + + assert response.status_code == 422 + + +def test_predict_cyclical_out_of_range(client): + """Test that cyclical encodings outside [-1, 1] are rejected.""" + payload = get_valid_payload() + payload["hour_sin"] = 1.5 + + response = client.post("/api/predict", json=payload) + + assert response.status_code == 422 + + +def test_cors_headers(client): + """Test that CORS headers are set on preflight requests.""" + # TestClient doesn't fully trigger CORS middleware, but the app has CORS configured + # Test that the health endpoint works (CORS is configured in create_app) + response = client.get("/api/health") + assert response.status_code == 200 + # CORS headers may not appear in TestClient due to how it handles middleware + # but they will appear in real HTTP requests; this just ensures no errors + + +def test_predict_hhi_range(client): + """Test that HHI must be between 0 and 1.""" + payload = get_valid_payload() + payload["sender_counterparty_hhi"] = 1.5 + + response = client.post("/api/predict", json=payload) + + assert response.status_code == 422 + + +def test_predict_invalid_json(client): + """Test prediction with invalid JSON.""" + response = client.post( + "/api/predict", + content="not valid json", + headers={"Content-Type": "application/json"}, + ) + + # FastAPI returns 422 for parsing errors + assert response.status_code in [400, 422, 503] + + +def test_predict_non_dict_payload(client): + """Test prediction with non-dictionary JSON.""" + response = client.post( + "/api/predict", + content=json.dumps(["not", "a", "dict"]), + headers={"Content-Type": "application/json"}, + ) + + assert response.status_code == 422 diff --git a/tests/test_api_schemas.py b/tests/test_api_schemas.py new file mode 100644 index 0000000..de40a86 --- /dev/null +++ b/tests/test_api_schemas.py @@ -0,0 +1,158 @@ +"""Tests for Pydantic request validation models.""" + +import pytest +from pydantic import ValidationError + +from src.api.models import FeatureVector + + +def create_valid_payload() -> dict: + """Create a valid feature vector payload.""" + return { + "Amount": 1000.0, + "log_amount": 6.91, + "hour_sin": 0.5, + "hour_cos": 0.866, + "dow_sin": 0.0, + "dow_cos": 1.0, + "month_sin": 0.259, + "month_cos": 0.966, + "is_weekend": 0, + "is_night": 0, + "currency_mismatch": 0, + "cross_border": 0, + "is_round_amount": 1, + "sender_txn_count_24h": 5, + "sender_amount_sum_24h": 5000.0, + "sender_amount_mean_30d": 1000.0, + "sender_amount_std_30d": 500.0, + "sender_amount_zscore": 0.0, + "receiver_txn_count_24h": 3, + "receiver_amount_sum_24h": 3000.0, + "seconds_since_sender_txn": 3600, + "sender_txn_count_lifetime": 100, + "receiver_txn_count_lifetime": 50, + "sender_out_degree": 25, + "receiver_in_degree": 20, + "pair_transaction_count": 5, + "sender_counterparty_hhi": 0.2, + "Payment_type": "Transfer", + "Payment_currency": "USD", + "Received_currency": "USD", + "Sender_bank_location": "US", + "Receiver_bank_location": "US", + } + + +def test_valid_feature_vector(): + """Test that valid feature vectors pass validation.""" + payload = create_valid_payload() + + result = FeatureVector.model_validate(payload) + + assert result.Amount == 1000.0 + assert result.Payment_currency == "USD" + + +def test_missing_required_field(): + """Test that missing required fields raise ValidationError.""" + payload = create_valid_payload() + del payload["Amount"] + + with pytest.raises(ValidationError) as exc_info: + FeatureVector.model_validate(payload) + + errors = exc_info.value.errors() + assert any("Amount" in str(err["loc"]) for err in errors) + + +def test_invalid_numeric_range(): + """Test that out-of-range numeric values are rejected.""" + payload = create_valid_payload() + payload["Amount"] = -100.0 + + with pytest.raises(ValidationError): + FeatureVector.model_validate(payload) + + +def test_invalid_cyclical_encoding(): + """Test that cyclical encodings outside [-1, 1] are rejected.""" + payload = create_valid_payload() + payload["hour_sin"] = 1.5 + + with pytest.raises(ValidationError): + FeatureVector.model_validate(payload) + + +def test_invalid_binary_field(): + """Test that binary fields reject non-0/1 values.""" + payload = create_valid_payload() + payload["is_weekend"] = 2 + + with pytest.raises(ValidationError): + FeatureVector.model_validate(payload) + + +def test_unknown_fields_rejected(): + """Test that unknown fields are rejected (extra='forbid').""" + payload = create_valid_payload() + payload["unknown_field"] = "should_fail" + + with pytest.raises(ValidationError) as exc_info: + FeatureVector.model_validate(payload) + + errors = exc_info.value.errors() + assert any(err["type"] == "extra_forbidden" for err in errors) + + +def test_string_for_numeric_field(): + """Test that string values for numeric fields are rejected.""" + payload = create_valid_payload() + payload["Amount"] = "not_a_number" + + with pytest.raises(ValidationError): + FeatureVector.model_validate(payload) + + +def test_nullable_fields(): + """Test that some fields allow None values.""" + payload = create_valid_payload() + payload["sender_amount_mean_30d"] = None + payload["sender_amount_std_30d"] = None + + result = FeatureVector.model_validate(payload) + + assert result.sender_amount_mean_30d is None + assert result.sender_amount_std_30d is None + + +def test_all_required_features_present(): + """Test that all required fields are present and validated.""" + payload = create_valid_payload() + + result = FeatureVector.model_validate(payload) + + assert result.Amount == payload["Amount"] + assert result.Payment_type == payload["Payment_type"] + assert result.sender_counterparty_hhi == payload["sender_counterparty_hhi"] + + +def test_hhi_range(): + """Test that HHI is constrained to [0, 1].""" + payload = create_valid_payload() + payload["sender_counterparty_hhi"] = 1.5 + + with pytest.raises(ValidationError): + FeatureVector.model_validate(payload) + + +def test_model_dump(): + """Test model_dump() serialization.""" + payload = create_valid_payload() + model = FeatureVector.model_validate(payload) + + dumped = model.model_dump() + + assert dumped["Amount"] == 1000.0 + assert dumped["Payment_currency"] == "USD" + assert "unknown_field" not in dumped diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..95f9f01 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,117 @@ +"""Tests for configuration management.""" + +import pytest + +from src.config import ( + XGBoostConfig, + PreprocessingConfig, + TrainingConfig, + get_config, + get_fast_config, + get_production_config, +) + + +def test_xgboost_config_defaults(): + """Test XGBoost config has sensible defaults.""" + config = XGBoostConfig() + + assert config.n_estimators == 500 + assert config.max_depth == 5 + assert config.learning_rate == 0.05 + assert config.random_state == 42 + + +def test_xgboost_config_to_dict(): + """Test XGBoost config can be converted to dict.""" + config = XGBoostConfig(n_estimators=100) + + config_dict = config.to_dict() + + assert isinstance(config_dict, dict) + assert config_dict["n_estimators"] == 100 + assert config_dict["learning_rate"] == 0.05 + + +def test_preprocessing_config_defaults(): + """Test preprocessing config has sensible defaults.""" + config = PreprocessingConfig() + + assert config.numeric_impute_strategy == "median" + assert config.categorical_impute_strategy == "most_frequent" + assert config.categorical_min_frequency == 20 + + +def test_training_config_auto_initialization(): + """Test that TrainingConfig initializes sub-configs automatically.""" + config = TrainingConfig() + + assert isinstance(config.xgboost, XGBoostConfig) + assert isinstance(config.preprocessing, PreprocessingConfig) + assert config.xgboost.n_estimators == 500 + + +def test_fast_config(): + """Test fast/smoke test configuration.""" + config = get_fast_config() + + assert config.xgboost.n_estimators == 10 + assert isinstance(config, TrainingConfig) + + +def test_production_config(): + """Test production configuration.""" + config = get_production_config() + + assert config.xgboost.n_estimators == 500 + + +def test_get_config_with_fast_mode(): + """Test get_config function with fast mode.""" + fast_config = get_config(fast=True) + prod_config = get_config(fast=False) + + assert fast_config.xgboost.n_estimators == 10 + assert prod_config.xgboost.n_estimators == 500 + + +def test_alert_config_initialization(): + """Test AlertConfig post_init.""" + from src.config import AlertConfig + + config = AlertConfig() + + assert config.alert_rate == 0.005 + assert isinstance(config.alert_rates_for_metrics, list) + assert len(config.alert_rates_for_metrics) == 3 + assert 0.001 in config.alert_rates_for_metrics + + +def test_alert_config_custom_rates(): + """Test AlertConfig with custom rates.""" + from src.config import AlertConfig + + custom_rates = [0.01, 0.02, 0.05] + config = AlertConfig(alert_rates_for_metrics=custom_rates) + + assert config.alert_rates_for_metrics == custom_rates + + +def test_config_customization(): + """Test that configs can be customized.""" + config = TrainingConfig() + config.xgboost.n_estimators = 200 + config.xgboost.learning_rate = 0.1 + + assert config.xgboost.n_estimators == 200 + assert config.xgboost.learning_rate == 0.1 + + +def test_config_isolation(): + """Test that configs are isolated (no shared state).""" + config1 = get_config(fast=True) + config2 = get_config(fast=True) + + config1.xgboost.n_estimators = 50 + + assert config2.xgboost.n_estimators == 10 diff --git a/tests/test_data_validation.py b/tests/test_data_validation.py new file mode 100644 index 0000000..6b3ee33 --- /dev/null +++ b/tests/test_data_validation.py @@ -0,0 +1,128 @@ +"""Tests for data quality validation.""" + +import pytest +import pandas as pd +import numpy as np + +from src.data.validation import DataQualityValidator, validate_features +from src.models.train import MODEL_FEATURES, TARGET + + +def create_valid_dataframe(n_rows: int = 100) -> pd.DataFrame: + """Create a valid feature dataframe for testing.""" + data = {feature: np.random.rand(n_rows) for feature in MODEL_FEATURES} + data["timestamp"] = pd.date_range("2026-01-01", periods=n_rows, freq="H") + data[TARGET] = np.random.randint(0, 2, n_rows) + + return pd.DataFrame(data) + + +def test_validator_accepts_valid_data(): + """Test that validator accepts valid data.""" + df = create_valid_dataframe() + + validator = DataQualityValidator() + assert validator.validate(df) + assert len(validator.issues) == 0 + + +def test_validator_detects_missing_features(): + """Test that validator detects missing required features.""" + df = create_valid_dataframe() + df = df.drop(columns=["Amount"]) + + validator = DataQualityValidator() + assert not validator.validate(df) + assert any("Missing" in issue for issue in validator.issues) + + +def test_validator_detects_missing_target(): + """Test that validator detects missing target column.""" + df = create_valid_dataframe() + df = df.drop(columns=[TARGET]) + + validator = DataQualityValidator() + assert not validator.validate(df) + assert any(TARGET in issue for issue in validator.issues) + + +def test_validator_detects_excessive_nans(): + """Test that validator detects columns with too many NaNs.""" + df = create_valid_dataframe(n_rows=100) + + # Introduce 10% NaN in one column (exceeds default 5% threshold) + df.loc[:10, "Amount"] = np.nan + + validator = DataQualityValidator(nan_threshold=0.05) + assert not validator.validate(df) + assert any("Amount" in issue and "NaN" in issue for issue in validator.issues) + + +def test_validator_detects_non_chronological_data(): + """Test that validator detects non-chronological ordering.""" + df = create_valid_dataframe(n_rows=10) + + # Shuffle timestamps + df = df.sample(frac=1).reset_index(drop=True) + + validator = DataQualityValidator() + assert not validator.validate(df) + assert any("chronologically" in issue for issue in validator.issues) + + +def test_validator_warns_on_extreme_class_imbalance(): + """Test that validator warns on extreme class imbalance.""" + df = create_valid_dataframe(n_rows=1000) + + # Create extreme imbalance (99.9% negative) + df[TARGET] = 0 + df.loc[:1, TARGET] = 1 + + validator = DataQualityValidator() + # Should still pass validation but with warning + validator.validate(df) + + +def test_validator_detects_infinite_values(): + """Test that validator detects infinite values.""" + df = create_valid_dataframe(n_rows=100) + + df.loc[0, "Amount"] = np.inf + df.loc[1, "log_amount"] = -np.inf + + validator = DataQualityValidator() + assert not validator.validate(df) + assert any("infinite" in issue for issue in validator.issues) + + +def test_validate_features_function(): + """Test the high-level validate_features function.""" + df = create_valid_dataframe() + + # Should not raise + assert validate_features(df) + + +def test_validate_features_raises_on_missing_schema(): + """Test that validate_features raises on missing required columns.""" + df = create_valid_dataframe() + df = df.drop(columns=["Amount"]) + + with pytest.raises(ValueError, match="Critical validation failure"): + validate_features(df) + + +def test_validator_with_custom_nan_threshold(): + """Test validator with custom NaN threshold.""" + df = create_valid_dataframe(n_rows=100) + + # Introduce 5% NaN + df.loc[:5, "Amount"] = np.nan + + # Should pass with 10% threshold + validator = DataQualityValidator(nan_threshold=0.10) + assert validator.validate(df) + + # Should fail with 1% threshold + validator = DataQualityValidator(nan_threshold=0.01) + assert not validator.validate(df) diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py new file mode 100644 index 0000000..3e718b5 --- /dev/null +++ b/tests/test_end_to_end.py @@ -0,0 +1,146 @@ +from pathlib import Path + +import joblib +import pandas as pd +from fastapi.testclient import TestClient + +from src.api.main import create_app +from src.config import get_config +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 +from src.models.calibration import ProbabilityCalibrator +from src.models.train import MODEL_FEATURES, TARGET, fit_model, temporal_split + + +def synthetic_features(rows=48): + import numpy as np + + records = [] + for index in range(rows): + # Create valid synthetic features + record = {} + + # Numeric features with valid ranges + record["Amount"] = float(100 * ((index % 7) + 1)) + record["log_amount"] = np.log1p(record["Amount"]) + + # Cyclical encodings must be in [-1, 1] + record["hour_sin"] = np.sin(2 * np.pi * (index % 24) / 24) + record["hour_cos"] = np.cos(2 * np.pi * (index % 24) / 24) + record["dow_sin"] = np.sin(2 * np.pi * (index % 7) / 7) + record["dow_cos"] = np.cos(2 * np.pi * (index % 7) / 7) + record["month_sin"] = np.sin(2 * np.pi * (index % 12) / 12) + record["month_cos"] = np.cos(2 * np.pi * (index % 12) / 12) + + # Binary features + record["is_weekend"] = int(index % 7 >= 5) + record["is_night"] = int((index % 24) in [22, 23, 0, 1, 2, 3, 4, 5, 6]) + record["currency_mismatch"] = int(index % 3 == 0) + record["cross_border"] = int(index % 2 == 0) + record["is_round_amount"] = int(index % 5 == 0) + + # Behavioral features + record["sender_txn_count_24h"] = max(0, index % 10) + record["sender_amount_sum_24h"] = float(index * 100 % 5000) + record["sender_amount_mean_30d"] = float(index * 50 % 1000) + record["sender_amount_std_30d"] = float(index * 30 % 500) + record["sender_amount_zscore"] = float((index - 5) / 5) + + record["receiver_txn_count_24h"] = max(0, index % 8) + record["receiver_amount_sum_24h"] = float(index * 80 % 4000) + record["seconds_since_sender_txn"] = max(-1, (index - 1) * 3600) + + # Network features + record["sender_txn_count_lifetime"] = max(0, index * 2) + record["receiver_txn_count_lifetime"] = max(0, index * 2 - 5) + record["sender_out_degree"] = max(0, index % 30) + record["receiver_in_degree"] = max(0, index % 25) + record["pair_transaction_count"] = max(0, index % 10) + record["sender_counterparty_hhi"] = float((index % 100) / 100) + + # Categorical features + record["Payment_type"] = "Transfer" + record["Payment_currency"] = "USD" + record["Received_currency"] = "USD" + record["Sender_bank_location"] = "US" + record["Receiver_bank_location"] = "US" + + # Metadata + record["timestamp"] = pd.Timestamp("2026-01-01") + pd.to_timedelta( + index, + unit="D", + ) + record[TARGET] = index % 2 + + records.append(record) + + return pd.DataFrame(records) + + +def test_training_artifact_and_api_prediction(tmp_path: Path): + raw_frame = pd.DataFrame( + { + "Time": ["12:00:00", "12:01:00"], + "Date": ["2026-01-01", "2026-01-02"], + "Sender_account": ["sender-1", "sender-1"], + "Receiver_account": ["receiver-1", "receiver-2"], + "Amount": [100.0, 200.0], + "Payment_currency": ["USD", "USD"], + "Received_currency": ["USD", "USD"], + "Sender_bank_location": ["US", "US"], + "Receiver_bank_location": ["US", "US"], + "Payment_type": ["Transfer", "Transfer"], + "Is_laundering": [0, 1], + "Laundering_type": ["None", "Structuring"], + } + ) + raw_path = tmp_path / "mini_saml.csv" + raw_frame.to_csv(raw_path, index=False) + loaded = load_saml_data(raw_path) + engineered = add_behavioral_features(add_transaction_features(loaded)) + assert "sender_counterparty_hhi" in engineered + + frame = synthetic_features() + train, calibration, _, test = temporal_split( + frame, + train_fraction=0.5, + calibration_fraction=0.2, + validation_fraction=0.1, + ) + config = get_config(fast=True) + preprocessor, model, calibration_x, calibration_y = fit_model( + train, + calibration, + config=config, + ) + calibration_probabilities = model.predict_proba(calibration_x)[:, 1] + calibrator = ProbabilityCalibrator().fit( + calibration_probabilities, + calibration_y, + ) + artifact_path = tmp_path / "risk_model.joblib" + joblib.dump( + { + "preprocessor": preprocessor, + "model": model, + "calibrator": calibrator, + "features": MODEL_FEATURES, + "decision_threshold": 0.5, + "alert_rate": 0.005, + "test_metrics": {}, + "model_version": "test", + }, + artifact_path, + ) + + client = TestClient(create_app(artifact_path)) + response = client.post( + "/api/predict", + json=test.iloc[0][MODEL_FEATURES].to_dict(), + ) + + assert response.status_code == 200 + body = response.json() + assert 0 <= body["risk_probability"] <= 1 + assert isinstance(body["requires_review"], bool) diff --git a/tests/test_features.py b/tests/test_features.py index 8182bb2..57b155c 100644 --- a/tests/test_features.py +++ b/tests/test_features.py @@ -1,6 +1,9 @@ +import pytest import pandas as pd +import numpy as np from src.features.transaction_features import add_transaction_features +from src.features.behavioral_features import add_behavioral_features def test_transaction_features_do_not_encode_account_ids(): @@ -23,3 +26,108 @@ def test_transaction_features_do_not_encode_account_ids(): assert "currency_mismatch" in result assert result["Sender_account"].dtype == object assert result["Receiver_account"].dtype == object + + +def test_transaction_features_handles_zero_amounts(): + """Test that zero and near-zero amounts are handled correctly.""" + frame = pd.DataFrame( + { + "timestamp": pd.to_datetime(["2026-01-01", "2026-01-02"]), + "Sender_account": ["s1", "s2"], + "Receiver_account": ["r1", "r2"], + "Amount": [0.0, 0.01], + "Payment_currency": ["USD", "USD"], + "Received_currency": ["USD", "USD"], + "Sender_bank_location": ["US", "US"], + "Receiver_bank_location": ["US", "US"], + } + ) + + result = add_transaction_features(frame) + + assert result["log_amount"].notna().all() + assert not np.isinf(result["log_amount"]).any() + + +def test_transaction_features_cyclical_encoding(): + """Test that cyclical time encoding produces values in [-1, 1].""" + frame = pd.DataFrame( + { + "timestamp": pd.to_datetime([ + "2026-01-01 00:00:00", + "2026-01-01 12:00:00", + "2026-01-01 23:59:59", + "2026-12-31 23:59:59", + ]), + "Sender_account": ["s1", "s2", "s3", "s4"], + "Receiver_account": ["r1", "r2", "r3", "r4"], + "Amount": [100.0] * 4, + "Payment_currency": ["USD"] * 4, + "Received_currency": ["USD"] * 4, + "Sender_bank_location": ["US"] * 4, + "Receiver_bank_location": ["US"] * 4, + } + ) + + result = add_transaction_features(frame) + + cyclical_cols = ["hour_sin", "hour_cos", "dow_sin", "dow_cos", "month_sin", "month_cos"] + for col in cyclical_cols: + assert (result[col].abs() <= 1.0).all(), f"{col} outside [-1, 1]" + + +def test_transaction_features_preserves_row_count(): + """Test that feature engineering doesn't add or remove rows.""" + frame = pd.DataFrame( + { + "timestamp": pd.to_datetime([f"2026-01-{i:02d}" for i in range(1, 11)]), + "Sender_account": [f"s{i}" for i in range(10)], + "Receiver_account": [f"r{i}" for i in range(10)], + "Amount": np.random.rand(10) * 1000, + "Payment_currency": ["USD"] * 10, + "Received_currency": ["USD"] * 10, + "Sender_bank_location": ["US"] * 10, + "Receiver_bank_location": ["US"] * 10, + } + ) + + result = add_transaction_features(frame) + + assert len(result) == len(frame) + + +def test_behavioral_features_handles_missing_history(): + """Test that behavioral features handle transactions with no prior history.""" + 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": ["USD"], + "Received_currency": ["USD"], + "Sender_bank_location": ["US"], + "Receiver_bank_location": ["US"], + "Payment_type": ["Transfer"], + "Is_laundering": [0], + "Laundering_type": ["None"], + } + ) + + # Add transaction features first + frame = add_transaction_features(frame) + + # Then behavioral features + result = add_behavioral_features(frame) + + # Check that all numeric behavioral features are populated + behavioral_cols = [ + "sender_txn_count_24h", + "sender_amount_sum_24h", + "seconds_since_sender_txn", + "sender_amount_zscore", + ] + + for col in behavioral_cols: + if col in result.columns: + assert result[col].notna().all(), f"{col} has NaN values" diff --git a/tests/test_inference.py b/tests/test_inference.py index 3634e8a..c037f12 100644 --- a/tests/test_inference.py +++ b/tests/test_inference.py @@ -1,27 +1,38 @@ +import asyncio from pathlib import Path import pytest +from fastapi.testclient import TestClient -from src.api.app import create_app -from src.models.inference import model_input_from_features +from src.api.main import create_app +from src.api.inference import model_input_from_features -def test_inference_rejects_identifiers_and_targets(): +@pytest.mark.asyncio +async def test_inference_rejects_identifiers_and_targets(): + """Test that async inference rejects account identifiers.""" with pytest.raises(ValueError, match="Identifier or target fields"): - model_input_from_features( + await model_input_from_features( {"Amount": 100.0, "Sender_account": "not-a-feature"}, ["Amount"], ) -def test_inference_requires_feature_store_output(): +@pytest.mark.asyncio +async def test_inference_requires_feature_store_output(): + """Test that async inference requires all model features.""" with pytest.raises(ValueError, match="Missing model features"): - model_input_from_features({"Amount": 100.0}, ["Amount", "sender_txn_count_24h"]) + await model_input_from_features( + {"Amount": 100.0}, + ["Amount", "sender_txn_count_24h"], + ) def test_api_reports_missing_model_artifact(tmp_path: Path): + """Test that API reports missing model artifact correctly.""" app = create_app(tmp_path / "absent.joblib") - response = app.test_client().get("/api/health") + client = TestClient(app) + response = client.get("/api/health") assert response.status_code == 200 - assert response.get_json()["model_loaded"] is False + assert response.json()["model_loaded"] is False diff --git a/tests/test_leakage.py b/tests/test_leakage.py index c0a8b23..90f992f 100644 --- a/tests/test_leakage.py +++ b/tests/test_leakage.py @@ -1,8 +1,8 @@ import pandas as pd +from src.evaluation.metrics import top_k_alert_mask from src.features.behavioral_features import add_behavioral_features -from src.models.train import MODEL_FEATURES, chronological_split - +from src.models.train import MODEL_FEATURES, chronological_split, temporal_split FORBIDDEN_FEATURES = { "Is_laundering", @@ -83,3 +83,40 @@ def test_amount_history_is_currency_aware(): assert result.loc[1, "sender_amount_sum_24h"] == 0 assert result.loc[1, "sender_amount_mean_30d"] == 0 assert result.loc[1, "sender_counterparty_hhi"] == 0 + + +def test_alert_budget_selects_exact_top_k_on_ties(): + mask = top_k_alert_mask([0.9, 0.5, 0.5, 0.1], alert_rate=0.5) + + assert mask.tolist() == [True, True, False, False] + + +def test_temporal_split_keeps_timestamp_groups_together(): + frame = pd.DataFrame( + { + "timestamp": pd.to_datetime( + [ + "2026-01-01", + "2026-01-01", + "2026-01-02", + "2026-01-03", + "2026-01-04", + "2026-01-05", + "2026-01-06", + "2026-01-07", + "2026-01-08", + "2026-01-09", + ] + ), + "value": range(10), + } + ) + + train, calibration, validation, test = temporal_split(frame) + + periods = [train, calibration, validation, test] + assert all(not period.empty for period in periods) + assert all( + left["timestamp"].max() < right["timestamp"].min() + for left, right in zip(periods, periods[1:]) + )