| """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" | |