///|
/// Threshold calibration curve for production decisions with asymmetric costs.
pub struct CalibrationPoint {
threshold : Double
precision : Double
recall : Double
f1 : Double
cost : Double
predicted_positive : Int
}
///|
pub struct CalibrationCurve {
points : Array[CalibrationPoint]
best_f1 : Double
best_cost : Double
selected_threshold : Double
}
///|
pub fn calibration_point(
threshold : Double,
labels : Array[Bool],
probabilities : Array[Double],
false_positive_cost : Double,
false_negative_cost : Double,
) -> CalibrationPoint {
let predicted = classification_labels(probabilities, threshold)
let matrix = confusion_matrix(labels, predicted)
let safe_fp = if false_positive_cost < 0.0 {
-false_positive_cost
} else {
false_positive_cost
}
let safe_fn = if false_negative_cost < 0.0 {
-false_negative_cost
} else {
false_negative_cost
}
{
threshold,
precision: precision(matrix),
recall: recall(matrix),
f1: f1_score(matrix),
cost: classification_cost(
matrix,
classification_threshold_rule(safe_fp, safe_fn, 0.0, 0.0),
),
predicted_positive: classification_positive_count(predicted),
}
}
///|
pub fn calibration_curve(
labels : Array[Bool],
probabilities : Array[Double],
steps : Int,
false_positive_cost : Double,
false_negative_cost : Double,
) -> CalibrationCurve {
let thresholds = classification_threshold_grid(steps)
let points = []
for threshold in thresholds {
points.push(
calibration_point(
threshold, labels, probabilities, false_positive_cost, false_negative_cost,
),
)
}
if points.length() == 0 {
return { points: [], best_f1: 0.0, best_cost: 0.0, selected_threshold: 0.5 }
}
let mut best_f1 = points[0]
let mut best_cost = points[0]
for point in points {
if point.f1 > best_f1.f1 {
best_f1 = point
}
if point.cost < best_cost.cost {
best_cost = point
}
}
{
points,
best_f1: best_f1.f1,
best_cost: best_cost.cost,
selected_threshold: best_f1.threshold,
}
}
///|
pub fn calibration_best_f1_threshold(curve : CalibrationCurve) -> Double {
let mut result = curve.selected_threshold
let mut score = -1.0
for point in curve.points {
if point.f1 > score {
score = point.f1
result = point.threshold
}
}
result
}
///|
pub fn calibration_best_cost_threshold(curve : CalibrationCurve) -> Double {
if curve.points.length() == 0 {
return 0.5
}
let mut result = curve.points[0].threshold
let mut cost = curve.points[0].cost
for point in curve.points {
if point.cost < cost {
cost = point.cost
result = point.threshold
}
}
result
}
///|
pub fn calibration_operating_point(
curve : CalibrationCurve,
threshold : Double,
) -> CalibrationPoint {
if curve.points.length() == 0 {
return {
threshold,
precision: 0.0,
recall: 0.0,
f1: 0.0,
cost: 0.0,
predicted_positive: 0,
}
}
let mut best = curve.points[0]
let mut distance = abs_double(best.threshold - threshold)
for point in curve.points {
let current = abs_double(point.threshold - threshold)
if current < distance {
distance = current
best = point
}
}
best
}
///|
pub fn calibration_thresholds(curve : CalibrationCurve) -> Array[Double] {
let result = []
for point in curve.points {
result.push(point.threshold)
}
result
}
///|
pub fn calibration_f1_values(curve : CalibrationCurve) -> Array[Double] {
let result = []
for point in curve.points {
result.push(point.f1)
}
result
}
///|
pub fn calibration_cost_values(curve : CalibrationCurve) -> Array[Double] {
let result = []
for point in curve.points {
result.push(point.cost)
}
result
}
///|
pub fn calibration_ece(
probabilities : Array[Double],
labels : Array[Bool],
bins : Int,
) -> Double {
probability_expected_calibration_error(
probability_calibration(probabilities, labels, bins),
)
}
///|
pub fn calibration_mce(
probabilities : Array[Double],
labels : Array[Bool],
bins : Int,
) -> Double {
probability_max_calibration_error(
probability_calibration(probabilities, labels, bins),
)
}
///|
pub fn calibration_brier_score(
probabilities : Array[Double],
labels : Array[Bool],
) -> Double {
let count = if probabilities.length() < labels.length() {
probabilities.length()
} else {
labels.length()
}
if count == 0 {
return 0.0
}
let mut total = 0.0
for index = 0; index < count; index = index + 1 {
let target = if labels[index] { 1.0 } else { 0.0 }
let delta = probabilities[index] - target
total += delta * delta
}
total / count.to_double()
}
///|
pub fn calibration_curve_vector(curve : CalibrationCurve) -> Array[Double] {
[
curve.best_f1,
curve.best_cost,
curve.selected_threshold,
curve.points.length().to_double(),
]
}
///|
pub fn calibration_curve_lines(curve : CalibrationCurve) -> Array[String] {
let lines = [
"best_f1=" + curve.best_f1.to_string(),
"best_cost=" + curve.best_cost.to_string(),
"selected_threshold=" + curve.selected_threshold.to_string(),
]
for point in curve.points {
lines.push(
point.threshold.to_string() +
"|" +
point.precision.to_string() +
"|" +
point.recall.to_string() +
"|" +
point.f1.to_string() +
"|" +
point.cost.to_string(),
)
}
lines
}
///|
pub fn calibration_curve_string(curve : CalibrationCurve) -> String {
calibration_curve_lines(curve).join("\n")
}
///|
pub fn calibration_is_usable(
curve : CalibrationCurve,
minimum_f1 : Double,
) -> Bool {
curve.points.length() > 0 && curve.best_f1 >= minimum_f1
}
///|
pub fn calibration_compare(
left : CalibrationCurve,
right : CalibrationCurve,
) -> Array[Double] {
[
left.best_f1,
right.best_f1,
left.best_cost,
right.best_cost,
left.selected_threshold,
right.selected_threshold,
]
}
///|
pub fn calibration_summary(
probabilities : Array[Double],
labels : Array[Bool],
) -> Array[Double] {
let curve = calibration_curve(labels, probabilities, 21, 1.0, 1.0)
[
curve.best_f1,
curve.best_cost,
curve.selected_threshold,
calibration_ece(probabilities, labels, 10),
calibration_mce(probabilities, labels, 10),
calibration_brier_score(probabilities, labels),
]
}