File size: 1,834 Bytes
42029e4
 
 
 
 
 
 
 
 
 
 
8c66d1b
 
42029e4
8c66d1b
42029e4
 
 
 
 
 
 
8c66d1b
42029e4
 
 
8c66d1b
42029e4
 
 
8c66d1b
42029e4
 
 
 
8c66d1b
42029e4
 
8c66d1b
42029e4
 
 
 
8c66d1b
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from ..utils.logging import get_logger
from .base import FraudModel
log = get_logger(__name__)

class IsolationForestFraudModel(FraudModel):
    name = 'isolation_forest'
    DEFAULT_PARAMS = {'n_estimators': 200, 'contamination': 0.02, 'max_samples': 0.5, 'n_jobs': -1, 'random_state': 42}

    def __init__(self, params: dict | None=None):
        self.params = {**self.DEFAULT_PARAMS, **(params or {})}
        self.scaler = StandardScaler()
        self.model: IsolationForest | None = None
        self.score_min_: float = 0.0
        self.score_max_: float = 1.0
        self._trained = False

    def fit(self, X: pd.DataFrame, y: pd.Series | None=None, **kwargs) -> 'IsolationForestFraudModel':
        X_scaled = self.scaler.fit_transform(X.values)
        self.model = IsolationForest(**self.params)
        self.model.fit(X_scaled)
        raw = -self.model.score_samples(X_scaled)
        self.score_min_ = float(np.percentile(raw, 1))
        self.score_max_ = float(np.percentile(raw, 99))
        self._trained = True
        log.info(f'[IF] trained. score range [{self.score_min_:.3f}, {self.score_max_:.3f}]')
        return self

    def predict_proba(self, X: pd.DataFrame) -> np.ndarray:
        if not self._trained or self.model is None:
            raise RuntimeError('Model not trained')
        X_scaled = self.scaler.transform(X.values)
        raw = -self.model.score_samples(X_scaled)
        span = max(self.score_max_ - self.score_min_, 1e-09)
        prob = (raw - self.score_min_) / span
        return np.clip(prob, 0.0, 1.0)

    def get_params(self) -> dict[str, Any]:
        return {'params': self.params}