Skip to content

Commit 2d7f271

Browse files
Merge pull request #51 from Climate-Vision/feature/governance-calibration
feat(governance): add calibration metrics for segmentation confidence
2 parents 765732d + 47f6619 commit 2d7f271

3 files changed

Lines changed: 341 additions & 0 deletions

File tree

src/climatevision/governance/__init__.py

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
Provides responsible AI capabilities:
55
- SHAP-based explainability for segmentation predictions
66
- Regional bias and fairness auditing
7+
- Calibration metrics for confidence reliability
78
- Anomaly detection for inference inputs/outputs
89
- Model audit trails and version tracking
910
"""
@@ -42,6 +43,16 @@
4243
check_fairness_gate,
4344
SUPPORTED_REGIONS,
4445
)
46+
from .calibration import (
47+
CalibrationReport,
48+
ReliabilityBin,
49+
brier_score,
50+
evaluate_calibration,
51+
expected_calibration_error,
52+
maximum_calibration_error,
53+
reliability_bins,
54+
write_calibration_report,
55+
)
4556

4657
__all__ = [
4758
# Explainability
@@ -73,4 +84,13 @@
7384
"RegionMetrics",
7485
"check_fairness_gate",
7586
"SUPPORTED_REGIONS",
87+
# Calibration
88+
"CalibrationReport",
89+
"ReliabilityBin",
90+
"brier_score",
91+
"evaluate_calibration",
92+
"expected_calibration_error",
93+
"maximum_calibration_error",
94+
"reliability_bins",
95+
"write_calibration_report",
7696
]
Lines changed: 196 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,196 @@
1+
"""
2+
Calibration metrics for ClimateVision segmentation models.
3+
4+
A model that reports a confidence of 0.9 should be correct about 90% of the
5+
time — anything else is miscalibration. For NGO-facing alerts driven by
6+
threshold logic on confidence, miscalibration directly mistranslates into
7+
either missed events or false alarms, so the calibration of every released
8+
model needs to be measured alongside the headline accuracy.
9+
10+
This module computes the standard reliability-diagram metrics for binary
11+
segmentation outputs:
12+
13+
- Reliability bins: bucket pixel predictions by confidence, record the
14+
observed positive-rate in each bucket against the bucket's mean confidence.
15+
- Expected Calibration Error (ECE): support-weighted mean of the absolute gap
16+
between confidence and accuracy across bins.
17+
- Maximum Calibration Error (MCE): the worst single-bin gap.
18+
- Brier score: mean squared error between probability and binary target.
19+
20+
All metrics operate on flat numpy arrays so they slot into the existing
21+
governance pipeline (model card generator, release CI gate) without
22+
introducing a torch dependency at evaluation time.
23+
"""
24+
25+
from __future__ import annotations
26+
27+
import json
28+
import logging
29+
from dataclasses import asdict, dataclass, field
30+
from pathlib import Path
31+
from typing import List, Union
32+
33+
import numpy as np
34+
35+
logger = logging.getLogger(__name__)
36+
37+
DEFAULT_N_BINS = 15
38+
39+
40+
@dataclass
41+
class ReliabilityBin:
42+
"""One bucket of the reliability diagram."""
43+
44+
lower: float
45+
upper: float
46+
count: int
47+
mean_confidence: float
48+
observed_positive_rate: float
49+
50+
51+
@dataclass
52+
class CalibrationReport:
53+
"""Calibration evaluation summary for a single model run."""
54+
55+
model_version: str
56+
n_samples: int
57+
n_bins: int
58+
ece: float
59+
mce: float
60+
brier_score: float
61+
bins: List[ReliabilityBin] = field(default_factory=list)
62+
63+
def to_dict(self) -> dict:
64+
d = asdict(self)
65+
d["bins"] = [asdict(b) for b in self.bins]
66+
return d
67+
68+
def is_well_calibrated(self, ece_threshold: float = 0.05) -> bool:
69+
"""Default release-gate threshold: ECE under 5%."""
70+
return self.ece <= ece_threshold
71+
72+
73+
def _validate_inputs(probabilities: np.ndarray, targets: np.ndarray) -> None:
74+
if probabilities.shape != targets.shape:
75+
raise ValueError(
76+
f"probabilities and targets must have the same shape, got "
77+
f"{probabilities.shape} and {targets.shape}"
78+
)
79+
if probabilities.size == 0:
80+
raise ValueError("probabilities array is empty")
81+
if probabilities.min() < 0.0 or probabilities.max() > 1.0:
82+
raise ValueError("probabilities must lie in [0, 1]")
83+
unique_targets = np.unique(targets)
84+
if not np.all(np.isin(unique_targets, [0, 1])):
85+
raise ValueError(
86+
f"targets must be binary {{0, 1}}, got values {unique_targets}"
87+
)
88+
89+
90+
def reliability_bins(
91+
probabilities: np.ndarray,
92+
targets: np.ndarray,
93+
n_bins: int = DEFAULT_N_BINS,
94+
) -> List[ReliabilityBin]:
95+
"""Bucket predictions by confidence and return per-bin reliability."""
96+
probs = np.asarray(probabilities, dtype=np.float64).ravel()
97+
tgts = np.asarray(targets, dtype=np.int32).ravel()
98+
_validate_inputs(probs, tgts)
99+
100+
edges = np.linspace(0.0, 1.0, n_bins + 1)
101+
bins: List[ReliabilityBin] = []
102+
for i in range(n_bins):
103+
lower, upper = edges[i], edges[i + 1]
104+
if i == n_bins - 1:
105+
mask = (probs >= lower) & (probs <= upper)
106+
else:
107+
mask = (probs >= lower) & (probs < upper)
108+
count = int(mask.sum())
109+
if count == 0:
110+
bins.append(
111+
ReliabilityBin(
112+
lower=float(lower),
113+
upper=float(upper),
114+
count=0,
115+
mean_confidence=0.0,
116+
observed_positive_rate=0.0,
117+
)
118+
)
119+
continue
120+
bins.append(
121+
ReliabilityBin(
122+
lower=float(lower),
123+
upper=float(upper),
124+
count=count,
125+
mean_confidence=float(probs[mask].mean()),
126+
observed_positive_rate=float(tgts[mask].mean()),
127+
)
128+
)
129+
return bins
130+
131+
132+
def expected_calibration_error(bins: List[ReliabilityBin]) -> float:
133+
"""Support-weighted mean gap between confidence and observed accuracy."""
134+
total = sum(b.count for b in bins)
135+
if total == 0:
136+
return 0.0
137+
weighted = sum(
138+
(b.count / total) * abs(b.mean_confidence - b.observed_positive_rate)
139+
for b in bins
140+
if b.count > 0
141+
)
142+
return float(weighted)
143+
144+
145+
def maximum_calibration_error(bins: List[ReliabilityBin]) -> float:
146+
"""Worst single-bin gap between confidence and observed accuracy."""
147+
populated = [b for b in bins if b.count > 0]
148+
if not populated:
149+
return 0.0
150+
return float(
151+
max(abs(b.mean_confidence - b.observed_positive_rate) for b in populated)
152+
)
153+
154+
155+
def brier_score(
156+
probabilities: np.ndarray, targets: np.ndarray
157+
) -> float:
158+
"""Mean squared error between probability and binary target."""
159+
probs = np.asarray(probabilities, dtype=np.float64).ravel()
160+
tgts = np.asarray(targets, dtype=np.float64).ravel()
161+
_validate_inputs(probs, tgts.astype(np.int32))
162+
return float(np.mean((probs - tgts) ** 2))
163+
164+
165+
def evaluate_calibration(
166+
probabilities: np.ndarray,
167+
targets: np.ndarray,
168+
*,
169+
model_version: str,
170+
n_bins: int = DEFAULT_N_BINS,
171+
) -> CalibrationReport:
172+
"""Run the full calibration evaluation and return a report dataclass."""
173+
probs = np.asarray(probabilities, dtype=np.float64).ravel()
174+
tgts = np.asarray(targets, dtype=np.int32).ravel()
175+
_validate_inputs(probs, tgts)
176+
bins = reliability_bins(probs, tgts, n_bins=n_bins)
177+
return CalibrationReport(
178+
model_version=model_version,
179+
n_samples=int(probs.size),
180+
n_bins=n_bins,
181+
ece=expected_calibration_error(bins),
182+
mce=maximum_calibration_error(bins),
183+
brier_score=brier_score(probs, tgts),
184+
bins=bins,
185+
)
186+
187+
188+
def write_calibration_report(
189+
report: CalibrationReport, path: Union[str, Path]
190+
) -> Path:
191+
"""Persist a CalibrationReport to disk as JSON."""
192+
out = Path(path)
193+
out.parent.mkdir(parents=True, exist_ok=True)
194+
out.write_text(json.dumps(report.to_dict(), indent=2))
195+
logger.info("Wrote calibration report to %s", out)
196+
return out

tests/test_calibration.py

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
"""Tests for governance.calibration."""
2+
3+
from __future__ import annotations
4+
5+
import json
6+
7+
import numpy as np
8+
import pytest
9+
10+
from climatevision.governance.calibration import (
11+
CalibrationReport,
12+
brier_score,
13+
evaluate_calibration,
14+
expected_calibration_error,
15+
maximum_calibration_error,
16+
reliability_bins,
17+
write_calibration_report,
18+
)
19+
20+
21+
def _perfectly_calibrated(n: int = 10_000, seed: int = 0):
22+
rng = np.random.default_rng(seed)
23+
probs = rng.uniform(0.0, 1.0, size=n)
24+
targets = (rng.uniform(0.0, 1.0, size=n) < probs).astype(np.int32)
25+
return probs, targets
26+
27+
28+
def _overconfident(n: int = 10_000, seed: int = 1):
29+
rng = np.random.default_rng(seed)
30+
probs = rng.uniform(0.8, 1.0, size=n)
31+
targets = (rng.uniform(0.0, 1.0, size=n) < 0.5).astype(np.int32)
32+
return probs, targets
33+
34+
35+
def test_reliability_bins_partition_inputs():
36+
probs, targets = _perfectly_calibrated()
37+
bins = reliability_bins(probs, targets, n_bins=10)
38+
assert len(bins) == 10
39+
assert sum(b.count for b in bins) == probs.size
40+
for b in bins:
41+
assert 0.0 <= b.lower < b.upper <= 1.0
42+
43+
44+
def test_perfectly_calibrated_has_low_ece():
45+
probs, targets = _perfectly_calibrated()
46+
bins = reliability_bins(probs, targets, n_bins=15)
47+
assert expected_calibration_error(bins) < 0.05
48+
49+
50+
def test_overconfident_has_high_ece():
51+
probs, targets = _overconfident()
52+
bins = reliability_bins(probs, targets, n_bins=15)
53+
assert expected_calibration_error(bins) > 0.2
54+
55+
56+
def test_mce_is_at_least_ece():
57+
probs, targets = _overconfident()
58+
bins = reliability_bins(probs, targets, n_bins=15)
59+
assert maximum_calibration_error(bins) >= expected_calibration_error(bins)
60+
61+
62+
def test_brier_score_zero_for_certain_correct_predictions():
63+
probs = np.array([1.0, 0.0, 1.0, 0.0])
64+
targets = np.array([1, 0, 1, 0])
65+
assert brier_score(probs, targets) == pytest.approx(0.0)
66+
67+
68+
def test_brier_score_one_for_certain_wrong_predictions():
69+
probs = np.array([1.0, 0.0, 1.0, 0.0])
70+
targets = np.array([0, 1, 0, 1])
71+
assert brier_score(probs, targets) == pytest.approx(1.0)
72+
73+
74+
def test_evaluate_calibration_returns_report_with_bins():
75+
probs, targets = _perfectly_calibrated()
76+
report = evaluate_calibration(
77+
probs, targets, model_version="unet-test-1", n_bins=10
78+
)
79+
assert isinstance(report, CalibrationReport)
80+
assert report.model_version == "unet-test-1"
81+
assert report.n_samples == probs.size
82+
assert report.n_bins == 10
83+
assert len(report.bins) == 10
84+
assert 0.0 <= report.ece <= 1.0
85+
assert 0.0 <= report.brier_score <= 1.0
86+
87+
88+
def test_well_calibrated_threshold():
89+
probs, targets = _perfectly_calibrated()
90+
report = evaluate_calibration(probs, targets, model_version="v")
91+
assert report.is_well_calibrated(ece_threshold=0.05)
92+
bad_probs, bad_targets = _overconfident()
93+
bad = evaluate_calibration(bad_probs, bad_targets, model_version="v")
94+
assert not bad.is_well_calibrated(ece_threshold=0.05)
95+
96+
97+
def test_validates_probability_range():
98+
with pytest.raises(ValueError, match="probabilities must lie in"):
99+
evaluate_calibration(
100+
np.array([1.5, 0.5]), np.array([1, 0]), model_version="v"
101+
)
102+
103+
104+
def test_validates_binary_targets():
105+
with pytest.raises(ValueError, match="targets must be binary"):
106+
evaluate_calibration(
107+
np.array([0.5, 0.5]), np.array([1, 2]), model_version="v"
108+
)
109+
110+
111+
def test_validates_shape_match():
112+
with pytest.raises(ValueError, match="same shape"):
113+
evaluate_calibration(
114+
np.array([0.5, 0.5]), np.array([1, 0, 1]), model_version="v"
115+
)
116+
117+
118+
def test_write_calibration_report_round_trips_json(tmp_path):
119+
probs, targets = _perfectly_calibrated(n=1000)
120+
report = evaluate_calibration(probs, targets, model_version="v0.1")
121+
out = write_calibration_report(report, tmp_path / "calib.json")
122+
loaded = json.loads(out.read_text())
123+
assert loaded["model_version"] == "v0.1"
124+
assert loaded["n_samples"] == 1000
125+
assert len(loaded["bins"]) == report.n_bins

0 commit comments

Comments
 (0)