File size: 1,381 Bytes
0655115
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Population Stability Index (PSI) helpers."""

import numpy as np

EPS = 1e-4


def reference_bins(values, n_bins: int = 10) -> dict:
    """Quantile bin edges + reference proportions from training data."""
    v = np.asarray(values, dtype=float)
    v = v[np.isfinite(v)]
    inner = np.unique(np.quantile(v, np.linspace(0, 1, n_bins + 1)[1:-1]))
    edges = np.r_[-np.inf, inner, np.inf]
    counts, _ = np.histogram(v, bins=edges)
    props = counts / counts.sum()
    return {"edges": [None if np.isinf(e) else float(e) for e in edges], "proportions": props.tolist()}


def _edges(ref: dict) -> np.ndarray:
    e = ref["edges"]
    last = len(e) - 1
    return np.array([-np.inf if i == 0 else np.inf if i == last else x for i, x in enumerate(e)], dtype=float)


def psi(ref: dict, new_values) -> float:
    v = np.asarray(new_values, dtype=float)
    v = v[np.isfinite(v)]
    if v.size == 0:
        raise ValueError("no finite values to compare")
    counts, _ = np.histogram(v, bins=_edges(ref))
    actual = np.clip(counts / counts.sum(), EPS, None)
    expected = np.clip(np.asarray(ref["proportions"], dtype=float), EPS, None)
    return float(np.sum((actual - expected) * np.log(actual / expected)))


def psi_status(value: float) -> str:
    if value < 0.1:
        return "stable"
    if value < 0.25:
        return "moderate_shift"
    return "significant_shift"