fraud_detection.py

"""
fraud_detection.py

Wraps scikit-learn's IsolationForest with a small, consistent interface used
by both the training script (train_model.py) and the Streamlit app (app.py).

IsolationForest is an unsupervised anomaly detector. Internally it returns:
    - decision_function(X): higher values = more "normal", lower/negative = more anomalous
    - predict(X): 1 = normal (inlier), -1 = anomaly (outlier)

For this project we convert everything into a single "anomaly score" where
HIGHER = MORE LIKELY FRAUD, which makes threshold selection, PR curves and
F1 optimization straightforward and consistent.
"""

from dataclasses import dataclass, field
from typing import Optional

import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.metrics import (
    average_precision_score,
    confusion_matrix,
    f1_score,
    precision_recall_curve,
)


@dataclass
class FraudDetector:
    """Thin wrapper around IsolationForest for fraud anomaly detection."""

    n_estimators: int = 100
    max_samples: str = "auto"
    contamination: float = 0.001727
    random_state: int = 42
    model: Optional[IsolationForest] = field(default=None, repr=False)
    threshold: float = 0.0  # anomaly-score threshold used to flag fraud

    def build(self) -> IsolationForest:
        self.model = IsolationForest(
            n_estimators=self.n_estimators,
            max_samples=self.max_samples,
            contamination=self.contamination,
            random_state=self.random_state,
            n_jobs=-1,
        )
        return self.model

    def fit(self, X):
        if self.model is None:
            self.build()
        self.model.fit(X)
        return self

    def anomaly_scores(self, X) -> np.ndarray:
        """Higher score = more anomalous (more likely fraud)."""
        return -self.model.decision_function(X)

    def predict_with_threshold(self, X, threshold: Optional[float] = None) -> np.ndarray:
        """Return binary predictions (1 = fraud, 0 = normal) using the given/stored threshold."""
        thr = self.threshold if threshold is None else threshold
        scores = self.anomaly_scores(X)
        return (scores >= thr).astype(int)

    @staticmethod
    def find_best_threshold(y_true, scores) -> float:
        """
        Scan the precision-recall curve to find the anomaly-score threshold
        that maximizes F1-score for the minority (fraud) class.
        """
        precision, recall, thresholds = precision_recall_curve(y_true, scores)
        # precision_recall_curve returns len(thresholds) == len(precision) - 1
        f1_scores = np.divide(
            2 * precision[:-1] * recall[:-1],
            precision[:-1] + recall[:-1],
            out=np.zeros_like(precision[:-1]),
            where=(precision[:-1] + recall[:-1]) != 0,
        )
        best_idx = int(np.argmax(f1_scores))
        return float(thresholds[best_idx])

    @staticmethod
    def evaluate(y_true, scores, y_pred) -> dict:
        """Compute the required evaluation metrics + supporting curve/matrix data."""
        pr_auc = average_precision_score(y_true, scores)
        f1 = f1_score(y_true, y_pred, pos_label=1)
        cm = confusion_matrix(y_true, y_pred)
        precision, recall, thresholds = precision_recall_curve(y_true, scores)

        return {
            "pr_auc": pr_auc,
            "f1_score": f1,
            "confusion_matrix": cm,
            "precision_curve": precision,
            "recall_curve": recall,
            "pr_thresholds": thresholds,
        }