Skip to content

Models

Metrics

compute_ece(y_true, proba, labels, n_bins=10)

Compute macro-averaged multiclass Expected Calibration Error (one-vs-rest).

Source code in src/models/metrics.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def compute_ece(
    y_true: np.ndarray,
    proba: np.ndarray,
    labels: list,
    n_bins: int = 10,
) -> float:
    """Compute macro-averaged multiclass Expected Calibration Error (one-vs-rest)."""
    y_arr = np.asarray(y_true)
    ece_total = 0.0
    bins = np.linspace(0.0, 1.0, n_bins + 1)
    for i, label in enumerate(labels):
        y_bin = (y_arr == label).astype(float)
        p_bin = proba[:, i]
        bin_indices = np.digitize(p_bin, bins[1:-1])
        ece_label = 0.0
        for b in range(n_bins):
            mask = bin_indices == b
            if mask.sum() == 0:
                continue
            acc = y_bin[mask].mean()
            conf = p_bin[mask].mean()
            ece_label += mask.sum() * abs(acc - conf)
        ece_total += ece_label / len(y_arr)
    return ece_total / len(labels)

extract_feature_importance(pipe, X_cols)

Extract feature importances from the 'clf' step of a sklearn Pipeline.

Source code in src/models/metrics.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def extract_feature_importance(pipe, X_cols: list[str]) -> pd.DataFrame | None:
    """Extract feature importances from the 'clf' step of a sklearn Pipeline."""
    clf = getattr(pipe, "named_steps", {}).get("clf")
    if clf is None:
        return None
    if hasattr(clf, "feature_importances_"):
        importances = clf.feature_importances_
    elif hasattr(clf, "coef_"):
        importances = np.abs(clf.coef_).mean(axis=0)
    else:
        return None
    if len(importances) != len(X_cols):
        return None
    return (
        pd.DataFrame({"feature": X_cols, "importance": importances})
        .sort_values("importance", ascending=False)
        .reset_index(drop=True)
    )

plot_feature_importance(df_imp, top_n=20, title='Feature Importance')

Horizontal bar chart of top-N feature importances.

Source code in src/models/metrics.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def plot_feature_importance(
    df_imp: pd.DataFrame,
    top_n: int = 20,
    title: str = "Feature Importance",
) -> plt.Figure:
    """Horizontal bar chart of top-N feature importances."""
    df_plot = df_imp.head(top_n).iloc[::-1]
    fig, ax = plt.subplots(figsize=(8, max(4, top_n * 0.35)))
    ax.barh(df_plot["feature"], df_plot["importance"])
    ax.set_xlabel("Importance")
    ax.set_title(title)
    plt.tight_layout()
    return fig

compute_segment_metrics(y_true, proba, labels, segments, segment_cols, min_samples=1)

Compute logloss and brier score per segment value for each segment column.

Source code in src/models/metrics.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def compute_segment_metrics(
    y_true: pd.Series,
    proba: np.ndarray,
    labels: list,
    segments: pd.DataFrame,
    segment_cols: list[str],
    min_samples: int = 1,
) -> pd.DataFrame:
    """Compute logloss and brier score per segment value for each segment column."""
    y_arr = y_true.to_numpy()
    rows = []
    for col in segment_cols:
        if col not in segments.columns:
            logger.warning(
                "Segment column '%s' not found in segments DataFrame; skipping.", col
            )
            continue
        seg_vals = segments[col].to_numpy()
        for val in np.unique(seg_vals[~pd.isnull(seg_vals)]):
            mask = seg_vals == val
            if mask.sum() < min_samples:
                continue
            try:
                ll = log_loss(y_arr[mask], proba[mask], labels=labels)
            except Exception:
                ll = float("nan")
            brier = multiclass_brier_score(y_arr[mask], proba[mask], labels)
            rows.append(
                {
                    "segment_col": col,
                    "segment_value": val,
                    "n": int(mask.sum()),
                    "logloss": float(ll),
                    "brier": float(brier),
                }
            )
    return pd.DataFrame(
        rows, columns=["segment_col", "segment_value", "n", "logloss", "brier"]
    )

plot_calibration_curves(y_true, proba, labels, label_names)

One-vs-rest calibration reliability diagrams for each class.

Source code in src/models/metrics.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def plot_calibration_curves(
    y_true: pd.Series,
    proba: np.ndarray,
    labels: list,
    label_names: dict,
) -> plt.Figure:
    """One-vs-rest calibration reliability diagrams for each class."""
    n_classes = len(labels)
    fig, axes = plt.subplots(1, n_classes, figsize=(5 * n_classes, 4))
    if n_classes == 1:
        axes = [axes]
    y_arr = y_true.to_numpy() if hasattr(y_true, "to_numpy") else np.asarray(y_true)
    for i, label in enumerate(labels):
        y_bin = (y_arr == label).astype(int)
        fraction_pos, mean_pred = calibration_curve(y_bin, proba[:, i], n_bins=10)
        ax = axes[i]
        ax.plot(mean_pred, fraction_pos, "s-", label="Model")
        ax.plot([0, 1], [0, 1], "k--", label="Perfect")
        ax.set_xlabel("Mean predicted probability")
        ax.set_ylabel("Fraction of positives")
        ax.set_title(label_names.get(label, str(label)))
        ax.legend(loc="upper left")
    plt.tight_layout()
    return fig

Pipelines (sklearn)

WeightedXGBClassifier

Bases: XGBClassifier

XGBClassifier extended to accept the sklearn-style class_weight parameter.

XGBoost does not support class_weight natively in its constructor. This wrapper converts class_weight to a sample_weight array on each :meth:fit call, mirroring the behaviour of sklearn estimators.

Source code in src/models/pipelines.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class WeightedXGBClassifier(XGBClassifier):
    """XGBClassifier extended to accept the sklearn-style ``class_weight`` parameter.

    XGBoost does not support ``class_weight`` natively in its constructor.
    This wrapper converts ``class_weight`` to a ``sample_weight`` array on
    each :meth:`fit` call, mirroring the behaviour of sklearn estimators.
    """

    def __init__(self, *, class_weight=None, **kwargs):
        super().__init__(**kwargs)
        self.class_weight = class_weight

    def get_xgb_params(self):
        """Return XGBoost booster params, excluding sklearn-only ``class_weight``."""
        params = super().get_xgb_params()
        params.pop("class_weight", None)
        return params

    def fit(self, X, y, sample_weight=None, **kwargs):
        if self.class_weight is not None and sample_weight is None:
            sample_weight = compute_sample_weight(self.class_weight, y)
        return super().fit(X, y, sample_weight=sample_weight, **kwargs)

get_xgb_params()

Return XGBoost booster params, excluding sklearn-only class_weight.

Source code in src/models/pipelines.py 
24
25
26
27
28
def get_xgb_params(self):
    """Return XGBoost booster params, excluding sklearn-only ``class_weight``."""
    params = super().get_xgb_params()
    params.pop("class_weight", None)
    return params

Hyperparameter Tuning