"""Agreement, error and bootstrap utilities (no modelling here).""" from __future__ import annotations import numpy as np import pandas as pd def cohen_kappa(a: pd.Series, b: pd.Series, weights: str | None = None) -> float: """Cohen's kappa; weights=None (nominal), 'linear' or 'quadratic' for ordinal codes (ints).""" m = a.notna() & b.notna() a, b = a[m], b[m] if len(a) == 0: return np.nan cats = sorted(set(a) | set(b)) idx = {c: i for i, c in enumerate(cats)} n = len(cats) O = np.zeros((n, n)) for x, y in zip(a, b): O[idx[x], idx[y]] += 1 O /= O.sum() E = np.outer(O.sum(1), O.sum(0)) if weights is None: W = 1 - np.eye(n) else: i, j = np.meshgrid(np.arange(n), np.arange(n), indexing="ij") W = np.abs(i - j) if weights == "linear" else (i - j) ** 2 W = W / W.max() if W.max() > 0 else W po = 1 - (W * O).sum() pe = 1 - (W * E).sum() return (po - pe) / (1 - pe) if pe < 1 else np.nan def krippendorff_alpha_nominal(labels: np.ndarray) -> float: """labels: 2-D array raters × units with None/NaN for missing. Nominal metric.""" units = [] for j in range(labels.shape[1]): col = [x for x in labels[:, j] if x is not None and not (isinstance(x, float) and np.isnan(x))] if len(col) >= 2: units.append(col) if not units: return np.nan n = sum(len(u) for u in units) vals = sorted({x for u in units for x in u}) Do = 0.0 for u in units: m = len(u) c = pd.Series(u).value_counts() Do += sum(c[v] * (m - c[v]) for v in c.index) / (m - 1) Do /= n tot = pd.Series([x for u in units for x in u]).value_counts() De = sum(tot[v] * (n - tot[v]) for v in tot.index) / (n - 1) / n return 1 - Do / De if De > 0 else np.nan def agreement_table(pred: pd.Series, ref: pd.Series, noninfo=("not_visible", "not_applicable", "not_stated", None)) -> dict: """Coverage (share of informative predictions), agreement among informative pairs, macro-F1, kappa.""" inf = ~pred.isin(noninfo) & pred.notna() refok = ~ref.isin(noninfo) & ref.notna() both = inf & refok out = {"n_ref": int(refok.sum()), "coverage": float(inf[refok].mean()) if refok.sum() else np.nan, "n_compared": int(both.sum())} if both.sum() == 0: out.update(accuracy=np.nan, kappa=np.nan, macro_f1=np.nan) return out p, r = pred[both], ref[both] out["accuracy"] = float((p == r).mean()) out["kappa"] = cohen_kappa(p, r) cats = sorted(set(p) | set(r)) f1s = [] for c in cats: tp = ((p == c) & (r == c)).sum(); fp = ((p == c) & (r != c)).sum(); fn = ((p != c) & (r == c)).sum() if tp + fp + fn == 0: continue prec = tp / (tp + fp) if tp + fp else 0.0 rec = tp / (tp + fn) if tp + fn else 0.0 f1s.append(2 * prec * rec / (prec + rec) if prec + rec else 0.0) out["macro_f1"] = float(np.mean(f1s)) if f1s else np.nan return out def cluster_bootstrap(df: pd.DataFrame, stat, cluster: str, B: int = 999, seed: int = 20261020, weight: str | None = None): """Percentile 95% CI of stat(df) under cluster resampling. stat: callable DataFrame → float.""" rng = np.random.default_rng(seed) clusters = df[cluster].unique() groups = {c: g for c, g in df.groupby(cluster)} point = stat(df) vals = [] for _ in range(B): pick = rng.choice(clusters, size=len(clusters), replace=True) vals.append(stat(pd.concat([groups[c] for c in pick]))) lo, hi = np.nanpercentile(vals, [2.5, 97.5]) return point, lo, hi def mdape(pred, ref): return float(np.nanmedian(np.abs(pred / ref - 1.0)) * 100) def ratio_stats(pred, ref): r = (pred / ref).replace([np.inf, -np.inf], np.nan).dropna() return {"n": int(len(r)), "median_ratio": float(r.median()), "mean_ratio": float(r.mean()), "mdape": float((r - 1).abs().median() * 100), "cod": float((r - r.median()).abs().mean() / r.median() * 100), "p10": float(r.quantile(.1)), "p90": float(r.quantile(.9)), "within_10pct": float(((r - 1).abs() <= 0.10).mean()), "within_20pct": float(((r - 1).abs() <= 0.20).mean())}